diff --git a/ssh/access.py b/ssh/access.py index 7f09bc2..0ee676a 100644 --- a/ssh/access.py +++ b/ssh/access.py @@ -294,7 +294,15 @@ def verify_identity(self, request, email): ssh_public_key = request["ssh_pub_key"] if not ssh_public_key or ssh_public_key == '': return {} - return {"ssh_public_key": ssh_public_key} + # Reject anything that is not a single, well-formed OpenSSH public key. + # The value is later interpolated into remote shell commands, so an + # unvalidated key is an OS command injection vector (CWE-78). + if not helpers.is_valid_ssh_public_key(ssh_public_key): + logger.warning( + "SSHModule: rejected malformed ssh public key from %s", email + ) + return {} + return {"ssh_public_key": ssh_public_key.strip()} def can_auto_approve(self): return False diff --git a/ssh/helpers.py b/ssh/helpers.py index ca8a0da..7e42e0b 100644 --- a/ssh/helpers.py +++ b/ssh/helpers.py @@ -1,5 +1,7 @@ """ Helper methods for ssh module """ +import re +import shlex import traceback import logging from fabric import Connection @@ -11,6 +13,13 @@ global ssh_machine_list ssh_machine_list = {} +# A POSIX/Linux login name: starts with a lowercase letter or underscore, +# followed by lowercase letters, digits, underscores or hyphens, optionally +# ending with a trailing '$'. Max 32 chars (useradd limit). This is an +# allow-list so that no shell metacharacter (``$(`` , backticks, ``;`` , ``/`` , +# spaces, ...) can ever reach a remote command via the username. +_USERNAME_RE = re.compile(r"^[a-z_][a-z0-9_-]{0,31}\$?$") + class SSHModuleError(Exception): """ Custom error class """ @@ -19,6 +28,47 @@ def __init__(self, message): self.message = message +def is_valid_username(username): + """Return True only for a well-formed Linux login name. + + Usernames flow into remote shell commands and filesystem paths + (``/home//...``); restricting them to a strict allow-list + prevents both command injection and path traversal. + """ + return bool(username) and bool(_USERNAME_RE.match(username)) + + +# An OpenSSH public key line: " [ optional comment]". +# The key type and base64 body are drawn from fixed character sets, so a valid +# key can never contain shell metacharacters. The optional comment is validated +# separately (below) to keep it free of newlines and shell metacharacters. +_SSH_KEY_TYPE_RE = ( + r"(?:ssh-ed25519|ssh-rsa|ssh-dss|" + r"ecdsa-sha2-nistp(?:256|384|521)|" + r"sk-ssh-ed25519@openssh\.com|sk-ecdsa-sha2-nistp256@openssh\.com)" +) +_SSH_KEY_RE = re.compile( + r"^" + _SSH_KEY_TYPE_RE + r" [A-Za-z0-9+/]+={0,3}( [\x20-\x7e]*)?$" +) + + +def is_valid_ssh_public_key(ssh_key): + """Return True only for a single, well-formed OpenSSH public key line. + + Rejects empty values, multi-line input, and anything carrying characters + outside the `` `` grammar. This is the source-side + guard against OS command injection through the ``ssh_public_key`` field; the + remote commands additionally ``shlex.quote`` the value as defence in depth. + """ + if not ssh_key or not isinstance(ssh_key, str): + return False + # Reject any newline/carriage-return so a payload cannot smuggle a second + # authorized_keys line or shell command. + if "\n" in ssh_key or "\r" in ssh_key: + return False + return bool(_SSH_KEY_RE.match(ssh_key.strip())) + + def _get_inventory_file_path(): if "ssh" in ACCESS_MODULES: if "inventory_file_path" in ACCESS_MODULES["ssh"]: @@ -88,6 +138,14 @@ def sshHelper(labels, user_identity, user, action): username = get_username(access_level, user) ssh_key = user_identity.identity["ssh_public_key"] + if not is_valid_username(username): + logger.error("SSHModule: rejected malformed username %r", username) + return False, "Invalid username" + + if not is_valid_ssh_public_key(ssh_key): + logger.error("SSHModule: rejected malformed ssh public key") + return False, "Invalid SSH public key" + if action == "grant": if access_level in ["sudo", "nonsudo"]: return add_user(hostname, ip, ssh_key, username, access_level) @@ -104,9 +162,10 @@ def add_key_existing_user(ip, ssh_key, access_level, username): return False, "Authentication failed to machine." try: + authorized_keys = "/home/{}/.ssh/authorized_keys".format(username) connection.sudo( - 'echo "{}" | sudo tee -a /home/{}/.ssh/authorized_keys > /dev/null'.format( - ssh_key, username + "echo {} | sudo tee -a {} > /dev/null".format( + shlex.quote(ssh_key), shlex.quote(authorized_keys) ) ) except Exception as e: @@ -126,43 +185,52 @@ def add_user(hostname, ip, ssh_key, username, access_level): return False, "Authentication failed to machine." # Check if the user already exists, if so, return and do nothing - if not connection.sudo("id {}".format(username), warn=True).failed: + quoted_username = shlex.quote(username) + if not connection.sudo("id {}".format(quoted_username), warn=True).failed: logger.info("User already exists") return False, "User already exists" try: + ssh_dir = "/home/{}/.ssh".format(username) + authorized_keys = "{}/authorized_keys".format(ssh_dir) + # Create the user - connection.sudo("useradd -m {}".format(username)) + connection.sudo("useradd -m {}".format(quoted_username)) # Set the password to nothing - connection.sudo("passwd -d {}".format(username)) + connection.sudo("passwd -d {}".format(quoted_username)) # Check if the user should be a root user or a basic user if access_level == "sudo": connection.sudo( "usermod -aG {} {}".format( - ACCESS_MODULES["ssh"]["common_sudo_group"], username + shlex.quote(ACCESS_MODULES["ssh"]["common_sudo_group"]), + quoted_username, ) ) # Create the .ssh directory - connection.sudo("mkdir /home/{}/.ssh".format(username)) + connection.sudo("mkdir {}".format(shlex.quote(ssh_dir))) # Create the authorized_keys file - connection.sudo("touch /home/{}/.ssh/authorized_keys".format(username)) + connection.sudo("touch {}".format(shlex.quote(authorized_keys))) # Add the user's SSH key to the authorized_keys file on the remote machine connection.sudo( - 'echo "{}" | sudo tee -a /home/{}/.ssh/authorized_keys > /dev/null'.format( - ssh_key, username + "echo {} | sudo tee -a {} > /dev/null".format( + shlex.quote(ssh_key), shlex.quote(authorized_keys) ) ) # Change the permissions of the authorized_keys file to 600 # (only the user can read and write) - connection.sudo("chmod 600 /home/{}/.ssh/authorized_keys".format(username)) + connection.sudo("chmod 600 {}".format(shlex.quote(authorized_keys))) # change the ownership of the /home/ directory to the user and # group of the user (username:username) - connection.sudo("chown -R {}:{} /home/{}".format(username, username, username)) + connection.sudo( + "chown -R {}:{} {}".format( + quoted_username, quoted_username, shlex.quote("/home/{}".format(username)) + ) + ) except Exception as e: logger.error("Exception occured while adding user: " + str(e)) traceback.print_exc() @@ -182,17 +250,24 @@ def replace_user_key(hostname, ip, new_ssh_key, old_ssh_key, username): if not connection: return False, "Authentication failed to machine." - # Replace the / with \/ in the old SSH key and the new SSH key so that it can - # be used in the sed command below (sed command uses / as a delimiter) - old_ssh_key = old_ssh_key.replace("/", "\/") # noqa - new_ssh_key = new_ssh_key.replace("/", "\/") # noqa + if not is_valid_username(username): + logger.error("SSHModule: rejected malformed username %r", username) + return False, "Invalid username" + + # Escape sed's regex metacharacters so the keys are matched/substituted + # literally (the security guard against shell injection is shlex.quote on + # the whole sed expression below; this only preserves sed correctness). + old_ssh_key = re.sub(r"([/\\&.*[\]^$])", r"\\\1", old_ssh_key) + new_ssh_key = re.sub(r"([/\\&])", r"\\\1", new_ssh_key) try: # Replace the old SSH key with the new SSH key in # the authorized_keys file on the remote machine + sed_expr = "s/{}/{}/g".format(old_ssh_key, new_ssh_key) + authorized_keys = "/home/{}/.ssh/authorized_keys".format(username) connection.sudo( - 'sed -i "s/{}/{}/g" /home/{}/.ssh/authorized_keys'.format( - old_ssh_key, new_ssh_key, username + "sed -i {} {}".format( + shlex.quote(sed_expr), shlex.quote(authorized_keys) ) ) except Exception as e: diff --git a/ssh/test_ssh_injection.py b/ssh/test_ssh_injection.py new file mode 100644 index 0000000..9b2a8f6 --- /dev/null +++ b/ssh/test_ssh_injection.py @@ -0,0 +1,173 @@ +"""SSH access module — OS command injection regression tests (GHSA-grpj-9ghj-xhv8). + +These tests lock in the fix for the second-order OS command injection where a +requester-controlled SSH public key / username was interpolated verbatim into +remote shell commands. They cover both layers of the fix: + + * source-side validation (``is_valid_ssh_public_key`` / ``is_valid_username`` + and ``SSHAccess.verify_identity``), and + * sink-side ``shlex.quote`` hardening in the ``connection.sudo`` calls. +""" + +import shlex + +import pytest + +from . import helpers, access + + +BENIGN_KEY = ( + "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIExampleBenignKeyDataXXXXXX user@laptop" +) +# Command-substitution payload embedded in the key / username fields. +MALICIOUS_KEY = "ssh-ed25519 AAAATEST$(touch /tmp/pwned)attacker@example.com" +MALICIOUS_USERNAME = "app;touch /tmp/pwned" + + +# --------------------------------------------------------------------------- # +# Pure validators +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize( + "key", + [ + BENIGN_KEY, + "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgExampleRsaBody+/09 me@host", + "ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTY= me@host", + ], +) +def test_valid_keys_accepted(key): + assert helpers.is_valid_ssh_public_key(key) is True + + +@pytest.mark.parametrize( + "key", + [ + "", + None, + MALICIOUS_KEY, + "ssh-ed25519 AAAA`whoami` c", + "ssh-ed25519 AAAA; rm -rf / #", + "ssh-ed25519 AAAA\ntouch /tmp/pwned", # newline-smuggled second line + "not-a-key-type AAAABBBB", + "ssh-ed25519 AAAA$body more", # '$' is not valid base64 + ], +) +def test_malicious_or_malformed_keys_rejected(key): + assert helpers.is_valid_ssh_public_key(key) is False + + +@pytest.mark.parametrize("username", ["app", "svc_1", "build-agent", "_sys"]) +def test_valid_usernames_accepted(username): + assert helpers.is_valid_username(username) is True + + +@pytest.mark.parametrize( + "username", + ["", None, MALICIOUS_USERNAME, "app$(id)", "../root", "a b", "UPPER", "x" * 40], +) +def test_malicious_or_malformed_usernames_rejected(username): + assert helpers.is_valid_username(username) is False + + +# --------------------------------------------------------------------------- # +# Source: verify_identity +# --------------------------------------------------------------------------- # +def test_verify_identity_rejects_malicious_key(): + result = access.SSHAccess().verify_identity({"ssh_pub_key": MALICIOUS_KEY}, "u@x.com") + assert result == {} + + +def test_verify_identity_accepts_benign_key(): + result = access.SSHAccess().verify_identity({"ssh_pub_key": BENIGN_KEY}, "u@x.com") + assert result == {"ssh_public_key": BENIGN_KEY} + + +# --------------------------------------------------------------------------- # +# Dispatcher: malicious input never reaches a shell +# --------------------------------------------------------------------------- # +def _identity(mocker, ssh_key): + ident = mocker.MagicMock() + ident.identity = {"ssh_public_key": ssh_key} + return ident + + +def _user(mocker, username): + user = mocker.MagicMock() + user.user.username = username + return user + + +def test_ssh_helper_rejects_malicious_key_without_connecting(mocker): + conn = mocker.patch.object(helpers, "get_connection_to_host") + labels = [{"access_level": "app", "machine": "h", "ip": "127.0.0.1"}] + ok, msg = helpers.sshHelper( + labels, _identity(mocker, MALICIOUS_KEY), _user(mocker, "app"), "grant" + ) + assert ok is False + assert msg == "Invalid SSH public key" + conn.assert_not_called() + + +def test_ssh_helper_rejects_malicious_username_without_connecting(mocker): + conn = mocker.patch.object(helpers, "get_connection_to_host") + # access_level "sudo" makes get_username fall back to user.user.username + labels = [{"access_level": "sudo", "machine": "h", "ip": "127.0.0.1"}] + ok, msg = helpers.sshHelper( + labels, _identity(mocker, BENIGN_KEY), _user(mocker, MALICIOUS_USERNAME), "grant" + ) + assert ok is False + assert msg == "Invalid username" + conn.assert_not_called() + + +# --------------------------------------------------------------------------- # +# Sink hardening: even if validation is bypassed, payloads are shell-quoted +# --------------------------------------------------------------------------- # +def _sudo_commands(conn_mock): + return [c.args[0] for c in conn_mock.sudo.call_args_list] + + +def test_add_key_existing_user_quotes_payload(mocker): + conn = mocker.MagicMock() + mocker.patch.object(helpers, "get_connection_to_host", return_value=conn) + + helpers.add_key_existing_user("127.0.0.1", MALICIOUS_KEY, "app", "app") + + cmd = _sudo_commands(conn)[0] + # The raw command-substitution must NOT appear unquoted… + assert "$(touch /tmp/pwned)" not in cmd.replace(shlex.quote(MALICIOUS_KEY), "") + # …because the whole key is passed as one shell-quoted token. + assert shlex.quote(MALICIOUS_KEY) in cmd + + +def test_add_user_quotes_username_and_key(mocker): + conn = mocker.MagicMock() + conn.sudo.return_value.failed = True # "id " -> user does not exist yet + mocker.patch.object(helpers, "get_connection_to_host", return_value=conn) + + helpers.add_user("h", "127.0.0.1", MALICIOUS_KEY, MALICIOUS_USERNAME, "sudo") + + cmds = _sudo_commands(conn) + assert cmds, "expected sudo commands to be issued" + for cmd in cmds: + assert "; touch" not in cmd # bare metacharacter never reaches the shell + assert "$(touch" not in cmd or shlex.quote(MALICIOUS_KEY) in cmd + + +def test_replace_user_key_quotes_sed_expression(mocker): + conn = mocker.MagicMock() + mocker.patch.object(helpers, "get_connection_to_host", return_value=conn) + + ok, _ = helpers.replace_user_key("h", "127.0.0.1", "", MALICIOUS_KEY, "app") + + cmd = _sudo_commands(conn)[0] + assert cmd.startswith("sed -i ") + # sed expression is a single shell-quoted token — no unquoted $() + assert "$(touch /tmp/pwned)" not in cmd + + +def test_replace_user_key_rejects_bad_username(mocker): + conn = mocker.patch.object(helpers, "get_connection_to_host") + ok, msg = helpers.replace_user_key("h", "127.0.0.1", "", BENIGN_KEY, MALICIOUS_USERNAME) + assert ok is False + assert msg == "Invalid username"