From d30ada9b04ad5b592a5a445f051f862be7a45c0a Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Tue, 1 Sep 2026 16:00:15 +0530 Subject: [PATCH 1/4] Add shell-specific password validation to prevent command injection --- keepercommander/plugins/commands.py | 63 +++++++++++++++++++++++++---- 1 file changed, 56 insertions(+), 7 deletions(-) diff --git a/keepercommander/plugins/commands.py b/keepercommander/plugins/commands.py index 93b7a1f77..c46068d8d 100644 --- a/keepercommander/plugins/commands.py +++ b/keepercommander/plugins/commands.py @@ -62,8 +62,8 @@ 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 (;)', @@ -71,10 +71,26 @@ def register_command_info(aliases, command_info): '--': '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 (`)', + '--': 'double hyphen (--)', + '\n': 'newline', +} -def validate_user_supplied_rotation_password(new_password): + +def validate_database_rotation_password(new_password): # type: (str) -> bool - matches = UNSAFE_ROTATION_PASSWORD_PATTERN.findall(new_password) + matches = UNSAFE_DATABASE_ROTATION_PASSWORD_PATTERN.findall(new_password) if not matches: return True @@ -84,7 +100,7 @@ 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(_UNSAFE_DATABASE_ROTATION_PASSWORD_LABELS.get(match, repr(match))) if len(labels) == 1: logging.error( @@ -99,6 +115,33 @@ def validate_user_supplied_rotation_password(new_password): return False +def validate_shell_rotation_password(new_password): + # type: (str) -> bool + matches = UNSAFE_SHELL_ROTATION_PASSWORD_PATTERN.findall(new_password) + if not matches: + return True + + labels = [] + seen = set() + for match in matches: + if match in seen: + continue + seen.add(match) + labels.append(_UNSAFE_SHELL_ROTATION_PASSWORD_LABELS.get(match, repr(match))) + + if len(labels) == 1: + logging.error( + 'Password contains character unsafe for shell rotation: %s', + labels[0], + ) + else: + logging.error( + 'Password contains characters unsafe for shell rotation: %s', + ', '.join(labels), + ) + return False + + def adjust_password(password): # type: (str) -> str if not password: return password @@ -214,8 +257,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): - return False + else: + # Use shell-specific validation for plugins that build shell commands + if plugin_name in ('ssh', 'pspasswd'): + if not validate_shell_rotation_password(new_password): + return False + else: + if not validate_database_rotation_password(new_password): + return False if plugin_kwargs.get('password') == new_password: logging.warning('Rotation aborted because the old and new passwords are the same.') From 82c96587772a3aeef59a206d0e1ef75ac56de70d Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Tue, 1 Sep 2026 17:25:05 +0530 Subject: [PATCH 2/4] Add backward compatibility --- keepercommander/plugins/commands.py | 78 +++++++++++----------- keepercommander/plugins/windows/windows.py | 4 +- 2 files changed, 41 insertions(+), 41 deletions(-) diff --git a/keepercommander/plugins/commands.py b/keepercommander/plugins/commands.py index c46068d8d..6c1c0ca4d 100644 --- a/keepercommander/plugins/commands.py +++ b/keepercommander/plugins/commands.py @@ -71,7 +71,7 @@ def register_command_info(aliases, command_info): '--': 'double hyphen (--)', } -UNSAFE_SHELL_ROTATION_PASSWORD_PATTERN = re.compile(r"""[';"\\&|<>^`]|--|\n""") +UNSAFE_SHELL_ROTATION_PASSWORD_PATTERN = re.compile(r"""[';"\\&|<>^`$(){}!]|--|\n""") _UNSAFE_SHELL_ROTATION_PASSWORD_LABELS = { "'": "single quote (')", '"': 'double quote (")', @@ -83,14 +83,30 @@ def register_command_info(aliases, command_info): '>': 'greater-than (>)', '^': 'caret (^)', '`': 'backtick (`)', + '$': 'dollar sign ($)', + '(': 'opening parenthesis (()', + ')': 'closing parenthesis ())', + '{': 'opening brace ({)', + '}': 'closing brace (})', + '!': 'exclamation mark (!)', '--': 'double hyphen (--)', '\n': 'newline', } +SHELL_ROTATION_PLUGINS = {'ssh', 'pspasswd'} -def validate_database_rotation_password(new_password): - # type: (str) -> bool - matches = UNSAFE_DATABASE_ROTATION_PASSWORD_PATTERN.findall(new_password) + +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 + + 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 @@ -100,46 +116,29 @@ def validate_database_rotation_password(new_password): if match in seen: continue seen.add(match) - labels.append(_UNSAFE_DATABASE_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_shell_rotation_password(new_password): +def validate_database_rotation_password(new_password): # type: (str) -> bool - matches = UNSAFE_SHELL_ROTATION_PASSWORD_PATTERN.findall(new_password) - if not matches: - return True + return validate_rotation_password(new_password, is_shell_plugin=False) - labels = [] - seen = set() - for match in matches: - if match in seen: - continue - seen.add(match) - labels.append(_UNSAFE_SHELL_ROTATION_PASSWORD_LABELS.get(match, repr(match))) - if len(labels) == 1: - logging.error( - 'Password contains character unsafe for shell rotation: %s', - labels[0], - ) - else: - logging.error( - 'Password contains characters unsafe for shell rotation: %s', - ', '.join(labels), - ) - return 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 @@ -257,14 +256,15 @@ 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) - else: - # Use shell-specific validation for plugins that build shell commands - if plugin_name in ('ssh', 'pspasswd'): - if not validate_shell_rotation_password(new_password): - return False - else: - if not validate_database_rotation_password(new_password): - return False + + # 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: logging.warning('Rotation aborted because the old and new passwords are the same.') diff --git a/keepercommander/plugins/windows/windows.py b/keepercommander/plugins/windows/windows.py index d8d7c00de..dced6f7b6 100644 --- a/keepercommander/plugins/windows/windows.py +++ b/keepercommander/plugins/windows/windows.py @@ -14,8 +14,8 @@ import subprocess -# These characters don't work for Windows password rotation -DISALLOW_WINDOWS_SPECIAL_CHARACTERS = '<>^&|' +# These characters don't work for Windows password rotation or shell command execution +DISALLOW_WINDOWS_SPECIAL_CHARACTERS = '<>^&|$(){}!;"\'\\\n' class Rotator: From 311be7dd0070c2e430a1f795132920418b91aeba Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Tue, 1 Sep 2026 17:47:31 +0530 Subject: [PATCH 3/4] Fix review comments --- keepercommander/plugins/commands.py | 19 +++++++++++++++++++ keepercommander/plugins/pspasswd/pspasswd.py | 8 +++++++- keepercommander/plugins/ssh/ssh.py | 10 ++++++++-- keepercommander/plugins/windows/windows.py | 4 ++-- 4 files changed, 36 insertions(+), 5 deletions(-) diff --git a/keepercommander/plugins/commands.py b/keepercommander/plugins/commands.py index 6c1c0ca4d..3f0414728 100644 --- a/keepercommander/plugins/commands.py +++ b/keepercommander/plugins/commands.py @@ -95,6 +95,22 @@ def register_command_info(aliases, command_info): SHELL_ROTATION_PLUGINS = {'ssh', 'pspasswd'} +# Shell-unsafe characters: guard against command injection in shell-interpolated commands +UNSAFE_SHELL_CHARACTERS = '<>^&|$(){}!;"\'\\\n' + + +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 @@ -102,6 +118,9 @@ def validate_rotation_password(new_password, is_shell_plugin): 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' diff --git a/keepercommander/plugins/pspasswd/pspasswd.py b/keepercommander/plugins/pspasswd/pspasswd.py index ae2c17718..11ec212ab 100644 --- a/keepercommander/plugins/pspasswd/pspasswd.py +++ b/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}"') diff --git a/keepercommander/plugins/ssh/ssh.py b/keepercommander/plugins/ssh/ssh.py index c219bd865..2eb351d38 100644 --- a/keepercommander/plugins/ssh/ssh.py +++ b/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) diff --git a/keepercommander/plugins/windows/windows.py b/keepercommander/plugins/windows/windows.py index dced6f7b6..cbec3a97d 100644 --- a/keepercommander/plugins/windows/windows.py +++ b/keepercommander/plugins/windows/windows.py @@ -14,8 +14,8 @@ import subprocess -# These characters don't work for Windows password rotation or shell command execution -DISALLOW_WINDOWS_SPECIAL_CHARACTERS = '<>^&|$(){}!;"\'\\\n' +# Characters that break net user arg parsing (subprocess list, no shell) +DISALLOW_WINDOWS_SPECIAL_CHARACTERS = '<>^&|' class Rotator: From 9d42eb1e03b4c7bea3246fda9ecf7bda9961a3bf Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Thu, 3 Sep 2026 18:51:19 +0530 Subject: [PATCH 4/4] Add unixpasswd in the ssh list and white space in regex --- keepercommander/plugins/commands.py | 8 +++++--- keepercommander/plugins/unixpasswd/unixpasswd.py | 10 ++++++++-- 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/keepercommander/plugins/commands.py b/keepercommander/plugins/commands.py index 3f0414728..0953677b2 100644 --- a/keepercommander/plugins/commands.py +++ b/keepercommander/plugins/commands.py @@ -93,10 +93,12 @@ def register_command_info(aliases, command_info): '\n': 'newline', } -SHELL_ROTATION_PLUGINS = {'ssh', 'pspasswd'} +SHELL_ROTATION_PLUGINS = {'ssh', 'pspasswd', 'unixpasswd'} -# Shell-unsafe characters: guard against command injection in shell-interpolated commands -UNSAFE_SHELL_CHARACTERS = '<>^&|$(){}!;"\'\\\n' +# 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): diff --git a/keepercommander/plugins/unixpasswd/unixpasswd.py b/keepercommander/plugins/unixpasswd/unixpasswd.py index 4d4c3324a..9903541a4 100644 --- a/keepercommander/plugins/unixpasswd/unixpasswd.py +++ b/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')