Restrict Service Mode host file I/O and pam tunnel commands

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
+    return wrapper

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
             )

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):
         """

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'])
+            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']))

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