fix(ssh): prevent OS command injection via ssh key and username (CTO-5387)

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

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/<username>/...``); 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: "<type> <base64-blob>[ 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 ``<type> <base64> <comment>`` 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/<username> 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:

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>" -> 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"