KC-1470: Protect Integration config records from Service Mode API access

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:

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,
     }

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()
 

keepercommander/service/decorators/min_commander_version.py

@@ -23,8 +23,11 @@
 
 # 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'
+# 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]:

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
+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
@@ -195,15 +201,28 @@ 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,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
+                }
+
+            # 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)

keepercommander/service/util/protected_records.py

@@ -9,27 +9,76 @@
 # 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, Tuple
+from typing import Dict, FrozenSet, Iterable, Optional, Set, 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')
 
+# 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')
+
+# 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_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
+
+    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, ...]:
     """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,31 +87,99 @@ 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."""
-    found: 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."""
+    from ..commands.integrations.approvals_setup import is_valid_keeper_uid
+    from ..decorators.logging import logger
 
-    docker_uid = (os.environ.get(_DOCKER_RECORD_UID_ENV) or '').strip()
-    if docker_uid:
-        found[docker_uid] = '<Docker config record>'
+    found: Dict[str, str] = {}
+    for env_name, label in _PINNED_RECORD_UID_ENVS.items():
+        uid = (os.environ.get(env_name) or '').strip()
+        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()
+    # 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)
         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 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_legacy_attachment(record):
+            found[uid] = '<record with a reserved config attachment>'
+
+        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
 
 
+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
+    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,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_commands().get(command_tokens[0].lower())
+    if not env_name:
+        return None
+
+    if '--sync-down' not 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/|=."""
 
@@ -131,3 +248,66 @@ 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)
+    # {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 = {}
+
+    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
+
+        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, index) in removed_from_parents.items():
+            try:
+                if uid not in subfolders:
+                    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__})')