diff --git a/keepercommander/plugins/commands.py b/keepercommander/plugins/commands.py index 93b7a1f77..0953677b2 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,63 @@ 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 (`)', + '$': '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: 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/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') diff --git a/keepercommander/plugins/windows/windows.py b/keepercommander/plugins/windows/windows.py index d8d7c00de..cbec3a97d 100644 --- a/keepercommander/plugins/windows/windows.py +++ b/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 = '<>^&|'