KC-1446: Add shell-specific password validation to prevent command injection

keepercommander/plugins/commands.py

@@ -62,19 +62,72 @@ def register_command_info(aliases, command_info):
 rotate_parser.error = raise_parse_exception
 rotate_parser.exit = suppress_exit
 
-UNSAFE_ROTATION_PASSWORD_PATTERN = re.compile(r"""[';"\\]|--""")
-_UNSAFE_ROTATION_PASSWORD_LABELS = {
+UNSAFE_DATABASE_ROTATION_PASSWORD_PATTERN = re.compile(r"""[';"\\]|--""")
+_UNSAFE_DATABASE_ROTATION_PASSWORD_LABELS = {
     "'": "single quote (')",
     '"': 'double quote (")',
     ';': 'semicolon (;)',
     '\\': 'backslash (\\)',
     '--': 'double hyphen (--)',
 }
 
+UNSAFE_SHELL_ROTATION_PASSWORD_PATTERN = re.compile(r"""[';"\\&|<>^`$(){}!]|--|\n""")
+_UNSAFE_SHELL_ROTATION_PASSWORD_LABELS = {
+    "'": "single quote (')",
+    '"': 'double quote (")',
+    ';': 'semicolon (;)',
+    '\\': 'backslash (\\)',
+    '&': 'ampersand (&)',
+    '|': 'pipe (|)',
+    '<': 'less-than (<)',
+    '>': 'greater-than (>)',
+    '^': 'caret (^)',
+    '`': 'backtick (`)',
+    '$': 'dollar sign ($)',
+    '(': 'opening parenthesis (()',
+    ')': 'closing parenthesis ())',
+    '{': 'opening brace ({)',
+    '}': 'closing brace (})',
+    '!': 'exclamation mark (!)',
+    '--': 'double hyphen (--)',
+    '\n': 'newline',
+}
 
-def validate_user_supplied_rotation_password(new_password):
-    # type: (str) -> bool
-    matches = UNSAFE_ROTATION_PASSWORD_PATTERN.findall(new_password)
+SHELL_ROTATION_PLUGINS = {'ssh', 'pspasswd', 'unixpasswd'}
+
+# Shell-unsafe characters: guard against command/argument injection in shell-interpolated commands.
+# Whitespace is included because unquoted values are split into separate command arguments
+# (e.g. "net user {user} {password}" - a password containing a space injects extra flags).
+UNSAFE_SHELL_CHARACTERS = '<>^&|$(){}!;"\'\\\n \t'
+
+
+def validate_shell_command_parameter(param, param_name):
+    # type: (str, str) -> bool
+    if not isinstance(param, str):
+        logging.error(f'{param_name} must be a string')
+        return False
+
+    for char in UNSAFE_SHELL_CHARACTERS:
+        if char in param:
+            logging.error(f'{param_name} contains shell metacharacter: {repr(char)}')
+            return False
+    return True
+
+
+def validate_rotation_password(new_password, is_shell_plugin):
+    # type: (str, bool) -> bool
+    if not isinstance(new_password, str):
+        logging.error('Password must be a string')
+        return False
+
+    if is_shell_plugin and not validate_shell_command_parameter(new_password, 'Password'):
+        return False
+
+    pattern = UNSAFE_SHELL_ROTATION_PASSWORD_PATTERN if is_shell_plugin else UNSAFE_DATABASE_ROTATION_PASSWORD_PATTERN
+    labels_map = _UNSAFE_SHELL_ROTATION_PASSWORD_LABELS if is_shell_plugin else _UNSAFE_DATABASE_ROTATION_PASSWORD_LABELS
+    context = 'shell' if is_shell_plugin else 'database'
+
+    matches = pattern.findall(new_password)
     if not matches:
         return True
 
@@ -84,21 +137,31 @@ def validate_user_supplied_rotation_password(new_password):
         if match in seen:
             continue
         seen.add(match)
-        labels.append(_UNSAFE_ROTATION_PASSWORD_LABELS.get(match, repr(match)))
+        labels.append(labels_map.get(match, repr(match)))
 
     if len(labels) == 1:
         logging.error(
-            'Password contains character unsafe for database rotation: %s',
-            labels[0],
+            'Password contains character unsafe for %s rotation: %s',
+            context, labels[0],
         )
     else:
         logging.error(
-            'Password contains characters unsafe for database rotation: %s',
-            ', '.join(labels),
+            'Password contains characters unsafe for %s rotation: %s',
+            context, ', '.join(labels),
         )
     return False
 
 
+def validate_database_rotation_password(new_password):
+    # type: (str) -> bool
+    return validate_rotation_password(new_password, is_shell_plugin=False)
+
+
+def validate_shell_rotation_password(new_password):
+    # type: (str) -> bool
+    return validate_rotation_password(new_password, is_shell_plugin=True)
+
+
 def adjust_password(password):   # type: (str) -> str
     if not password:
         return password
@@ -214,7 +277,14 @@ def rotate_password(params, record_uid, rotate_name=None, plugin_name=None, host
         if not length:
             length = plugin_kwargs.get('length')
         new_password = get_new_password(plugin, rules, length)
-    elif not validate_user_supplied_rotation_password(new_password):
+
+    # Validate password regardless of source (auto-generated or user-supplied)
+    if not new_password or not isinstance(new_password, str):
+        logging.error('Password generation or validation failed: invalid password')
+        return False
+
+    is_shell_plugin = plugin_name in SHELL_ROTATION_PLUGINS
+    if not validate_rotation_password(new_password, is_shell_plugin):
         return False
 
     if plugin_kwargs.get('password') == new_password:

keepercommander/plugins/pspasswd/pspasswd.py

@@ -29,8 +29,14 @@ def rotate_start_msg(self):
 
     def rotate(self, record, new_password):
         """Rotate Windows account password"""
+        from ..commands import validate_shell_command_parameter
+
+        if not validate_shell_command_parameter(self.login, 'User login'):
+            return False
+        if not validate_shell_command_parameter(new_password, 'New password'):
+            return False
+
         host_arg = f'\\\\{self.host} ' if self.host else ''
-        # the characters below mess with windows command line
         escape_quote_password = new_password.replace('"', '""')
         error_code = subprocess.call(f'pspasswd {host_arg}{self.login} "{escape_quote_password}"')
 

keepercommander/plugins/ssh/ssh.py

@@ -58,9 +58,15 @@ def rotate_ssh(host, port, user, old_password, new_password, timeout=5, revert=F
     old_password(str): old password
     new_password(str): new password
     timeout(int): SSH connection timeout in seconds
-    revert(bool): True if the new_password is the original password to revert a previous rotation.
-                  This is used to print log messages that make more sense.
+    revert(bool): True to revert a previous rotation
     """
+    from ..commands import validate_shell_command_parameter
+
+    if not validate_shell_command_parameter(user, 'User login'):
+        return False
+    if not validate_shell_command_parameter(new_password, 'New password'):
+        return False
+
     rotate_success = False
     ssh_logger = logging.getLogger('paramiko')
     ssh_logger.setLevel(logging.WARNING)

keepercommander/plugins/unixpasswd/unixpasswd.py

@@ -27,12 +27,18 @@
 
 
 def rotate(record, newpassword):
+    from ..commands import validate_shell_command_parameter
+
     user = RecordMixin.get_record_field(record, 'login')
     oldpassword = RecordMixin.get_record_field(record, 'password')
 
+    if not validate_shell_command_parameter(user, 'User login'):
+        return False
+    if not validate_shell_command_parameter(newpassword, 'New password'):
+        return False
+
     logging.info('Connecting to super user %s', user)
-    user = user.replace("\\", "\\\\").replace("\"", "\\\"").replace(";", "\\;")
-    p = pexpect.spawn(f'su - "{user}"', timeout=5)
+    p = pexpect.spawn('su', ['-', user], timeout=5)
     p.expect('[Pp]assword')
     if not p.waitnoecho(1):
         raise Exception('Password prompt is expected')

keepercommander/plugins/windows/windows.py

@@ -14,7 +14,7 @@
 import subprocess
 
 
-# These characters don't work for Windows password rotation
+# Characters that break net user arg parsing (subprocess list, no shell)
 DISALLOW_WINDOWS_SPECIAL_CHARACTERS = '<>^&|'