Sign Up
Log In
Log In
or
Sign Up
Places
All Projects
Status Monitor
Collapse sidebar
openSUSE:Leap:15.5:Update
salt.34212
fix-cve-2023-34049-bsc-1215157.patch
Overview
Repositories
Revisions
Requests
Users
Attributes
Meta
File fix-cve-2023-34049-bsc-1215157.patch of Package salt.34212
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
Locations
Projects
Search
Status Monitor
Help
OpenBuildService.org
Documentation
API Documentation
Code of Conduct
Contact
Support
@OBShq
Terms
openSUSE Build Service is sponsored by
The Open Build Service is an
openSUSE project
.
Sign Up
Log In
Places
Places
All Projects
Status Monitor
