diff --git a/keepercommander/service/util/command_util.py b/keepercommander/service/util/command_util.py index 910c2c42f..72180e83d 100644 --- a/keepercommander/service/util/command_util.py +++ b/keepercommander/service/util/command_util.py @@ -25,6 +25,7 @@ is_throttle_error, throttle_error_response, ) +from .protected_records import get_protected_record_uids, hide_from_record_cache 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 @@ -172,16 +173,18 @@ def execute(cls, command: str, temp_files: Optional[list] = None) -> Tuple[Any, # Mode will treat as safe, not the whole shared OS temp root. request_temp_dir = os.path.dirname(temp_files[0]) if temp_files else None + def blocked(error): + logger.warning( + f"Service Mode blocked command '{command_tokens[0] if command_tokens else ''}': {error}" + ) + return {"status": "error", "error": error}, 403 + # Same tokens the CLI will run — do not use raw HTTP split(" ") service_mode_error = Verifycommand.validate_service_mode_restrictions( command_tokens, request_temp_dir ) if service_mode_error: - logger.warning( - f"Service Mode blocked command '{command_tokens[0] if command_tokens else ''}': " - f"{service_mode_error}" - ) - return {"status": "error", "error": service_mode_error}, 403 + return blocked(service_mode_error) force_error = Verifycommand.validate_enterprise_user_add_role_force( command_tokens, params @@ -189,16 +192,27 @@ def execute(cls, command: str, temp_files: Optional[list] = None) -> Tuple[Any, if force_error: return {"status": "error", "error": force_error}, 400 + # 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) + protected_command_error = Verifycommand.validate_service_mode_protected_record_command( + command_tokens, protected_uids + ) + if protected_command_error: + return blocked(protected_command_error) + sailpoint_enabled = bool((os.environ.get('SAILPOINT_RECORD') or '').strip()) - if sailpoint_enabled: - from ..commands.integrations.sailpoint.service import SailPointService - command, sailpoint_response = SailPointService.handle_command(params, command) - if sailpoint_response is not None: - response, status_code = sailpoint_response - response = CommandExecutor.encrypt_response(response) - return response, status_code - - return_value, printed_output, log_output = CommandExecutor.capture_output_and_logs(params, command) + + with hide_from_record_cache(params, protected_uids): + if sailpoint_enabled: + from ..commands.integrations.sailpoint.service import SailPointService + command, sailpoint_response = SailPointService.handle_command(params, command) + if sailpoint_response is not None: + response, status_code = sailpoint_response + response = CommandExecutor.encrypt_response(response) + return response, status_code + + return_value, printed_output, log_output = CommandExecutor.capture_output_and_logs(params, command) response = return_value if return_value else printed_output # Debug logging with sanitization diff --git a/keepercommander/service/util/protected_records.py b/keepercommander/service/util/protected_records.py new file mode 100644 index 000000000..dc5ff0497 --- /dev/null +++ b/keepercommander/service/util/protected_records.py @@ -0,0 +1,133 @@ +# _ __ +# | |/ /___ ___ _ __ ___ _ _ ® +# | ' 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) + + +def get_protected_record_title_set() -> FrozenSet[str]: + """Lower-cased titles of Service Mode's own config records, for literal matching.""" + return frozenset(t.lower() for t in _protected_titles()) + + +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] = {} + + docker_uid = (os.environ.get(_DOCKER_RECORD_UID_ENV) or '').strip() + if docker_uid: + found[docker_uid] = '' + + 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: + 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: + found[uid] = record.title + return found + + +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/|=.""" + + def __init__(self, source, protected_uids: Iterable[str]): + self._protected_uids = frozenset(protected_uids) + super().__init__({k: v for k, v in source.items() if k not in self._protected_uids}) + + def __setitem__(self, key, value): + if key in self._protected_uids: + return + super().__setitem__(key, value) + + +@contextlib.contextmanager +def hide_from_record_cache(params, protected_uids: Dict[str, str]): + """For the with-block, guards record_cache/nested_share_records/nested_share_record_data against reintroduction (not just a one-time pop) and strips protected UIDs from subfolder_record_cache, restoring everything on exit; relies on Service Mode commands running one at a time (same assumption capture_output_and_logs already makes) and may leave record_cache stale until the next sync if one lands mid-command.""" + if params is None or not protected_uids: + yield + return + + protected_uid_set = frozenset(protected_uids) + + original_caches = {} + saved_entries = {} + for attr in _GUARDED_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)) + + subfolder_cache = getattr(params, 'subfolder_record_cache', None) + removed_from_folders: Dict[str, set] = {} + if isinstance(subfolder_cache, dict): + for folder_uid, uids in subfolder_cache.items(): + if not isinstance(uids, set): + continue + hit = uids & protected_uid_set + if hit: + removed_from_folders[folder_uid] = hit + uids -= hit + + 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_record_cache: failed to restore {attr} ({type(e).__name__}); restoring protected entries only') + try: + setattr(params, attr, dict(saved_entries[attr])) + except Exception: + pass + + if isinstance(subfolder_cache, dict): + for folder_uid, hit in removed_from_folders.items(): + try: + uids = subfolder_cache.get(folder_uid) + if isinstance(uids, set): + uids |= hit + except Exception as e: + logger.debug(f'hide_from_record_cache: failed to restore subfolder {folder_uid} ({type(e).__name__})') diff --git a/keepercommander/service/util/verified_command.py b/keepercommander/service/util/verified_command.py index 20bbb6fcb..9d8ea389b 100644 --- a/keepercommander/service/util/verified_command.py +++ b/keepercommander/service/util/verified_command.py @@ -17,6 +17,10 @@ class Verifycommand: # Aliases from record.py — CommandExecutor checks tokens before cli expands them. _RECORD_EDIT_COMMANDS = frozenset({'record-add', 'ra', 'record-update', 'ru'}) + _PROTECTED_RECORD_MSG = ( + 'Service Mode configuration records are not accessible through Service Mode' + ) + # Legacy Commands category — plugin-based rotation/connection commands have no safe Service Mode form # and are blocked unconditionally, regardless of what an API key's command_list allows. _LEGACY_COMMANDS = frozenset({ @@ -99,6 +103,33 @@ def validate_service_mode_restrictions(command_tokens, request_temp_dir=None): return error return None + @staticmethod + def _record_reference_candidates(tok): + """tok itself, plus its value if tok is a --flag=value (or -f=value) option.""" + if '=' in tok: + _, _, value = tok.partition('=') + if value: + return (tok, value) + return (tok,) + + @staticmethod + def validate_service_mode_protected_record_command(command_tokens, protected_uids=None): + """Reject any command with a protected title/UID as a whole token or --flag=value; indirect forms (comma lists, path-qualified titles) rely on protected_records.hide_from_record_cache instead.""" + if not command_tokens: + return None + + from .protected_records import get_protected_record_title_set + protected_titles = get_protected_record_title_set() + + uid_set = set(protected_uids) if protected_uids else set() + for tok in command_tokens[1:]: + for candidate in Verifycommand._record_reference_candidates(tok): + if candidate.lower() in protected_titles: + return Verifycommand._PROTECTED_RECORD_MSG + if candidate in uid_set: + return Verifycommand._PROTECTED_RECORD_MSG + return None + @staticmethod def validate_service_mode_double_dash(command_tokens, request_temp_dir=None): """Block bare '--' anywhere in Service Mode input; error or None.""" diff --git a/unit-tests/service/test_command.py b/unit-tests/service/test_command.py index 2df0637b1..fdd3415f5 100644 --- a/unit-tests/service/test_command.py +++ b/unit-tests/service/test_command.py @@ -1,10 +1,39 @@ +import json import unittest from unittest import TestCase, mock from flask import Flask +from keepercommander import params as params_module 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 +from keepercommander.service.util.protected_records import get_protected_record_uids + +PROTECTED_TITLE = 'Commander Service Mode Config' +PROTECTED_UID = 'PROTECTED_CONFIG_UID' +NORMAL_UID = 'NORMAL_RECORD_UID' + + +def _record_cache_entry(uid, title): + return { + 'record_uid': uid, + 'version': 2, + 'revision': 1, + 'client_modified_time': 0, + 'shared': False, + 'record_key_unencrypted': b'0' * 32, + 'data_unencrypted': json.dumps({'title': title}).encode('utf-8'), + } + + +def _params_with_protected_and_normal_record(): + 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'), + } + return p class TestCommandAPI(TestCase): def setUp(self): @@ -115,4 +144,131 @@ def test_integration_command_flow(self): response, status_code = CommandExecutor.execute(test_command) self.assertEqual(status_code, 200) - self.assertIsNotNone(response) \ No newline at end of file + self.assertIsNotNone(response) + + +class TestProtectedRecordCommandExecution(TestCase): + """End-to-end proof, through CommandExecutor.execute, that Service Mode's own config record stays blocked while unrelated records work.""" + + def _run(self, command, params=None): + params = params or _params_with_protected_and_normal_record() + 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, params + + def test_blocked_commands_never_reach_cli_dispatch(self): + for command in ( + f'get {PROTECTED_UID}', + f'get "{PROTECTED_TITLE}"', + f'list {PROTECTED_UID}', + f'search {PROTECTED_UID}', + f'record-update --record {PROTECTED_UID} title=x', + f'record-update --record={PROTECTED_UID} title=x', + f'rm {PROTECTED_UID}', + f'share-record {PROTECTED_UID} --email a@b.com', + f'share-folder --record {PROTECTED_UID} -e a@b.com', + f'share-folder --record={PROTECTED_UID} -e a@b.com', + f'ls "{PROTECTED_TITLE}"', + f'tree "{PROTECTED_TITLE}"', + f'nsf-get {PROTECTED_UID}', + # Not one of the commands anyone would think to curate a list around -- + # proves the check isn't gated by command name at all. + f'keep-alive {PROTECTED_UID}', + ): + with self.subTest(command=command): + response, status_code, mock_capture, _ = self._run(command) + self.assertEqual(status_code, 403) + self.assertEqual(response.get('status'), 'error') + mock_capture.assert_not_called() + + def test_unrelated_record_commands_are_unaffected(self): + for command in ( + f'get {NORMAL_UID}', + f'list {NORMAL_UID}', + f'search {NORMAL_UID}', + f'record-update --record {NORMAL_UID} title=x', + f'rm {NORMAL_UID}', + f'share-record {NORMAL_UID} --email a@b.com', + ): + with self.subTest(command=command): + response, status_code, mock_capture, _ = self._run(command) + self.assertEqual(status_code, 200) + mock_capture.assert_called_once() + + def test_record_cache_is_popped_during_dispatch_and_restored_after(self): + params = _params_with_protected_and_normal_record() + seen_during_call = {} + + def fake_capture(p, command): + seen_during_call['keys'] = set(p.record_cache.keys()) + 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'get {NORMAL_UID}') + + self.assertEqual(status_code, 200) + self.assertNotIn(PROTECTED_UID, seen_during_call['keys']) + self.assertIn(NORMAL_UID, seen_during_call['keys']) + # Restored after the call returns -- the shared params singleton must + # not stay permanently blind to its own config record. + self.assertIn(PROTECTED_UID, params.record_cache) + + def test_record_cache_restored_even_if_dispatch_raises(self): + params = _params_with_protected_and_normal_record() + with mock.patch( + 'keepercommander.service.core.globals.ensure_params_loaded', return_value=params + ), mock.patch.object( + CommandExecutor, 'capture_output_and_logs', side_effect=CommandExecutionError('boom') + ): + response, status_code = CommandExecutor.execute(f'get {NORMAL_UID}') + + self.assertEqual(status_code, 400) + self.assertIn(PROTECTED_UID, params.record_cache) + + def test_protected_record_check_runs_for_every_command(self): + """The check is unconditional -- even a command with no known relationship + to records (whoami) still triggers it, so no command can be missed.""" + params = _params_with_protected_and_normal_record() + 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', '') + ), mock.patch( + 'keepercommander.service.util.command_util.get_protected_record_uids', + wraps=get_protected_record_uids, + ) as mock_get_uids: + CommandExecutor.execute('whoami') + mock_get_uids.assert_called_once() + + def test_sailpoint_handling_runs_inside_the_record_cache_guard(self): + """SailPoint's own pre-processing (handle_command) can resolve/act on + records before cli.do_command ever runs -- it must run with the guard + already active, not before it, or a folder/recursive share under + SailPoint mode could reach the protected record before it's hidden.""" + params = _params_with_protected_and_normal_record() + seen_during_handle_command = {} + + def fake_handle_command(p, command): + seen_during_handle_command['keys'] = set(p.record_cache.keys()) + return command, None + + with mock.patch( + 'keepercommander.service.core.globals.ensure_params_loaded', return_value=params + ), mock.patch.dict('os.environ', {'SAILPOINT_RECORD': 'sailpoint-uid'}), mock.patch( + 'keepercommander.service.commands.integrations.sailpoint.service.SailPointService.handle_command', + side_effect=fake_handle_command, + ), mock.patch.object( + CommandExecutor, 'capture_output_and_logs', return_value=('ok', 'ok', '') + ): + CommandExecutor.execute(f'get {NORMAL_UID}') + + 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 diff --git a/unit-tests/service/test_protected_records.py b/unit-tests/service/test_protected_records.py new file mode 100644 index 000000000..0d25fdb24 --- /dev/null +++ b/unit-tests/service/test_protected_records.py @@ -0,0 +1,237 @@ +import json +import os +from unittest import TestCase, mock + +from keepercommander import params as params_module +from keepercommander.service.util.protected_records import ( + get_protected_record_title_set, + get_protected_record_uids, + hide_from_record_cache, +) + +PROTECTED_TITLE = 'Commander Service Mode Config' +PROTECTED_DOCKER_TITLE = 'Commander Service Mode Docker Config' + + +def _record_cache_entry(uid, title, version=2): + return { + 'record_uid': uid, + 'version': version, + 'revision': 1, + 'client_modified_time': 0, + 'shared': False, + 'record_key_unencrypted': b'0' * 32, + 'data_unencrypted': json.dumps({'title': title}).encode('utf-8'), + } + + +def _params_with_records(entries): + p = params_module.KeeperParams() + p.record_cache = {uid: _record_cache_entry(uid, title) for uid, title in entries.items()} + return p + + +class TestGetProtectedRecordTitleSet(TestCase): + def test_contains_expected_titles_lowercased(self): + titles = get_protected_record_title_set() + self.assertIn('commander service mode config', titles) + self.assertIn('commander service mode docker config', titles) + self.assertIn('commander service mode', titles) + + +class TestGetProtectedRecordUids(TestCase): + def test_returns_only_matching_protected_records(self): + p = _params_with_records({ + 'UID_CONFIG': PROTECTED_TITLE, + 'UID_DOCKER': PROTECTED_DOCKER_TITLE, + 'UID_OTHER': 'My Normal Record', + }) + result = get_protected_record_uids(p) + self.assertEqual(set(result.keys()), {'UID_CONFIG', 'UID_DOCKER'}) + self.assertEqual(result['UID_CONFIG'], PROTECTED_TITLE) + self.assertEqual(result['UID_DOCKER'], PROTECTED_DOCKER_TITLE) + + def test_no_protected_records_present(self): + p = _params_with_records({'UID_OTHER': 'My Normal Record'}) + self.assertEqual(get_protected_record_uids(p), {}) + + def test_empty_record_cache(self): + p = params_module.KeeperParams() + p.record_cache = {} + self.assertEqual(get_protected_record_uids(p), {}) + + def test_params_none(self): + self.assertEqual(get_protected_record_uids(None), {}) + + def test_record_cache_not_a_dict_is_ignored(self): + """A Mock/non-dict record_cache (as some tests construct) must fail safe to {}.""" + class FakeParams: + record_cache = object() + + self.assertEqual(get_protected_record_uids(FakeParams()), {}) + + def test_similar_but_not_exact_title_is_not_matched(self): + """A title that shares every token with a protected title but isn't an exact match must not be protected.""" + p = _params_with_records({'UID_OTHER': 'Commander Service Mode Config Backup'}) + self.assertEqual(get_protected_record_uids(p), {}) + + 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'}) + result = get_protected_record_uids(p) + self.assertIn('DOCKER_CUSTOM_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)) + + def test_no_docker_env_var_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}) + p.record_cache['MALFORMED_UID'] = { + 'record_uid': 'MALFORMED_UID', 'version': 2, 'revision': 1, + 'data_unencrypted': json.dumps({'title': 'whatever'}).encode('utf-8'), + # record_key_unencrypted deliberately missing. + } + result = get_protected_record_uids(p) + self.assertEqual(set(result.keys()), {'UID_CONFIG'}) + + +class TestGetProtectedRecordUidsNotCached(TestCase): + """Not memoized -- a newly added protected record must be picked up immediately, without needing a revision change.""" + + def test_newly_added_record_is_found_without_a_revision_change(self): + p = _params_with_records({'UID_CONFIG': PROTECTED_TITLE}) + p.revision = 100 + first = get_protected_record_uids(p) + self.assertEqual(set(first.keys()), {'UID_CONFIG'}) + + p.record_cache['UID_DOCKER'] = _record_cache_entry('UID_DOCKER', PROTECTED_DOCKER_TITLE) + second = get_protected_record_uids(p) + self.assertEqual(set(second.keys()), {'UID_CONFIG', 'UID_DOCKER'}) + + def test_removed_record_is_no_longer_found(self): + p = _params_with_records({'UID_CONFIG': PROTECTED_TITLE, 'UID_DOCKER': PROTECTED_DOCKER_TITLE}) + p.revision = 100 + first = get_protected_record_uids(p) + self.assertEqual(set(first.keys()), {'UID_CONFIG', 'UID_DOCKER'}) + + del p.record_cache['UID_DOCKER'] + second = get_protected_record_uids(p) + self.assertEqual(set(second.keys()), {'UID_CONFIG'}) + + +class TestHideFromRecordCache(TestCase): + def test_hides_protected_uid_inside_the_block(self): + p = _params_with_records({'PROTECTED': PROTECTED_TITLE, 'NORMAL': 'Other'}) + with hide_from_record_cache(p, {'PROTECTED': PROTECTED_TITLE}): + self.assertNotIn('PROTECTED', p.record_cache) + self.assertIn('NORMAL', p.record_cache) + + def test_restores_original_entry_and_plain_dict_after_block(self): + p = _params_with_records({'PROTECTED': PROTECTED_TITLE, 'NORMAL': 'Other'}) + original_entry = p.record_cache['PROTECTED'] + with hide_from_record_cache(p, {'PROTECTED': PROTECTED_TITLE}): + pass + self.assertIs(type(p.record_cache), dict) + self.assertEqual(p.record_cache['PROTECTED'], original_entry) + + def test_restores_even_if_block_raises(self): + p = _params_with_records({'PROTECTED': PROTECTED_TITLE}) + with self.assertRaises(ValueError): + with hide_from_record_cache(p, {'PROTECTED': PROTECTED_TITLE}): + raise ValueError('boom') + self.assertIs(type(p.record_cache), dict) + self.assertIn('PROTECTED', p.record_cache) + + def test_reintroduction_during_block_is_blocked(self): + """A forced sync-down writing the protected UID back into record_cache mid-command must not make it visible.""" + p = _params_with_records({'PROTECTED': PROTECTED_TITLE, 'NORMAL': 'Other'}) + with hide_from_record_cache(p, {'PROTECTED': PROTECTED_TITLE}): + p.record_cache['PROTECTED'] = _record_cache_entry('PROTECTED', 'reintroduced') + self.assertNotIn('PROTECTED', p.record_cache) + + p.record_cache.update({'PROTECTED': _record_cache_entry('PROTECTED', 'via-update'), + 'NEW': _record_cache_entry('NEW', 'legit new record')}) + self.assertNotIn('PROTECTED', p.record_cache) + self.assertIn('NEW', p.record_cache) + + # Legit mutations made during the block survive; the protected entry is restored. + self.assertIn('NEW', p.record_cache) + self.assertIn('PROTECTED', p.record_cache) + + def test_no_protected_uids_is_a_noop(self): + p = _params_with_records({'NORMAL': 'Other'}) + with hide_from_record_cache(p, {}): + self.assertIn('NORMAL', p.record_cache) + + def test_params_none_is_a_noop(self): + with hide_from_record_cache(None, {'PROTECTED': PROTECTED_TITLE}): + pass + + def test_nested_share_caches_are_guarded_too(self): + """load_pam_record falls back to nested_share_records, so guarding record_cache alone isn't enough.""" + p = _params_with_records({'NORMAL': 'Other'}) + p.nested_share_records = {'PROTECTED': {'title': 'nsf copy'}, 'OTHER_NSF': {'title': 'x'}} + p.nested_share_record_data = {'PROTECTED': {'data_json': {'title': 'nsf data copy'}}} + + with hide_from_record_cache(p, {'PROTECTED': PROTECTED_TITLE}): + self.assertNotIn('PROTECTED', p.nested_share_records) + self.assertIn('OTHER_NSF', p.nested_share_records) + self.assertNotIn('PROTECTED', p.nested_share_record_data) + + # Reintroduction mid-command must be blocked here too. + p.nested_share_records['PROTECTED'] = {'title': 'reintroduced'} + self.assertNotIn('PROTECTED', p.nested_share_records) + + self.assertIn('PROTECTED', p.nested_share_records) + self.assertIn('PROTECTED', p.nested_share_record_data) + self.assertIs(type(p.nested_share_records), dict) + + def test_subfolder_record_cache_uid_is_stripped_for_the_duration(self): + """_build_folder_json leaks the bare UID from a folder's set even when the record fails to load.""" + p = _params_with_records({'NORMAL': 'Other'}) + p.subfolder_record_cache = {'FOLDER1': {'PROTECTED', 'NORMAL'}, 'FOLDER2': {'OTHER'}} + + with hide_from_record_cache(p, {'PROTECTED': PROTECTED_TITLE}): + self.assertNotIn('PROTECTED', p.subfolder_record_cache['FOLDER1']) + self.assertIn('NORMAL', p.subfolder_record_cache['FOLDER1']) + self.assertEqual(p.subfolder_record_cache['FOLDER2'], {'OTHER'}) + + self.assertIn('PROTECTED', p.subfolder_record_cache['FOLDER1']) + self.assertIn('NORMAL', p.subfolder_record_cache['FOLDER1']) + + def test_missing_optional_caches_do_not_raise(self): + """A params fixture with only default-empty NSF/subfolder caches must not break the guard.""" + p = _params_with_records({'NORMAL': 'Other'}) + with hide_from_record_cache(p, {'PROTECTED': PROTECTED_TITLE}): + pass + + def test_attribute_replaced_with_none_mid_command_does_not_mask_the_original_error(self): + """A command that clears/replaces a guarded attribute mid-execution must not + turn a real error into a confusing 'NoneType is not iterable' one, and the + protected entry must still be restored on a best-effort basis.""" + p = _params_with_records({'PROTECTED': PROTECTED_TITLE}) + with self.assertRaises(ValueError) as ctx: + with hide_from_record_cache(p, {'PROTECTED': PROTECTED_TITLE}): + p.record_cache = None + raise ValueError('original command error') + self.assertEqual(str(ctx.exception), 'original command error') + self.assertIn('PROTECTED', p.record_cache) + + def test_attribute_replaced_with_plain_dict_preserves_its_contents(self): + """A command that replaces the guarded attribute with a fresh plain dict + (not just mutating the guarded one) must not lose that dict's entries.""" + p = _params_with_records({'PROTECTED': PROTECTED_TITLE, 'NORMAL': 'Other'}) + with hide_from_record_cache(p, {'PROTECTED': PROTECTED_TITLE}): + p.record_cache = {'REPLACED': _record_cache_entry('REPLACED', 'from a full reassignment')} + + self.assertIn('REPLACED', p.record_cache) + self.assertIn('PROTECTED', p.record_cache) diff --git a/unit-tests/service/test_verified_command.py b/unit-tests/service/test_verified_command.py index 4d4ca1c01..e1a94337a 100644 --- a/unit-tests/service/test_verified_command.py +++ b/unit-tests/service/test_verified_command.py @@ -373,3 +373,115 @@ def test_is_record_file_attachment_arg(self): self.assertFalse(is_file('--title')) self.assertFalse(is_file('profile=x')) self.assertFalse(is_file('my.file=x')) # not a file-type field after parse_field + + +class TestProtectedServiceConfigRecords(TestCase): + """No Service Mode command may touch Service Mode's own config records by literal title or UID -- checked + unconditionally for every command (not a curated list) so no current or future command can slip through.""" + + PROTECTED_TITLE = 'Commander Service Mode Config' + PROTECTED_UID = 'PROTECTED_UID_1234' + + def _check(self, cmd, protected_uids=None): + return Verifycommand.validate_service_mode_protected_record_command( + _tokens(cmd), protected_uids or {self.PROTECTED_UID} + ) + + def test_blocks_by_title_across_classic_commands(self): + for cmd in ( + f'get "{self.PROTECTED_TITLE}"', + f'g "{self.PROTECTED_TITLE}"', + f'list "{self.PROTECTED_TITLE}"', + f'l "{self.PROTECTED_TITLE}"', + f'search "{self.PROTECTED_TITLE}"', + f's "{self.PROTECTED_TITLE}"', + f'record-update --record "{self.PROTECTED_TITLE}" title=x', + f'ru --record "{self.PROTECTED_TITLE}" title=x', + f'share-record "{self.PROTECTED_TITLE}" --email a@b.com', + f'sr "{self.PROTECTED_TITLE}" --email a@b.com', + f'rm "{self.PROTECTED_TITLE}"', + f'share-folder --record "{self.PROTECTED_TITLE}" -e a@b.com', + f'record-history "{self.PROTECTED_TITLE}"', + f'clipboard-copy "{self.PROTECTED_TITLE}"', + f'totp "{self.PROTECTED_TITLE}"', + f'one-time-share "{self.PROTECTED_TITLE}"', + f'ls "{self.PROTECTED_TITLE}"', + f'tree "{self.PROTECTED_TITLE}"', + ): + with self.subTest(cmd=cmd): + self.assertIsNotNone(self._check(cmd)) + + def test_blocks_by_title_across_nsf_commands(self): + for cmd in ( + f'nsf-get "{self.PROTECTED_TITLE}"', + f'nsf-share-record "{self.PROTECTED_TITLE}" --email a@b.com', + f'nsf-record-update --record "{self.PROTECTED_TITLE}" title=x', + f'nsf-transfer-record "{self.PROTECTED_TITLE}" a@b.com', + f'nsf-record-details "{self.PROTECTED_TITLE}"', + f'nsf-rm "{self.PROTECTED_TITLE}"', + f'nsf-move "{self.PROTECTED_TITLE}" root', + f'nsf-ln "{self.PROTECTED_TITLE}" SomeFolder', + f'nsf-shortcut keep "{self.PROTECTED_TITLE}"', + ): + with self.subTest(cmd=cmd): + self.assertIsNotNone(self._check(cmd)) + + def test_blocks_by_title_case_insensitive(self): + self.assertIsNotNone(self._check('get "COMMANDER service MODE config"')) + + def test_blocks_by_uid_regardless_of_command(self): + for cmd in ( + f'get {self.PROTECTED_UID}', + f'list {self.PROTECTED_UID}', + f'search {self.PROTECTED_UID}', + f'record-update --record {self.PROTECTED_UID} title=x', + f'share-record {self.PROTECTED_UID} --email a@b.com', + f'share-folder --record {self.PROTECTED_UID} -e a@b.com', + f'rm {self.PROTECTED_UID}', + f'nsf-get {self.PROTECTED_UID}', + f'nsf-transfer-record {self.PROTECTED_UID} a@b.com', + # A command with no known relationship to records at all -- still + # caught, since the check is unconditional, not command-specific. + f'keep-alive {self.PROTECTED_UID}', + ): + with self.subTest(cmd=cmd): + self.assertIsNotNone(self._check(cmd)) + + def test_uid_match_is_case_sensitive(self): + self.assertIsNone(self._check(f'get {self.PROTECTED_UID.lower()}')) + + def test_unrelated_title_and_uid_are_allowed(self): + self.assertIsNone(self._check('get "My Normal Record"')) + self.assertIsNone(self._check('rm SOME_OTHER_UID')) + self.assertIsNone(self._check('record-add --title "My Normal Record"')) + + def test_no_protected_uids_still_blocks_by_title(self): + err = Verifycommand.validate_service_mode_protected_record_command( + _tokens(f'get "{self.PROTECTED_TITLE}"'), None + ) + self.assertIsNotNone(err) + + def test_empty_tokens_returns_none(self): + self.assertIsNone(Verifycommand.validate_service_mode_protected_record_command([])) + + def test_blocks_equals_form_uid(self): + for cmd in ( + f'record-update --record={self.PROTECTED_UID} title=x', + f'get --record-uid={self.PROTECTED_UID}', + f'share-folder --record={self.PROTECTED_UID} -e a@b.com', + f'share-folder -r={self.PROTECTED_UID} -e a@b.com', + ): + with self.subTest(cmd=cmd): + self.assertIsNotNone(self._check(cmd)) + + def test_blocks_equals_form_title(self): + self.assertIsNotNone( + self._check(f'record-update --record="{self.PROTECTED_TITLE}" title=x') + ) + + def test_equals_form_unrelated_value_is_allowed(self): + self.assertIsNone(self._check('record-update --record=SOME_OTHER_UID title=x')) + self.assertIsNone(self._check('record-update --title="My Normal Record" x=y')) + + def test_equals_form_with_no_value_does_not_crash(self): + self.assertIsNone(self._check('get --record-uid='))