From 7f50b0caf3a062f728433c5c49dccec6ce87cdc8 Mon Sep 17 00:00:00 2001
From: amangalampalli-ks <amangalampalli@keepersecurity.com>
Date: Mon, 21 Sep 2026 11:19:35 +0530
Subject: [PATCH 1/7] Protect Integration config records from Service Mode API
 access

---
 keepercommander/service/README.md             | 10 +++
 .../service/commands/terraform_app_setup.py   |  2 +-
 .../decorators/min_commander_version.py       |  4 +-
 keepercommander/service/util/command_util.py  | 10 ++-
 .../service/util/protected_records.py         | 58 +++++++++++---
 unit-tests/service/test_command.py            | 79 ++++++++++++++++++-
 unit-tests/service/test_protected_records.py  | 75 ++++++++++++++++++
 unit-tests/service/test_runtime_policy.py     |  5 +-
 .../service/test_terraform_app_setup.py       |  3 +-
 9 files changed, 229 insertions(+), 17 deletions(-)

diff --git a/keepercommander/service/README.md b/keepercommander/service/README.md
index 18b8e4e28..a04b01174 100644
--- a/keepercommander/service/README.md
+++ b/keepercommander/service/README.md
@@ -502,6 +502,16 @@ This automates the complete setup for Slack App integration:
 
 The command generates a complete `docker-compose.yml` with both Commander service and Slack App service configured.
 
+**Generated compose environment for the Slack service:**
+
+| Env var | Value |
+|---------|-------|
+| `KSM_CONFIG` | Base64 KSM config |
+| `COMMANDER_RECORD` | Commander Docker config record UID |
+| `SLACK_RECORD` | Slack config record UID |
+
+Image name used in compose: `keeper/slack-app:latest`.
+
 ### Google Chat App Integration Setup
 
 For integrating Commander Service Mode with Google Chat, use the `gchat-app-setup` command:
diff --git a/keepercommander/service/commands/terraform_app_setup.py b/keepercommander/service/commands/terraform_app_setup.py
index 85d4b6be6..811f3e5da 100644
--- a/keepercommander/service/commands/terraform_app_setup.py
+++ b/keepercommander/service/commands/terraform_app_setup.py
@@ -136,7 +136,7 @@ def generate_docker_compose_yaml(self, setup_result: SetupResult, config: Docker
             asdict(config),
             commander_service_name=TerraformSetupConstants.COMMANDER_SERVICE_NAME,
             commander_container_name=TerraformSetupConstants.COMMANDER_CONTAINER_NAME,
-            commander_environment={TERRAFORM_DOCKER_ENV: '1'},
+            commander_environment={TERRAFORM_DOCKER_ENV: setup_result.record_uid},
         )
         return builder.build()
 
diff --git a/keepercommander/service/decorators/min_commander_version.py b/keepercommander/service/decorators/min_commander_version.py
index 4d484caf3..1c4eca6c0 100644
--- a/keepercommander/service/decorators/min_commander_version.py
+++ b/keepercommander/service/decorators/min_commander_version.py
@@ -23,8 +23,8 @@
 
 # Hyphenated only: Werkzeug/WSGI silently drops headers that contain underscores.
 MIN_COMMANDER_VERSION_HEADER = 'Min-Commander-Version'
-# Set on terraform-app-setup compose; not a secret — instance identity only.
-TERRAFORM_DOCKER_ENV = 'KEEPER_TERRAFORM'
+# Set on terraform-app-setup compose to the Terraform config record's UID.
+TERRAFORM_DOCKER_ENV = 'TERRAFORM_RECORD'
 
 
 def _parse_version(version_str: str) -> Optional[Version]:
diff --git a/keepercommander/service/util/command_util.py b/keepercommander/service/util/command_util.py
index 72180e83d..a9917547a 100644
--- a/keepercommander/service/util/command_util.py
+++ b/keepercommander/service/util/command_util.py
@@ -25,7 +25,7 @@
     is_throttle_error,
     throttle_error_response,
 )
-from .protected_records import get_protected_record_uids, hide_from_record_cache
+from .protected_records import get_protected_record_uids, hide_from_record_cache, resolve_sync_down_exempt_uid
 from .verified_command import Verifycommand
 from ..core.globals import get_current_params
 from ..decorators.logging import logger, debug_decorator, sanitize_debug_data, sanitize_command_fields
@@ -195,6 +195,14 @@ def blocked(error):
             # Checked for every command (not a curated list) so no current or future
             # command can be missed as a way to reference these records.
             protected_uids = get_protected_record_uids(params)
+
+            # {slack,teams,gchat}-app-setup --sync-down needs its own config record reachable.
+            sync_down_exempt_uid = resolve_sync_down_exempt_uid(command_tokens)
+            if sync_down_exempt_uid is not None:
+                protected_uids = {
+                    uid: title for uid, title in protected_uids.items() if uid != sync_down_exempt_uid
+                }
+
             protected_command_error = Verifycommand.validate_service_mode_protected_record_command(
                 command_tokens, protected_uids
             )
diff --git a/keepercommander/service/util/protected_records.py b/keepercommander/service/util/protected_records.py
index dc5ff0497..30b6b5eb0 100644
--- a/keepercommander/service/util/protected_records.py
+++ b/keepercommander/service/util/protected_records.py
@@ -16,10 +16,16 @@
 import contextlib
 import os
 from collections import UserDict
-from typing import Dict, FrozenSet, Iterable, Tuple
+from typing import Dict, FrozenSet, Iterable, Optional, Tuple
 
-# Docker mode passes the config record's UID here regardless of what title --record-name gave it at setup time.
-_DOCKER_RECORD_UID_ENV = 'COMMANDER_RECORD'
+# Each integration's own setup pins its config record's UID here, regardless of its title. Extend when a new integration gets an always-hidden record.
+_PINNED_RECORD_UID_ENVS: Dict[str, str] = {
+    'COMMANDER_RECORD': '<Docker config record>',
+    'TERRAFORM_RECORD': '<Terraform config record>',
+    'SLACK_RECORD': '<Slack config record>',
+    'TEAMS_RECORD': '<Teams config record>',
+    'GCHAT_RECORD': '<GChat config record>',
+}
 
 # uid-keyed caches resolve_single_record/load_pam_record fall back to when a UID isn't in record_cache.
 _GUARDED_CACHE_ATTRS = ('record_cache', 'nested_share_records', 'nested_share_record_data')
@@ -28,8 +34,18 @@
 def _protected_titles() -> Tuple[str, ...]:
     """The literal titles of Service Mode's own config records; imported lazily to avoid a circular import through verified_command."""
     from ..config.file_handler import SERVICE_CONFIG_RECORD_TITLES
-    from ..docker.models import DockerSetupConstants
-    return (*SERVICE_CONFIG_RECORD_TITLES, DockerSetupConstants.DEFAULT_RECORD_NAME)
+    from ..commands.terraform_app_setup import TerraformSetupConstants
+    from ..commands.integrations.slack_app_setup import SlackAppSetupCommand
+    from ..commands.integrations.teams_app_setup import TeamsAppSetupCommand
+    from ..docker.models import DockerSetupConstants, GChatConstants
+    return (
+        *SERVICE_CONFIG_RECORD_TITLES,
+        DockerSetupConstants.DEFAULT_RECORD_NAME,
+        TerraformSetupConstants.DEFAULT_RECORD_NAME,
+        GChatConstants.DEFAULT_RECORD_NAME,
+        SlackAppSetupCommand().get_default_record_name(),
+        TeamsAppSetupCommand().get_default_record_name(),
+    )
 
 
 def get_protected_record_title_set() -> FrozenSet[str]:
@@ -38,12 +54,13 @@ def get_protected_record_title_set() -> FrozenSet[str]:
 
 
 def get_protected_record_uids(params) -> Dict[str, str]:
-    """Resolve current UIDs of Service Mode's own config records ({uid: title}), matching by title plus the Docker record's UID from COMMANDER_RECORD; not cached, since a stale result on this security check is worse than the cost of a full-vault scan."""
+    """Resolve current UIDs of Service Mode's own config records ({uid: title}), matching by title plus each integration's pinned UID env var (_PINNED_RECORD_UID_ENVS); not cached, since a stale result on this security check is worse than the cost of a full-vault scan."""
     found: Dict[str, str] = {}
 
-    docker_uid = (os.environ.get(_DOCKER_RECORD_UID_ENV) or '').strip()
-    if docker_uid:
-        found[docker_uid] = '<Docker config record>'
+    for env_name, label in _PINNED_RECORD_UID_ENVS.items():
+        uid = (os.environ.get(env_name) or '').strip()
+        if uid:
+            found[uid] = label
 
     if params is None or not isinstance(getattr(params, 'record_cache', None), dict) or not params.record_cache:
         return found
@@ -63,6 +80,29 @@ def get_protected_record_uids(params) -> Dict[str, str]:
     return found
 
 
+# {command: pinned-UID env var} for the live, intentionally-exposed feature that must keep working
+# ({slack,gchat}-app-setup --sync-down);
+_SYNC_DOWN_EXEMPT_ENV_BY_COMMAND: Dict[str, str] = {
+    'slack-app-setup': 'SLACK_RECORD',
+    'gchat-app-setup': 'GCHAT_RECORD',
+}
+
+
+def resolve_sync_down_exempt_uid(command_tokens) -> Optional[str]:
+    """For '{slack,teams,gchat}-app-setup ... --sync-down ...', the one UID this dispatch may bypass Layers A/B for -- always this integration's own pinned-env UID, never derived from what the admin passes (e.g. -r/--integration-record), so a different integration's protected record can never be reached this way."""
+    if not command_tokens:
+        return None
+
+    env_name = _SYNC_DOWN_EXEMPT_ENV_BY_COMMAND.get(command_tokens[0].lower())
+    if not env_name:
+        return None
+
+    if not any(tok == '--sync-down' for tok in command_tokens[1:]):
+        return None
+
+    return (os.environ.get(env_name) or '').strip() or None
+
+
 class _GuardedRecordCache(UserDict):
     """A uid-keyed cache view that can never hold the given protected UIDs; UserDict (not dict) so every mutation reliably routes through __setitem__, even C-level ones like setdefault/|=."""
 
diff --git a/unit-tests/service/test_command.py b/unit-tests/service/test_command.py
index fdd3415f5..6f9c6ac6e 100644
--- a/unit-tests/service/test_command.py
+++ b/unit-tests/service/test_command.py
@@ -271,4 +271,81 @@ def fake_handle_command(p, command):
         self.assertNotIn(PROTECTED_UID, seen_during_handle_command['keys'])
         self.assertIn(NORMAL_UID, seen_during_handle_command['keys'])
         # Restored after the whole guarded block exits, same as the non-SailPoint case.
-        self.assertIn(PROTECTED_UID, params.record_cache)
\ No newline at end of file
+        self.assertIn(PROTECTED_UID, params.record_cache)
+
+
+class TestSyncDownExemptionCommandExecution(TestCase):
+    """slack-app-setup --sync-down must keep working for its OWN config record while every
+    other protected record (including a different integration's) stays blocked."""
+
+    SLACK_UID = 'SLACK_CONFIG_UID'
+    GCHAT_UID = 'GCHAT_CONFIG_UID'
+
+    def _params(self):
+        p = params_module.KeeperParams()
+        p.service_mode = False
+        p.record_cache = {
+            self.SLACK_UID: _record_cache_entry(self.SLACK_UID, 'Commander Service Mode Slack App Config'),
+            self.GCHAT_UID: _record_cache_entry(self.GCHAT_UID, 'Commander Service Mode Google Chat App Config'),
+            NORMAL_UID: _record_cache_entry(NORMAL_UID, 'My Normal Record'),
+        }
+        return p
+
+    def _run(self, command, params, capture_side_effect=None):
+        env = {'SLACK_RECORD': self.SLACK_UID, 'GCHAT_RECORD': self.GCHAT_UID}
+        with mock.patch.dict('os.environ', env), mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch.object(
+            CommandExecutor, 'capture_output_and_logs',
+            side_effect=capture_side_effect, return_value=('ok', 'ok', '') if capture_side_effect is None else None,
+        ) as mock_capture:
+            response, status_code = CommandExecutor.execute(command)
+        return response, status_code, mock_capture
+
+    def test_get_slack_record_is_blocked(self):
+        params = self._params()
+        response, status_code, mock_capture = self._run(f'get {self.SLACK_UID}', params)
+        self.assertEqual(status_code, 403)
+        mock_capture.assert_not_called()
+
+    def test_slack_sync_down_default_flow_is_not_blocked_and_sees_its_own_record(self):
+        params = self._params()
+        seen = {}
+
+        def fake_capture(p, command):
+            seen['keys'] = set(p.record_cache.keys())
+            return 'ok', 'ok', ''
+
+        response, status_code, mock_capture = self._run(
+            'slack-app-setup --sync-down', params, capture_side_effect=fake_capture
+        )
+        self.assertEqual(status_code, 200)
+        mock_capture.assert_called_once()
+        self.assertIn(self.SLACK_UID, seen['keys'])         # not hidden for this one dispatch
+        self.assertIn(self.GCHAT_UID, params.record_cache)  # untouched throughout
+
+    def test_slack_sync_down_with_explicit_own_uid_is_not_blocked(self):
+        params = self._params()
+        response, status_code, mock_capture = self._run(
+            f'slack-app-setup --sync-down -r {self.SLACK_UID}', params
+        )
+        self.assertEqual(status_code, 200)
+        mock_capture.assert_called_once()
+
+    def test_slack_sync_down_with_a_different_integrations_uid_is_still_blocked(self):
+        params = self._params()
+        response, status_code, mock_capture = self._run(
+            f'slack-app-setup --sync-down -r {self.GCHAT_UID}', params
+        )
+        self.assertEqual(status_code, 403)
+        mock_capture.assert_not_called()
+
+    def test_slack_setup_without_sync_down_gets_no_exemption(self):
+        """The main (non --sync-down) flow must not get the record cache exemption
+        just because it names the record's own UID."""
+        params = self._params()
+        response, status_code, mock_capture = self._run(
+            f'slack-app-setup --slack-record-name {self.SLACK_UID}', params
+        )
+        self.assertEqual(status_code, 403)
+        mock_capture.assert_not_called()
\ No newline at end of file
diff --git a/unit-tests/service/test_protected_records.py b/unit-tests/service/test_protected_records.py
index 0d25fdb24..36989ff9e 100644
--- a/unit-tests/service/test_protected_records.py
+++ b/unit-tests/service/test_protected_records.py
@@ -7,6 +7,7 @@
     get_protected_record_title_set,
     get_protected_record_uids,
     hide_from_record_cache,
+    resolve_sync_down_exempt_uid,
 )
 
 PROTECTED_TITLE = 'Commander Service Mode Config'
@@ -38,6 +39,13 @@ def test_contains_expected_titles_lowercased(self):
         self.assertIn('commander service mode docker config', titles)
         self.assertIn('commander service mode', titles)
 
+    def test_contains_terraform_slack_teams_gchat_titles(self):
+        titles = get_protected_record_title_set()
+        self.assertIn('commander service mode terraform config', titles)
+        self.assertIn('commander service mode slack app config', titles)
+        self.assertIn('commander service mode teams app config', titles)
+        self.assertIn('commander service mode google chat app config', titles)
+
 
 class TestGetProtectedRecordUids(TestCase):
     def test_returns_only_matching_protected_records(self):
@@ -92,6 +100,32 @@ def test_no_docker_env_var_falls_back_to_title_only(self):
             result = get_protected_record_uids(p)
         self.assertEqual(set(result.keys()), {'UID_CONFIG'})
 
+    def test_terraform_record_protected_by_uid_even_with_custom_title(self):
+        with mock.patch.dict(os.environ, {'TERRAFORM_RECORD': 'TF_CUSTOM_UID'}):
+            p = _params_with_records({'TF_CUSTOM_UID': 'My Totally Custom Terraform Title'})
+            self.assertIn('TF_CUSTOM_UID', get_protected_record_uids(p))
+
+    def test_slack_record_protected_by_uid_even_with_custom_title(self):
+        with mock.patch.dict(os.environ, {'SLACK_RECORD': 'SLACK_CUSTOM_UID'}):
+            p = _params_with_records({'SLACK_CUSTOM_UID': 'My Totally Custom Slack Title'})
+            self.assertIn('SLACK_CUSTOM_UID', get_protected_record_uids(p))
+
+    def test_teams_record_protected_by_uid_even_with_custom_title(self):
+        with mock.patch.dict(os.environ, {'TEAMS_RECORD': 'TEAMS_CUSTOM_UID'}):
+            p = _params_with_records({'TEAMS_CUSTOM_UID': 'My Totally Custom Teams Title'})
+            self.assertIn('TEAMS_CUSTOM_UID', get_protected_record_uids(p))
+
+    def test_gchat_record_protected_by_uid_even_with_custom_title(self):
+        with mock.patch.dict(os.environ, {'GCHAT_RECORD': 'GCHAT_CUSTOM_UID'}):
+            p = _params_with_records({'GCHAT_CUSTOM_UID': 'My Totally Custom GChat Title'})
+            self.assertIn('GCHAT_CUSTOM_UID', get_protected_record_uids(p))
+
+    def test_no_pinned_env_vars_falls_back_to_title_only(self):
+        with mock.patch.dict(os.environ, {}, clear=True):
+            p = _params_with_records({'UID_CONFIG': PROTECTED_TITLE})
+            result = get_protected_record_uids(p)
+        self.assertEqual(set(result.keys()), {'UID_CONFIG'})
+
     def test_malformed_record_entry_is_skipped_not_raised(self):
         """A record missing an expected key must not break the scan for every other record."""
         p = _params_with_records({'UID_CONFIG': PROTECTED_TITLE})
@@ -235,3 +269,44 @@ def test_attribute_replaced_with_plain_dict_preserves_its_contents(self):
 
         self.assertIn('REPLACED', p.record_cache)
         self.assertIn('PROTECTED', p.record_cache)
+
+
+class TestResolveSyncDownExemptUid(TestCase):
+    def test_returns_env_uid_for_slack_sync_down(self):
+        with mock.patch.dict(os.environ, {'SLACK_RECORD': 'SLACK_UID'}, clear=True):
+            self.assertEqual(
+                resolve_sync_down_exempt_uid(['slack-app-setup', '--sync-down']), 'SLACK_UID'
+            )
+
+    def test_returns_env_uid_regardless_of_explicit_dash_r_value(self):
+        """Never derived from the admin's own -r value -- always the env-pinned UID."""
+        with mock.patch.dict(os.environ, {'SLACK_RECORD': 'SLACK_UID'}, clear=True):
+            self.assertEqual(
+                resolve_sync_down_exempt_uid(['slack-app-setup', '--sync-down', '-r', 'SOME_OTHER_UID']),
+                'SLACK_UID',
+            )
+
+    def test_none_without_sync_down_token(self):
+        with mock.patch.dict(os.environ, {'SLACK_RECORD': 'SLACK_UID'}, clear=True):
+            self.assertIsNone(resolve_sync_down_exempt_uid(['slack-app-setup']))
+
+    def test_none_for_unrelated_command(self):
+        with mock.patch.dict(os.environ, {'SLACK_RECORD': 'SLACK_UID'}, clear=True):
+            self.assertIsNone(resolve_sync_down_exempt_uid(['get', '--sync-down']))
+
+    def test_none_when_env_var_unset(self):
+        with mock.patch.dict(os.environ, {}, clear=True):
+            self.assertIsNone(resolve_sync_down_exempt_uid(['slack-app-setup', '--sync-down']))
+
+    def test_gchat_uses_its_own_env_var(self):
+        with mock.patch.dict(os.environ, {'GCHAT_RECORD': 'G_UID'}, clear=True):
+            self.assertEqual(resolve_sync_down_exempt_uid(['gchat-app-setup', '--sync-down']), 'G_UID')
+
+    def test_teams_has_no_sync_down_exemption_yet(self):
+        """Teams has no approvals profile yet, so --sync-down isn't even a registered flag for it;
+        this must stay None rather than exempting a UID for a flow that can't actually run."""
+        with mock.patch.dict(os.environ, {'TEAMS_RECORD': 'T_UID'}, clear=True):
+            self.assertIsNone(resolve_sync_down_exempt_uid(['teams-app-setup', '--sync-down']))
+
+    def test_empty_tokens_returns_none(self):
+        self.assertIsNone(resolve_sync_down_exempt_uid([]))
diff --git a/unit-tests/service/test_runtime_policy.py b/unit-tests/service/test_runtime_policy.py
index debdb4724..4c77f7879 100644
--- a/unit-tests/service/test_runtime_policy.py
+++ b/unit-tests/service/test_runtime_policy.py
@@ -20,6 +20,7 @@
 from keepercommander.service.commands.integrations.sailpoint_app_setup import SailPointAppSetupCommand
 from keepercommander.service.commands.integrations.slack_app_setup import SlackAppSetupCommand
 from keepercommander.service.commands.terraform_app_setup import TerraformSetupConstants
+from keepercommander.service.decorators.min_commander_version import TERRAFORM_DOCKER_ENV
 from keepercommander.service.util.exceptions import ValidationError
 
 
@@ -67,7 +68,7 @@ def test_gchat_record_env_confines_to_gchat_allowlist(self):
         self.assertEqual(set(args.commands.split(',')), allowed)
 
     def test_terraform_env_confines_to_terraform_allowlist(self):
-        with mock.patch.dict(os.environ, {'KEEPER_TERRAFORM': '1'}, clear=True):
+        with mock.patch.dict(os.environ, {TERRAFORM_DOCKER_ENV: 'tf-record-uid'}, clear=True):
             args = _Args(commands=TerraformSetupConstants.SERVICE_COMMANDS + ',clipboard-copy')
             apply_runtime_command_policy(args)
         self.assertEqual(
@@ -85,7 +86,7 @@ def test_sailpoint_record_env_confines_to_sailpoint_allowlist(self):
 
     def test_multiple_integration_env_vars_raises(self):
         with mock.patch.dict(
-            os.environ, {'SLACK_RECORD': 'uid-1', 'KEEPER_TERRAFORM': '1'}, clear=True
+            os.environ, {'SLACK_RECORD': 'uid-1', TERRAFORM_DOCKER_ENV: 'tf-record-uid'}, clear=True
         ):
             args = _Args(commands='search,malicious-command')
             with self.assertRaises(ValidationError):
diff --git a/unit-tests/service/test_terraform_app_setup.py b/unit-tests/service/test_terraform_app_setup.py
index ec7c25b60..4607a40b9 100644
--- a/unit-tests/service/test_terraform_app_setup.py
+++ b/unit-tests/service/test_terraform_app_setup.py
@@ -58,7 +58,8 @@ def test_compose_uses_terraform_service_and_container_names(self):
         self.assertIn('container_name: keeper-service-terraform', yaml_content)
         self.assertNotIn('container_name: keeper-service\n', yaml_content)
         self.assertIn(f'{TERRAFORM_DOCKER_ENV}:', yaml_content)
-        self.assertRegex(yaml_content, rf"{TERRAFORM_DOCKER_ENV}:\s*'?1'?")
+        # Now carries the record UID (so protected_records.py can pin it), not a bare '1' flag.
+        self.assertRegex(yaml_content, rf"{TERRAFORM_DOCKER_ENV}:\s*'?{setup_result.record_uid}'?")
 
     @mock.patch(
         'keepercommander.service.commands.terraform_app_setup.RuntimeServiceConfig'

From 1cd4485efb0f2e3f30e4194516ac3dd2857e2ea8 Mon Sep 17 00:00:00 2001
From: amangalampalli-ks <amangalampalli@keepersecurity.com>
Date: Mon, 21 Sep 2026 12:51:06 +0530
Subject: [PATCH 2/7] Add terraform backward compatability and test cases

---
 .../commands/integrations/runtime_policy.py   | 10 +--
 .../decorators/min_commander_version.py       | 10 ++-
 .../service/util/protected_records.py         | 33 +++++----
 unit-tests/service/test_command.py            | 71 ++++++++++++++++++-
 .../service/test_min_commander_version.py     | 26 ++++++-
 unit-tests/service/test_protected_records.py  | 56 ++++++++++-----
 unit-tests/service/test_runtime_policy.py     | 12 +++-
 7 files changed, 179 insertions(+), 39 deletions(-)

diff --git a/keepercommander/service/commands/integrations/runtime_policy.py b/keepercommander/service/commands/integrations/runtime_policy.py
index 109e2213b..e773c4828 100644
--- a/keepercommander/service/commands/integrations/runtime_policy.py
+++ b/keepercommander/service/commands/integrations/runtime_policy.py
@@ -50,19 +50,21 @@ def _integration_sanitizers():
     from .sailpoint_app_setup import SailPointAppSetupCommand
     from .slack_app_setup import SlackAppSetupCommand
     from .teams_app_setup import TeamsAppSetupCommand
-    from ...decorators.min_commander_version import TERRAFORM_DOCKER_ENV
+    from ...decorators.min_commander_version import TERRAFORM_DOCKER_ENV, TERRAFORM_DOCKER_ENV_LEGACY
 
     slack = SlackAppSetupCommand()
     teams = TeamsAppSetupCommand()
     gchat = GChatAppSetupCommand()
     sailpoint = SailPointAppSetupCommand()
+    terraform_sanitizer = lambda commands: sanitize_commands(
+        commands, TerraformSetupConstants.SERVICE_COMMANDS_LIST
+    )
 
     return {
         slack.get_record_env_key(): slack.sanitize_service_commands,
         teams.get_record_env_key(): teams.sanitize_service_commands,
         gchat.get_record_env_key(): gchat.sanitize_service_commands,
         sailpoint.get_record_env_key(): sailpoint.sanitize_service_commands,
-        TERRAFORM_DOCKER_ENV: lambda commands: sanitize_commands(
-            commands, TerraformSetupConstants.SERVICE_COMMANDS_LIST
-        ),
+        TERRAFORM_DOCKER_ENV: terraform_sanitizer,
+        TERRAFORM_DOCKER_ENV_LEGACY: terraform_sanitizer,
     }
diff --git a/keepercommander/service/decorators/min_commander_version.py b/keepercommander/service/decorators/min_commander_version.py
index 1c4eca6c0..64f336b78 100644
--- a/keepercommander/service/decorators/min_commander_version.py
+++ b/keepercommander/service/decorators/min_commander_version.py
@@ -25,6 +25,9 @@
 MIN_COMMANDER_VERSION_HEADER = 'Min-Commander-Version'
 # Set on terraform-app-setup compose to the Terraform config record's UID.
 TERRAFORM_DOCKER_ENV = 'TERRAFORM_RECORD'
+# Pre-rename value (was '1', not a UID). Recognized so containers upgraded without re-running
+# terraform-app-setup don't silently lose enforcement; drop after a migration period.
+TERRAFORM_DOCKER_ENV_LEGACY = 'KEEPER_TERRAFORM'
 
 
 def _parse_version(version_str: str) -> Optional[Version]:
@@ -41,8 +44,11 @@ def _parse_version(version_str: str) -> Optional[Version]:
 
 
 def _is_terraform_docker() -> bool:
-    """True when this process was started from terraform-app-setup compose."""
-    return bool((os.environ.get(TERRAFORM_DOCKER_ENV) or '').strip())
+    """True when this process was started from terraform-app-setup compose (new or pre-rename env var)."""
+    return bool(
+        (os.environ.get(TERRAFORM_DOCKER_ENV) or '').strip()
+        or (os.environ.get(TERRAFORM_DOCKER_ENV_LEGACY) or '').strip()
+    )
 
 
 def _read_min_commander_version_header() -> Optional[str]:
diff --git a/keepercommander/service/util/protected_records.py b/keepercommander/service/util/protected_records.py
index 30b6b5eb0..40df4d82c 100644
--- a/keepercommander/service/util/protected_records.py
+++ b/keepercommander/service/util/protected_records.py
@@ -55,18 +55,23 @@ def get_protected_record_title_set() -> FrozenSet[str]:
 
 def get_protected_record_uids(params) -> Dict[str, str]:
     """Resolve current UIDs of Service Mode's own config records ({uid: title}), matching by title plus each integration's pinned UID env var (_PINNED_RECORD_UID_ENVS); not cached, since a stale result on this security check is worse than the cost of a full-vault scan."""
-    found: Dict[str, str] = {}
+    from ..commands.integrations.approvals_setup import is_valid_keeper_uid
+    from ..decorators.logging import logger
 
+    found: Dict[str, str] = {}
     for env_name, label in _PINNED_RECORD_UID_ENVS.items():
         uid = (os.environ.get(env_name) or '').strip()
-        if uid:
-            found[uid] = label
+        if not uid:
+            continue
+        if not is_valid_keeper_uid(uid):
+            logger.warning(f'protected_records: {env_name} is set but not a valid record UID; falling back to title matching for it')
+            continue
+        found[uid] = label
 
     if params is None or not isinstance(getattr(params, 'record_cache', None), dict) or not params.record_cache:
         return found
 
     from ... import vault
-    from ..decorators.logging import logger
 
     protected_titles = get_protected_record_title_set()
     for uid in params.record_cache:
@@ -80,24 +85,26 @@ def get_protected_record_uids(params) -> Dict[str, str]:
     return found
 
 
-# {command: pinned-UID env var} for the live, intentionally-exposed feature that must keep working
-# ({slack,gchat}-app-setup --sync-down);
-_SYNC_DOWN_EXEMPT_ENV_BY_COMMAND: Dict[str, str] = {
-    'slack-app-setup': 'SLACK_RECORD',
-    'gchat-app-setup': 'GCHAT_RECORD',
-}
+def _sync_down_exempt_commands() -> Dict[str, str]:
+    """{command name: pinned-UID env var}, derived from each integration's own class instead of duplicated
+    literals. Only integrations with a live --sync-down flow are listed (Teams has no approvals profile yet,
+    so it has no --sync-down flag to exempt), so an unlisted command fails closed with no exemption at all."""
+    from ..commands.integrations.gchat_app_setup import GChatAppSetupCommand
+    from ..commands.integrations.slack_app_setup import SlackAppSetupCommand
+    return {cmd.get_command_name(): cmd.get_record_env_key() for cmd in (SlackAppSetupCommand(), GChatAppSetupCommand())}
 
 
 def resolve_sync_down_exempt_uid(command_tokens) -> Optional[str]:
-    """For '{slack,teams,gchat}-app-setup ... --sync-down ...', the one UID this dispatch may bypass Layers A/B for -- always this integration's own pinned-env UID, never derived from what the admin passes (e.g. -r/--integration-record), so a different integration's protected record can never be reached this way."""
+    """For '{slack,gchat}-app-setup ... --sync-down ...', the one UID this dispatch may bypass Layers A/B for -- always this integration's own pinned-env UID, never derived from what the admin passes (e.g. -r/--integration-record), so a different integration's protected record can never be reached this way."""
     if not command_tokens:
         return None
 
-    env_name = _SYNC_DOWN_EXEMPT_ENV_BY_COMMAND.get(command_tokens[0].lower())
+    env_name = _sync_down_exempt_commands().get(command_tokens[0].lower())
     if not env_name:
         return None
 
-    if not any(tok == '--sync-down' for tok in command_tokens[1:]):
+    from .verified_command import Verifycommand
+    if not Verifycommand._has_option(command_tokens, '--sync-down'):
         return None
 
     return (os.environ.get(env_name) or '').strip() or None
diff --git a/unit-tests/service/test_command.py b/unit-tests/service/test_command.py
index 6f9c6ac6e..8a27c1bc5 100644
--- a/unit-tests/service/test_command.py
+++ b/unit-tests/service/test_command.py
@@ -348,4 +348,73 @@ def test_slack_setup_without_sync_down_gets_no_exemption(self):
             f'slack-app-setup --slack-record-name {self.SLACK_UID}', params
         )
         self.assertEqual(status_code, 403)
-        mock_capture.assert_not_called()
\ No newline at end of file
+        mock_capture.assert_not_called()
+
+    def test_gchat_sync_down_default_flow_is_not_blocked_and_sees_its_own_record(self):
+        """Same as Slack's own-record flow, but for GChat -- proves the exemption isn't Slack-specific."""
+        params = self._params()
+        seen = {}
+
+        def fake_capture(p, command):
+            seen['keys'] = set(p.record_cache.keys())
+            return 'ok', 'ok', ''
+
+        response, status_code, mock_capture = self._run(
+            'gchat-app-setup --sync-down', params, capture_side_effect=fake_capture
+        )
+        self.assertEqual(status_code, 200)
+        mock_capture.assert_called_once()
+        self.assertIn(self.GCHAT_UID, seen['keys'])
+        self.assertIn(self.SLACK_UID, params.record_cache)
+
+    def test_gchat_sync_down_with_a_different_integrations_uid_is_still_blocked(self):
+        params = self._params()
+        response, status_code, mock_capture = self._run(
+            f'gchat-app-setup --sync-down -r {self.SLACK_UID}', params
+        )
+        self.assertEqual(status_code, 403)
+        mock_capture.assert_not_called()
+
+
+class TestTerraformRecordProtectionCommandExecution(TestCase):
+    """Same UID-pinning protection Docker/Slack/GChat get, proven at the CommandExecutor.execute() boundary."""
+
+    TERRAFORM_UID = 'TERRAFORM_CONFIG_UID'
+
+    def _params(self):
+        p = params_module.KeeperParams()
+        p.service_mode = False
+        p.record_cache = {
+            self.TERRAFORM_UID: _record_cache_entry(self.TERRAFORM_UID, 'Commander Service Mode Terraform Config'),
+            NORMAL_UID: _record_cache_entry(NORMAL_UID, 'My Normal Record'),
+        }
+        return p
+
+    def _run(self, command, params):
+        with mock.patch.dict('os.environ', {'TERRAFORM_RECORD': self.TERRAFORM_UID}), mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch.object(
+            CommandExecutor, 'capture_output_and_logs', return_value=('ok', 'ok', '')
+        ) as mock_capture:
+            response, status_code = CommandExecutor.execute(command)
+        return response, status_code, mock_capture
+
+    def test_get_terraform_record_is_blocked(self):
+        params = self._params()
+        response, status_code, mock_capture = self._run(f'get {self.TERRAFORM_UID}', params)
+        self.assertEqual(status_code, 403)
+        mock_capture.assert_not_called()
+
+    def test_share_record_on_terraform_record_is_blocked(self):
+        params = self._params()
+        response, status_code, mock_capture = self._run(
+            f'share-record {self.TERRAFORM_UID} --email a@b.com', params
+        )
+        self.assertEqual(status_code, 403)
+        mock_capture.assert_not_called()
+
+    def test_normal_record_is_unaffected(self):
+        params = self._params()
+        response, status_code, mock_capture = self._run(f'get {NORMAL_UID}', params)
+        self.assertEqual(status_code, 200)
+        mock_capture.assert_called_once()
\ No newline at end of file
diff --git a/unit-tests/service/test_min_commander_version.py b/unit-tests/service/test_min_commander_version.py
index 6e0aa90fd..eb69fb5fc 100644
--- a/unit-tests/service/test_min_commander_version.py
+++ b/unit-tests/service/test_min_commander_version.py
@@ -18,6 +18,7 @@
 from keepercommander.service.decorators.min_commander_version import (
     MIN_COMMANDER_VERSION_HEADER,
     TERRAFORM_DOCKER_ENV,
+    TERRAFORM_DOCKER_ENV_LEGACY,
     check_min_commander_version,
     min_commander_version_check,
     _parse_version,
@@ -166,7 +167,10 @@ def test_rejects_when_running_version_unparseable(self):
         '17.0.0',
     )
     def test_non_terraform_docker_ignores_min_version_header(self):
-        env = {k: v for k, v in os.environ.items() if k != TERRAFORM_DOCKER_ENV}
+        env = {
+            k: v for k, v in os.environ.items()
+            if k not in (TERRAFORM_DOCKER_ENV, TERRAFORM_DOCKER_ENV_LEGACY)
+        }
         with mock.patch.dict(os.environ, env, clear=True):
             with self.app.test_request_context(
                 '/api/v2/executecommand-async',
@@ -175,6 +179,26 @@ def test_non_terraform_docker_ignores_min_version_header(self):
             ):
                 self.assertIsNone(check_min_commander_version())
 
+    @mock.patch.dict(os.environ, {TERRAFORM_DOCKER_ENV_LEGACY: '1'}, clear=True)
+    @mock.patch(
+        'keepercommander.service.decorators.min_commander_version._RUNNING_VERSION',
+        Version('17.0.0'),
+    )
+    @mock.patch(
+        'keepercommander.service.decorators.min_commander_version._RUNNING_VERSION_RAW',
+        '17.0.0',
+    )
+    def test_legacy_terraform_env_var_still_enforces(self):
+        """A container upgraded without re-running terraform-app-setup still has the old
+        KEEPER_TERRAFORM marker -- enforcement must not silently disable itself."""
+        with self.app.test_request_context(
+            '/api/v2/executecommand-async',
+            method='POST',
+            headers={MIN_COMMANDER_VERSION_HEADER: '18.1.0'},
+        ):
+            body, status = check_min_commander_version()
+            self.assertEqual(status, 426)
+
     @mock.patch.dict(os.environ, _TERRAFORM_DOCKER_ENV, clear=False)
     @mock.patch(
         'keepercommander.service.decorators.min_commander_version._RUNNING_VERSION',
diff --git a/unit-tests/service/test_protected_records.py b/unit-tests/service/test_protected_records.py
index 36989ff9e..903950a32 100644
--- a/unit-tests/service/test_protected_records.py
+++ b/unit-tests/service/test_protected_records.py
@@ -3,6 +3,7 @@
 from unittest import TestCase, mock
 
 from keepercommander import params as params_module
+from keepercommander.utils import generate_uid
 from keepercommander.service.util.protected_records import (
     get_protected_record_title_set,
     get_protected_record_uids,
@@ -85,14 +86,16 @@ def test_similar_but_not_exact_title_is_not_matched(self):
 
     def test_docker_record_protected_by_uid_even_with_custom_title(self):
         """--record-name can give the Docker config record a custom title; COMMANDER_RECORD must still identify it."""
-        with mock.patch.dict(os.environ, {'COMMANDER_RECORD': 'DOCKER_CUSTOM_UID'}):
-            p = _params_with_records({'DOCKER_CUSTOM_UID': 'My Totally Custom Docker Title'})
+        uid = generate_uid()
+        with mock.patch.dict(os.environ, {'COMMANDER_RECORD': uid}):
+            p = _params_with_records({uid: 'My Totally Custom Docker Title'})
             result = get_protected_record_uids(p)
-        self.assertIn('DOCKER_CUSTOM_UID', result)
+        self.assertIn(uid, result)
 
     def test_docker_env_uid_present_even_without_params(self):
-        with mock.patch.dict(os.environ, {'COMMANDER_RECORD': 'DOCKER_CUSTOM_UID'}):
-            self.assertIn('DOCKER_CUSTOM_UID', get_protected_record_uids(None))
+        uid = generate_uid()
+        with mock.patch.dict(os.environ, {'COMMANDER_RECORD': uid}):
+            self.assertIn(uid, get_protected_record_uids(None))
 
     def test_no_docker_env_var_falls_back_to_title_only(self):
         with mock.patch.dict(os.environ, {}, clear=True):
@@ -100,25 +103,36 @@ def test_no_docker_env_var_falls_back_to_title_only(self):
             result = get_protected_record_uids(p)
         self.assertEqual(set(result.keys()), {'UID_CONFIG'})
 
+    def test_malformed_env_uid_falls_back_to_title_matching(self):
+        """A misconfigured pinning env var (not a real record UID) must not become a phantom protected token."""
+        with mock.patch.dict(os.environ, {'TERRAFORM_RECORD': '1'}):
+            p = _params_with_records({'UID_CONFIG': PROTECTED_TITLE})
+            result = get_protected_record_uids(p)
+        self.assertEqual(set(result.keys()), {'UID_CONFIG'})
+
     def test_terraform_record_protected_by_uid_even_with_custom_title(self):
-        with mock.patch.dict(os.environ, {'TERRAFORM_RECORD': 'TF_CUSTOM_UID'}):
-            p = _params_with_records({'TF_CUSTOM_UID': 'My Totally Custom Terraform Title'})
-            self.assertIn('TF_CUSTOM_UID', get_protected_record_uids(p))
+        uid = generate_uid()
+        with mock.patch.dict(os.environ, {'TERRAFORM_RECORD': uid}):
+            p = _params_with_records({uid: 'My Totally Custom Terraform Title'})
+            self.assertIn(uid, get_protected_record_uids(p))
 
     def test_slack_record_protected_by_uid_even_with_custom_title(self):
-        with mock.patch.dict(os.environ, {'SLACK_RECORD': 'SLACK_CUSTOM_UID'}):
-            p = _params_with_records({'SLACK_CUSTOM_UID': 'My Totally Custom Slack Title'})
-            self.assertIn('SLACK_CUSTOM_UID', get_protected_record_uids(p))
+        uid = generate_uid()
+        with mock.patch.dict(os.environ, {'SLACK_RECORD': uid}):
+            p = _params_with_records({uid: 'My Totally Custom Slack Title'})
+            self.assertIn(uid, get_protected_record_uids(p))
 
     def test_teams_record_protected_by_uid_even_with_custom_title(self):
-        with mock.patch.dict(os.environ, {'TEAMS_RECORD': 'TEAMS_CUSTOM_UID'}):
-            p = _params_with_records({'TEAMS_CUSTOM_UID': 'My Totally Custom Teams Title'})
-            self.assertIn('TEAMS_CUSTOM_UID', get_protected_record_uids(p))
+        uid = generate_uid()
+        with mock.patch.dict(os.environ, {'TEAMS_RECORD': uid}):
+            p = _params_with_records({uid: 'My Totally Custom Teams Title'})
+            self.assertIn(uid, get_protected_record_uids(p))
 
     def test_gchat_record_protected_by_uid_even_with_custom_title(self):
-        with mock.patch.dict(os.environ, {'GCHAT_RECORD': 'GCHAT_CUSTOM_UID'}):
-            p = _params_with_records({'GCHAT_CUSTOM_UID': 'My Totally Custom GChat Title'})
-            self.assertIn('GCHAT_CUSTOM_UID', get_protected_record_uids(p))
+        uid = generate_uid()
+        with mock.patch.dict(os.environ, {'GCHAT_RECORD': uid}):
+            p = _params_with_records({uid: 'My Totally Custom GChat Title'})
+            self.assertIn(uid, get_protected_record_uids(p))
 
     def test_no_pinned_env_vars_falls_back_to_title_only(self):
         with mock.patch.dict(os.environ, {}, clear=True):
@@ -310,3 +324,11 @@ def test_teams_has_no_sync_down_exemption_yet(self):
 
     def test_empty_tokens_returns_none(self):
         self.assertIsNone(resolve_sync_down_exempt_uid([]))
+
+    def test_matches_unambiguous_abbreviation_of_sync_down(self):
+        """Reuses Verifycommand's abbreviation-aware flag matching, so an abbreviated
+        --sync-down (as argparse's own allow_abbrev would accept) isn't missed."""
+        with mock.patch.dict(os.environ, {'SLACK_RECORD': 'SLACK_UID'}, clear=True):
+            self.assertEqual(
+                resolve_sync_down_exempt_uid(['slack-app-setup', '--sync-d']), 'SLACK_UID'
+            )
diff --git a/unit-tests/service/test_runtime_policy.py b/unit-tests/service/test_runtime_policy.py
index 4c77f7879..04c1d414e 100644
--- a/unit-tests/service/test_runtime_policy.py
+++ b/unit-tests/service/test_runtime_policy.py
@@ -20,7 +20,7 @@
 from keepercommander.service.commands.integrations.sailpoint_app_setup import SailPointAppSetupCommand
 from keepercommander.service.commands.integrations.slack_app_setup import SlackAppSetupCommand
 from keepercommander.service.commands.terraform_app_setup import TerraformSetupConstants
-from keepercommander.service.decorators.min_commander_version import TERRAFORM_DOCKER_ENV
+from keepercommander.service.decorators.min_commander_version import TERRAFORM_DOCKER_ENV, TERRAFORM_DOCKER_ENV_LEGACY
 from keepercommander.service.util.exceptions import ValidationError
 
 
@@ -75,6 +75,16 @@ def test_terraform_env_confines_to_terraform_allowlist(self):
             set(args.commands.split(',')), set(TerraformSetupConstants.SERVICE_COMMANDS_LIST)
         )
 
+    def test_legacy_terraform_env_confines_to_terraform_allowlist(self):
+        """A container upgraded without re-running terraform-app-setup still has the old
+        KEEPER_TERRAFORM marker -- the startup sanitizer must not silently skip it."""
+        with mock.patch.dict(os.environ, {TERRAFORM_DOCKER_ENV_LEGACY: '1'}, clear=True):
+            args = _Args(commands=TerraformSetupConstants.SERVICE_COMMANDS + ',clipboard-copy')
+            apply_runtime_command_policy(args)
+        self.assertEqual(
+            set(args.commands.split(',')), set(TerraformSetupConstants.SERVICE_COMMANDS_LIST)
+        )
+
     def test_sailpoint_record_env_confines_to_sailpoint_allowlist(self):
         # Fallback path: applies even if SailPointService.maybe_enable() skipped sanitizing.
         allowed = set(SailPointAppSetupCommand().get_service_commands().split(','))

From c3ebd914c764061f11d6c07eb136d54320f08128 Mon Sep 17 00:00:00 2001
From: amangalampalli-ks <amangalampalli@keepersecurity.com>
Date: Mon, 21 Sep 2026 13:07:09 +0530
Subject: [PATCH 3/7] Fix review comments

---
 keepercommander/service/util/command_util.py      |  2 +-
 keepercommander/service/util/protected_records.py |  7 ++-----
 unit-tests/service/test_protected_records.py      | 12 ++++++------
 3 files changed, 9 insertions(+), 12 deletions(-)

diff --git a/keepercommander/service/util/command_util.py b/keepercommander/service/util/command_util.py
index a9917547a..f6f1d4f30 100644
--- a/keepercommander/service/util/command_util.py
+++ b/keepercommander/service/util/command_util.py
@@ -196,7 +196,7 @@ def blocked(error):
             # command can be missed as a way to reference these records.
             protected_uids = get_protected_record_uids(params)
 
-            # {slack,teams,gchat}-app-setup --sync-down needs its own config record reachable.
+            # {slack,gchat}-app-setup --sync-down needs its own config record reachable.
             sync_down_exempt_uid = resolve_sync_down_exempt_uid(command_tokens)
             if sync_down_exempt_uid is not None:
                 protected_uids = {
diff --git a/keepercommander/service/util/protected_records.py b/keepercommander/service/util/protected_records.py
index 40df4d82c..f0c7d8f7b 100644
--- a/keepercommander/service/util/protected_records.py
+++ b/keepercommander/service/util/protected_records.py
@@ -86,9 +86,7 @@ def get_protected_record_uids(params) -> Dict[str, str]:
 
 
 def _sync_down_exempt_commands() -> Dict[str, str]:
-    """{command name: pinned-UID env var}, derived from each integration's own class instead of duplicated
-    literals. Only integrations with a live --sync-down flow are listed (Teams has no approvals profile yet,
-    so it has no --sync-down flag to exempt), so an unlisted command fails closed with no exemption at all."""
+    """{command name: pinned-UID env var}, derived from each integration's own class instead of duplicated literals."""
     from ..commands.integrations.gchat_app_setup import GChatAppSetupCommand
     from ..commands.integrations.slack_app_setup import SlackAppSetupCommand
     return {cmd.get_command_name(): cmd.get_record_env_key() for cmd in (SlackAppSetupCommand(), GChatAppSetupCommand())}
@@ -103,8 +101,7 @@ def resolve_sync_down_exempt_uid(command_tokens) -> Optional[str]:
     if not env_name:
         return None
 
-    from .verified_command import Verifycommand
-    if not Verifycommand._has_option(command_tokens, '--sync-down'):
+    if '--sync-down' not in command_tokens[1:]:
         return None
 
     return (os.environ.get(env_name) or '').strip() or None
diff --git a/unit-tests/service/test_protected_records.py b/unit-tests/service/test_protected_records.py
index 903950a32..f98605c08 100644
--- a/unit-tests/service/test_protected_records.py
+++ b/unit-tests/service/test_protected_records.py
@@ -325,10 +325,10 @@ def test_teams_has_no_sync_down_exemption_yet(self):
     def test_empty_tokens_returns_none(self):
         self.assertIsNone(resolve_sync_down_exempt_uid([]))
 
-    def test_matches_unambiguous_abbreviation_of_sync_down(self):
-        """Reuses Verifycommand's abbreviation-aware flag matching, so an abbreviated
-        --sync-down (as argparse's own allow_abbrev would accept) isn't missed."""
+    def test_abbreviated_flag_does_not_grant_the_exemption(self):
+        """Exact match only -- granting an exemption is the permissive direction, so an
+        abbreviation like '--s' (ambiguous with --skip-device-setup on the real parser
+        anyway) must not be treated as --sync-down."""
         with mock.patch.dict(os.environ, {'SLACK_RECORD': 'SLACK_UID'}, clear=True):
-            self.assertEqual(
-                resolve_sync_down_exempt_uid(['slack-app-setup', '--sync-d']), 'SLACK_UID'
-            )
+            self.assertIsNone(resolve_sync_down_exempt_uid(['slack-app-setup', '--s']))
+            self.assertIsNone(resolve_sync_down_exempt_uid(['slack-app-setup', '--sync']))

From 2ffc3df7d6cdcb59cb3bb90b6a999930637581f5 Mon Sep 17 00:00:00 2001
From: amangalampalli-ks <amangalampalli@keepersecurity.com>
Date: Mon, 21 Sep 2026 13:17:47 +0530
Subject: [PATCH 4/7] Fix failing test case in windows

---
 unit-tests/service/test_tunneling.py | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/unit-tests/service/test_tunneling.py b/unit-tests/service/test_tunneling.py
index 6de58970d..528fbe68e 100644
--- a/unit-tests/service/test_tunneling.py
+++ b/unit-tests/service/test_tunneling.py
@@ -227,7 +227,11 @@ class TestDownloadCloudflared(unittest.TestCase):
     def test_lookup_failure_is_logged_not_swallowed_silently(self):
         # Force platform.system() to an unsupported value so _download_cloudflared
         # raises right after the (logged) lookup failure, without attempting a real download.
-        with mock.patch('keepercommander.service.util.tunneling.subprocess.run',
+        # sys.platform is pinned to a POSIX value too -- the PATH lookup this test exercises
+        # is skipped entirely on real win32 (see test_windows_never_searches_path_or_cwd), so
+        # this must not depend on which OS actually runs the test.
+        with mock.patch('keepercommander.service.util.tunneling.sys.platform', 'darwin'), \
+             mock.patch('keepercommander.service.util.tunneling.subprocess.run',
                          side_effect=OSError("cloudflared not found")), \
              mock.patch('keepercommander.service.util.tunneling.logging.debug') as mock_debug, \
              mock.patch('platform.system', return_value='unsupported'):

From 899a501cb750694938d12dd2daa505fb46209792 Mon Sep 17 00:00:00 2001
From: amangalampalli-ks <amangalampalli@keepersecurity.com>
Date: Mon, 21 Sep 2026 16:26:12 +0530
Subject: [PATCH 5/7] Restrict folder access via Service Mode for all *-setup
 commands

---
 keepercommander/service/util/command_util.py  |  17 ++-
 .../service/util/protected_records.py         |  79 ++++++++++-
 unit-tests/service/test_command.py            |  99 +++++++++++++-
 unit-tests/service/test_protected_records.py  | 128 ++++++++++++++++++
 4 files changed, 317 insertions(+), 6 deletions(-)

diff --git a/keepercommander/service/util/command_util.py b/keepercommander/service/util/command_util.py
index f6f1d4f30..f04a7152b 100644
--- a/keepercommander/service/util/command_util.py
+++ b/keepercommander/service/util/command_util.py
@@ -25,7 +25,13 @@
     is_throttle_error,
     throttle_error_response,
 )
-from .protected_records import get_protected_record_uids, hide_from_record_cache, resolve_sync_down_exempt_uid
+from .protected_records import (
+    get_protected_folder_uids,
+    get_protected_record_uids,
+    hide_from_folder_cache,
+    hide_from_record_cache,
+    resolve_sync_down_exempt_uid,
+)
 from .verified_command import Verifycommand
 from ..core.globals import get_current_params
 from ..decorators.logging import logger, debug_decorator, sanitize_debug_data, sanitize_command_fields
@@ -203,15 +209,20 @@ def blocked(error):
                     uid: title for uid, title in protected_uids.items() if uid != sync_down_exempt_uid
                 }
 
+            # Derived from the record set so the exemption above reaches the exempted integration's own folder too.
+            protected_folder_uids = get_protected_folder_uids(params, protected_uids)
+
             protected_command_error = Verifycommand.validate_service_mode_protected_record_command(
-                command_tokens, protected_uids
+                command_tokens,
+                {**protected_uids, **{uid: '<protected folder>' for uid in protected_folder_uids}},
             )
             if protected_command_error:
                 return blocked(protected_command_error)
 
             sailpoint_enabled = bool((os.environ.get('SAILPOINT_RECORD') or '').strip())
 
-            with hide_from_record_cache(params, protected_uids):
+            with hide_from_record_cache(params, protected_uids), \
+                 hide_from_folder_cache(params, protected_folder_uids):
                 if sailpoint_enabled:
                     from ..commands.integrations.sailpoint.service import SailPointService
                     command, sailpoint_response = SailPointService.handle_command(params, command)
diff --git a/keepercommander/service/util/protected_records.py b/keepercommander/service/util/protected_records.py
index f0c7d8f7b..9d4f2aeaa 100644
--- a/keepercommander/service/util/protected_records.py
+++ b/keepercommander/service/util/protected_records.py
@@ -9,14 +9,14 @@
 # Contact: ops@keepersecurity.com
 #
 
-"""Identify Service Mode's own config records so they can be hidden from commands."""
+"""Identify Service Mode's own config records and folders so they can be hidden from commands."""
 
 from __future__ import annotations
 
 import contextlib
 import os
 from collections import UserDict
-from typing import Dict, FrozenSet, Iterable, Optional, Tuple
+from typing import Dict, FrozenSet, Iterable, Optional, Set, Tuple
 
 # Each integration's own setup pins its config record's UID here, regardless of its title. Extend when a new integration gets an always-hidden record.
 _PINNED_RECORD_UID_ENVS: Dict[str, str] = {
@@ -30,6 +30,10 @@
 # uid-keyed caches resolve_single_record/load_pam_record fall back to when a UID isn't in record_cache.
 _GUARDED_CACHE_ATTRS = ('record_cache', 'nested_share_records', 'nested_share_record_data')
 
+# Raw + derived folder caches every folder-resolving command reads through,
+# directly or via subfolder.try_resolve_path/get_folder_uids.
+_GUARDED_FOLDER_CACHE_ATTRS = ('folder_cache', 'shared_folder_cache', 'subfolder_cache')
+
 
 def _protected_titles() -> Tuple[str, ...]:
     """The literal titles of Service Mode's own config records; imported lazily to avoid a circular import through verified_command."""
@@ -85,6 +89,19 @@ def get_protected_record_uids(params) -> Dict[str, str]:
     return found
 
 
+def get_protected_folder_uids(params, protected_record_uids: Dict[str, str]) -> Set[str]:
+    """Folders directly containing an already-protected record, via subfolder_record_cache (folder_uid -> set of record UIDs) -- derived from record protection rather than a separate per-integration title list, so it stays correct even if a folder is renamed."""
+    subfolder_record_cache = getattr(params, 'subfolder_record_cache', None)
+    if params is None or not protected_record_uids or not isinstance(subfolder_record_cache, dict):
+        return set()
+
+    record_uids = protected_record_uids.keys()
+    return {
+        folder_uid for folder_uid, uids in subfolder_record_cache.items()
+        if folder_uid and isinstance(uids, (set, frozenset)) and uids & record_uids
+    }
+
+
 def _sync_down_exempt_commands() -> Dict[str, str]:
     """{command name: pinned-UID env var}, derived from each integration's own class instead of duplicated literals."""
     from ..commands.integrations.gchat_app_setup import GChatAppSetupCommand
@@ -175,3 +192,61 @@ def hide_from_record_cache(params, protected_uids: Dict[str, str]):
                         uids |= hit
                 except Exception as e:
                     logger.debug(f'hide_from_record_cache: failed to restore subfolder {folder_uid} ({type(e).__name__})')
+
+
+@contextlib.contextmanager
+def hide_from_folder_cache(params, protected_folder_uids: Set[str]):
+    """For the with-block, hides protected_folder_uids from folder_cache/shared_folder_cache/subfolder_cache and
+    from their parent's (or root_folder's) .subfolders list, restoring everything on exit """
+    if params is None or not protected_folder_uids:
+        yield
+        return
+
+    protected_uid_set = frozenset(protected_folder_uids)
+
+    folder_cache = getattr(params, 'folder_cache', None)
+    root_folder = getattr(params, 'root_folder', None)
+    removed_from_parents: Dict[str, list] = {}
+    if isinstance(folder_cache, dict):
+        for uid in protected_uid_set:
+            node = folder_cache.get(uid)
+            parent_uid = getattr(node, 'parent_uid', None) if node is not None else None
+            parent = folder_cache.get(parent_uid) if parent_uid else root_folder
+            subfolders = getattr(parent, 'subfolders', None) if parent is not None else None
+            if isinstance(subfolders, list) and uid in subfolders:
+                subfolders.remove(uid)
+                removed_from_parents[uid] = subfolders
+
+    original_caches = {}
+    saved_entries = {}
+    for attr in _GUARDED_FOLDER_CACHE_ATTRS:
+        source = getattr(params, attr, None)
+        if not isinstance(source, dict):
+            continue
+        original_caches[attr] = source
+        saved_entries[attr] = {uid: source[uid] for uid in protected_uid_set if uid in source}
+        setattr(params, attr, _GuardedRecordCache(source, protected_uid_set))
+
+    try:
+        yield
+    finally:
+        from ..decorators.logging import logger
+
+        for attr in original_caches:
+            try:
+                restored = dict(getattr(params, attr, None) or {})
+                restored.update(saved_entries[attr])
+                setattr(params, attr, restored)
+            except Exception as e:
+                logger.debug(f'hide_from_folder_cache: failed to restore {attr} ({type(e).__name__}); restoring protected entries only')
+                try:
+                    setattr(params, attr, dict(saved_entries[attr]))
+                except Exception:
+                    pass
+
+        for uid, subfolders in removed_from_parents.items():
+            try:
+                if uid not in subfolders:
+                    subfolders.append(uid)
+            except Exception as e:
+                logger.debug(f'hide_from_folder_cache: failed to restore subfolders entry for {uid} ({type(e).__name__})')
diff --git a/unit-tests/service/test_command.py b/unit-tests/service/test_command.py
index 8a27c1bc5..14637adde 100644
--- a/unit-tests/service/test_command.py
+++ b/unit-tests/service/test_command.py
@@ -4,6 +4,7 @@
 from unittest import TestCase, mock
 from flask import Flask
 from keepercommander import params as params_module
+from keepercommander.subfolder import RootFolderNode, SharedFolderNode
 from keepercommander.service.util.command_util import CommandExecutor
 from keepercommander.service.util.exceptions import CommandExecutionError
 from keepercommander.service.util.parse_keeper_response import parse_keeper_response
@@ -417,4 +418,100 @@ def test_normal_record_is_unaffected(self):
         params = self._params()
         response, status_code, mock_capture = self._run(f'get {NORMAL_UID}', params)
         self.assertEqual(status_code, 200)
-        mock_capture.assert_called_once()
\ No newline at end of file
+        mock_capture.assert_called_once()
+
+class TestProtectedFolderCommandExecution(TestCase):
+    """The shared folder holding a protected config record must be just as unreachable
+    as the record itself -- ls/tree/rndir/mv/share-folder all resolve folders through the
+    same caches hide_from_folder_cache guards."""
+
+    PROTECTED_FOLDER_UID = 'PROTECTED_FOLDER_UID'
+    PROTECTED_FOLDER_TITLE = 'Commander Service Mode - Docker'
+    NORMAL_FOLDER_UID = 'NORMAL_FOLDER_UID'
+
+    def _params(self):
+        p = params_module.KeeperParams()
+        p.service_mode = False
+        p.record_cache = {
+            PROTECTED_UID: _record_cache_entry(PROTECTED_UID, PROTECTED_TITLE),
+            NORMAL_UID: _record_cache_entry(NORMAL_UID, 'My Normal Record'),
+        }
+        p.root_folder = RootFolderNode()
+
+        protected_node = SharedFolderNode()
+        protected_node.uid = self.PROTECTED_FOLDER_UID
+        protected_node.name = self.PROTECTED_FOLDER_TITLE
+        normal_node = SharedFolderNode()
+        normal_node.uid = self.NORMAL_FOLDER_UID
+        normal_node.name = 'My Normal Folder'
+
+        p.folder_cache = {self.PROTECTED_FOLDER_UID: protected_node, self.NORMAL_FOLDER_UID: normal_node}
+        p.root_folder.subfolders = [self.PROTECTED_FOLDER_UID, self.NORMAL_FOLDER_UID]
+        p.shared_folder_cache = {
+            self.PROTECTED_FOLDER_UID: {'name_unencrypted': self.PROTECTED_FOLDER_TITLE},
+            self.NORMAL_FOLDER_UID: {'name_unencrypted': 'My Normal Folder'},
+        }
+        p.subfolder_cache = {
+            self.PROTECTED_FOLDER_UID: {'type': 'shared_folder', 'shared_folder_uid': self.PROTECTED_FOLDER_UID},
+            self.NORMAL_FOLDER_UID: {'type': 'shared_folder', 'shared_folder_uid': self.NORMAL_FOLDER_UID},
+        }
+        p.subfolder_record_cache = {
+            self.PROTECTED_FOLDER_UID: {PROTECTED_UID},
+            self.NORMAL_FOLDER_UID: {NORMAL_UID},
+        }
+        return p
+
+    def _run(self, command, params):
+        with mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch.object(
+            CommandExecutor, 'capture_output_and_logs', return_value=('ok', 'ok', '')
+        ) as mock_capture:
+            response, status_code = CommandExecutor.execute(command)
+        return response, status_code, mock_capture
+
+    def test_blocked_folder_commands_never_reach_cli_dispatch(self):
+        """Layer B's literal-token scan is UID-only for folders (by design -- see the plan);
+        a title-only reference isn't caught here, it fails to resolve at all once Layer A
+        hides the folder from folder_cache/shared_folder_cache, proven separately in
+        TestHideFromFolderCache and test_folder_caches_hidden_during_dispatch_and_restored_after."""
+        for command in (
+            f'ls {self.PROTECTED_FOLDER_UID}',
+            f'tree {self.PROTECTED_FOLDER_UID}',
+            f'rndir {self.PROTECTED_FOLDER_UID} x',
+            f'mv {self.PROTECTED_FOLDER_UID} /',
+            f'share-folder {self.PROTECTED_FOLDER_UID} -e a@b.com',
+        ):
+            with self.subTest(command=command):
+                params = self._params()
+                response, status_code, mock_capture = self._run(command, params)
+                self.assertEqual(status_code, 403)
+                mock_capture.assert_not_called()
+
+    def test_normal_folder_commands_are_unaffected(self):
+        params = self._params()
+        response, status_code, mock_capture = self._run(f'ls {self.NORMAL_FOLDER_UID}', params)
+        self.assertEqual(status_code, 200)
+        mock_capture.assert_called_once()
+
+    def test_folder_caches_hidden_during_dispatch_and_restored_after(self):
+        params = self._params()
+        seen = {}
+
+        def fake_capture(p, command):
+            seen['folder_keys'] = set(p.folder_cache.keys())
+            seen['subfolders'] = list(p.root_folder.subfolders)
+            return 'ok', 'ok', ''
+
+        with mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch.object(CommandExecutor, 'capture_output_and_logs', side_effect=fake_capture):
+            response, status_code = CommandExecutor.execute(f'ls {self.NORMAL_FOLDER_UID}')
+
+        self.assertEqual(status_code, 200)
+        self.assertNotIn(self.PROTECTED_FOLDER_UID, seen['folder_keys'])
+        self.assertIn(self.NORMAL_FOLDER_UID, seen['folder_keys'])
+        self.assertNotIn(self.PROTECTED_FOLDER_UID, seen['subfolders'])
+
+        self.assertIn(self.PROTECTED_FOLDER_UID, params.folder_cache)
+        self.assertIn(self.PROTECTED_FOLDER_UID, params.root_folder.subfolders)
diff --git a/unit-tests/service/test_protected_records.py b/unit-tests/service/test_protected_records.py
index f98605c08..e6fa24f59 100644
--- a/unit-tests/service/test_protected_records.py
+++ b/unit-tests/service/test_protected_records.py
@@ -3,10 +3,13 @@
 from unittest import TestCase, mock
 
 from keepercommander import params as params_module
+from keepercommander.subfolder import RootFolderNode, SharedFolderNode
 from keepercommander.utils import generate_uid
 from keepercommander.service.util.protected_records import (
+    get_protected_folder_uids,
     get_protected_record_title_set,
     get_protected_record_uids,
+    hide_from_folder_cache,
     hide_from_record_cache,
     resolve_sync_down_exempt_uid,
 )
@@ -332,3 +335,128 @@ def test_abbreviated_flag_does_not_grant_the_exemption(self):
         with mock.patch.dict(os.environ, {'SLACK_RECORD': 'SLACK_UID'}, clear=True):
             self.assertIsNone(resolve_sync_down_exempt_uid(['slack-app-setup', '--s']))
             self.assertIsNone(resolve_sync_down_exempt_uid(['slack-app-setup', '--sync']))
+
+
+def _folder_node(uid, name, parent_uid=None):
+    node = SharedFolderNode()
+    node.uid = uid
+    node.parent_uid = parent_uid
+    node.name = name
+    return node
+
+
+def _params_with_folder(folder_uid='FOLDER1', record_uid='PROTECTED', parent_uid=None):
+    p = params_module.KeeperParams()
+    p.root_folder = RootFolderNode()
+    node = _folder_node(folder_uid, 'Commander Service Mode - Docker', parent_uid)
+    p.folder_cache = {folder_uid: node}
+    parent_list = p.root_folder.subfolders if not parent_uid else None
+    if parent_list is not None:
+        parent_list.append(folder_uid)
+    p.shared_folder_cache = {folder_uid: {'name_unencrypted': node.name}}
+    p.subfolder_cache = {folder_uid: {'type': 'shared_folder', 'shared_folder_uid': folder_uid}}
+    p.subfolder_record_cache = {folder_uid: {record_uid}, 'OTHER_FOLDER': {'OTHER_RECORD'}}
+    return p
+
+
+class TestGetProtectedFolderUids(TestCase):
+    def test_returns_folder_containing_a_protected_record(self):
+        p = _params_with_folder(folder_uid='FOLDER1', record_uid='PROTECTED')
+        result = get_protected_folder_uids(p, {'PROTECTED': 'Commander Service Mode Docker Config'})
+        self.assertEqual(result, {'FOLDER1'})
+
+    def test_no_protected_records_present(self):
+        p = _params_with_folder(folder_uid='FOLDER1', record_uid='PROTECTED')
+        self.assertEqual(get_protected_folder_uids(p, {'UNRELATED': 'x'}), set())
+
+    def test_empty_protected_record_uids_is_a_noop(self):
+        p = _params_with_folder()
+        self.assertEqual(get_protected_folder_uids(p, {}), set())
+
+    def test_params_none(self):
+        self.assertEqual(get_protected_folder_uids(None, {'PROTECTED': 'x'}), set())
+
+    def test_missing_subfolder_record_cache_is_ignored(self):
+        p = params_module.KeeperParams()
+        self.assertEqual(get_protected_folder_uids(p, {'PROTECTED': 'x'}), set())
+
+    def test_renamed_folder_is_still_found(self):
+        """Derived from record containment, not a title list -- a rename doesn't lose protection."""
+        p = _params_with_folder(folder_uid='FOLDER1', record_uid='PROTECTED')
+        p.folder_cache['FOLDER1'].name = 'My Totally Renamed Folder'
+        p.shared_folder_cache['FOLDER1']['name_unencrypted'] = 'My Totally Renamed Folder'
+        result = get_protected_folder_uids(p, {'PROTECTED': 'Commander Service Mode Docker Config'})
+        self.assertEqual(result, {'FOLDER1'})
+
+
+class TestHideFromFolderCache(TestCase):
+    def test_hides_protected_folder_from_all_three_caches_inside_the_block(self):
+        p = _params_with_folder(folder_uid='FOLDER1')
+        with hide_from_folder_cache(p, {'FOLDER1'}):
+            self.assertNotIn('FOLDER1', p.folder_cache)
+            self.assertNotIn('FOLDER1', p.shared_folder_cache)
+            self.assertNotIn('FOLDER1', p.subfolder_cache)
+
+    def test_strips_uid_from_root_folder_subfolders_during_the_block(self):
+        p = _params_with_folder(folder_uid='FOLDER1')
+        with hide_from_folder_cache(p, {'FOLDER1'}):
+            self.assertNotIn('FOLDER1', p.root_folder.subfolders)
+
+    def test_strips_uid_from_parent_folders_subfolders_when_nested(self):
+        p = _params_with_folder(folder_uid='FOLDER1', parent_uid='PARENT')
+        parent = _folder_node('PARENT', 'Some Parent Folder')
+        p.folder_cache['PARENT'] = parent
+        parent.subfolders.append('FOLDER1')
+        with hide_from_folder_cache(p, {'FOLDER1'}):
+            self.assertNotIn('FOLDER1', parent.subfolders)
+        self.assertIn('FOLDER1', parent.subfolders)
+
+    def test_restores_everything_after_the_block(self):
+        p = _params_with_folder(folder_uid='FOLDER1')
+        with hide_from_folder_cache(p, {'FOLDER1'}):
+            pass
+        self.assertIn('FOLDER1', p.folder_cache)
+        self.assertIn('FOLDER1', p.shared_folder_cache)
+        self.assertIn('FOLDER1', p.subfolder_cache)
+        self.assertIn('FOLDER1', p.root_folder.subfolders)
+
+    def test_restores_even_if_block_raises(self):
+        p = _params_with_folder(folder_uid='FOLDER1')
+        with self.assertRaises(ValueError):
+            with hide_from_folder_cache(p, {'FOLDER1'}):
+                raise ValueError('boom')
+        self.assertIn('FOLDER1', p.folder_cache)
+        self.assertIn('FOLDER1', p.root_folder.subfolders)
+
+    def test_reintroduction_during_block_is_blocked(self):
+        p = _params_with_folder(folder_uid='FOLDER1')
+        with hide_from_folder_cache(p, {'FOLDER1'}):
+            p.folder_cache['FOLDER1'] = _folder_node('FOLDER1', 'reintroduced')
+            self.assertNotIn('FOLDER1', p.folder_cache)
+        self.assertIn('FOLDER1', p.folder_cache)
+
+    def test_no_protected_folders_is_a_noop(self):
+        p = _params_with_folder(folder_uid='FOLDER1')
+        with hide_from_folder_cache(p, set()):
+            self.assertIn('FOLDER1', p.folder_cache)
+
+    def test_params_none_is_a_noop(self):
+        with hide_from_folder_cache(None, {'FOLDER1'}):
+            pass
+
+    def test_missing_root_folder_does_not_raise(self):
+        """A params fixture with no root_folder set (e.g. never synced) must not crash the guard."""
+        p = _params_with_folder(folder_uid='FOLDER1')
+        p.root_folder = None
+        with hide_from_folder_cache(p, {'FOLDER1'}):
+            self.assertNotIn('FOLDER1', p.folder_cache)
+
+    def test_a_resync_mid_block_self_heals_without_reintroducing_the_folder(self):
+        """A forced resync mid-command rebuilds folder_cache/root_folder from the (still-guarded)
+        raw subfolder_cache/shared_folder_cache, so the protected folder must not reappear."""
+        p = _params_with_folder(folder_uid='FOLDER1')
+        with hide_from_folder_cache(p, {'FOLDER1'}):
+            from keepercommander.sync_down import prepare_folder_tree
+            prepare_folder_tree(p)
+            self.assertNotIn('FOLDER1', p.folder_cache)
+            self.assertNotIn('FOLDER1', p.root_folder.subfolders)

From c4f992b77ac8e8556e6101c972e9c54e9bef87a5 Mon Sep 17 00:00:00 2001
From: amangalampalli-ks <amangalampalli@keepersecurity.com>
Date: Mon, 21 Sep 2026 19:17:23 +0530
Subject: [PATCH 6/7] Prevent config and service config json file attachments

---
 .../service/util/protected_records.py         |  67 +++++++-
 unit-tests/service/test_command.py            |  60 +++++++
 unit-tests/service/test_protected_records.py  | 155 +++++++++++++++++-
 3 files changed, 280 insertions(+), 2 deletions(-)

diff --git a/keepercommander/service/util/protected_records.py b/keepercommander/service/util/protected_records.py
index 9d4f2aeaa..8d6728d40 100644
--- a/keepercommander/service/util/protected_records.py
+++ b/keepercommander/service/util/protected_records.py
@@ -34,6 +34,64 @@
 # directly or via subfolder.try_resolve_path/get_folder_uids.
 _GUARDED_FOLDER_CACHE_ATTRS = ('folder_cache', 'shared_folder_cache', 'subfolder_cache')
 
+# Commander's own session (config.json) and Service Mode's own runtime (service_config.json)
+# config files -- always attached under these exact, hardcoded names, never user-choosable.
+_RESERVED_ATTACHMENT_NAMES = frozenset({'config.json', 'service_config.json'})
+
+
+def _attachment_file_uids(record) -> list:
+    """UIDs of every file attachment on record -- legacy PasswordRecord.attachments' ids, or a
+    typed record's fileRef entries (each its own separate FileRecord, loadable/gettable by that UID)."""
+    from ... import vault
+
+    if isinstance(record, vault.PasswordRecord):
+        return [atta.id for atta in (record.attachments or []) if atta.id]
+    if isinstance(record, vault.TypedRecord):
+        typed_field = record.get_typed_field('fileRef')
+        if typed_field and isinstance(typed_field.value, list):
+            return [uid for uid in typed_field.value if isinstance(uid, str)]
+    return []
+
+
+def _has_reserved_attachment(params, record) -> bool:
+    """True if record carries a legacy or typed-record attachment named config.json/service_config.json."""
+    from ... import vault
+
+    if isinstance(record, vault.PasswordRecord):
+        return any(
+            (atta.title or atta.name or '').lower() in _RESERVED_ATTACHMENT_NAMES
+            for atta in (record.attachments or [])
+        )
+    for file_uid in _attachment_file_uids(record):
+        try:
+            file_record = vault.KeeperRecord.load(params, file_uid)
+        except Exception:
+            continue
+        if isinstance(file_record, vault.FileRecord) and \
+                (file_record.title or file_record.name or '').lower() in _RESERVED_ATTACHMENT_NAMES:
+            return True
+    return False
+
+
+def _expand_with_attachment_uids(params, found: Dict[str, str]) -> None:
+    """Any file attachment on an already-protected record is protected too -- otherwise get can still
+    reach the attachment's own FileRecord UID directly, even though the parent record is unreachable."""
+    record_cache = getattr(params, 'record_cache', None)
+    if not isinstance(record_cache, dict):
+        return
+
+    from ... import vault
+
+    for uid in list(found.keys()):
+        try:
+            record = vault.KeeperRecord.load(params, uid)
+        except Exception:
+            continue
+        if record is None:
+            continue
+        for file_uid in _attachment_file_uids(record):
+            found.setdefault(file_uid, '<attachment on a protected record>')
+
 
 def _protected_titles() -> Tuple[str, ...]:
     """The literal titles of Service Mode's own config records; imported lazily to avoid a circular import through verified_command."""
@@ -73,6 +131,7 @@ def get_protected_record_uids(params) -> Dict[str, str]:
         found[uid] = label
 
     if params is None or not isinstance(getattr(params, 'record_cache', None), dict) or not params.record_cache:
+        _expand_with_attachment_uids(params, found)
         return found
 
     from ... import vault
@@ -84,8 +143,14 @@ def get_protected_record_uids(params) -> Dict[str, str]:
         except Exception as e:
             logger.debug(f'protected_records: could not load record {uid} ({type(e).__name__}); skipping')
             continue
-        if record and record.title.lower() in protected_titles:
+        if not record:
+            continue
+        if record.title.lower() in protected_titles:
             found[uid] = record.title
+        elif _has_reserved_attachment(params, record):
+            found[uid] = '<record with a reserved config attachment>'
+
+    _expand_with_attachment_uids(params, found)
     return found
 
 
diff --git a/unit-tests/service/test_command.py b/unit-tests/service/test_command.py
index 14637adde..dad1ee643 100644
--- a/unit-tests/service/test_command.py
+++ b/unit-tests/service/test_command.py
@@ -515,3 +515,63 @@ def fake_capture(p, command):
 
         self.assertIn(self.PROTECTED_FOLDER_UID, params.folder_cache)
         self.assertIn(self.PROTECTED_FOLDER_UID, params.root_folder.subfolders)
+
+
+class TestReservedAttachmentCommandExecution(TestCase):
+    """A record with an arbitrary, non-default title must still be blocked if it carries
+    one of Commander's own reserved config-file attachments (config.json/service_config.json)."""
+
+    ARBITRARY_UID = 'ARBITRARY_TITLED_RECORD_UID'
+
+    def _params(self):
+        p = params_module.KeeperParams()
+        p.service_mode = False
+        p.record_cache = {
+            self.ARBITRARY_UID: _record_cache_entry(self.ARBITRARY_UID, 'My Totally Unrelated Title'),
+            NORMAL_UID: _record_cache_entry(NORMAL_UID, 'My Normal Record'),
+        }
+        return p
+
+    def _run(self, command, params, reserved_uids=(ARBITRARY_UID,)):
+        with mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch(
+            'keepercommander.service.util.protected_records._has_reserved_attachment',
+            side_effect=lambda p, record: record.record_uid in reserved_uids,
+        ), mock.patch.object(
+            CommandExecutor, 'capture_output_and_logs', return_value=('ok', 'ok', '')
+        ) as mock_capture:
+            response, status_code = CommandExecutor.execute(command)
+        return response, status_code, mock_capture
+
+    def test_get_on_record_with_reserved_attachment_is_blocked(self):
+        params = self._params()
+        response, status_code, mock_capture = self._run(f'get {self.ARBITRARY_UID}', params)
+        self.assertEqual(status_code, 403)
+        mock_capture.assert_not_called()
+
+    def test_file_report_omits_record_with_reserved_attachment(self):
+        params = self._params()
+        seen = {}
+
+        def fake_capture(p, command):
+            seen['keys'] = set(p.record_cache.keys())
+            return 'ok', 'ok', ''
+
+        with mock.patch(
+            'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
+        ), mock.patch(
+            'keepercommander.service.util.protected_records._has_reserved_attachment',
+            side_effect=lambda p, record: record.record_uid == self.ARBITRARY_UID,
+        ), mock.patch.object(CommandExecutor, 'capture_output_and_logs', side_effect=fake_capture):
+            response, status_code = CommandExecutor.execute('file-report')
+
+        self.assertEqual(status_code, 200)
+        self.assertNotIn(self.ARBITRARY_UID, seen['keys'])
+        self.assertIn(NORMAL_UID, seen['keys'])
+
+    def test_record_without_reserved_attachment_is_unaffected(self):
+        params = self._params()
+        response, status_code, mock_capture = self._run(f'get {NORMAL_UID}', params, reserved_uids=())
+        self.assertEqual(status_code, 200)
+        mock_capture.assert_called_once()
diff --git a/unit-tests/service/test_protected_records.py b/unit-tests/service/test_protected_records.py
index e6fa24f59..41d723617 100644
--- a/unit-tests/service/test_protected_records.py
+++ b/unit-tests/service/test_protected_records.py
@@ -2,10 +2,13 @@
 import os
 from unittest import TestCase, mock
 
-from keepercommander import params as params_module
+from keepercommander import params as params_module, vault
 from keepercommander.subfolder import RootFolderNode, SharedFolderNode
 from keepercommander.utils import generate_uid
 from keepercommander.service.util.protected_records import (
+    _attachment_file_uids,
+    _expand_with_attachment_uids,
+    _has_reserved_attachment,
     get_protected_folder_uids,
     get_protected_record_title_set,
     get_protected_record_uids,
@@ -460,3 +463,153 @@ def test_a_resync_mid_block_self_heals_without_reintroducing_the_folder(self):
             prepare_folder_tree(p)
             self.assertNotIn('FOLDER1', p.folder_cache)
             self.assertNotIn('FOLDER1', p.root_folder.subfolders)
+
+
+class TestHasReservedAttachment(TestCase):
+    def test_password_record_with_reserved_attachment_name(self):
+        record = vault.PasswordRecord()
+        record.attachments = [vault.AttachmentFile({'name': 'config.json', 'title': 'config.json'})]
+        self.assertTrue(_has_reserved_attachment(None, record))
+
+    def test_password_record_with_reserved_title_but_different_name(self):
+        """attachment.py itself checks title OR name -- match either."""
+        record = vault.PasswordRecord()
+        record.attachments = [vault.AttachmentFile({'name': 'file123', 'title': 'service_config.json'})]
+        self.assertTrue(_has_reserved_attachment(None, record))
+
+    def test_password_record_with_unrelated_attachment(self):
+        record = vault.PasswordRecord()
+        record.attachments = [vault.AttachmentFile({'name': 'notes.pdf', 'title': 'notes.pdf'})]
+        self.assertFalse(_has_reserved_attachment(None, record))
+
+    def test_password_record_with_no_attachments(self):
+        record = vault.PasswordRecord()
+        self.assertFalse(_has_reserved_attachment(None, record))
+
+    def test_typed_record_with_reserved_file_ref(self):
+        record = vault.TypedRecord()
+        record.fields = [vault.TypedField({'type': 'fileRef', 'value': ['FILE_UID_1']})]
+        file_record = vault.FileRecord()
+        file_record.title = 'config.json'
+        file_record.name = 'config.json'
+        with mock.patch('keepercommander.vault.KeeperRecord.load', return_value=file_record):
+            self.assertTrue(_has_reserved_attachment(object(), record))
+
+    def test_typed_record_with_unrelated_file_ref(self):
+        record = vault.TypedRecord()
+        record.fields = [vault.TypedField({'type': 'fileRef', 'value': ['FILE_UID_1']})]
+        file_record = vault.FileRecord()
+        file_record.title = 'notes.pdf'
+        file_record.name = 'notes.pdf'
+        with mock.patch('keepercommander.vault.KeeperRecord.load', return_value=file_record):
+            self.assertFalse(_has_reserved_attachment(object(), record))
+
+    def test_typed_record_with_no_file_ref(self):
+        record = vault.TypedRecord()
+        self.assertFalse(_has_reserved_attachment(object(), record))
+
+    def test_neither_password_nor_typed_record(self):
+        record = vault.FileRecord()
+        self.assertFalse(_has_reserved_attachment(None, record))
+
+
+class TestGetProtectedRecordUidsWithReservedAttachments(TestCase):
+    def test_arbitrary_titled_record_with_reserved_attachment_is_protected(self):
+        """A record with none of the known titles is still protected if it carries
+        one of Commander's own reserved config-file attachments."""
+        p = _params_with_records({'UID_ARBITRARY': 'My Totally Unrelated Title'})
+        with mock.patch(
+            'keepercommander.service.util.protected_records._has_reserved_attachment',
+            side_effect=lambda params, record: record.record_uid == 'UID_ARBITRARY',
+        ):
+            result = get_protected_record_uids(p)
+        self.assertIn('UID_ARBITRARY', result)
+
+    def test_arbitrary_titled_record_without_reserved_attachment_is_not_protected(self):
+        p = _params_with_records({'UID_ARBITRARY': 'My Totally Unrelated Title'})
+        with mock.patch(
+            'keepercommander.service.util.protected_records._has_reserved_attachment',
+            return_value=False,
+        ):
+            result = get_protected_record_uids(p)
+        self.assertEqual(result, {})
+
+
+class TestExpandWithAttachmentUids(TestCase):
+    """Regression test for a live gap: get 'service_config.json' returned the attachment's own
+    FileRecord (a separate record_cache entry, its own UID) directly, since only the parent
+    record was ever added to protected_uids -- the attachment's own UID was never blocked."""
+
+    def test_typed_record_file_ref_uid_is_protected(self):
+        parent = vault.TypedRecord()
+        parent.record_uid = 'PARENT_UID'
+        parent.fields = [vault.TypedField({'type': 'fileRef', 'value': ['FILE_UID_1']})]
+        found = {'PARENT_UID': 'Commander Service Mode Slack App Config'}
+
+        p = _params_with_records({'PARENT_UID': 'x', 'FILE_UID_1': 'y'})
+        with mock.patch('keepercommander.vault.KeeperRecord.load', return_value=parent):
+            _expand_with_attachment_uids(p, found)
+        self.assertIn('FILE_UID_1', found)
+
+    def test_legacy_password_record_attachment_id_is_protected(self):
+        parent = vault.PasswordRecord()
+        parent.record_uid = 'PARENT_UID'
+        parent.attachments = [vault.AttachmentFile({'id': 'ATTA_ID_1', 'name': 'config.json'})]
+        found = {'PARENT_UID': 'Commander Service Mode Docker Config'}
+
+        p = _params_with_records({'PARENT_UID': 'x'})
+        with mock.patch('keepercommander.vault.KeeperRecord.load', return_value=parent):
+            _expand_with_attachment_uids(p, found)
+        self.assertIn('ATTA_ID_1', found)
+
+    def test_noop_without_record_cache(self):
+        found = {'PARENT_UID': 'x'}
+        _expand_with_attachment_uids(params_module.KeeperParams(), found)
+        self.assertEqual(found, {'PARENT_UID': 'x'})
+
+    def test_noop_when_protected_uid_is_not_loadable(self):
+        found = {'PARENT_UID': 'x'}
+        p = _params_with_records({'PARENT_UID': 'x'})
+        with mock.patch('keepercommander.vault.KeeperRecord.load', return_value=None):
+            _expand_with_attachment_uids(p, found)
+        self.assertEqual(found, {'PARENT_UID': 'x'})
+
+    def test_end_to_end_arbitrary_titled_parent_with_attachment_protects_the_file_uid(self):
+        """Reproduces the exact reported scenario: a record with a normal title carrying a
+        service_config.json fileRef attachment -- both the parent and the attachment's own
+        FileRecord UID must come back protected."""
+        parent = vault.TypedRecord()
+        parent.record_uid = 'PARENT_UID'
+        parent.title = 'My Totally Unrelated Title'
+        parent.fields = [vault.TypedField({'type': 'fileRef', 'value': ['FILE_UID_1']})]
+
+        file_record = vault.FileRecord()
+        file_record.record_uid = 'FILE_UID_1'
+        file_record.title = 'service_config.json'
+        file_record.name = 'service_config.json'
+
+        records = {'PARENT_UID': parent, 'FILE_UID_1': file_record}
+        p = _params_with_records({'PARENT_UID': 'x', 'FILE_UID_1': 'y'})
+        with mock.patch('keepercommander.vault.KeeperRecord.load', side_effect=lambda params, uid: records.get(uid)):
+            result = get_protected_record_uids(p)
+        self.assertIn('PARENT_UID', result)
+        self.assertIn('FILE_UID_1', result)
+
+
+class TestAttachmentFileUids(TestCase):
+    def test_password_record_returns_attachment_ids(self):
+        record = vault.PasswordRecord()
+        record.attachments = [
+            vault.AttachmentFile({'id': 'A1', 'name': 'x'}),
+            vault.AttachmentFile({'id': 'A2', 'name': 'y'}),
+        ]
+        self.assertEqual(set(_attachment_file_uids(record)), {'A1', 'A2'})
+
+    def test_typed_record_returns_file_ref_values(self):
+        record = vault.TypedRecord()
+        record.fields = [vault.TypedField({'type': 'fileRef', 'value': ['F1', 'F2']})]
+        self.assertEqual(set(_attachment_file_uids(record)), {'F1', 'F2'})
+
+    def test_record_with_no_attachments_returns_empty(self):
+        self.assertEqual(_attachment_file_uids(vault.PasswordRecord()), [])
+        self.assertEqual(_attachment_file_uids(vault.TypedRecord()), [])

From ca19f196e343d1222d935d98471cd4c5ca7206a4 Mon Sep 17 00:00:00 2001
From: amangalampalli-ks <amangalampalli@keepersecurity.com>
Date: Mon, 21 Sep 2026 19:46:54 +0530
Subject: [PATCH 7/7] Fix claude review comments

---
 .../service/util/protected_records.py         | 114 ++++++------
 unit-tests/service/test_command.py            |  38 +++-
 unit-tests/service/test_protected_records.py  | 168 +++++++++---------
 3 files changed, 168 insertions(+), 152 deletions(-)

diff --git a/keepercommander/service/util/protected_records.py b/keepercommander/service/util/protected_records.py
index 8d6728d40..7d1b18c4b 100644
--- a/keepercommander/service/util/protected_records.py
+++ b/keepercommander/service/util/protected_records.py
@@ -53,44 +53,15 @@ def _attachment_file_uids(record) -> list:
     return []
 
 
-def _has_reserved_attachment(params, record) -> bool:
-    """True if record carries a legacy or typed-record attachment named config.json/service_config.json."""
+def _has_reserved_legacy_attachment(record) -> bool:
+    """True if a PasswordRecord's own .attachments (no extra load -- filenames live on the attachment
+    object itself) include one literally named config.json/service_config.json."""
     from ... import vault
 
-    if isinstance(record, vault.PasswordRecord):
-        return any(
-            (atta.title or atta.name or '').lower() in _RESERVED_ATTACHMENT_NAMES
-            for atta in (record.attachments or [])
-        )
-    for file_uid in _attachment_file_uids(record):
-        try:
-            file_record = vault.KeeperRecord.load(params, file_uid)
-        except Exception:
-            continue
-        if isinstance(file_record, vault.FileRecord) and \
-                (file_record.title or file_record.name or '').lower() in _RESERVED_ATTACHMENT_NAMES:
-            return True
-    return False
-
-
-def _expand_with_attachment_uids(params, found: Dict[str, str]) -> None:
-    """Any file attachment on an already-protected record is protected too -- otherwise get can still
-    reach the attachment's own FileRecord UID directly, even though the parent record is unreachable."""
-    record_cache = getattr(params, 'record_cache', None)
-    if not isinstance(record_cache, dict):
-        return
-
-    from ... import vault
-
-    for uid in list(found.keys()):
-        try:
-            record = vault.KeeperRecord.load(params, uid)
-        except Exception:
-            continue
-        if record is None:
-            continue
-        for file_uid in _attachment_file_uids(record):
-            found.setdefault(file_uid, '<attachment on a protected record>')
+    return isinstance(record, vault.PasswordRecord) and any(
+        (atta.title or atta.name or '').lower() in _RESERVED_ATTACHMENT_NAMES
+        for atta in (record.attachments or [])
+    )
 
 
 def _protected_titles() -> Tuple[str, ...]:
@@ -131,12 +102,16 @@ def get_protected_record_uids(params) -> Dict[str, str]:
         found[uid] = label
 
     if params is None or not isinstance(getattr(params, 'record_cache', None), dict) or not params.record_cache:
-        _expand_with_attachment_uids(params, found)
         return found
 
     from ... import vault
 
     protected_titles = get_protected_record_title_set()
+    # One load per record_cache entry, no more -- a FileRecord attachment target is itself an
+    # entry in this same cache, so its name is picked up by this same pass rather than a second,
+    # per-attachment load 
+    reserved_file_uids: Set[str] = set()
+    pending_attachments: Dict[str, list] = {}
     for uid in params.record_cache:
         try:
             record = vault.KeeperRecord.load(params, uid)
@@ -145,12 +120,28 @@ def get_protected_record_uids(params) -> Dict[str, str]:
             continue
         if not record:
             continue
+
+        if isinstance(record, vault.FileRecord):
+            if (record.title or record.name or '').lower() in _RESERVED_ATTACHMENT_NAMES:
+                reserved_file_uids.add(uid)
+            continue
+
         if record.title.lower() in protected_titles:
             found[uid] = record.title
-        elif _has_reserved_attachment(params, record):
+        elif _has_reserved_legacy_attachment(record):
             found[uid] = '<record with a reserved config attachment>'
 
-    _expand_with_attachment_uids(params, found)
+        file_uids = _attachment_file_uids(record)
+        if file_uids:
+            pending_attachments[uid] = file_uids
+
+    for parent_uid, file_uids in pending_attachments.items():
+        if parent_uid not in found and any(file_uid in reserved_file_uids for file_uid in file_uids):
+            found[parent_uid] = '<record with a reserved config attachment>'
+        if parent_uid in found:
+            for file_uid in file_uids:
+                found.setdefault(file_uid, '<attachment on a protected record>')
+
     return found
 
 
@@ -271,28 +262,33 @@ def hide_from_folder_cache(params, protected_folder_uids: Set[str]):
 
     folder_cache = getattr(params, 'folder_cache', None)
     root_folder = getattr(params, 'root_folder', None)
-    removed_from_parents: Dict[str, list] = {}
-    if isinstance(folder_cache, dict):
-        for uid in protected_uid_set:
-            node = folder_cache.get(uid)
-            parent_uid = getattr(node, 'parent_uid', None) if node is not None else None
-            parent = folder_cache.get(parent_uid) if parent_uid else root_folder
-            subfolders = getattr(parent, 'subfolders', None) if parent is not None else None
-            if isinstance(subfolders, list) and uid in subfolders:
-                subfolders.remove(uid)
-                removed_from_parents[uid] = subfolders
-
+    # {uid: (subfolders_list, original_index)} -- built up incrementally inside the try below so a
+    # failure partway through setup still leaves whatever was already removed restorable in finally,
+    # rather than mutating this live list before there's any guarantee finally will run at all.
+    removed_from_parents: Dict[str, tuple] = {}
     original_caches = {}
     saved_entries = {}
-    for attr in _GUARDED_FOLDER_CACHE_ATTRS:
-        source = getattr(params, attr, None)
-        if not isinstance(source, dict):
-            continue
-        original_caches[attr] = source
-        saved_entries[attr] = {uid: source[uid] for uid in protected_uid_set if uid in source}
-        setattr(params, attr, _GuardedRecordCache(source, protected_uid_set))
 
     try:
+        if isinstance(folder_cache, dict):
+            for uid in protected_uid_set:
+                node = folder_cache.get(uid)
+                parent_uid = getattr(node, 'parent_uid', None) if node is not None else None
+                parent = folder_cache.get(parent_uid) if parent_uid else root_folder
+                subfolders = getattr(parent, 'subfolders', None) if parent is not None else None
+                if isinstance(subfolders, list) and uid in subfolders:
+                    index = subfolders.index(uid)
+                    subfolders.remove(uid)
+                    removed_from_parents[uid] = (subfolders, index)
+
+        for attr in _GUARDED_FOLDER_CACHE_ATTRS:
+            source = getattr(params, attr, None)
+            if not isinstance(source, dict):
+                continue
+            original_caches[attr] = source
+            saved_entries[attr] = {uid: source[uid] for uid in protected_uid_set if uid in source}
+            setattr(params, attr, _GuardedRecordCache(source, protected_uid_set))
+
         yield
     finally:
         from ..decorators.logging import logger
@@ -309,9 +305,9 @@ def hide_from_folder_cache(params, protected_folder_uids: Set[str]):
                 except Exception:
                     pass
 
-        for uid, subfolders in removed_from_parents.items():
+        for uid, (subfolders, index) in removed_from_parents.items():
             try:
                 if uid not in subfolders:
-                    subfolders.append(uid)
+                    subfolders.insert(min(index, len(subfolders)), uid)
             except Exception as e:
                 logger.debug(f'hide_from_folder_cache: failed to restore subfolders entry for {uid} ({type(e).__name__})')
diff --git a/unit-tests/service/test_command.py b/unit-tests/service/test_command.py
index dad1ee643..de0736de8 100644
--- a/unit-tests/service/test_command.py
+++ b/unit-tests/service/test_command.py
@@ -3,7 +3,7 @@
 
 from unittest import TestCase, mock
 from flask import Flask
-from keepercommander import params as params_module
+from keepercommander import params as params_module, vault
 from keepercommander.subfolder import RootFolderNode, SharedFolderNode
 from keepercommander.service.util.command_util import CommandExecutor
 from keepercommander.service.util.exceptions import CommandExecutionError
@@ -532,12 +532,35 @@ def _params(self):
         }
         return p
 
+    @staticmethod
+    def _record_with_reserved_attachment(uid, title):
+        record = vault.PasswordRecord()
+        record.record_uid = uid
+        record.title = title
+        record.attachments = [vault.AttachmentFile({'id': f'{uid}_ATTA', 'name': 'config.json'})]
+        return record
+
+    @staticmethod
+    def _plain_record(uid, title):
+        record = vault.PasswordRecord()
+        record.record_uid = uid
+        record.title = title
+        return record
+
+    _TITLES = {ARBITRARY_UID: 'My Totally Unrelated Title', NORMAL_UID: 'My Normal Record'}
+
     def _run(self, command, params, reserved_uids=(ARBITRARY_UID,)):
+        records = {
+            uid: (
+                self._record_with_reserved_attachment(uid, self._TITLES[uid])
+                if uid in reserved_uids else self._plain_record(uid, self._TITLES[uid])
+            )
+            for uid in params.record_cache
+        }
         with mock.patch(
             'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
         ), mock.patch(
-            'keepercommander.service.util.protected_records._has_reserved_attachment',
-            side_effect=lambda p, record: record.record_uid in reserved_uids,
+            'keepercommander.vault.KeeperRecord.load', side_effect=lambda p, uid: records.get(uid)
         ), mock.patch.object(
             CommandExecutor, 'capture_output_and_logs', return_value=('ok', 'ok', '')
         ) as mock_capture:
@@ -552,6 +575,12 @@ def test_get_on_record_with_reserved_attachment_is_blocked(self):
 
     def test_file_report_omits_record_with_reserved_attachment(self):
         params = self._params()
+        records = {
+            self.ARBITRARY_UID: self._record_with_reserved_attachment(
+                self.ARBITRARY_UID, self._TITLES[self.ARBITRARY_UID]
+            ),
+            NORMAL_UID: self._plain_record(NORMAL_UID, self._TITLES[NORMAL_UID]),
+        }
         seen = {}
 
         def fake_capture(p, command):
@@ -561,8 +590,7 @@ def fake_capture(p, command):
         with mock.patch(
             'keepercommander.service.core.globals.ensure_params_loaded', return_value=params
         ), mock.patch(
-            'keepercommander.service.util.protected_records._has_reserved_attachment',
-            side_effect=lambda p, record: record.record_uid == self.ARBITRARY_UID,
+            'keepercommander.vault.KeeperRecord.load', side_effect=lambda p, uid: records.get(uid)
         ), mock.patch.object(CommandExecutor, 'capture_output_and_logs', side_effect=fake_capture):
             response, status_code = CommandExecutor.execute('file-report')
 
diff --git a/unit-tests/service/test_protected_records.py b/unit-tests/service/test_protected_records.py
index 41d723617..9fe2a6940 100644
--- a/unit-tests/service/test_protected_records.py
+++ b/unit-tests/service/test_protected_records.py
@@ -7,8 +7,7 @@
 from keepercommander.utils import generate_uid
 from keepercommander.service.util.protected_records import (
     _attachment_file_uids,
-    _expand_with_attachment_uids,
-    _has_reserved_attachment,
+    _has_reserved_legacy_attachment,
     get_protected_folder_uids,
     get_protected_record_title_set,
     get_protected_record_uids,
@@ -465,122 +464,92 @@ def test_a_resync_mid_block_self_heals_without_reintroducing_the_folder(self):
             self.assertNotIn('FOLDER1', p.root_folder.subfolders)
 
 
-class TestHasReservedAttachment(TestCase):
+class TestHasReservedLegacyAttachment(TestCase):
     def test_password_record_with_reserved_attachment_name(self):
         record = vault.PasswordRecord()
         record.attachments = [vault.AttachmentFile({'name': 'config.json', 'title': 'config.json'})]
-        self.assertTrue(_has_reserved_attachment(None, record))
+        self.assertTrue(_has_reserved_legacy_attachment(record))
 
     def test_password_record_with_reserved_title_but_different_name(self):
         """attachment.py itself checks title OR name -- match either."""
         record = vault.PasswordRecord()
         record.attachments = [vault.AttachmentFile({'name': 'file123', 'title': 'service_config.json'})]
-        self.assertTrue(_has_reserved_attachment(None, record))
+        self.assertTrue(_has_reserved_legacy_attachment(record))
 
     def test_password_record_with_unrelated_attachment(self):
         record = vault.PasswordRecord()
         record.attachments = [vault.AttachmentFile({'name': 'notes.pdf', 'title': 'notes.pdf'})]
-        self.assertFalse(_has_reserved_attachment(None, record))
+        self.assertFalse(_has_reserved_legacy_attachment(record))
 
     def test_password_record_with_no_attachments(self):
-        record = vault.PasswordRecord()
-        self.assertFalse(_has_reserved_attachment(None, record))
+        self.assertFalse(_has_reserved_legacy_attachment(vault.PasswordRecord()))
 
-    def test_typed_record_with_reserved_file_ref(self):
-        record = vault.TypedRecord()
-        record.fields = [vault.TypedField({'type': 'fileRef', 'value': ['FILE_UID_1']})]
-        file_record = vault.FileRecord()
-        file_record.title = 'config.json'
-        file_record.name = 'config.json'
-        with mock.patch('keepercommander.vault.KeeperRecord.load', return_value=file_record):
-            self.assertTrue(_has_reserved_attachment(object(), record))
+    def test_non_password_record_is_always_false(self):
+        """TypedRecord's fileRef attachments are handled inline in get_protected_record_uids
+        (via the reserved_file_uids/pending_attachments cross-reference), not here."""
+        self.assertFalse(_has_reserved_legacy_attachment(vault.TypedRecord()))
+        self.assertFalse(_has_reserved_legacy_attachment(vault.FileRecord()))
 
-    def test_typed_record_with_unrelated_file_ref(self):
-        record = vault.TypedRecord()
-        record.fields = [vault.TypedField({'type': 'fileRef', 'value': ['FILE_UID_1']})]
-        file_record = vault.FileRecord()
-        file_record.title = 'notes.pdf'
-        file_record.name = 'notes.pdf'
-        with mock.patch('keepercommander.vault.KeeperRecord.load', return_value=file_record):
-            self.assertFalse(_has_reserved_attachment(object(), record))
 
-    def test_typed_record_with_no_file_ref(self):
-        record = vault.TypedRecord()
-        self.assertFalse(_has_reserved_attachment(object(), record))
+class TestGetProtectedRecordUidsWithReservedAttachments(TestCase):
+    """Records are loaded exactly once each -- attachment detection must not add extra
+    per-attachment KeeperRecord.load calls (previously N extra loads per fileRef attachment)."""
 
-    def test_neither_password_nor_typed_record(self):
-        record = vault.FileRecord()
-        self.assertFalse(_has_reserved_attachment(None, record))
+    @staticmethod
+    def _params_with(records: dict):
+        """records: {uid: KeeperRecord-like object}, each already carrying its own .record_uid."""
+        p = _params_with_records({uid: 'placeholder' for uid in records})
+        return p, records
 
+    def test_arbitrary_titled_typed_record_with_reserved_file_ref_is_protected(self):
+        parent = vault.TypedRecord()
+        parent.record_uid = 'PARENT_UID'
+        parent.title = 'My Totally Unrelated Title'
+        parent.fields = [vault.TypedField({'type': 'fileRef', 'value': ['FILE_UID_1']})]
 
-class TestGetProtectedRecordUidsWithReservedAttachments(TestCase):
-    def test_arbitrary_titled_record_with_reserved_attachment_is_protected(self):
-        """A record with none of the known titles is still protected if it carries
-        one of Commander's own reserved config-file attachments."""
-        p = _params_with_records({'UID_ARBITRARY': 'My Totally Unrelated Title'})
-        with mock.patch(
-            'keepercommander.service.util.protected_records._has_reserved_attachment',
-            side_effect=lambda params, record: record.record_uid == 'UID_ARBITRARY',
-        ):
-            result = get_protected_record_uids(p)
-        self.assertIn('UID_ARBITRARY', result)
+        file_record = vault.FileRecord()
+        file_record.record_uid = 'FILE_UID_1'
+        file_record.title = 'service_config.json'
+        file_record.name = 'service_config.json'
 
-    def test_arbitrary_titled_record_without_reserved_attachment_is_not_protected(self):
-        p = _params_with_records({'UID_ARBITRARY': 'My Totally Unrelated Title'})
-        with mock.patch(
-            'keepercommander.service.util.protected_records._has_reserved_attachment',
-            return_value=False,
-        ):
+        p, records = self._params_with({'PARENT_UID': parent, 'FILE_UID_1': file_record})
+        with mock.patch('keepercommander.vault.KeeperRecord.load', side_effect=lambda params, uid: records.get(uid)):
             result = get_protected_record_uids(p)
-        self.assertEqual(result, {})
+        self.assertIn('PARENT_UID', result)
+        self.assertIn('FILE_UID_1', result)
 
+    def test_arbitrary_titled_password_record_with_reserved_attachment_is_protected(self):
+        parent = vault.PasswordRecord()
+        parent.record_uid = 'PARENT_UID'
+        parent.title = 'My Totally Unrelated Title'
+        parent.attachments = [vault.AttachmentFile({'id': 'ATTA_1', 'name': 'config.json'})]
 
-class TestExpandWithAttachmentUids(TestCase):
-    """Regression test for a live gap: get 'service_config.json' returned the attachment's own
-    FileRecord (a separate record_cache entry, its own UID) directly, since only the parent
-    record was ever added to protected_uids -- the attachment's own UID was never blocked."""
+        p, records = self._params_with({'PARENT_UID': parent})
+        with mock.patch('keepercommander.vault.KeeperRecord.load', side_effect=lambda params, uid: records.get(uid)):
+            result = get_protected_record_uids(p)
+        self.assertIn('PARENT_UID', result)
+        self.assertIn('ATTA_1', result)
 
-    def test_typed_record_file_ref_uid_is_protected(self):
+    def test_unrelated_attachment_name_is_not_protected(self):
         parent = vault.TypedRecord()
         parent.record_uid = 'PARENT_UID'
+        parent.title = 'My Totally Unrelated Title'
         parent.fields = [vault.TypedField({'type': 'fileRef', 'value': ['FILE_UID_1']})]
-        found = {'PARENT_UID': 'Commander Service Mode Slack App Config'}
 
-        p = _params_with_records({'PARENT_UID': 'x', 'FILE_UID_1': 'y'})
-        with mock.patch('keepercommander.vault.KeeperRecord.load', return_value=parent):
-            _expand_with_attachment_uids(p, found)
-        self.assertIn('FILE_UID_1', found)
+        file_record = vault.FileRecord()
+        file_record.record_uid = 'FILE_UID_1'
+        file_record.title = 'notes.pdf'
+        file_record.name = 'notes.pdf'
 
-    def test_legacy_password_record_attachment_id_is_protected(self):
-        parent = vault.PasswordRecord()
-        parent.record_uid = 'PARENT_UID'
-        parent.attachments = [vault.AttachmentFile({'id': 'ATTA_ID_1', 'name': 'config.json'})]
-        found = {'PARENT_UID': 'Commander Service Mode Docker Config'}
-
-        p = _params_with_records({'PARENT_UID': 'x'})
-        with mock.patch('keepercommander.vault.KeeperRecord.load', return_value=parent):
-            _expand_with_attachment_uids(p, found)
-        self.assertIn('ATTA_ID_1', found)
-
-    def test_noop_without_record_cache(self):
-        found = {'PARENT_UID': 'x'}
-        _expand_with_attachment_uids(params_module.KeeperParams(), found)
-        self.assertEqual(found, {'PARENT_UID': 'x'})
-
-    def test_noop_when_protected_uid_is_not_loadable(self):
-        found = {'PARENT_UID': 'x'}
-        p = _params_with_records({'PARENT_UID': 'x'})
-        with mock.patch('keepercommander.vault.KeeperRecord.load', return_value=None):
-            _expand_with_attachment_uids(p, found)
-        self.assertEqual(found, {'PARENT_UID': 'x'})
-
-    def test_end_to_end_arbitrary_titled_parent_with_attachment_protects_the_file_uid(self):
-        """Reproduces the exact reported scenario: a record with a normal title carrying a
-        service_config.json fileRef attachment -- both the parent and the attachment's own
-        FileRecord UID must come back protected."""
+        p, records = self._params_with({'PARENT_UID': parent, 'FILE_UID_1': file_record})
+        with mock.patch('keepercommander.vault.KeeperRecord.load', side_effect=lambda params, uid: records.get(uid)):
+            result = get_protected_record_uids(p)
+        self.assertEqual(result, {})
+
+    def test_reserved_attachment_on_an_already_title_protected_record_is_still_swept_in(self):
         parent = vault.TypedRecord()
         parent.record_uid = 'PARENT_UID'
-        parent.title = 'My Totally Unrelated Title'
+        parent.title = PROTECTED_TITLE
         parent.fields = [vault.TypedField({'type': 'fileRef', 'value': ['FILE_UID_1']})]
 
         file_record = vault.FileRecord()
@@ -588,13 +557,36 @@ def test_end_to_end_arbitrary_titled_parent_with_attachment_protects_the_file_ui
         file_record.title = 'service_config.json'
         file_record.name = 'service_config.json'
 
-        records = {'PARENT_UID': parent, 'FILE_UID_1': file_record}
-        p = _params_with_records({'PARENT_UID': 'x', 'FILE_UID_1': 'y'})
+        p, records = self._params_with({'PARENT_UID': parent, 'FILE_UID_1': file_record})
         with mock.patch('keepercommander.vault.KeeperRecord.load', side_effect=lambda params, uid: records.get(uid)):
             result = get_protected_record_uids(p)
         self.assertIn('PARENT_UID', result)
         self.assertIn('FILE_UID_1', result)
 
+    def test_load_is_called_exactly_once_per_record_cache_entry(self):
+        """Regression test for the N-vs-3N perf issue: attachment detection must not add
+        extra per-attachment loads on top of the one load every record already gets."""
+        parent = vault.TypedRecord()
+        parent.record_uid = 'PARENT_UID'
+        parent.title = 'Unrelated'
+        parent.fields = [vault.TypedField({'type': 'fileRef', 'value': ['FILE_UID_1', 'FILE_UID_2']})]
+
+        file_record_1 = vault.FileRecord()
+        file_record_1.record_uid = 'FILE_UID_1'
+        file_record_1.title = 'notes.pdf'
+        file_record_2 = vault.FileRecord()
+        file_record_2.record_uid = 'FILE_UID_2'
+        file_record_2.title = 'photo.png'
+
+        p, records = self._params_with(
+            {'PARENT_UID': parent, 'FILE_UID_1': file_record_1, 'FILE_UID_2': file_record_2}
+        )
+        with mock.patch(
+            'keepercommander.vault.KeeperRecord.load', side_effect=lambda params, uid: records.get(uid)
+        ) as mock_load:
+            get_protected_record_uids(p)
+        self.assertEqual(mock_load.call_count, len(records))
+
 
 class TestAttachmentFileUids(TestCase):
     def test_password_record_returns_attachment_ids(self):