SLFO-pool/salt
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
main
salt/fix-cve-2023-34049-bsc-1215157.patch

1164 lines
40 KiB
Diff
Raw Permalink Blame History

From b2baafcc96a2807cf7d34374904e1710a4f58b9f Mon Sep 17 00:00:00 2001
From: Alexander Graul <agraul@suse.com>
Date: Tue, 31 Oct 2023 11:26:15 +0100
Subject: [PATCH] Fix CVE-2023-34049 (bsc#1215157)
Backport of https://github.com/saltstack/salt/pull/65482
---
salt/client/ssh/__init__.py | 56 +++-
tests/integration/modules/test_ssh.py | 3 +-
tests/integration/ssh/test_pre_flight.py | 132 --------
.../integration/ssh/test_pre_flight.py | 315 ++++++++++++++++++
tests/pytests/unit/client/ssh/test_single.py | 296 +++++++++++++---
tests/pytests/unit/client/ssh/test_ssh.py | 110 ++++++
6 files changed, 727 insertions(+), 185 deletions(-)
delete mode 100644 tests/integration/ssh/test_pre_flight.py
create mode 100644 tests/pytests/integration/ssh/test_pre_flight.py
diff --git a/salt/client/ssh/__init__.py b/salt/client/ssh/__init__.py
index b120e0002e8..1e143f9e30c 100644
--- a/salt/client/ssh/__init__.py
+++ b/salt/client/ssh/__init__.py
@@ -12,9 +12,11 @@ import hashlib
import logging
import multiprocessing
import os
+import pathlib
import queue
import re
import shlex
+import shutil
import subprocess
import sys
import tarfile
@@ -515,7 +517,14 @@ class SSH(MultiprocessingStateMixin):
if target.get("passwd", False) or self.opts["ssh_passwd"]:
self._key_deploy_run(host, target, False)
return ret
- if ret[host].get("stderr", "").count("Permission denied"):
+ stderr = ret[host].get("stderr", "")
+ # -failed to upload file- is detecting scp errors
+ # Errors to ignore when Permission denied is in the stderr. For example
+ # scp can get a permission denied on the target host, but they where
+ # able to accurate authenticate against the box
+ ignore_err = ["failed to upload file"]
+ check_err = [x for x in ignore_err if stderr.count(x)]
+ if "Permission denied" in stderr and not check_err:
target = self.targets[host]
# permission denied, attempt to auto deploy ssh key
print(
@@ -1137,11 +1146,30 @@ class Single:
"""
Run our pre_flight script before running any ssh commands
"""
- script = os.path.join(tempfile.gettempdir(), self.ssh_pre_file)
-
- self.shell.send(self.ssh_pre_flight, script)
-
- return self.execute_script(script, script_args=self.ssh_pre_flight_args)
+ with tempfile.NamedTemporaryFile() as temp:
+ # ensure we use copyfile to not copy the file attributes
+ # we want to ensure we use the perms set by the secure
+ # NamedTemporaryFile
+ try:
+ shutil.copyfile(self.ssh_pre_flight, temp.name)
+ except OSError as err:
+ return (
+ "",
+ "Could not copy pre flight script to temporary path",
+ 1,
+ )
+ target_script = f".{pathlib.Path(temp.name).name}"
+ log.trace("Copying the pre flight script to target")
+ stdout, stderr, retcode = self.shell.send(temp.name, target_script)
+ if retcode != 0:
+ # We could not copy the script to the target
+ log.error("Could not copy the pre flight script to target")
+ return stdout, stderr, retcode
+
+ log.trace("Executing the pre flight script on target")
+ return self.execute_script(
+ target_script, script_args=self.ssh_pre_flight_args
+ )
def check_thin_dir(self):
"""
@@ -1531,18 +1559,20 @@ ARGS = {arguments}\n'''.format(
return self.shell.exec_cmd(cmd_str)
# Write the shim to a temporary file in the default temp directory
- with tempfile.NamedTemporaryFile(
- mode="w+b", prefix="shim_", delete=False
- ) as shim_tmp_file:
+ with tempfile.NamedTemporaryFile(mode="w+b", delete=False) as shim_tmp_file:
shim_tmp_file.write(salt.utils.stringutils.to_bytes(cmd_str))
# Copy shim to target system, under $HOME/.<randomized name>
- target_shim_file = ".{}.{}".format(
- binascii.hexlify(os.urandom(6)).decode("ascii"), extension
- )
+ target_shim_file = f".{pathlib.Path(shim_tmp_file.name).name}"
+
if self.winrm:
target_shim_file = saltwinshell.get_target_shim_file(self, target_shim_file)
- self.shell.send(shim_tmp_file.name, target_shim_file, makedirs=True)
+ stdout, stderr, retcode = self.shell.send(
+ shim_tmp_file.name, target_shim_file, makedirs=True
+ )
+ if retcode != 0:
+ log.error("Could not copy the shim script to target")
+ return stdout, stderr, retcode
# Remove our shim file
try:
diff --git a/tests/integration/modules/test_ssh.py b/tests/integration/modules/test_ssh.py
index 0817877c86b..55586211622 100644
--- a/tests/integration/modules/test_ssh.py
+++ b/tests/integration/modules/test_ssh.py
@@ -26,7 +26,8 @@ def check_status():
return False
-@pytest.mark.windows_whitelisted
+# @pytest.mark.windows_whitelisted
+# De-whitelist windows since it's hanging on the newer windows golden images
@pytest.mark.skip_if_binaries_missing("ssh", "ssh-keygen", check_all=True)
class SSHModuleTest(ModuleCase):
"""
diff --git a/tests/integration/ssh/test_pre_flight.py b/tests/integration/ssh/test_pre_flight.py
deleted file mode 100644
index 1598b3d51b5..00000000000
--- a/tests/integration/ssh/test_pre_flight.py
+++ /dev/null
@@ -1,132 +0,0 @@
-"""
-Test for ssh_pre_flight roster option
-"""
-
-import os
-
-import pytest
-
-import salt.utils.files
-from tests.support.case import SSHCase
-from tests.support.runtests import RUNTIME_VARS
-
-
-class SSHPreFlightTest(SSHCase):
- """
- Test ssh_pre_flight roster option
- """
-
- def setUp(self):
- super().setUp()
- self.roster = os.path.join(RUNTIME_VARS.TMP, "pre_flight_roster")
- self.data = {
- "ssh_pre_flight": os.path.join(RUNTIME_VARS.TMP, "ssh_pre_flight.sh")
- }
- self.test_script = os.path.join(
- RUNTIME_VARS.TMP, "test-pre-flight-script-worked.txt"
- )
-
- def _create_roster(self, pre_flight_script_args=None):
- data = dict(self.data)
- if pre_flight_script_args:
- data["ssh_pre_flight_args"] = pre_flight_script_args
-
- self.custom_roster(self.roster, data)
-
- with salt.utils.files.fopen(data["ssh_pre_flight"], "w") as fp_:
- fp_.write("touch {}".format(self.test_script))
-
- @pytest.mark.slow_test
- def test_ssh_pre_flight(self):
- """
- test ssh when ssh_pre_flight is set
- ensure the script runs successfully
- """
- self._create_roster()
- assert self.run_function("test.ping", roster_file=self.roster)
-
- assert os.path.exists(self.test_script)
-
- @pytest.mark.slow_test
- def test_ssh_run_pre_flight(self):
- """
- test ssh when --pre-flight is passed to salt-ssh
- to ensure the script runs successfully
- """
- self._create_roster()
- # make sure we previously ran a command so the thin dir exists
- self.run_function("test.ping", wipe=False)
- assert not os.path.exists(self.test_script)
-
- assert self.run_function(
- "test.ping", ssh_opts="--pre-flight", roster_file=self.roster, wipe=False
- )
- assert os.path.exists(self.test_script)
-
- @pytest.mark.slow_test
- def test_ssh_run_pre_flight_args(self):
- """
- test ssh when --pre-flight is passed to salt-ssh
- to ensure the script runs successfully passing some args
- """
- self._create_roster(pre_flight_script_args="foobar test")
- # make sure we previously ran a command so the thin dir exists
- self.run_function("test.ping", wipe=False)
- assert not os.path.exists(self.test_script)
-
- assert self.run_function(
- "test.ping", ssh_opts="--pre-flight", roster_file=self.roster, wipe=False
- )
- assert os.path.exists(self.test_script)
-
- @pytest.mark.slow_test
- def test_ssh_run_pre_flight_args_prevent_injection(self):
- """
- test ssh when --pre-flight is passed to salt-ssh
- and evil arguments are used in order to produce shell injection
- """
- injected_file = os.path.join(RUNTIME_VARS.TMP, "injection")
- self._create_roster(
- pre_flight_script_args="foobar; echo injected > {}".format(injected_file)
- )
- # make sure we previously ran a command so the thin dir exists
- self.run_function("test.ping", wipe=False)
- assert not os.path.exists(self.test_script)
- assert not os.path.isfile(injected_file)
-
- assert self.run_function(
- "test.ping", ssh_opts="--pre-flight", roster_file=self.roster, wipe=False
- )
-
- assert not os.path.isfile(
- injected_file
- ), "File injection suceeded. This shouldn't happend"
-
- @pytest.mark.slow_test
- def test_ssh_run_pre_flight_failure(self):
- """
- test ssh_pre_flight when there is a failure
- in the script.
- """
- self._create_roster()
- with salt.utils.files.fopen(self.data["ssh_pre_flight"], "w") as fp_:
- fp_.write("exit 2")
-
- ret = self.run_function(
- "test.ping", ssh_opts="--pre-flight", roster_file=self.roster, wipe=False
- )
- assert ret["retcode"] == 2
-
- def tearDown(self):
- """
- make sure to clean up any old ssh directories
- """
- files = [
- self.roster,
- self.data["ssh_pre_flight"],
- self.test_script,
- os.path.join(RUNTIME_VARS.TMP, "injection"),
- ]
- for fp_ in files:
- if os.path.exists(fp_):
- os.remove(fp_)
diff --git a/tests/pytests/integration/ssh/test_pre_flight.py b/tests/pytests/integration/ssh/test_pre_flight.py
new file mode 100644
index 00000000000..09c65d29430
--- /dev/null
+++ b/tests/pytests/integration/ssh/test_pre_flight.py
@@ -0,0 +1,315 @@
+"""
+Test for ssh_pre_flight roster option
+"""
+
+try:
+ import grp
+ import pwd
+except ImportError:
+ # windows stacktraces on import of these modules
+ pass
+import os
+import pathlib
+import shutil
+import subprocess
+
+import pytest
+import yaml
+from saltfactories.utils import random_string
+
+import salt.utils.files
+
+pytestmark = pytest.mark.skip_on_windows(reason="Salt-ssh not available on Windows")
+
+
+def _custom_roster(roster_file, roster_data):
+ with salt.utils.files.fopen(roster_file, "r") as fp:
+ data = salt.utils.yaml.safe_load(fp)
+ for key, item in roster_data.items():
+ data["localhost"][key] = item
+ with salt.utils.files.fopen(roster_file, "w") as fp:
+ yaml.safe_dump(data, fp)
+
+
+@pytest.fixture
+def _create_roster(salt_ssh_roster_file, tmp_path):
+ ret = {}
+ ret["roster"] = salt_ssh_roster_file
+ ret["data"] = {"ssh_pre_flight": str(tmp_path / "ssh_pre_flight.sh")}
+ ret["test_script"] = str(tmp_path / "test-pre-flight-script-worked.txt")
+ ret["thin_dir"] = tmp_path / "thin_dir"
+
+ with salt.utils.files.fopen(salt_ssh_roster_file, "r") as fp:
+ data = salt.utils.yaml.safe_load(fp)
+ pre_flight_script = ret["data"]["ssh_pre_flight"]
+ data["localhost"]["ssh_pre_flight"] = pre_flight_script
+ data["localhost"]["thin_dir"] = str(ret["thin_dir"])
+ with salt.utils.files.fopen(salt_ssh_roster_file, "w") as fp:
+ yaml.safe_dump(data, fp)
+
+ with salt.utils.files.fopen(pre_flight_script, "w") as fp:
+ fp.write("touch {}".format(ret["test_script"]))
+
+ yield ret
+ if ret["thin_dir"].exists():
+ shutil.rmtree(ret["thin_dir"])
+
+
+@pytest.mark.slow_test
+def test_ssh_pre_flight(salt_ssh_cli, caplog, _create_roster):
+ """
+ test ssh when ssh_pre_flight is set
+ ensure the script runs successfully
+ """
+ ret = salt_ssh_cli.run("test.ping")
+ assert ret.returncode == 0
+
+ assert pathlib.Path(_create_roster["test_script"]).exists()
+
+
+@pytest.mark.slow_test
+def test_ssh_run_pre_flight(salt_ssh_cli, _create_roster):
+ """
+ test ssh when --pre-flight is passed to salt-ssh
+ to ensure the script runs successfully
+ """
+ # make sure we previously ran a command so the thin dir exists
+ ret = salt_ssh_cli.run("test.ping")
+ assert pathlib.Path(_create_roster["test_script"]).exists()
+
+ # Now remeove the script to ensure pre_flight doesn't run
+ # without --pre-flight
+ pathlib.Path(_create_roster["test_script"]).unlink()
+
+ assert salt_ssh_cli.run("test.ping").returncode == 0
+ assert not pathlib.Path(_create_roster["test_script"]).exists()
+
+ # Now ensure
+ ret = salt_ssh_cli.run(
+ "test.ping",
+ "--pre-flight",
+ )
+ assert ret.returncode == 0
+ assert pathlib.Path(_create_roster["test_script"]).exists()
+
+
+@pytest.mark.slow_test
+def test_ssh_run_pre_flight_args(salt_ssh_cli, _create_roster):
+ """
+ test ssh when --pre-flight is passed to salt-ssh
+ to ensure the script runs successfully passing some args
+ """
+ _custom_roster(salt_ssh_cli.roster_file, {"ssh_pre_flight_args": "foobar test"})
+ # Create pre_flight script that accepts args
+ test_script = _create_roster["test_script"]
+ test_script_1 = pathlib.Path(test_script + "-foobar")
+ test_script_2 = pathlib.Path(test_script + "-test")
+ with salt.utils.files.fopen(_create_roster["data"]["ssh_pre_flight"], "w") as fp:
+ fp.write(
+ f"""
+ touch {str(test_script)}-$1
+ touch {str(test_script)}-$2
+ """
+ )
+ ret = salt_ssh_cli.run("test.ping")
+ assert ret.returncode == 0
+ assert test_script_1.exists()
+ assert test_script_2.exists()
+ pathlib.Path(test_script_1).unlink()
+ pathlib.Path(test_script_2).unlink()
+
+ ret = salt_ssh_cli.run("test.ping")
+ assert ret.returncode == 0
+ assert not test_script_1.exists()
+ assert not test_script_2.exists()
+
+ ret = salt_ssh_cli.run(
+ "test.ping",
+ "--pre-flight",
+ )
+ assert ret.returncode == 0
+ assert test_script_1.exists()
+ assert test_script_2.exists()
+
+
+@pytest.mark.slow_test
+def test_ssh_run_pre_flight_args_prevent_injection(
+ salt_ssh_cli, _create_roster, tmp_path
+):
+ """
+ test ssh when --pre-flight is passed to salt-ssh
+ and evil arguments are used in order to produce shell injection
+ """
+ injected_file = tmp_path / "injection"
+ _custom_roster(
+ salt_ssh_cli.roster_file,
+ {"ssh_pre_flight_args": f"foobar; echo injected > {str(injected_file)}"},
+ )
+ # Create pre_flight script that accepts args
+ test_script = _create_roster["test_script"]
+ test_script_1 = pathlib.Path(test_script + "-echo")
+ test_script_2 = pathlib.Path(test_script + "-foobar;")
+ with salt.utils.files.fopen(_create_roster["data"]["ssh_pre_flight"], "w") as fp:
+ fp.write(
+ f"""
+ touch {str(test_script)}-$1
+ touch {str(test_script)}-$2
+ """
+ )
+
+ # make sure we previously ran a command so the thin dir exists
+ ret = salt_ssh_cli.run("test.ping")
+ assert ret.returncode == 0
+ assert test_script_1.exists()
+ assert test_script_2.exists()
+ test_script_1.unlink()
+ test_script_2.unlink()
+ assert not injected_file.is_file()
+
+ ret = salt_ssh_cli.run(
+ "test.ping",
+ "--pre-flight",
+ )
+ assert ret.returncode == 0
+
+ assert test_script_1.exists()
+ assert test_script_2.exists()
+ assert not pathlib.Path(
+ injected_file
+ ).is_file(), "File injection suceeded. This shouldn't happend"
+
+
+@pytest.mark.flaky(max_runs=4)
+@pytest.mark.slow_test
+def test_ssh_run_pre_flight_failure(salt_ssh_cli, _create_roster):
+ """
+ test ssh_pre_flight when there is a failure
+ in the script.
+ """
+ with salt.utils.files.fopen(_create_roster["data"]["ssh_pre_flight"], "w") as fp_:
+ fp_.write("exit 2")
+
+ ret = salt_ssh_cli.run(
+ "test.ping",
+ "--pre-flight",
+ )
+ assert ret.data["retcode"] == 2
+
+
+@pytest.fixture
+def account():
+ username = random_string("test-account-", uppercase=False)
+ with pytest.helpers.create_account(username=username) as account:
+ yield account
+
+
+@pytest.mark.slow_test
+def test_ssh_pre_flight_script(salt_ssh_cli, caplog, _create_roster, tmp_path, account):
+ """
+ Test to ensure user cannot create and run a script
+ with the expected pre_flight script path on target.
+ """
+ try:
+ script = pathlib.Path.home() / "hacked"
+ tmp_preflight = pathlib.Path("/tmp", "ssh_pre_flight.sh")
+ tmp_preflight.write_text(f"touch {script}")
+ os.chown(tmp_preflight, account.info.uid, account.info.gid)
+ ret = salt_ssh_cli.run("test.ping")
+ assert not script.is_file()
+ assert ret.returncode == 0
+ assert ret.stdout == '{\n"localhost": true\n}\n'
+ finally:
+ for _file in [script, tmp_preflight]:
+ if _file.is_file():
+ _file.unlink()
+
+
+def demote(user_uid, user_gid):
+ def result():
+ # os.setgid does not remove group membership, so we remove them here so they are REALLY non-root
+ os.setgroups([])
+ os.setgid(user_gid)
+ os.setuid(user_uid)
+
+ return result
+
+
+@pytest.mark.slow_test
+def test_ssh_pre_flight_perms(salt_ssh_cli, caplog, _create_roster, account):
+ """
+ Test to ensure standard user cannot run pre flight script
+ on target when user sets wrong permissions (777) on
+ ssh_pre_flight script.
+ """
+ try:
+ script = pathlib.Path("/tmp", "itworked")
+ preflight = pathlib.Path("/ssh_pre_flight.sh")
+ preflight.write_text(f"touch {str(script)}")
+ tmp_preflight = pathlib.Path("/tmp", preflight.name)
+
+ _custom_roster(salt_ssh_cli.roster_file, {"ssh_pre_flight": str(preflight)})
+ preflight.chmod(0o0777)
+ run_script = pathlib.Path("/run_script")
+ run_script.write_text(
+ f"""
+ x=1
+ while [ $x -le 200000 ]; do
+ SCRIPT=`bash {str(tmp_preflight)} 2> /dev/null; echo $?`
+ if [ ${{SCRIPT}} == 0 ]; then
+ break
+ fi
+ x=$(( $x + 1 ))
+ done
+ """
+ )
+ run_script.chmod(0o0777)
+ # pylint: disable=W1509
+ ret = subprocess.Popen(
+ ["sh", f"{run_script}"],
+ preexec_fn=demote(account.info.uid, account.info.gid),
+ stdout=None,
+ stderr=None,
+ stdin=None,
+ universal_newlines=True,
+ )
+ # pylint: enable=W1509
+ ret = salt_ssh_cli.run("test.ping")
+ assert ret.returncode == 0
+
+ # Lets make sure a different user other than root
+ # Didn't run the script
+ assert os.stat(script).st_uid != account.info.uid
+ assert script.is_file()
+ finally:
+ for _file in [script, preflight, tmp_preflight, run_script]:
+ if _file.is_file():
+ _file.unlink()
+
+
+@pytest.mark.slow_test
+def test_ssh_run_pre_flight_target_file_perms(salt_ssh_cli, _create_roster, tmp_path):
+ """
+ test ssh_pre_flight to ensure the target pre flight script
+ has the correct perms
+ """
+ perms_file = tmp_path / "perms"
+ with salt.utils.files.fopen(_create_roster["data"]["ssh_pre_flight"], "w") as fp_:
+ fp_.write(
+ f"""
+ SCRIPT_NAME=$0
+ stat -L -c "%a %G %U" $SCRIPT_NAME > {perms_file}
+ """
+ )
+
+ ret = salt_ssh_cli.run(
+ "test.ping",
+ "--pre-flight",
+ )
+ assert ret.returncode == 0
+ with salt.utils.files.fopen(perms_file) as fp:
+ data = fp.read()
+ assert data.split()[0] == "600"
+ uid = os.getuid()
+ gid = os.getgid()
+ assert data.split()[1] == grp.getgrgid(gid).gr_name
+ assert data.split()[2] == pwd.getpwuid(uid).pw_name
diff --git a/tests/pytests/unit/client/ssh/test_single.py b/tests/pytests/unit/client/ssh/test_single.py
index f97519d5cc2..c88a1c2127f 100644
--- a/tests/pytests/unit/client/ssh/test_single.py
+++ b/tests/pytests/unit/client/ssh/test_single.py
@@ -1,6 +1,5 @@
-import os
+import logging
import re
-import tempfile
from textwrap import dedent
import pytest
@@ -16,6 +15,8 @@ import salt.utils.yaml
from salt.client import ssh
from tests.support.mock import MagicMock, call, patch
+log = logging.getLogger(__name__)
+
@pytest.fixture
def opts(tmp_path):
@@ -59,7 +60,7 @@ def test_single_opts(opts, target):
fsclient=None,
thin=salt.utils.thin.thin_path(opts["cachedir"]),
mine=False,
- **target
+ **target,
)
assert single.shell._ssh_opts() == ""
@@ -87,7 +88,7 @@ def test_run_with_pre_flight(opts, target, tmp_path):
fsclient=None,
thin=salt.utils.thin.thin_path(opts["cachedir"]),
mine=False,
- **target
+ **target,
)
cmd_ret = ("Success", "", 0)
@@ -122,7 +123,7 @@ def test_run_with_pre_flight_with_args(opts, target, tmp_path):
fsclient=None,
thin=salt.utils.thin.thin_path(opts["cachedir"]),
mine=False,
- **target
+ **target,
)
cmd_ret = ("Success", "foobar", 0)
@@ -156,7 +157,7 @@ def test_run_with_pre_flight_stderr(opts, target, tmp_path):
fsclient=None,
thin=salt.utils.thin.thin_path(opts["cachedir"]),
mine=False,
- **target
+ **target,
)
cmd_ret = ("", "Error running script", 1)
@@ -190,7 +191,7 @@ def test_run_with_pre_flight_script_doesnot_exist(opts, target, tmp_path):
fsclient=None,
thin=salt.utils.thin.thin_path(opts["cachedir"]),
mine=False,
- **target
+ **target,
)
cmd_ret = ("Success", "", 0)
@@ -224,7 +225,7 @@ def test_run_with_pre_flight_thin_dir_exists(opts, target, tmp_path):
fsclient=None,
thin=salt.utils.thin.thin_path(opts["cachedir"]),
mine=False,
- **target
+ **target,
)
cmd_ret = ("", "", 0)
@@ -242,6 +243,39 @@ def test_run_with_pre_flight_thin_dir_exists(opts, target, tmp_path):
assert ret == cmd_ret
+def test_run_ssh_pre_flight(opts, target, tmp_path):
+ """
+ test Single.run_ssh_pre_flight function
+ """
+ target["ssh_pre_flight"] = str(tmp_path / "script.sh")
+ single = ssh.Single(
+ opts,
+ opts["argv"],
+ "localhost",
+ mods={},
+ fsclient=None,
+ thin=salt.utils.thin.thin_path(opts["cachedir"]),
+ mine=False,
+ **target,
+ )
+
+ cmd_ret = ("Success", "", 0)
+ mock_flight = MagicMock(return_value=cmd_ret)
+ mock_cmd = MagicMock(return_value=cmd_ret)
+ patch_flight = patch("salt.client.ssh.Single.run_ssh_pre_flight", mock_flight)
+ patch_cmd = patch("salt.client.ssh.Single.cmd_block", mock_cmd)
+ patch_exec_cmd = patch(
+ "salt.client.ssh.shell.Shell.exec_cmd", return_value=("", "", 1)
+ )
+ patch_os = patch("os.path.exists", side_effect=[True])
+
+ with patch_os, patch_flight, patch_cmd, patch_exec_cmd:
+ ret = single.run()
+ mock_cmd.assert_called()
+ mock_flight.assert_called()
+ assert ret == cmd_ret
+
+
def test_execute_script(opts, target, tmp_path):
"""
test Single.execute_script()
@@ -255,7 +289,7 @@ def test_execute_script(opts, target, tmp_path):
thin=salt.utils.thin.thin_path(opts["cachedir"]),
mine=False,
winrm=False,
- **target
+ **target,
)
exp_ret = ("Success", "", 0)
@@ -273,7 +307,7 @@ def test_execute_script(opts, target, tmp_path):
] == mock_cmd.call_args_list
-def test_shim_cmd(opts, target):
+def test_shim_cmd(opts, target, tmp_path):
"""
test Single.shim_cmd()
"""
@@ -287,7 +321,7 @@ def test_shim_cmd(opts, target):
mine=False,
winrm=False,
tty=True,
- **target
+ **target,
)
exp_ret = ("Success", "", 0)
@@ -295,21 +329,24 @@ def test_shim_cmd(opts, target):
patch_cmd = patch("salt.client.ssh.shell.Shell.exec_cmd", mock_cmd)
patch_send = patch("salt.client.ssh.shell.Shell.send", return_value=("", "", 0))
patch_rand = patch("os.urandom", return_value=b"5\xd9l\xca\xc2\xff")
+ tmp_file = tmp_path / "tmp_file"
+ mock_tmp = MagicMock()
+ patch_tmp = patch("tempfile.NamedTemporaryFile", mock_tmp)
+ mock_tmp.return_value.__enter__.return_value.name = tmp_file
- with patch_cmd, patch_rand, patch_send:
+ with patch_cmd, patch_tmp, patch_send:
ret = single.shim_cmd(cmd_str="echo test")
assert ret == exp_ret
assert [
- call("/bin/sh '.35d96ccac2ff.py'"),
- call("rm '.35d96ccac2ff.py'"),
+ call(f"/bin/sh '.{tmp_file.name}'"),
+ call(f"rm '.{tmp_file.name}'"),
] == mock_cmd.call_args_list
-def test_run_ssh_pre_flight(opts, target, tmp_path):
+def test_shim_cmd_copy_fails(opts, target, caplog):
"""
- test Single.run_ssh_pre_flight
+ test Single.shim_cmd() when copying the file fails
"""
- target["ssh_pre_flight"] = str(tmp_path / "script.sh")
single = ssh.Single(
opts,
opts["argv"],
@@ -320,24 +357,202 @@ def test_run_ssh_pre_flight(opts, target, tmp_path):
mine=False,
winrm=False,
tty=True,
- **target
+ **target,
)
- exp_ret = ("Success", "", 0)
- mock_cmd = MagicMock(return_value=exp_ret)
+ ret_cmd = ("Success", "", 0)
+ mock_cmd = MagicMock(return_value=ret_cmd)
patch_cmd = patch("salt.client.ssh.shell.Shell.exec_cmd", mock_cmd)
- patch_send = patch("salt.client.ssh.shell.Shell.send", return_value=exp_ret)
- exp_tmp = os.path.join(
- tempfile.gettempdir(), os.path.basename(target["ssh_pre_flight"])
+ ret_send = ("", "General error in file copy", 1)
+ patch_send = patch("salt.client.ssh.shell.Shell.send", return_value=ret_send)
+ patch_rand = patch("os.urandom", return_value=b"5\xd9l\xca\xc2\xff")
+
+ with patch_cmd, patch_rand, patch_send:
+ ret = single.shim_cmd(cmd_str="echo test")
+ assert ret == ret_send
+ assert "Could not copy the shim script to target" in caplog.text
+ mock_cmd.assert_not_called()
+
+
+def test_run_ssh_pre_flight_no_connect(opts, target, tmp_path, caplog):
+ """
+ test Single.run_ssh_pre_flight when you
+ cannot connect to the target
+ """
+ pre_flight = tmp_path / "script.sh"
+ pre_flight.write_text("")
+ target["ssh_pre_flight"] = str(pre_flight)
+ single = ssh.Single(
+ opts,
+ opts["argv"],
+ "localhost",
+ mods={},
+ fsclient=None,
+ thin=salt.utils.thin.thin_path(opts["cachedir"]),
+ mine=False,
+ winrm=False,
+ tty=True,
+ **target,
)
+ mock_exec_cmd = MagicMock(return_value=("", "", 1))
+ patch_exec_cmd = patch("salt.client.ssh.shell.Shell.exec_cmd", mock_exec_cmd)
+ tmp_file = tmp_path / "tmp_file"
+ mock_tmp = MagicMock()
+ patch_tmp = patch("tempfile.NamedTemporaryFile", mock_tmp)
+ mock_tmp.return_value.__enter__.return_value.name = tmp_file
+ ret_send = (
+ "",
+ "ssh: connect to host 192.168.1.186 port 22: No route to host\nscp: Connection closed\n",
+ 255,
+ )
+ send_mock = MagicMock(return_value=ret_send)
+ patch_send = patch("salt.client.ssh.shell.Shell.send", send_mock)
+
+ with caplog.at_level(logging.TRACE):
+ with patch_send, patch_exec_cmd, patch_tmp:
+ ret = single.run_ssh_pre_flight()
+ assert "Copying the pre flight script" in caplog.text
+ assert "Could not copy the pre flight script to target" in caplog.text
+ assert ret == ret_send
+ assert send_mock.call_args_list[0][0][0] == tmp_file
+ target_script = send_mock.call_args_list[0][0][1]
+ assert re.search(r".[a-z0-9]+", target_script)
+ mock_exec_cmd.assert_not_called()
+
+
+def test_run_ssh_pre_flight_permission_denied(opts, target, tmp_path):
+ """
+ test Single.run_ssh_pre_flight when you
+ cannot copy script to the target due to
+ a permission denied error
+ """
+ pre_flight = tmp_path / "script.sh"
+ pre_flight.write_text("")
+ target["ssh_pre_flight"] = str(pre_flight)
+ single = ssh.Single(
+ opts,
+ opts["argv"],
+ "localhost",
+ mods={},
+ fsclient=None,
+ thin=salt.utils.thin.thin_path(opts["cachedir"]),
+ mine=False,
+ winrm=False,
+ tty=True,
+ **target,
+ )
+ mock_exec_cmd = MagicMock(return_value=("", "", 1))
+ patch_exec_cmd = patch("salt.client.ssh.shell.Shell.exec_cmd", mock_exec_cmd)
+ tmp_file = tmp_path / "tmp_file"
+ mock_tmp = MagicMock()
+ patch_tmp = patch("tempfile.NamedTemporaryFile", mock_tmp)
+ mock_tmp.return_value.__enter__.return_value.name = tmp_file
+ ret_send = (
+ "",
+ 'scp: dest open "/tmp/preflight.sh": Permission denied\nscp: failed to upload file /etc/salt/preflight.sh to /tmp/preflight.sh\n',
+ 255,
+ )
+ send_mock = MagicMock(return_value=ret_send)
+ patch_send = patch("salt.client.ssh.shell.Shell.send", send_mock)
- with patch_cmd, patch_send:
+ with patch_send, patch_exec_cmd, patch_tmp:
ret = single.run_ssh_pre_flight()
- assert ret == exp_ret
- assert [
- call("/bin/sh '{}'".format(exp_tmp)),
- call("rm '{}'".format(exp_tmp)),
- ] == mock_cmd.call_args_list
+ assert ret == ret_send
+ assert send_mock.call_args_list[0][0][0] == tmp_file
+ target_script = send_mock.call_args_list[0][0][1]
+ assert re.search(r".[a-z0-9]+", target_script)
+ mock_exec_cmd.assert_not_called()
+
+
+def test_run_ssh_pre_flight_connect(opts, target, tmp_path, caplog):
+ """
+ test Single.run_ssh_pre_flight when you
+ can connect to the target
+ """
+ pre_flight = tmp_path / "script.sh"
+ pre_flight.write_text("")
+ target["ssh_pre_flight"] = str(pre_flight)
+ single = ssh.Single(
+ opts,
+ opts["argv"],
+ "localhost",
+ mods={},
+ fsclient=None,
+ thin=salt.utils.thin.thin_path(opts["cachedir"]),
+ mine=False,
+ winrm=False,
+ tty=True,
+ **target,
+ )
+ ret_exec_cmd = ("", "", 1)
+ mock_exec_cmd = MagicMock(return_value=ret_exec_cmd)
+ patch_exec_cmd = patch("salt.client.ssh.shell.Shell.exec_cmd", mock_exec_cmd)
+ tmp_file = tmp_path / "tmp_file"
+ mock_tmp = MagicMock()
+ patch_tmp = patch("tempfile.NamedTemporaryFile", mock_tmp)
+ mock_tmp.return_value.__enter__.return_value.name = tmp_file
+ ret_send = (
+ "",
+ "\rroot@192.168.1.187's password: \n\rpreflight.sh 0% 0 0.0KB/s --:-- ETA\rpreflight.sh 100% 20 2.7KB/s 00:00 \n",
+ 0,
+ )
+ send_mock = MagicMock(return_value=ret_send)
+ patch_send = patch("salt.client.ssh.shell.Shell.send", send_mock)
+
+ with caplog.at_level(logging.TRACE):
+ with patch_send, patch_exec_cmd, patch_tmp:
+ ret = single.run_ssh_pre_flight()
+
+ assert "Executing the pre flight script on target" in caplog.text
+ assert ret == ret_exec_cmd
+ assert send_mock.call_args_list[0][0][0] == tmp_file
+ target_script = send_mock.call_args_list[0][0][1]
+ assert re.search(r".[a-z0-9]+", target_script)
+ mock_exec_cmd.assert_called()
+
+
+def test_run_ssh_pre_flight_shutil_fails(opts, target, tmp_path):
+ """
+ test Single.run_ssh_pre_flight when cannot
+ copyfile with shutil
+ """
+ pre_flight = tmp_path / "script.sh"
+ pre_flight.write_text("")
+ target["ssh_pre_flight"] = str(pre_flight)
+ single = ssh.Single(
+ opts,
+ opts["argv"],
+ "localhost",
+ mods={},
+ fsclient=None,
+ thin=salt.utils.thin.thin_path(opts["cachedir"]),
+ mine=False,
+ winrm=False,
+ tty=True,
+ **target,
+ )
+ ret_exec_cmd = ("", "", 1)
+ mock_exec_cmd = MagicMock(return_value=ret_exec_cmd)
+ patch_exec_cmd = patch("salt.client.ssh.shell.Shell.exec_cmd", mock_exec_cmd)
+ tmp_file = tmp_path / "tmp_file"
+ mock_tmp = MagicMock()
+ patch_tmp = patch("tempfile.NamedTemporaryFile", mock_tmp)
+ mock_tmp.return_value.__enter__.return_value.name = tmp_file
+ send_mock = MagicMock()
+ mock_shutil = MagicMock(side_effect=IOError("Permission Denied"))
+ patch_shutil = patch("shutil.copyfile", mock_shutil)
+ patch_send = patch("salt.client.ssh.shell.Shell.send", send_mock)
+
+ with patch_send, patch_exec_cmd, patch_tmp, patch_shutil:
+ ret = single.run_ssh_pre_flight()
+
+ assert ret == (
+ "",
+ "Could not copy pre flight script to temporary path",
+ 1,
+ )
+ mock_exec_cmd.assert_not_called()
+ send_mock.assert_not_called()
@pytest.mark.skip_on_windows(reason="SSH_PY_SHIM not set on windows")
@@ -355,7 +570,7 @@ def test_cmd_run_set_path(opts, target):
fsclient=None,
thin=salt.utils.thin.thin_path(opts["cachedir"]),
mine=False,
- **target
+ **target,
)
ret = single._cmd_str()
@@ -376,7 +591,7 @@ def test_cmd_run_not_set_path(opts, target):
fsclient=None,
thin=salt.utils.thin.thin_path(opts["cachedir"]),
mine=False,
- **target
+ **target,
)
ret = single._cmd_str()
@@ -395,7 +610,7 @@ def test_cmd_block_python_version_error(opts, target):
thin=salt.utils.thin.thin_path(opts["cachedir"]),
mine=False,
winrm=False,
- **target
+ **target,
)
mock_shim = MagicMock(
return_value=(("", "ERROR: Unable to locate appropriate python command\n", 10))
@@ -434,7 +649,9 @@ def test_run_with_pre_flight_args(opts, target, test_opts, tmp_path):
and script successfully runs
"""
opts["ssh_run_pre_flight"] = True
- target["ssh_pre_flight"] = str(tmp_path / "script.sh")
+ pre_flight_script = tmp_path / "script.sh"
+ pre_flight_script.write_text("")
+ target["ssh_pre_flight"] = str(pre_flight_script)
if test_opts[0] is not None:
target["ssh_pre_flight_args"] = test_opts[0]
@@ -448,7 +665,7 @@ def test_run_with_pre_flight_args(opts, target, test_opts, tmp_path):
fsclient=None,
thin=salt.utils.thin.thin_path(opts["cachedir"]),
mine=False,
- **target
+ **target,
)
cmd_ret = ("Success", "", 0)
@@ -456,14 +673,15 @@ def test_run_with_pre_flight_args(opts, target, test_opts, tmp_path):
mock_exec_cmd = MagicMock(return_value=("", "", 0))
patch_cmd = patch("salt.client.ssh.Single.cmd_block", mock_cmd)
patch_exec_cmd = patch("salt.client.ssh.shell.Shell.exec_cmd", mock_exec_cmd)
- patch_shell_send = patch("salt.client.ssh.shell.Shell.send", return_value=None)
+ patch_shell_send = patch(
+ "salt.client.ssh.shell.Shell.send", return_value=("", "", 0)
+ )
patch_os = patch("os.path.exists", side_effect=[True])
with patch_os, patch_cmd, patch_exec_cmd, patch_shell_send:
- ret = single.run()
- assert mock_exec_cmd.mock_calls[0].args[
- 0
- ] == "/bin/sh '/tmp/script.sh'{}".format(expected_args)
+ single.run()
+ script_args = mock_exec_cmd.mock_calls[0].args[0]
+ assert re.search(r"\/bin\/sh '.[a-z0-9]+", script_args)
@pytest.mark.slow_test
diff --git a/tests/pytests/unit/client/ssh/test_ssh.py b/tests/pytests/unit/client/ssh/test_ssh.py
index 377aad9998c..cece16026cf 100644
--- a/tests/pytests/unit/client/ssh/test_ssh.py
+++ b/tests/pytests/unit/client/ssh/test_ssh.py
@@ -339,3 +339,113 @@ def test_extra_filerefs(tmp_path, opts):
with patch("salt.roster.get_roster_file", MagicMock(return_value=roster)):
ssh_obj = client._prep_ssh(**ssh_opts)
assert ssh_obj.opts.get("extra_filerefs", None) == "salt://foobar"
+
+
+def test_key_deploy_permission_denied_scp(tmp_path, opts):
+ """
+ test "key_deploy" function when
+ permission denied authentication error
+ when attempting to use scp to copy file
+ to target
+ """
+ host = "localhost"
+ passwd = "password"
+ usr = "ssh-usr"
+ opts["ssh_user"] = usr
+ opts["tgt"] = host
+
+ ssh_ret = {
+ host: {
+ "stdout": "\rroot@192.168.1.187's password: \n\rroot@192.168.1.187's password: \n\rroot@192.168.1.187's password: \n",
+ "stderr": "Permission denied, please try again.\nPermission denied, please try again.\nroot@192.168.1.187: Permission denied (publickey,gssapi-keyex,gssapi-with-micimport pudb; pu.dbassword).\nscp: Connection closed\n",
+ "retcode": 255,
+ }
+ }
+ key_run_ret = {
+ "localhost": {
+ "jid": "20230922155652279959",
+ "return": "test",
+ "retcode": 0,
+ "id": "test",
+ "fun": "cmd.run",
+ "fun_args": ["echo test"],
+ }
+ }
+ patch_roster_file = patch("salt.roster.get_roster_file", MagicMock(return_value=""))
+ with patch_roster_file:
+ client = ssh.SSH(opts)
+ patch_input = patch("builtins.input", side_effect=["y"])
+ patch_getpass = patch("getpass.getpass", return_value=["password"])
+ mock_key_run = MagicMock(return_value=key_run_ret)
+ patch_key_run = patch("salt.client.ssh.SSH._key_deploy_run", mock_key_run)
+ with patch_input, patch_getpass, patch_key_run:
+ ret = client.key_deploy(host, ssh_ret)
+ assert mock_key_run.call_args_list[0][0] == (
+ host,
+ {"passwd": [passwd], "host": host, "user": usr},
+ True,
+ )
+ assert ret == key_run_ret
+ assert mock_key_run.call_count == 1
+
+
+def test_key_deploy_permission_denied_file_scp(tmp_path, opts):
+ """
+ test "key_deploy" function when permission denied
+ due to not having access to copy the file to the target
+ We do not want to deploy the key, because this is not
+ an authentication to the target error.
+ """
+ host = "localhost"
+ passwd = "password"
+ usr = "ssh-usr"
+ opts["ssh_user"] = usr
+ opts["tgt"] = host
+
+ mock_key_run = MagicMock(return_value=False)
+ patch_key_run = patch("salt.client.ssh.SSH._key_deploy_run", mock_key_run)
+
+ ssh_ret = {
+ "localhost": {
+ "stdout": "",
+ "stderr": 'scp: dest open "/tmp/preflight.sh": Permission denied\nscp: failed to upload file /etc/salt/preflight.sh to /tmp/preflight.sh\n',
+ "retcode": 1,
+ }
+ }
+ patch_roster_file = patch("salt.roster.get_roster_file", MagicMock(return_value=""))
+ with patch_roster_file:
+ client = ssh.SSH(opts)
+ ret = client.key_deploy(host, ssh_ret)
+ assert ret == ssh_ret
+ assert mock_key_run.call_count == 0
+
+
+def test_key_deploy_no_permission_denied(tmp_path, opts):
+ """
+ test "key_deploy" function when no permission denied
+ is returned
+ """
+ host = "localhost"
+ passwd = "password"
+ usr = "ssh-usr"
+ opts["ssh_user"] = usr
+ opts["tgt"] = host
+
+ mock_key_run = MagicMock(return_value=False)
+ patch_key_run = patch("salt.client.ssh.SSH._key_deploy_run", mock_key_run)
+ ssh_ret = {
+ "localhost": {
+ "jid": "20230922161937998385",
+ "return": "test",
+ "retcode": 0,
+ "id": "test",
+ "fun": "cmd.run",
+ "fun_args": ["echo test"],
+ }
+ }
+ patch_roster_file = patch("salt.roster.get_roster_file", MagicMock(return_value=""))
+ with patch_roster_file:
+ client = ssh.SSH(opts)
+ ret = client.key_deploy(host, ssh_ret)
+ assert ret == ssh_ret
+ assert mock_key_run.call_count == 0
--
2.42.0
Reference in New Issue View Git Blame Copy Permalink