diff --git a/keepercommander/commands/base.py b/keepercommander/commands/base.py index b0e3d1ff3..861bb8567 100644 --- a/keepercommander/commands/base.py +++ b/keepercommander/commands/base.py @@ -713,6 +713,14 @@ def normalize_output_param(args: str) -> str: return args +def tokenize_command_args(args, envvars=None): + """Tokenize command arguments using the same transforms as Command.execute_args.""" + args = '' if args is None else args + if envvars is not None: + args = expand_cmd_args(args, envvars) + return shlex.split(normalize_output_param(args)) + + def as_boolean(value, default=None): # type: (Any, Optional[bool]) -> bool if isinstance(value, bool): return value @@ -769,14 +777,13 @@ def execute_args(self, params, args, **kwargs): alias_cmd = d.get('command') if alias_cmd and alias_cmd != parser.prog: parser.prog = alias_cmd - args = expand_cmd_args(args, envvars) - args = normalize_output_param(args) + command_tokens = tokenize_command_args(args, envvars) if self.support_extra_parameters(): - opts, extra_args = parser.parse_known_args(shlex.split(args)) + opts, extra_args = parser.parse_known_args(command_tokens) if extra_args: self.extra_parameters = ' '.join(extra_args) else: - opts = parser.parse_args(shlex.split(args)) + opts = parser.parse_args(command_tokens) d.update(opts.__dict__) return self.execute(params, **d) diff --git a/keepercommander/service/util/command_util.py b/keepercommander/service/util/command_util.py index d38ca9d2b..ef7162d11 100644 --- a/keepercommander/service/util/command_util.py +++ b/keepercommander/service/util/command_util.py @@ -270,21 +270,40 @@ def execute(cls, command: str, temp_files: Optional[list] = None) -> Tuple[Any, command = ensure_record_add_json_format(html.unescape(command)) try: - command_tokens = Verifycommand.tokenize_service_command(command) + # Check the raw command before environment expansion: Service + # Mode explicitly forbids this syntax, and expansion must not + # erase evidence of it before the restriction check. + raw_command_tokens = Verifycommand.tokenize_service_command(command) except ValueError: return {"status": "error", "error": "Invalid command syntax"}, 400 - # This request's own FILEDATA directory - the only paths Service - # Mode will treat as safe, not the whole shared OS temp root. - request_temp_dir = os.path.dirname(temp_files[0]) if temp_files else None + blocked_command = raw_command_tokens[0] if raw_command_tokens else '' def blocked(error): logger.warning( - f"Service Mode blocked command '{command_tokens[0] if command_tokens else ''}': {error}" + f"Service Mode blocked command '{blocked_command}': {error}" ) return {"status": "error", "error": error}, 403 - # Same tokens the CLI will run — do not use raw HTTP split(" ") + expansion_error = Verifycommand.validate_service_mode_env_var_expansion_command( + raw_command_tokens + ) + if expansion_error: + return blocked(expansion_error) + + try: + command_tokens = Verifycommand.tokenize_service_command( + command, params.environment_variables + ) + except ValueError: + return {"status": "error", "error": "Invalid command syntax"}, 400 + + # This request's own FILEDATA directory - the only paths Service + # Mode will treat as safe, not the whole shared OS temp root. + request_temp_dir = os.path.dirname(temp_files[0]) if temp_files else None + + # Apply every Service Mode restriction to the same tokens the + # CLI's Command.execute_args parser will receive. service_mode_error = Verifycommand.validate_service_mode_restrictions( command_tokens, request_temp_dir ) diff --git a/keepercommander/service/util/request_validation.py b/keepercommander/service/util/request_validation.py index c5d9cbf32..7bb6b939c 100644 --- a/keepercommander/service/util/request_validation.py +++ b/keepercommander/service/util/request_validation.py @@ -16,6 +16,7 @@ import tempfile import os import json +import shlex import shutil from ..decorators.logging import logger, sanitize_command_fields @@ -82,19 +83,35 @@ def process_file_data(request_data: Dict[str, Any], command: str) -> Tuple[str, temp_file_path = temp_file.name temp_files.append(temp_file_path) - # Replace FILEDATA placeholder with temporary file path, but only when it's used as a file parameter - # Supports: PAM import (--filename), Import (positional), Enterprise-push (positional) - # Callable replacement -- a plain string would let a Windows path's backslashes be parsed as regex backreferences. - file_param_pattern = r'(--filename[=\s]+["\']?)FILEDATA(["\']?)' + # Replace the entire FILEDATA token with a quoted path. + # Reusing the caller's quotes would break when the + # generated path itself contains a single quote. + file_param_pattern = ( + r'(--filename[=\s]+)(?P["\']?)FILEDATA(?P=quote)' + ) + + def replace_filename_placeholder(match): + prefix = match.group(1) + return f'{prefix}{shlex.quote(temp_file_path)}' + processed_command = re.sub( file_param_pattern, - lambda m: f'{m.group(1)}{temp_file_path}{m.group(2)}', + replace_filename_placeholder, processed_command, ) - # Also handle standalone FILEDATA (not in quotes) - standalone_pattern = r'\bFILEDATA\b(?!["\'])' - processed_command = re.sub(standalone_pattern, lambda _: temp_file_path, processed_command) + # Also handle positional FILEDATA (including a quoted + # placeholder), replacing its original quoting entirely. + standalone_pattern = r'(?P["\']?)\bFILEDATA\b(?P=quote)' + + def replace_standalone_placeholder(match): + return shlex.quote(temp_file_path) + + processed_command = re.sub( + standalone_pattern, + replace_standalone_placeholder, + processed_command, + ) logger.debug(f"Created temporary file {temp_file_path} for filedata") diff --git a/keepercommander/service/util/verified_command.py b/keepercommander/service/util/verified_command.py index 71c77e033..ba4165227 100644 --- a/keepercommander/service/util/verified_command.py +++ b/keepercommander/service/util/verified_command.py @@ -2,7 +2,6 @@ import contextlib import io import os -import shlex class Verifycommand: @@ -99,16 +98,20 @@ class Verifycommand: _HOST_FS_MSG = ( 'Local filesystem access is not permitted through Service Mode' ) + _AUTOMATOR_DATA_EXPANSION_MSG = ( + 'Automator file and environment expansions are not permitted through Service Mode' + ) @staticmethod def validate_service_mode_restrictions(command_tokens, request_temp_dir=None): - """Run Service Mode bans on executor tokens (shlex); error string or None.""" + """Run Service Mode bans over the exact tokens used by the command parser.""" if not command_tokens: return None for validator in ( Verifycommand.validate_service_mode_double_dash, Verifycommand.validate_service_mode_env_var_expansion_command, + Verifycommand.validate_service_mode_automator_file_access_command, Verifycommand.validate_service_mode_legacy_command, Verifycommand.validate_nsf_convert_command, Verifycommand.validate_service_mode_pam_tunnel_command, @@ -127,9 +130,10 @@ def validate_service_mode_restrictions(command_tokens, request_temp_dir=None): return None @staticmethod - def tokenize_service_command(command): - """Tokenize a command using the executor's backslash handling.""" - return shlex.split(command.replace('\\', '\\\\')) + def tokenize_service_command(command, envvars=None): + """Tokenize with the same argument transforms used by Command.execute_args.""" + from ...commands.base import tokenize_command_args + return tokenize_command_args(command, envvars) @staticmethod def validate_nsf_convert_command(command_tokens, request_temp_dir=None): @@ -186,6 +190,62 @@ def validate_service_mode_env_var_expansion_command(command_tokens, request_temp return Verifycommand._ENV_VAR_EXPANSION_MSG return None + @staticmethod + def validate_service_mode_automator_file_access_command(command_tokens, request_temp_dir=None): + """Block Automator operations that read local files or environment values.""" + if not command_tokens: + return None + + command = command_tokens[0].lower() + if command == 'automator-certificate': + return Verifycommand._HOST_FS_MSG + + if command != 'automator' or len(command_tokens) < 2: + return None + + action = command_tokens[1].lower() + if action == 'certificate': + # This is the registered group form in current Commander releases. + return Verifycommand._HOST_FS_MSG + + if action != 'edit': + return None + + # --set supports both ``--set KEY=VALUE`` and ``--set=KEY=VALUE``; + # argparse also accepts the unambiguous ``--se`` abbreviation. + # Extract only those settings instead of matching unrelated Automator + # arguments such as --url. + settings = [] + i = 2 + while i < len(command_tokens): + token = command_tokens[i] + option, separator, inline_value = token.partition('=') + if option in ('--set', '--se'): + if separator: + settings.append(inline_value) + elif i + 1 < len(command_tokens): + settings.append(command_tokens[i + 1]) + i += 1 + i += 1 + + for setting in settings: + separator = setting.find('=') + if separator <= 1: + # AutomatorEditCommand uses the same pos > 1 condition and + # returns before resolving expansions for malformed settings. + continue + value = setting[separator + 1:].strip() + if not (value.startswith('${') and value.endswith('}')): + continue + expansion_type, colon, source = value[2:-1].partition(':') + if ( + colon + and expansion_type.lower() in ('file', 'base64', 'env') + and source + and '}' not in source): + return Verifycommand._AUTOMATOR_DATA_EXPANSION_MSG + return None + @staticmethod def validate_service_mode_legacy_command(command_tokens, request_temp_dir=None): """Block legacy commands in Service Mode; error or None.""" @@ -254,7 +314,7 @@ def validate_service_mode_record_file_attachment_command(command_tokens, request @staticmethod def _is_record_file_attachment_arg(token): - """True when parse_field would treat this token as a file attachment field.""" + """True when parse_field would treat this executor token as a file field.""" if not token or '=' not in token: return False # Mirror RecordEditMixin.parse_field field-name normalization. diff --git a/unit-tests/service/test_verified_command.py b/unit-tests/service/test_verified_command.py index 2796a3cb5..f70a6ec75 100644 --- a/unit-tests/service/test_verified_command.py +++ b/unit-tests/service/test_verified_command.py @@ -1,4 +1,4 @@ -from unittest import TestCase +from unittest import TestCase, mock from html import unescape import os import shutil @@ -6,16 +6,14 @@ import shlex +from keepercommander.commands.base import normalize_output_param from keepercommander.service.util.verified_command import Verifycommand def _tokens(command: str): """Same normalization CommandExecutor uses before policy checks.""" command = unescape(command) - try: - return shlex.split(command.replace('\\', '\\\\')) - except ValueError: - return command.split() + return Verifycommand.tokenize_service_command(command) class TestServiceModeCommandPolicy(TestCase): @@ -108,6 +106,96 @@ def test_attachment_commands_blocked_for_remote_api(self): ) self.assertIsNone(check(_tokens('record-add --title t -rt login login=user'))) + def test_backslash_escaped_record_file_field_is_blocked(self): + """Escaped field names are checked after the same parsing as execution.""" + check = Verifycommand.validate_service_mode_restrictions + + # The HTTP JSON representation of this argument contains ``file\\=``; + # after JSON decoding there is one backslash, which shlex consumes and + # turns into the actual ``file=`` field name seen by record-add. + for field in ( + r'file\=/etc/passwd', + r'f.file\=/etc/passwd', + r'fi\le=/etc/passwd', + r'f.fi\le.doc\=/etc/passwd', + ): + command = f'record-add -t f01 -rt login --self-destruct=1d {field}' + with self.subTest(field=field): + self.assertEqual(_tokens(command), shlex.split(command)) + self.assertIsNotNone(check(_tokens(command))) + + # The ordinary spelling remains blocked as before. + self.assertIsNotNone(check(_tokens( + 'record-add -t f01 -rt login --self-destruct=1d file=/etc/passwd' + ))) + self.assertIsNone(check(_tokens( + r'record-add -t f01 -rt login lo\gin=user' + ))) + + def test_escaped_host_path_options_are_blocked(self): + """Filesystem flags and parser-driven file inputs use execution tokens.""" + check = Verifycommand.validate_service_mode_restrictions + ban = 'Local filesystem access' + for cmd in ( + r'generate --out\put=/tmp/sm-write-canary.txt', + r'generate --out\put /tmp/sm-write-canary.txt', + r'import --for\mat=json /etc/passwd', + r'import --file\-cache=/tmp/sm-cache', + r'pam project import --filen\ame=/etc/passwd', + r'audit-report --form\at=pdf', + ): + with self.subTest(cmd=cmd): + self.assertIn(ban, check(_tokens(cmd))) + + def test_automator_local_file_access_is_blocked(self): + check = Verifycommand.validate_service_mode_restrictions + host_fs_msg = 'Local filesystem access' + for cmd in ( + 'automator-certificate /etc/passwd', + 'automator certificate /etc/passwd', + ): + with self.subTest(cmd=cmd): + self.assertIn(host_fs_msg, check(_tokens(cmd))) + + def test_automator_local_data_expansions_are_blocked(self): + check = Verifycommand.validate_service_mode_restrictions + expansion_msg = 'Automator file and environment expansions' + for cmd in ( + 'automator edit TestAutomator --set debugUrl=${file:/etc/passwd}', + 'automator edit TestAutomator --set key=${base64:/etc/shadow}', + 'automator edit TestAutomator --set key=${env:AWS_SECRET_ACCESS_KEY}', + 'automator edit TestAutomator --set key=${FiLe:/etc/passwd}', + 'automator edit TestAutomator --set "debugUrl = ${file:/etc/passwd}"', + 'automator edit TestAutomator --set "debugUrl= ${file:/etc/passwd}"', + 'automator edit TestAutomator --set "debugUrl=${file:/etc/passwd} "', + 'automator edit TestAutomator --set="debugUrl = ${base64:/etc/shadow}"', + 'automator edit TestAutomator --se "debugUrl = ${file:/etc/passwd}"', + ): + with self.subTest(cmd=cmd): + self.assertIn(expansion_msg, check(_tokens(cmd))) + + # Ordinary Automator edits remain usable through Service Mode. + self.assertIsNone(check(_tokens( + 'automator edit TestAutomator --set debugUrl=https://example.invalid' + ))) + # The expansion syntax is only interpreted for --set values, not URL. + self.assertIsNone(check(_tokens( + 'automator edit TestAutomator --url=${file:/tmp/canary.txt}' + ))) + + def test_quoted_backslash_path_tokens_match_executor(self): + # The CLI normalizes --output before shlex on Windows; Service Mode + # must apply the same sequence and keep the path as one argument. + path = r'C:\service\files\payload.txt' + command = f'generate --output "{path}"' + with mock.patch('sys.platform', 'win32'): + expected_tokens = shlex.split(normalize_output_param(command)) + actual_tokens = Verifycommand.tokenize_service_command(command) + + self.assertEqual(actual_tokens, expected_tokens) + self.assertEqual(actual_tokens[-1], r'C:/service/files/payload.txt') + self.assertEqual(len(actual_tokens), 3) + def test_host_filesystem_commands_blocked(self): check = Verifycommand.validate_service_mode_restrictions ban = 'Local filesystem access' @@ -221,16 +309,29 @@ def test_filename_allows_temp_filedata_paths_only(self): temp_path = os.path.join(request_temp_dir, 'service_filedata_test.json') self.assertIsNone( - check(_tokens(f'pam project import --filename={temp_path}'), request_temp_dir) + check( + _tokens(f'pam project import --filename={shlex.quote(temp_path)}'), + request_temp_dir, + ) + ) + self.assertIsNone( + check(_tokens(f'pam project import -f {shlex.quote(temp_path)}'), request_temp_dir) ) - self.assertIsNone(check(_tokens(f'pam project import -f {temp_path}'), request_temp_dir)) - self.assertIsNone(check(_tokens(f'import --format=json {temp_path}'), request_temp_dir)) self.assertIsNone( - check(_tokens(f'enterprise-push {temp_path} --email user@example.com'), request_temp_dir) + check(_tokens(f'import --format=json {shlex.quote(temp_path)}'), request_temp_dir) + ) + self.assertIsNone( + check( + _tokens(f'enterprise-push {shlex.quote(temp_path)} --email user@example.com'), + request_temp_dir, + ) ) # PAM --config is a vault UID, not a host path self.assertIsNone( - check(_tokens('pam project extend --config=SOME_UID -f ' + temp_path), request_temp_dir) + check( + _tokens(f'pam project extend --config=SOME_UID -f {shlex.quote(temp_path)}'), + request_temp_dir, + ) ) def test_command_aliases_do_not_bypass_host_path_checks(self): @@ -254,8 +355,15 @@ def test_command_aliases_do_not_bypass_host_path_checks(self): self.addCleanup(shutil.rmtree, request_temp_dir, ignore_errors=True) temp_path = os.path.join(request_temp_dir, 'service_filedata_alias_test.json') self.assertIsNone(check(_tokens('gen --output stdout'))) - self.assertIsNone(check(_tokens(f'pam p i -f {temp_path}'), request_temp_dir)) - self.assertIsNone(check(_tokens(f'pam p e --filename={temp_path}'), request_temp_dir)) + self.assertIsNone( + check(_tokens(f'pam p i -f {shlex.quote(temp_path)}'), request_temp_dir) + ) + self.assertIsNone( + check( + _tokens(f'pam p e --filename={shlex.quote(temp_path)}'), + request_temp_dir, + ) + ) def test_non_request_temp_path_still_rejected(self): """A path under a DIFFERENT request's temp dir is not automatically safe, @@ -270,11 +378,15 @@ def test_non_request_temp_path_still_rejected(self): other_path = os.path.join(other_request_dir, 'someone_elses_file.json') self.assertIn( - ban, check(_tokens(f'pam project import -f {other_path}'), request_temp_dir) + ban, + check( + _tokens(f'pam project import -f {shlex.quote(other_path)}'), + request_temp_dir, + ), ) # No request_temp_dir at all (e.g. a request with no FILEDATA) trusts nothing. self.assertIn( - ban, check(_tokens(f'pam project import -f {other_path}')) + ban, check(_tokens(f'pam project import -f {shlex.quote(other_path)}')) ) def test_flag_abbreviations_do_not_bypass_host_path_checks(self): @@ -321,7 +433,22 @@ def test_import_bare_name_positional_requires_format_awareness(self): request_temp_dir = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, request_temp_dir, ignore_errors=True) temp_path = os.path.join(request_temp_dir, 'import_data.json') - self.assertIsNone(check(_tokens(f'import --format=json {temp_path}'), request_temp_dir)) + self.assertIsNone( + check(_tokens(f'import --format=json {shlex.quote(temp_path)}'), request_temp_dir) + ) + + def test_windows_output_normalization_is_checked_as_a_file_path(self): + check = Verifycommand.validate_service_mode_restrictions + normalized_cases = [] + with mock.patch('sys.platform', 'win32'): + for output_value in Verifycommand._NON_PATH_OUTPUT_VALUES: + command = rf'public-api-key list --format json --output \{output_value}' + normalized_cases.append((output_value, _tokens(command))) + + for output_value, tokens in normalized_cases: + with self.subTest(output_value=output_value): + self.assertEqual(tokens[-1], f'/{output_value}') + self.assertIn('Local filesystem access', check(tokens)) def test_filedata_substitution_happens_before_validation(self): """process_file_data must run before validate_service_mode_restrictions, @@ -336,6 +463,7 @@ def test_filedata_substitution_happens_before_validation(self): self.assertTrue(temp_files, 'expected a temp file to be created') self.assertNotIn('FILEDATA', processed_command) + self.assertTrue(any(token.endswith(temp_files[0]) for token in _tokens(processed_command))) request_temp_dir = os.path.dirname(temp_files[0]) tokens = _tokens(processed_command) @@ -343,6 +471,131 @@ def test_filedata_substitution_happens_before_validation(self): Verifycommand.validate_service_mode_restrictions(tokens, request_temp_dir) ) + def test_filedata_positional_placeholder_is_shell_quoted(self): + """A generated path remains one executor token, including with spaces.""" + from keepercommander.service.util.request_validation import RequestValidator + from unittest.mock import patch + + request_data = {'filedata': {'some': 'data'}} + command = 'import --format=json FILEDATA' + request_temp_dir = tempfile.mkdtemp(prefix='service mode ') + self.addCleanup(shutil.rmtree, request_temp_dir, ignore_errors=True) + with patch( + 'keepercommander.service.util.request_validation.tempfile.mkdtemp', + return_value=request_temp_dir): + processed_command, temp_files = RequestValidator.process_file_data( + request_data, command + ) + self.addCleanup(RequestValidator.cleanup_temp_files, temp_files) + + self.assertEqual( + shlex.split(processed_command), + ['import', '--format=json', temp_files[0]], + ) + + def test_filedata_filename_placeholder_is_shell_quoted(self): + from keepercommander.service.util.request_validation import RequestValidator + from unittest.mock import patch + + request_data = {'filedata': {'some': 'data'}} + command = 'pam project import --filename=FILEDATA' + request_temp_dir = tempfile.mkdtemp(prefix='service mode ') + self.addCleanup(shutil.rmtree, request_temp_dir, ignore_errors=True) + with patch( + 'keepercommander.service.util.request_validation.tempfile.mkdtemp', + return_value=request_temp_dir): + processed_command, temp_files = RequestValidator.process_file_data( + request_data, command + ) + self.addCleanup(RequestValidator.cleanup_temp_files, temp_files) + + tokens = shlex.split(processed_command) + self.assertEqual(tokens[-1], f'--filename={temp_files[0]}') + self.assertIsNone( + Verifycommand.validate_service_mode_restrictions( + tokens, request_temp_dir + ) + ) + + def test_filedata_windows_path_survives_executor_tokenization(self): + from keepercommander.service.util.request_validation import RequestValidator + from unittest.mock import MagicMock, patch + + request_data = {'filedata': {'some': 'data'}} + command = 'import --format=json FILEDATA' + temp_file_path = r'C:\service mode\payload.json' + temp_file = MagicMock() + temp_file.name = temp_file_path + with patch( + 'keepercommander.service.util.request_validation.tempfile.mkdtemp', + return_value=r'C:\service mode'), patch( + 'keepercommander.service.util.request_validation.tempfile.NamedTemporaryFile') as make_temp_file: + make_temp_file.return_value.__enter__.return_value = temp_file + processed_command, temp_files = RequestValidator.process_file_data( + request_data, command + ) + + self.assertEqual(temp_files, [temp_file_path]) + self.assertEqual( + shlex.split(processed_command), + ['import', '--format=json', temp_file_path], + ) + + def test_filedata_quoted_positional_placeholder_is_replaced(self): + from keepercommander.service.util.request_validation import RequestValidator + + request_data = {'filedata': {'some': 'data'}} + command = 'import --format=json "FILEDATA"' + processed_command, temp_files = RequestValidator.process_file_data( + request_data, command + ) + self.addCleanup(RequestValidator.cleanup_temp_files, temp_files) + + self.assertEqual( + shlex.split(processed_command), + ['import', '--format=json', temp_files[0]], + ) + + def test_quoted_filedata_placeholders_support_apostrophes_in_path(self): + from keepercommander.service.util.request_validation import RequestValidator + request_data = {'filedata': {'some': 'data'}} + commands = ( + "import --format=json 'FILEDATA'", + "pam project import --filename='FILEDATA'", + ) + + for command in commands: + with self.subTest(command=command), tempfile.TemporaryDirectory() as parent_dir: + request_temp_dir = os.path.join(parent_dir, "service user's files") + os.mkdir(request_temp_dir) + with mock.patch( + 'keepercommander.service.util.request_validation.tempfile.mkdtemp', + return_value=request_temp_dir): + processed_command, temp_files = RequestValidator.process_file_data( + request_data, command + ) + self.addCleanup(RequestValidator.cleanup_temp_files, temp_files) + + expected_last_token = ( + f'--filename={temp_files[0]}' + if '--filename' in command else temp_files[0] + ) + self.assertEqual(_tokens(processed_command)[-1], expected_last_token) + + def test_mismatched_filedata_quotes_remain_invalid_command_syntax(self): + """Malformed placeholder quoting must fail closed, not be repaired.""" + from keepercommander.service.util.request_validation import RequestValidator + + request_data = {'filedata': {'some': 'data'}} + command = 'pam project import --filename="FILEDATA\'' + processed_command, temp_files = RequestValidator.process_file_data( + request_data, command + ) + self.addCleanup(RequestValidator.cleanup_temp_files, temp_files) + + with self.assertRaises(ValueError): + Verifycommand.tokenize_service_command(processed_command) + def test_option_values_yields_dash_leading_values(self): is_file = Verifycommand._option_values self.assertEqual(list(is_file(['generate', '--output', '-'], '--output')), ['-'])