diff --git a/keepercommander/service/decorators/auth.py b/keepercommander/service/decorators/auth.py index 3ced7b0d4..2d21da195 100644 --- a/keepercommander/service/decorators/auth.py +++ b/keepercommander/service/decorators/auth.py @@ -128,6 +128,6 @@ def wrapper(*args, **kwargs): 'status': 'error', 'error': transform_folder_error }, 400 - + 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 2f62f5f3a..1e240ad05 100644 --- a/keepercommander/service/util/verified_command.py +++ b/keepercommander/service/util/verified_command.py @@ -1,4 +1,102 @@ 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'}) + # 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): + """Run Service Mode bans on executor tokens (shlex); error string or None.""" + if not command_tokens: + return None + + 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 (and ra/ru) file fields in Service Mode; error or None.""" + if not command_tokens: + return None + # 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 + return ( + 'File attachments by local path are not permitted through Service Mode' + ) + + @staticmethod + def _is_record_file_attachment_arg(token): + """True when parse_field would treat this token as a file attachment field.""" + if not token or '=' not in token: + return False + # 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 bbd94ba4b..68913db28 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,75 @@ 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_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( + '/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], 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 through policy_check""" + 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_service_mode_restrictions_pam_tunnel(self): + """Unit-level checks for pam tunnel Service Mode allowlist (post-shlex tokens)""" + ban = 'not available in Service Mode' + 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'])) + + 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.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 new file mode 100644 index 000000000..257ca41dd --- /dev/null +++ b/unit-tests/service/test_service_mode_pam_tunnel.py @@ -0,0 +1,103 @@ +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)}') + + 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"')) + ) + 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')) + ) + # 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): + 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.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