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/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/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..64f336b78 100644 --- a/keepercommander/service/decorators/min_commander_version.py +++ b/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]: diff --git a/keepercommander/service/util/command_util.py b/keepercommander/service/util/command_util.py index 72180e83d..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 +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: '' 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 dc5ff0497..7d1b18c4b 100644 --- a/keepercommander/service/util/protected_records.py +++ b/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': '', + 'TERRAFORM_RECORD': '', + 'SLACK_RECORD': '', + 'TEAMS_RECORD': '', + 'GCHAT_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] = '' + 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] = '' + + 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] = '' + if parent_uid in found: + for file_uid in file_uids: + found.setdefault(file_uid, '') + 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__})') diff --git a/unit-tests/service/test_command.py b/unit-tests/service/test_command.py index fdd3415f5..de0736de8 100644 --- a/unit-tests/service/test_command.py +++ b/unit-tests/service/test_command.py @@ -3,7 +3,8 @@ 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 from keepercommander.service.util.parse_keeper_response import parse_keeper_response @@ -271,4 +272,334 @@ 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() + + 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() + +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) + + +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 + + @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.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: + 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() + 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): + 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.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') + + 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_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 0d25fdb24..9fe2a6940 100644 --- a/unit-tests/service/test_protected_records.py +++ b/unit-tests/service/test_protected_records.py @@ -2,11 +2,18 @@ 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, + _has_reserved_legacy_attachment, + 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, ) PROTECTED_TITLE = 'Commander Service Mode Config' @@ -38,6 +45,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): @@ -77,14 +91,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): @@ -92,6 +108,43 @@ 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): + 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): + 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): + 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): + 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): + 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 +288,320 @@ 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([])) + + 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.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) + + +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_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_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_legacy_attachment(record)) + + def test_password_record_with_no_attachments(self): + self.assertFalse(_has_reserved_legacy_attachment(vault.PasswordRecord())) + + 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())) + + +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).""" + + @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']})] + + file_record = vault.FileRecord() + file_record.record_uid = 'FILE_UID_1' + file_record.title = 'service_config.json' + file_record.name = 'service_config.json' + + 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_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'})] + + 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_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']})] + + file_record = vault.FileRecord() + file_record.record_uid = 'FILE_UID_1' + file_record.title = 'notes.pdf' + file_record.name = 'notes.pdf' + + 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 = PROTECTED_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' + + 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): + 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()), []) diff --git a/unit-tests/service/test_runtime_policy.py b/unit-tests/service/test_runtime_policy.py index debdb4724..04c1d414e 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, TERRAFORM_DOCKER_ENV_LEGACY from keepercommander.service.util.exceptions import ValidationError @@ -67,7 +68,17 @@ 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( + 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( @@ -85,7 +96,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' 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'):