From 0725d91a970a1aeb84706e3e473c7a29deed4d70 Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Thu, 13 Aug 2026 16:42:36 +0530 Subject: [PATCH 1/4] Restrict Service Mode host file I/O and pam tunnel commands --- keepercommander/commands/record_edit.py | 14 ++++++ keepercommander/service/decorators/auth.py | 9 ++++ .../service/util/verified_command.py | 19 ++++++++ unit-tests/service/test_auth_security.py | 44 ++++++++++++++++++- unit-tests/test_command_record.py | 17 +++++++ 5 files changed, 102 insertions(+), 1 deletion(-) diff --git a/keepercommander/commands/record_edit.py b/keepercommander/commands/record_edit.py index c0750fdb1..470ae0383 100644 --- a/keepercommander/commands/record_edit.py +++ b/keepercommander/commands/record_edit.py @@ -773,6 +773,13 @@ def assign_typed_fields(self, record, fields): def upload_attachments(self, params, record, files, stop_on_error): # type: (KeeperParams, Union[vault.PasswordRecord, vault.TypedRecord], List[ParsedFieldValue], bool) -> None + if files and getattr(params, 'service_mode', False): + # Remote Service Mode callers must not open arbitrary host paths + # (e.g. file=@~/.keeper/config.json). Local CLI is unchanged. + raise CommandError( + '', + 'File attachments by local path are not permitted through Service Mode') + tasks = [] for file_attachment in files: if file_attachment.value.startswith('@'): @@ -1503,6 +1510,13 @@ def get_parser(self): return download_parser def execute(self, params, **kwargs): + if getattr(params, 'service_mode', False): + # Downloads write to the Commander host disk; remote API callers + # do not receive the file bytes in the HTTP response. + raise CommandError( + 'download-attachment', + 'Downloading attachments to the local filesystem is not permitted through Service Mode') + records = kwargs.get('records') if not records: self.get_parser().print_help() diff --git a/keepercommander/service/decorators/auth.py b/keepercommander/service/decorators/auth.py index 3ced7b0d4..25afb67fc 100644 --- a/keepercommander/service/decorators/auth.py +++ b/keepercommander/service/decorators/auth.py @@ -128,6 +128,15 @@ def wrapper(*args, **kwargs): 'status': 'error', 'error': transform_folder_error }, 400 + + # Block pam tunnel (except edit) in Service Mode — closes host shell via --run + pam_tunnel_error = Verifycommand.validate_pam_tunnel_command(command) + if pam_tunnel_error: + logger.debug(f"Command validation failed: {command[0]} - {pam_tunnel_error}") + return { + 'status': 'error', + 'error': pam_tunnel_error + }, 403 return fn(*args, **kwargs) return wrapper \ No newline at end of file diff --git a/keepercommander/service/util/verified_command.py b/keepercommander/service/util/verified_command.py index 2f62f5f3a..f8f574276 100644 --- a/keepercommander/service/util/verified_command.py +++ b/keepercommander/service/util/verified_command.py @@ -1,4 +1,23 @@ class Verifycommand: + @staticmethod + def validate_pam_tunnel_command(command): + """ + Service Mode: only 'pam tunnel edit' is allowed (aliases: pam t edit/e). + Blocks start/list/stop/diagnose (and aliases), including pam tunnel start --run. + Returns None if allowed or not a pam tunnel command; error string if blocked. + """ + if not command or len(command) < 2: + return None + if command[0] != 'pam': + return None + if command[1] not in ('tunnel', 't'): + return None + if len(command) >= 3 and command[2] in ('edit', 'e'): + return None + return ( + 'pam tunnel commands other than edit are not available in Service Mode' + ) + @staticmethod def validate_append_command(command): """ diff --git a/unit-tests/service/test_auth_security.py b/unit-tests/service/test_auth_security.py index bbd94ba4b..93a113e0e 100644 --- a/unit-tests/service/test_auth_security.py +++ b/unit-tests/service/test_auth_security.py @@ -3,6 +3,7 @@ from keepercommander.service.decorators.auth import auth_check, policy_check from keepercommander.service.decorators.security import security_check, is_allowed_ip from keepercommander.service.util.config_reader import ConfigReader +from keepercommander.service.util.verified_command import Verifycommand class TestAuthSecurity(TestCase): def setUp(self): @@ -96,4 +97,45 @@ def test_policy_check_denied_command(self, mock_read_config): response = policy_check(lambda *args, **kwargs: ({'status': 'success'}, 200))() self.assertEqual(response[1], 403) self.assertEqual(response[0]['status'], 'error') - self.assertIn('Not permitted', response[0]['error']) \ No newline at end of file + self.assertIn('Not permitted', response[0]['error']) + + @mock.patch.object(ConfigReader, 'read_config') + def test_policy_check_blocks_pam_tunnel_start(self, mock_read_config): + """pam tunnel start (incl. --run) is blocked in Service Mode""" + mock_read_config.return_value = "pam" + + with self.app.test_request_context( + '/test', method='POST', + json={"command": "pam tunnel start uid --run id"}, + headers={'api-key': 'test_key'}, + ): + response = policy_check(lambda *args, **kwargs: ({'status': 'success'}, 200))() + self.assertEqual(response[1], 403) + self.assertIn('pam tunnel', response[0]['error']) + + @mock.patch.object(ConfigReader, 'read_config') + def test_policy_check_allows_pam_tunnel_edit(self, mock_read_config): + """pam tunnel edit remains allowed in Service Mode""" + mock_read_config.return_value = "pam" + + with self.app.test_request_context( + '/test', method='POST', + json={"command": "pam tunnel edit uid --enable-tunneling"}, + headers={'api-key': 'test_key'}, + ): + response = policy_check(lambda *args, **kwargs: ({'status': 'success'}, 200))() + self.assertEqual(response[1], 200) + self.assertEqual(response[0]['status'], 'success') + + def test_validate_pam_tunnel_command(self): + """Unit-level checks for pam tunnel Service Mode allowlist""" + ban = 'not available in Service Mode' + self.assertIsNone(Verifycommand.validate_pam_tunnel_command(['pam', 'tunnel', 'edit', 'uid'])) + self.assertIsNone(Verifycommand.validate_pam_tunnel_command(['pam', 't', 'e', 'uid'])) + self.assertIsNone(Verifycommand.validate_pam_tunnel_command(['pam', 'rotation', 'list'])) + self.assertIn(ban, Verifycommand.validate_pam_tunnel_command( + ['pam', 'tunnel', 'start', 'uid', '--run', 'id'])) + self.assertIn(ban, Verifycommand.validate_pam_tunnel_command(['pam', 'tunnel', 'list'])) + self.assertIn(ban, Verifycommand.validate_pam_tunnel_command(['pam', 'tunnel', 'stop', 'uid'])) + self.assertIn(ban, Verifycommand.validate_pam_tunnel_command(['pam', 'tunnel', 'diagnose'])) + self.assertIn(ban, Verifycommand.validate_pam_tunnel_command(['pam', 't', 's', 'uid'])) diff --git a/unit-tests/test_command_record.py b/unit-tests/test_command_record.py index d21e3927f..57982d224 100644 --- a/unit-tests/test_command_record.py +++ b/unit-tests/test_command_record.py @@ -436,6 +436,23 @@ def test_append_notes_command(self): with self.assertRaises(CommandError): cmd.execute(params, notes='notes', record='invalid') + def test_upload_attachments_blocked_in_service_mode(self): + params = get_synced_params() + params.service_mode = True + mixin = record_edit.RecordEditMixin() + files = [record_edit.ParsedFieldValue('', 'file', '', '@/tmp/any-file.txt')] + with self.assertRaises(CommandError) as ctx: + mixin.upload_attachments(params, vault.TypedRecord(), files, True) + self.assertIn('Service Mode', str(ctx.exception)) + + def test_download_attachment_blocked_in_service_mode(self): + params = get_synced_params() + params.service_mode = True + cmd = record_edit.RecordDownloadAttachmentCommand() + with self.assertRaises(CommandError) as ctx: + cmd.execute(params, records=['any-uid']) + self.assertIn('Service Mode', str(ctx.exception)) + def test_download_attachment_command(self): params = get_synced_params() cmd = record_edit.RecordDownloadAttachmentCommand() From ae593dacd0f07e9056f4588f9c4b8328f8c372f4 Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Thu, 13 Aug 2026 20:05:41 +0530 Subject: [PATCH 2/4] Fix review comments --- keepercommander/service/decorators/auth.py | 11 +--- keepercommander/service/util/command_util.py | 8 +++ .../service/util/verified_command.py | 42 +++++++++----- unit-tests/service/test_auth_security.py | 35 ++++++------ .../service/test_service_mode_pam_tunnel.py | 57 +++++++++++++++++++ 5 files changed, 113 insertions(+), 40 deletions(-) create mode 100644 unit-tests/service/test_service_mode_pam_tunnel.py diff --git a/keepercommander/service/decorators/auth.py b/keepercommander/service/decorators/auth.py index 25afb67fc..2d21da195 100644 --- a/keepercommander/service/decorators/auth.py +++ b/keepercommander/service/decorators/auth.py @@ -129,14 +129,5 @@ def wrapper(*args, **kwargs): 'error': transform_folder_error }, 400 - # Block pam tunnel (except edit) in Service Mode — closes host shell via --run - pam_tunnel_error = Verifycommand.validate_pam_tunnel_command(command) - if pam_tunnel_error: - logger.debug(f"Command validation failed: {command[0]} - {pam_tunnel_error}") - return { - 'status': 'error', - 'error': pam_tunnel_error - }, 403 - return fn(*args, **kwargs) - return wrapper \ No newline at end of file + return wrapper diff --git a/keepercommander/service/util/command_util.py b/keepercommander/service/util/command_util.py index 5458b495c..2a70d8a96 100644 --- a/keepercommander/service/util/command_util.py +++ b/keepercommander/service/util/command_util.py @@ -168,6 +168,14 @@ def execute(cls, command: str) -> Tuple[Any, int]: command_tokens = shlex.split(command) except ValueError: command_tokens = command.split() + + # Same tokens the CLI will run — do not use raw HTTP split(" ") + service_mode_error = Verifycommand.validate_service_mode_restrictions( + command_tokens + ) + if service_mode_error: + return {"status": "error", "error": service_mode_error}, 403 + force_error = Verifycommand.validate_enterprise_user_add_role_force( command_tokens, params ) diff --git a/keepercommander/service/util/verified_command.py b/keepercommander/service/util/verified_command.py index f8f574276..87c8dec29 100644 --- a/keepercommander/service/util/verified_command.py +++ b/keepercommander/service/util/verified_command.py @@ -1,22 +1,36 @@ class Verifycommand: + # pam tunnel aliases: start=s, list=l, stop=x, edit=e, diagnose=d + # (see PAMTunnelCommand.register_command in tunnel_and_connections.py) + _PAM_TUNNEL_ALIASES = { + 's': 'start', + 'l': 'list', + 'x': 'stop', + 'e': 'edit', + 'd': 'diagnose', + } + _PAM_TUNNEL_ALLOWED = frozenset({'edit'}) + @staticmethod - def validate_pam_tunnel_command(command): + def validate_service_mode_restrictions(command_tokens): """ - Service Mode: only 'pam tunnel edit' is allowed (aliases: pam t edit/e). - Blocks start/list/stop/diagnose (and aliases), including pam tunnel start --run. - Returns None if allowed or not a pam tunnel command; error string if blocked. + Authoritative Service Mode command policy. + + Call on tokens from CommandExecutor (html.unescape + shlex.split), + not on raw HTTP split(" "). Returns error string if blocked, else None. """ - if not command or len(command) < 2: - return None - if command[0] != 'pam': - return None - if command[1] not in ('tunnel', 't'): - return None - if len(command) >= 3 and command[2] in ('edit', 'e'): + if not command_tokens: return None - return ( - 'pam tunnel commands other than edit are not available in Service Mode' - ) + + tokens_l = [t.lower() for t in command_tokens] + if tokens_l[0] == 'pam' and len(tokens_l) >= 2 and tokens_l[1] in ('tunnel', 't'): + # Default verb matches PAMTunnelCommand.default_verb ('list') + verb = tokens_l[2] if len(tokens_l) >= 3 else 'list' + verb = Verifycommand._PAM_TUNNEL_ALIASES.get(verb, verb) + if verb not in Verifycommand._PAM_TUNNEL_ALLOWED: + return ( + 'pam tunnel commands other than edit are not available in Service Mode' + ) + return None @staticmethod def validate_append_command(command): diff --git a/unit-tests/service/test_auth_security.py b/unit-tests/service/test_auth_security.py index 93a113e0e..f8cbda0a5 100644 --- a/unit-tests/service/test_auth_security.py +++ b/unit-tests/service/test_auth_security.py @@ -100,8 +100,8 @@ def test_policy_check_denied_command(self, mock_read_config): self.assertIn('Not permitted', response[0]['error']) @mock.patch.object(ConfigReader, 'read_config') - def test_policy_check_blocks_pam_tunnel_start(self, mock_read_config): - """pam tunnel start (incl. --run) is blocked in Service Mode""" + def test_policy_check_allows_pam_tunnel_start_at_http_layer(self, mock_read_config): + """Tunnel ban is enforced in CommandExecutor, not policy_check split(' ').""" mock_read_config.return_value = "pam" with self.app.test_request_context( @@ -110,12 +110,12 @@ def test_policy_check_blocks_pam_tunnel_start(self, mock_read_config): headers={'api-key': 'test_key'}, ): response = policy_check(lambda *args, **kwargs: ({'status': 'success'}, 200))() - self.assertEqual(response[1], 403) - self.assertIn('pam tunnel', response[0]['error']) + self.assertEqual(response[1], 200) + self.assertEqual(response[0]['status'], 'success') @mock.patch.object(ConfigReader, 'read_config') def test_policy_check_allows_pam_tunnel_edit(self, mock_read_config): - """pam tunnel edit remains allowed in Service Mode""" + """pam tunnel edit remains allowed through policy_check""" mock_read_config.return_value = "pam" with self.app.test_request_context( @@ -127,15 +127,18 @@ def test_policy_check_allows_pam_tunnel_edit(self, mock_read_config): self.assertEqual(response[1], 200) self.assertEqual(response[0]['status'], 'success') - def test_validate_pam_tunnel_command(self): - """Unit-level checks for pam tunnel Service Mode allowlist""" + def test_validate_service_mode_restrictions_pam_tunnel(self): + """Unit-level checks for pam tunnel Service Mode allowlist (post-shlex tokens)""" ban = 'not available in Service Mode' - self.assertIsNone(Verifycommand.validate_pam_tunnel_command(['pam', 'tunnel', 'edit', 'uid'])) - self.assertIsNone(Verifycommand.validate_pam_tunnel_command(['pam', 't', 'e', 'uid'])) - self.assertIsNone(Verifycommand.validate_pam_tunnel_command(['pam', 'rotation', 'list'])) - self.assertIn(ban, Verifycommand.validate_pam_tunnel_command( - ['pam', 'tunnel', 'start', 'uid', '--run', 'id'])) - self.assertIn(ban, Verifycommand.validate_pam_tunnel_command(['pam', 'tunnel', 'list'])) - self.assertIn(ban, Verifycommand.validate_pam_tunnel_command(['pam', 'tunnel', 'stop', 'uid'])) - self.assertIn(ban, Verifycommand.validate_pam_tunnel_command(['pam', 'tunnel', 'diagnose'])) - self.assertIn(ban, Verifycommand.validate_pam_tunnel_command(['pam', 't', 's', 'uid'])) + check = Verifycommand.validate_service_mode_restrictions + self.assertIsNone(check(['pam', 'tunnel', 'edit', 'uid'])) + self.assertIsNone(check(['pam', 't', 'e', 'uid'])) + self.assertIsNone(check(['pam', 'rotation', 'list'])) + self.assertIn(ban, check(['pam', 'tunnel', 'start', 'uid', '--run', 'id'])) + self.assertIn(ban, check(['pam', 'tunnel', 'list'])) + self.assertIn(ban, check(['pam', 'tunnel', 'stop', 'uid'])) + self.assertIn(ban, check(['pam', 'tunnel', 'diagnose'])) + self.assertIn(ban, check(['pam', 't', 's', 'uid'])) + # Case-insensitive (executor-normalized tokens) + self.assertIn(ban, check(['pam', 'TUNNEL', 'START', 'uid'])) + self.assertIsNone(check(['pam', 'TUNNEL', 'EDIT', 'uid'])) diff --git a/unit-tests/service/test_service_mode_pam_tunnel.py b/unit-tests/service/test_service_mode_pam_tunnel.py new file mode 100644 index 000000000..bc382c7fe --- /dev/null +++ b/unit-tests/service/test_service_mode_pam_tunnel.py @@ -0,0 +1,57 @@ +from unittest import TestCase +from html import unescape + +import shlex + +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) + except ValueError: + return command.split() + + +class TestServiceModeCommandPolicy(TestCase): + """Service Mode bans must use the same tokenize path as CommandExecutor.""" + + def test_pam_tunnel_blocked_except_edit(self): + for cmd in ( + 'pam tunnel start uid --run id', + 'pam tunnel list', + 'pam tunnel stop uid', + 'pam tunnel diagnose', + 'pam tunnel', # defaults to list + 'pam t s uid', + 'pam t l', + 'pam t x uid', + 'pam t d', + ): + with self.subTest(cmd=cmd): + err = Verifycommand.validate_service_mode_restrictions(_tokens(cmd)) + self.assertIsNotNone(err) + self.assertIn('pam tunnel', err) + + self.assertIsNone( + Verifycommand.validate_service_mode_restrictions( + _tokens('pam tunnel edit SOME_UID --enable-tunneling') + ) + ) + self.assertIsNone( + Verifycommand.validate_service_mode_restrictions(_tokens('pam t e SOME_UID')) + ) + + def test_pam_tunnel_bypass_vectors_normalized(self): + """Double space / case / HTML entities must still block after executor-style parse.""" + for raw in ( + 'pam tunnel start uid --run id', + 'pam TUNNEL start uid --run id', + 'pam tunnel start uid --run id', + 'pam t START uid --run id', + ): + with self.subTest(raw=raw): + err = Verifycommand.validate_service_mode_restrictions(_tokens(raw)) + self.assertIsNotNone(err, msg=f'should block: {raw!r} -> {_tokens(raw)}') From aadeb5723c9f90b80a7f16842e61f241e2fc07b6 Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Fri, 14 Aug 2026 11:44:13 +0530 Subject: [PATCH 3/4] Place all service mode restrictions at one place --- keepercommander/commands/record_edit.py | 14 --- .../service/util/verified_command.py | 89 +++++++++++++++---- unit-tests/service/test_auth_security.py | 14 +++ .../service/test_service_mode_pam_tunnel.py | 29 ++++++ unit-tests/test_command_record.py | 17 ---- 5 files changed, 117 insertions(+), 46 deletions(-) diff --git a/keepercommander/commands/record_edit.py b/keepercommander/commands/record_edit.py index 470ae0383..c0750fdb1 100644 --- a/keepercommander/commands/record_edit.py +++ b/keepercommander/commands/record_edit.py @@ -773,13 +773,6 @@ def assign_typed_fields(self, record, fields): def upload_attachments(self, params, record, files, stop_on_error): # type: (KeeperParams, Union[vault.PasswordRecord, vault.TypedRecord], List[ParsedFieldValue], bool) -> None - if files and getattr(params, 'service_mode', False): - # Remote Service Mode callers must not open arbitrary host paths - # (e.g. file=@~/.keeper/config.json). Local CLI is unchanged. - raise CommandError( - '', - 'File attachments by local path are not permitted through Service Mode') - tasks = [] for file_attachment in files: if file_attachment.value.startswith('@'): @@ -1510,13 +1503,6 @@ def get_parser(self): return download_parser def execute(self, params, **kwargs): - if getattr(params, 'service_mode', False): - # Downloads write to the Commander host disk; remote API callers - # do not receive the file bytes in the HTTP response. - raise CommandError( - 'download-attachment', - 'Downloading attachments to the local filesystem is not permitted through Service Mode') - records = kwargs.get('records') if not records: self.get_parser().print_help() diff --git a/keepercommander/service/util/verified_command.py b/keepercommander/service/util/verified_command.py index 87c8dec29..00dde3220 100644 --- a/keepercommander/service/util/verified_command.py +++ b/keepercommander/service/util/verified_command.py @@ -12,26 +12,85 @@ class Verifycommand: @staticmethod def validate_service_mode_restrictions(command_tokens): - """ - Authoritative Service Mode command policy. - - Call on tokens from CommandExecutor (html.unescape + shlex.split), - not on raw HTTP split(" "). Returns error string if blocked, else None. - """ + """Run Service Mode bans on executor tokens (shlex); error string or None.""" if not command_tokens: return None - tokens_l = [t.lower() for t in command_tokens] - if tokens_l[0] == 'pam' and len(tokens_l) >= 2 and tokens_l[1] in ('tunnel', 't'): - # Default verb matches PAMTunnelCommand.default_verb ('list') - verb = tokens_l[2] if len(tokens_l) >= 3 else 'list' - verb = Verifycommand._PAM_TUNNEL_ALIASES.get(verb, verb) - if verb not in Verifycommand._PAM_TUNNEL_ALLOWED: - return ( - 'pam tunnel commands other than edit are not available in Service Mode' - ) + for validator in ( + Verifycommand.validate_service_mode_pam_tunnel_command, + Verifycommand.validate_service_mode_download_attachment_command, + Verifycommand.validate_service_mode_upload_attachment_command, + Verifycommand.validate_service_mode_record_file_attachment_command, + ): + error = validator(command_tokens) + if error: + return error return None + @staticmethod + def validate_service_mode_pam_tunnel_command(command_tokens): + """Allow only pam tunnel edit in Service Mode; error string or None.""" + if not command_tokens or len(command_tokens) < 2: + return None + + tokens_l = [t.lower() for t in command_tokens] + if tokens_l[0] != 'pam' or tokens_l[1] not in ('tunnel', 't'): + return None + + # Default verb matches PAMTunnelCommand.default_verb ('list') + verb = tokens_l[2] if len(tokens_l) >= 3 else 'list' + verb = Verifycommand._PAM_TUNNEL_ALIASES.get(verb, verb) + if verb in Verifycommand._PAM_TUNNEL_ALLOWED: + return None + return ( + 'pam tunnel commands other than edit are not available in Service Mode' + ) + + @staticmethod + def validate_service_mode_download_attachment_command(command_tokens): + """Block download-attachment in Service Mode; error string or None.""" + if not command_tokens: + return None + if command_tokens[0].lower() not in ('download-attachment', 'da'): + return None + return ( + 'Downloading attachments to the local filesystem is not permitted ' + 'through Service Mode' + ) + + @staticmethod + def validate_service_mode_upload_attachment_command(command_tokens): + """Block upload-attachment in Service Mode; error string or None.""" + if not command_tokens: + return None + if command_tokens[0].lower() not in ('upload-attachment', 'ua'): + return None + return ( + 'Uploading attachments from the local filesystem is not permitted ' + 'through Service Mode' + ) + + @staticmethod + def validate_service_mode_record_file_attachment_command(command_tokens): + """Block record-add/update file=@ / f.file= in Service Mode; error or None.""" + if not command_tokens: + return None + if command_tokens[0].lower() not in ('record-add', 'record-update'): + return None + if not any(Verifycommand._is_record_file_attachment_arg(tok) for tok in command_tokens): + return None + return ( + 'File attachments by local path are not permitted through Service Mode' + ) + + @staticmethod + def _is_record_file_attachment_arg(token): + """True for file=@path, file.Label=path, or f.file='/path'.""" + if not token or '=' not in token: + return False + left = token.split('=', 1)[0].lower() + return left == 'file' or left.startswith('file.') or left.endswith('.file') + @staticmethod def validate_append_command(command): """ diff --git a/unit-tests/service/test_auth_security.py b/unit-tests/service/test_auth_security.py index f8cbda0a5..6c66ba7c9 100644 --- a/unit-tests/service/test_auth_security.py +++ b/unit-tests/service/test_auth_security.py @@ -142,3 +142,17 @@ def test_validate_service_mode_restrictions_pam_tunnel(self): # Case-insensitive (executor-normalized tokens) self.assertIn(ban, check(['pam', 'TUNNEL', 'START', 'uid'])) self.assertIsNone(check(['pam', 'TUNNEL', 'EDIT', 'uid'])) + + def test_validate_service_mode_restrictions_attachments(self): + check = Verifycommand.validate_service_mode_restrictions + self.assertIn('Download', check(['download-attachment', 'uid'])) + self.assertIn('Upload', check(['upload-attachment', '/tmp/x', '--record', 'uid'])) + self.assertIn( + 'File attachments', + check(['record-add', '--title', 't', '-rt', 'login', 'file=@/tmp/x']), + ) + self.assertIn( + 'File attachments', + check(['record-update', '--force', '--record', 'uid', "f.file=/tmp/service_config.json"]), + ) + self.assertIsNone(check(['record-add', '--title', 't', '-rt', 'login', 'login=user'])) diff --git a/unit-tests/service/test_service_mode_pam_tunnel.py b/unit-tests/service/test_service_mode_pam_tunnel.py index bc382c7fe..a8d39f8f4 100644 --- a/unit-tests/service/test_service_mode_pam_tunnel.py +++ b/unit-tests/service/test_service_mode_pam_tunnel.py @@ -55,3 +55,32 @@ def test_pam_tunnel_bypass_vectors_normalized(self): with self.subTest(raw=raw): err = Verifycommand.validate_service_mode_restrictions(_tokens(raw)) self.assertIsNotNone(err, msg=f'should block: {raw!r} -> {_tokens(raw)}') + + def test_attachment_commands_blocked_for_remote_api(self): + check = Verifycommand.validate_service_mode_restrictions + self.assertIsNotNone(check(_tokens('download-attachment SOME_UID'))) + self.assertIsNotNone(check(_tokens('da SOME_UID'))) + self.assertIsNotNone(check(_tokens('upload-attachment /tmp/x --record SOME_UID'))) + self.assertIsNotNone(check(_tokens('ua /tmp/x --record SOME_UID'))) + self.assertIsNotNone( + check(_tokens('record-add --title t -rt login "file=@/tmp/x"')) + ) + # Same shape as record_handler._update_or_add_record_attachment + self.assertIsNotNone( + check(_tokens( + "record-update --force --record UID --title t " + "--record-type=login f.file='/tmp/service_config.json'" + )) + ) + self.assertIsNotNone( + check(_tokens('record-add --title t -rt login file.Label=/etc/passwd')) + ) + self.assertIsNone(check(_tokens('record-add --title t -rt login login=user'))) + + def test_is_record_file_attachment_arg(self): + is_file = Verifycommand._is_record_file_attachment_arg + self.assertTrue(is_file('file=@/tmp/x')) + self.assertTrue(is_file("f.file='/path/service_config.json'")) + self.assertTrue(is_file('file.MyDoc=/tmp/x')) + self.assertFalse(is_file('login=user')) + self.assertFalse(is_file('--title')) diff --git a/unit-tests/test_command_record.py b/unit-tests/test_command_record.py index 57982d224..d21e3927f 100644 --- a/unit-tests/test_command_record.py +++ b/unit-tests/test_command_record.py @@ -436,23 +436,6 @@ def test_append_notes_command(self): with self.assertRaises(CommandError): cmd.execute(params, notes='notes', record='invalid') - def test_upload_attachments_blocked_in_service_mode(self): - params = get_synced_params() - params.service_mode = True - mixin = record_edit.RecordEditMixin() - files = [record_edit.ParsedFieldValue('', 'file', '', '@/tmp/any-file.txt')] - with self.assertRaises(CommandError) as ctx: - mixin.upload_attachments(params, vault.TypedRecord(), files, True) - self.assertIn('Service Mode', str(ctx.exception)) - - def test_download_attachment_blocked_in_service_mode(self): - params = get_synced_params() - params.service_mode = True - cmd = record_edit.RecordDownloadAttachmentCommand() - with self.assertRaises(CommandError) as ctx: - cmd.execute(params, records=['any-uid']) - self.assertIn('Service Mode', str(ctx.exception)) - def test_download_attachment_command(self): params = get_synced_params() cmd = record_edit.RecordDownloadAttachmentCommand() From b84f5b26c53c7b7c3f05f77618676f1682b51541 Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Fri, 14 Aug 2026 12:50:32 +0530 Subject: [PATCH 4/4] Add ra ru in restricted list --- .../service/util/verified_command.py | 16 +++++++++++----- unit-tests/service/test_auth_security.py | 13 +++++++++++++ .../service/test_service_mode_pam_tunnel.py | 19 ++++++++++++++++++- 3 files changed, 42 insertions(+), 6 deletions(-) diff --git a/keepercommander/service/util/verified_command.py b/keepercommander/service/util/verified_command.py index 00dde3220..1e240ad05 100644 --- a/keepercommander/service/util/verified_command.py +++ b/keepercommander/service/util/verified_command.py @@ -9,6 +9,8 @@ class Verifycommand: 'd': 'diagnose', } _PAM_TUNNEL_ALLOWED = frozenset({'edit'}) + # Aliases from record.py — CommandExecutor checks tokens before cli expands them. + _RECORD_EDIT_COMMANDS = frozenset({'record-add', 'ra', 'record-update', 'ru'}) @staticmethod def validate_service_mode_restrictions(command_tokens): @@ -72,10 +74,11 @@ def validate_service_mode_upload_attachment_command(command_tokens): @staticmethod def validate_service_mode_record_file_attachment_command(command_tokens): - """Block record-add/update file=@ / f.file= in Service Mode; error or None.""" + """Block record-add/update (and ra/ru) file fields in Service Mode; error or None.""" if not command_tokens: return None - if command_tokens[0].lower() not in ('record-add', 'record-update'): + # Tokens are checked before cli alias expansion, so list ra/ru explicitly. + if command_tokens[0].lower() not in Verifycommand._RECORD_EDIT_COMMANDS: return None if not any(Verifycommand._is_record_file_attachment_arg(tok) for tok in command_tokens): return None @@ -85,11 +88,14 @@ def validate_service_mode_record_file_attachment_command(command_tokens): @staticmethod def _is_record_file_attachment_arg(token): - """True for file=@path, file.Label=path, or f.file='/path'.""" + """True when parse_field would treat this token as a file attachment field.""" if not token or '=' not in token: return False - left = token.split('=', 1)[0].lower() - return left == 'file' or left.startswith('file.') or left.endswith('.file') + # Mirror RecordEditMixin.parse_field field-name normalization. + name = token.split('=', 1)[0].lower() + if name.startswith('f.') or name.startswith('c.'): + name = name[2:] + return name.split('.', 1)[0] == 'file' @staticmethod def validate_append_command(command): diff --git a/unit-tests/service/test_auth_security.py b/unit-tests/service/test_auth_security.py index 6c66ba7c9..68913db28 100644 --- a/unit-tests/service/test_auth_security.py +++ b/unit-tests/service/test_auth_security.py @@ -155,4 +155,17 @@ def test_validate_service_mode_restrictions_attachments(self): 'File attachments', check(['record-update', '--force', '--record', 'uid', "f.file=/tmp/service_config.json"]), ) + self.assertIn( + 'File attachments', + check(['ra', '--title', 't', '-rt', 'login', 'file=@/etc/passwd']), + ) + self.assertIn( + 'File attachments', + check(['ru', '--force', '--record', 'uid', 'f.file.doc=@/etc/passwd']), + ) + self.assertIn( + 'File attachments', + check(['record-add', '--title', 't', '-rt', 'login', 'c.file.doc=@/etc/passwd']), + ) self.assertIsNone(check(['record-add', '--title', 't', '-rt', 'login', 'login=user'])) + self.assertIsNone(check(['record-add', '--title', 't', '-rt', 'login', 'my.file=x'])) diff --git a/unit-tests/service/test_service_mode_pam_tunnel.py b/unit-tests/service/test_service_mode_pam_tunnel.py index a8d39f8f4..257ca41dd 100644 --- a/unit-tests/service/test_service_mode_pam_tunnel.py +++ b/unit-tests/service/test_service_mode_pam_tunnel.py @@ -65,7 +65,6 @@ def test_attachment_commands_blocked_for_remote_api(self): self.assertIsNotNone( check(_tokens('record-add --title t -rt login "file=@/tmp/x"')) ) - # Same shape as record_handler._update_or_add_record_attachment self.assertIsNotNone( check(_tokens( "record-update --force --record UID --title t " @@ -75,6 +74,20 @@ def test_attachment_commands_blocked_for_remote_api(self): self.assertIsNotNone( check(_tokens('record-add --title t -rt login file.Label=/etc/passwd')) ) + # Aliases checked before cli expansion + self.assertIsNotNone( + check(_tokens('ra --title t -rt login file=@/etc/passwd')) + ) + self.assertIsNotNone( + check(_tokens('ru --force --record UID f.file=/etc/shadow')) + ) + # Labeled f./c. file fields (parse_field type == file) + self.assertIsNotNone( + check(_tokens('record-add --title t -rt login f.file.doc=@/etc/passwd')) + ) + self.assertIsNotNone( + check(_tokens('record-add --title t -rt login c.file.doc=@/etc/passwd')) + ) self.assertIsNone(check(_tokens('record-add --title t -rt login login=user'))) def test_is_record_file_attachment_arg(self): @@ -82,5 +95,9 @@ def test_is_record_file_attachment_arg(self): self.assertTrue(is_file('file=@/tmp/x')) self.assertTrue(is_file("f.file='/path/service_config.json'")) self.assertTrue(is_file('file.MyDoc=/tmp/x')) + self.assertTrue(is_file('f.file.doc=@/etc/passwd')) + self.assertTrue(is_file('c.file.doc=@/etc/passwd')) self.assertFalse(is_file('login=user')) self.assertFalse(is_file('--title')) + self.assertFalse(is_file('profile=x')) + self.assertFalse(is_file('my.file=x')) # not a file-type field after parse_field