diff --git a/keepercommander/service/api/command.py b/keepercommander/service/api/command.py index c80c6b1ea..bb736bec7 100644 --- a/keepercommander/service/api/command.py +++ b/keepercommander/service/api/command.py @@ -99,7 +99,7 @@ def execute_command_direct(**kwargs) -> Tuple[Union[Response, bytes], int]: ) return response_data - response, status_code = CommandExecutor.execute(processed_command) + response, status_code = CommandExecutor.execute(processed_command, temp_files=temp_files) # If we get a busy response, add v1-specific message if (isinstance(response, dict) and diff --git a/keepercommander/service/core/request_queue.py b/keepercommander/service/core/request_queue.py index 44e387e19..175947418 100644 --- a/keepercommander/service/core/request_queue.py +++ b/keepercommander/service/core/request_queue.py @@ -325,7 +325,7 @@ def _process_request(self, request: QueuedRequest): try: # Execute the command using existing CommandExecutor - result, status_code = CommandExecutor.execute(request.command) + result, status_code = CommandExecutor.execute(request.command, temp_files=request.temp_files) # Mark as completed request.status = RequestStatus.COMPLETED diff --git a/keepercommander/service/util/command_util.py b/keepercommander/service/util/command_util.py index cb7d94bf6..0c7507f08 100644 --- a/keepercommander/service/util/command_util.py +++ b/keepercommander/service/util/command_util.py @@ -148,13 +148,12 @@ def _finalize_parsed_response(cls, response: Any) -> Tuple[Any, int]: return response, status_code @classmethod - def execute(cls, command: str) -> Tuple[Any, int]: + def execute(cls, command: str, temp_files: Optional[list] = None) -> Tuple[Any, int]: logger.debug(f"Executing command: {sanitize_command_fields(command)}") - validation_error = cls.validate_command(command) if validation_error: return validation_error - + from ..core.globals import ensure_params_loaded try: params = ensure_params_loaded() @@ -169,11 +168,19 @@ def execute(cls, command: str) -> Tuple[Any, int]: except ValueError: command_tokens = command.split() + # This request's own FILEDATA directory - the only paths Service + # 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 + # Same tokens the CLI will run — do not use raw HTTP split(" ") service_mode_error = Verifycommand.validate_service_mode_restrictions( - command_tokens + 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 force_error = Verifycommand.validate_enterprise_user_add_role_force( diff --git a/keepercommander/service/util/request_validation.py b/keepercommander/service/util/request_validation.py index cff1051f0..b0ad33cf6 100644 --- a/keepercommander/service/util/request_validation.py +++ b/keepercommander/service/util/request_validation.py @@ -15,6 +15,7 @@ import tempfile import os import json +import shutil from ..decorators.logging import logger, sanitize_command_fields @@ -71,9 +72,11 @@ def process_file_data(request_data: Dict[str, Any], command: str) -> Tuple[str, logger.warning("filedata must be a JSON object or array") return processed_command, temp_files + request_temp_dir = None try: - # Create temporary file with the filedata content - with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False, encoding='utf-8') as temp_file: + request_temp_dir = tempfile.mkdtemp(prefix='keeper_svc_') + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False, + encoding='utf-8', dir=request_temp_dir) as temp_file: json.dump(filedata, temp_file, indent=2) temp_file_path = temp_file.name temp_files.append(temp_file_path) @@ -91,23 +94,24 @@ def process_file_data(request_data: Dict[str, Any], command: str) -> Tuple[str, except Exception as e: logger.error(f"Error creating temporary file for filedata: {e}") - # Clean up any created temp files - for temp_path in temp_files: + if request_temp_dir: try: - os.unlink(temp_path) - except Exception: - pass + shutil.rmtree(request_temp_dir) + logger.debug(f"Cleaned up request temp directory: {request_temp_dir}") + except Exception as cleanup_error: + logger.warning(f"Failed to clean up request temp directory {request_temp_dir}: {cleanup_error}") return command, [] return processed_command, temp_files @staticmethod def cleanup_temp_files(temp_files: list) -> None: - """Clean up temporary files. - + """Clean up temporary files and their parent per-request directories. + Args: temp_files: List of temporary file paths to clean up """ + parent_dirs = set() for temp_path in temp_files: try: if os.path.exists(temp_path): @@ -115,6 +119,16 @@ def cleanup_temp_files(temp_files: list) -> None: logger.debug(f"Cleaned up temporary file: {temp_path}") except Exception as e: logger.warning(f"Failed to clean up temporary file {temp_path}: {e}") + parent_dirs.add(os.path.dirname(temp_path)) + + # Remove parent directories (each file has one dedicated per-request directory) + for parent_dir in parent_dirs: + try: + if os.path.exists(parent_dir): + shutil.rmtree(parent_dir) + logger.debug(f"Cleaned up request temp directory: {parent_dir}") + except Exception as e: + logger.warning(f"Failed to clean up request temp directory {parent_dir}: {e}") @staticmethod def validate_request_json() -> Optional[Tuple]: diff --git a/keepercommander/service/util/verified_command.py b/keepercommander/service/util/verified_command.py index 1e240ad05..631984afc 100644 --- a/keepercommander/service/util/verified_command.py +++ b/keepercommander/service/util/verified_command.py @@ -1,3 +1,8 @@ +import contextlib +import io +import os + + class Verifycommand: # pam tunnel aliases: start=s, list=l, stop=x, edit=e, diagnose=d # (see PAMTunnelCommand.register_command in tunnel_and_connections.py) @@ -12,8 +17,49 @@ class Verifycommand: # Aliases from record.py — CommandExecutor checks tokens before cli expands them. _RECORD_EDIT_COMMANDS = frozenset({'record-add', 'ra', 'record-update', 'ru'}) + # WARNING: everything below is a DENYLIST. Any command/flag that reads or + # writes a host file and is NOT enumerated here is allowed by default. + # Adding a new command with local file I/O? Add it here, or it silently + # bypasses Service Mode's "no host filesystem access" boundary. + # + # Commands that always read/write host files (no safe Service Mode form). + _HOST_FS_COMMANDS = frozenset({ + 'run-batch', 'run', + 'export', + 'download-membership', + 'download-record-types', + 'apply-membership', + 'load-record-types', + }) + # Positional file input; FILEDATA is rewritten to a temp path before execute. + _FILE_INPUT_COMMANDS = frozenset({'import', 'enterprise-push'}) + # import --format values that name an account/API/URL source, not a local + # file (importer/commands.py choices). Any format NOT in this set is treated as file-based by default. + _IMPORT_FORMATS_WITHOUT_FILE = frozenset({ + 'lastpass', 'manageengine', 'thycotic', 'cyberark', 'cyberark_portal', + }) + # generate's registered alias (commands/utils.py: aliases['gen'] = 'generate'). + _GENERATE_COMMAND_NAMES = frozenset({'generate', 'gen'}) + # pam's 'project' subcommand alias (discoveryrotation.py: register_command('project', ..., 'p')). + _PAM_PROJECT_ALIASES = {'p': 'project'} + # pam project's own subcommand aliases (pam_import/commands.py: register_command(...)). + _PAM_PROJECT_SUBCOMMAND_ALIASES = {'x': 'export', 'i': 'import', 'e': 'extend'} + # --output values that select a mode/format, not a host path. + _NON_PATH_OUTPUT_VALUES = frozenset({ + 'clipboard', 'stdout', 'stdouthidden', 'variable', + 'token', 'base64', 'json', 'k8s', 'text', + }) + # Extensions that indicate a local data file (avoid treating emails as paths). + _LOCAL_FILE_EXTENSIONS = frozenset({ + '.json', '.csv', '.txt', '.yaml', '.yml', '.xml', '.kdbx', '.zip', + '.pdf', '.ndjson', '.1pif', '.xlsx', '.xls', '.kdb', '.gpg', + }) + _HOST_FS_MSG = ( + 'Local filesystem access is not permitted through Service Mode' + ) + @staticmethod - def validate_service_mode_restrictions(command_tokens): + def validate_service_mode_restrictions(command_tokens, request_temp_dir=None): """Run Service Mode bans on executor tokens (shlex); error string or None.""" if not command_tokens: return None @@ -23,14 +69,17 @@ def validate_service_mode_restrictions(command_tokens): Verifycommand.validate_service_mode_download_attachment_command, Verifycommand.validate_service_mode_upload_attachment_command, Verifycommand.validate_service_mode_record_file_attachment_command, + Verifycommand.validate_service_mode_host_filesystem_command, + Verifycommand.validate_service_mode_host_path_args, + Verifycommand.validate_service_mode_file_input_command, ): - error = validator(command_tokens) + error = validator(command_tokens, request_temp_dir) if error: return error return None @staticmethod - def validate_service_mode_pam_tunnel_command(command_tokens): + def validate_service_mode_pam_tunnel_command(command_tokens, request_temp_dir=None): """Allow only pam tunnel edit in Service Mode; error string or None.""" if not command_tokens or len(command_tokens) < 2: return None @@ -49,7 +98,7 @@ def validate_service_mode_pam_tunnel_command(command_tokens): ) @staticmethod - def validate_service_mode_download_attachment_command(command_tokens): + def validate_service_mode_download_attachment_command(command_tokens, request_temp_dir=None): """Block download-attachment in Service Mode; error string or None.""" if not command_tokens: return None @@ -61,7 +110,7 @@ def validate_service_mode_download_attachment_command(command_tokens): ) @staticmethod - def validate_service_mode_upload_attachment_command(command_tokens): + def validate_service_mode_upload_attachment_command(command_tokens, request_temp_dir=None): """Block upload-attachment in Service Mode; error string or None.""" if not command_tokens: return None @@ -73,7 +122,7 @@ def validate_service_mode_upload_attachment_command(command_tokens): ) @staticmethod - def validate_service_mode_record_file_attachment_command(command_tokens): + def validate_service_mode_record_file_attachment_command(command_tokens, request_temp_dir=None): """Block record-add/update (and ra/ru) file fields in Service Mode; error or None.""" if not command_tokens: return None @@ -97,6 +146,236 @@ def _is_record_file_attachment_arg(token): name = name[2:] return name.split('.', 1)[0] == 'file' + @staticmethod + def validate_service_mode_host_filesystem_command(command_tokens, request_temp_dir=None): + """Block commands that always touch the host filesystem; error or None.""" + if not command_tokens: + return None + if command_tokens[0].lower() not in Verifycommand._HOST_FS_COMMANDS: + return None + return Verifycommand._HOST_FS_MSG + + @staticmethod + def validate_service_mode_host_path_args(command_tokens, request_temp_dir=None): + """Block host-path flags (--output, --filename, …); error or None.""" + if not command_tokens: + return None + + tokens = command_tokens + cmd0 = tokens[0].lower() + + # PDF always requires an on-disk output file. Routed through + # _option_values so abbreviations (--form, --fmt=pdf, …) are caught too. + for value in Verifycommand._option_values(tokens, '--format'): + if value.lower() == 'pdf': + return Verifycommand._HOST_FS_MSG + + for flag in ('--out-dir', '--output-dir', '--from-file', '--file-cache', + '--keepass-key-file', '-v3f'): + if Verifycommand._has_option(tokens, flag): + return Verifycommand._HOST_FS_MSG + + # credential-provision --config is a host YAML path; --config-base64 is OK. + # Do not ban --config globally — PAM uses it for configuration UIDs. + if cmd0 in ('credential-provision', 'cp') and Verifycommand._has_option(tokens, '--config'): + return Verifycommand._HOST_FS_MSG + + for value in Verifycommand._option_values(tokens, '--filename'): + if not Verifycommand._is_service_temp_path(value, request_temp_dir): + return Verifycommand._HOST_FS_MSG + + # pam project import/extend use -f as --filename (not --force). + if Verifycommand._is_pam_project_filename_cmd(tokens): + for value in Verifycommand._option_values(tokens, '-f'): + if not Verifycommand._is_service_temp_path(value, request_temp_dir): + return Verifycommand._HOST_FS_MSG + + for value in Verifycommand._option_values(tokens, '--file'): + if Verifycommand._looks_like_local_path(value): + if not Verifycommand._is_service_temp_path(value, request_temp_dir): + return Verifycommand._HOST_FS_MSG + + for value in Verifycommand._option_values(tokens, '--output'): + if value.lower() not in Verifycommand._NON_PATH_OUTPUT_VALUES: + return Verifycommand._HOST_FS_MSG + + # generate / pam project export use -o as a file path (not mode). + if cmd0 in Verifycommand._GENERATE_COMMAND_NAMES or Verifycommand._is_pam_project_export(tokens): + for value in Verifycommand._option_values(tokens, '-o'): + if value.lower() not in Verifycommand._NON_PATH_OUTPUT_VALUES: + return Verifycommand._HOST_FS_MSG + + # transfer-user (tu) @filename mapping files - scoped to that command + # only, since '@value' is a legitimate argument shape elsewhere. + if cmd0 in ('transfer-user', 'tu'): + for tok in tokens[1:]: + if tok.startswith('@') and Verifycommand._looks_like_local_path(tok[1:]): + return Verifycommand._HOST_FS_MSG + + return None + + @staticmethod + def validate_service_mode_file_input_command(command_tokens, request_temp_dir=None): + """Allow import/enterprise-push only with FILEDATA temp paths; error or None.""" + if not command_tokens: + return None + cmd0 = command_tokens[0].lower() + if cmd0 not in Verifycommand._FILE_INPUT_COMMANDS: + return None + + if cmd0 == 'import': + from ...importer.commands import import_parser + ns = Verifycommand._safe_parse(import_parser, command_tokens[1:]) + if ns is None: + # Malformed for the real parser too; it will reject this itself. + return None + fmt = (ns.format or '').lower() + if fmt in Verifycommand._IMPORT_FORMATS_WITHOUT_FILE: + return None + if ns.name and not Verifycommand._is_service_temp_path(ns.name, request_temp_dir): + return Verifycommand._HOST_FS_MSG + return None + + if cmd0 == 'enterprise-push': + from ...commands.enterprise_push import enterprise_push_parser + ns = Verifycommand._safe_parse(enterprise_push_parser, command_tokens[1:]) + if ns is None: + return None + if ns.file and not Verifycommand._is_service_temp_path(ns.file, request_temp_dir): + return Verifycommand._HOST_FS_MSG + return None + + return None + + @staticmethod + def _safe_parse(parser, tokens): + """Resolve tokens via the command's real argparse parser; None if it can't be resolved. + + Used instead of hand-scanning tokens so we get the parser's own + distinction between a positional value and a flag's value -- the + raw tokens alone don't tell you that. + """ + try: + with contextlib.redirect_stderr(io.StringIO()): + ns, _ = parser.parse_known_args(tokens) + return ns + except SystemExit: + return None + except Exception: + return None + + @staticmethod + def _pam_project_verb(command_tokens): + """Resolve 'pam ' (alias-aware) to the verb, or None.""" + if len(command_tokens) < 3: + return None + t = [x.lower() for x in command_tokens[:3]] + if t[0] != 'pam': + return None + project = Verifycommand._PAM_PROJECT_ALIASES.get(t[1], t[1]) + if project != 'project': + return None + return Verifycommand._PAM_PROJECT_SUBCOMMAND_ALIASES.get(t[2], t[2]) + + @staticmethod + def _is_pam_project_export(command_tokens): + return Verifycommand._pam_project_verb(command_tokens) == 'export' + + @staticmethod + def _is_pam_project_filename_cmd(command_tokens): + return Verifycommand._pam_project_verb(command_tokens) in ('import', 'extend') + + # Flags we separately check for. Some are literal prefixes of others + # (--output/--output-dir, --file/--filename, --file/--file-cache) - a + # token that's a *complete* match for one of these is a distinct flag in + # its own right, never an abbreviation attempt at a different one. + _KNOWN_DANGEROUS_FLAGS = frozenset({ + '--output', '--filename', '--file', '--format', + '--out-dir', '--output-dir', '--from-file', '--file-cache', + '--keepass-key-file', '--config', + }) + + @staticmethod + def _flag_matches(tok_name, flag): + """True if tok_name is exactly flag, or an unambiguous long-option + abbreviation of it (argparse's allow_abbrev default -- almost no + parser in this codebase opts out of it, so '--out' really does mean + '--output' to the command that will actually run). + + Short options ('-o', '-f') are never abbreviated by argparse, so + those only ever match exactly. + """ + if tok_name == flag: + return True + # Reject bare '--' (argparse's "end of options" marker), anything + # that isn't a '--long' option on both sides, and anything that's + # already a complete, distinct flag of its own. + if len(tok_name) <= 2 or not tok_name.startswith('--') or not flag.startswith('--'): + return False + if tok_name in Verifycommand._KNOWN_DANGEROUS_FLAGS: + return False + return flag.startswith(tok_name) + + @staticmethod + def _has_option(tokens, flag): + """True if flag, flag=value, or an abbreviation of flag appears in tokens.""" + flag_l = flag.lower() + for tok in tokens[1:]: + name = tok.lower().split('=', 1)[0] + if Verifycommand._flag_matches(name, flag_l): + return True + return False + + @staticmethod + def _option_values(tokens, flag): + """Yield values for --flag value / --flag=value (and short -o value), + matching flag itself or an unambiguous abbreviation of it.""" + flag_l = flag.lower() + i = 1 + while i < len(tokens): + tok = tokens[i] + lower = tok.lower() + name, sep, _ = lower.partition('=') + if sep and Verifycommand._flag_matches(name, flag_l): + yield tok.split('=', 1)[1] + i += 1 + continue + if not sep and Verifycommand._flag_matches(name, flag_l): + # Consume the next token as the value even if it starts with + # '-' (e.g. a path or FILEDATA placeholder) -- these flags are + # all required-argument (action='store'), so skipping a + # dash-leading value here let it slip past every check below. + if i + 1 < len(tokens): + yield tokens[i + 1] + i += 2 + continue + i += 1 + + @staticmethod + def _looks_like_local_path(value): + if not value: + return False + if value.startswith(('http://', 'https://')): + return False + if value.startswith('~') or '/' in value or '\\' in value: + return True + _, ext = os.path.splitext(value) + return ext.lower() in Verifycommand._LOCAL_FILE_EXTENSIONS + + @staticmethod + def _is_service_temp_path(path, request_temp_dir): + """True when path resolves under *this request's* own temp directory, not just anywhere under the + shared OS temp root -- other processes/users can also write there. + """ + if not path or not request_temp_dir: + return False + try: + resolved = os.path.realpath(os.path.expanduser(path)) + root = os.path.realpath(request_temp_dir) + return resolved == root or resolved.startswith(root + os.sep) + except (OSError, ValueError): + return False + @staticmethod def validate_append_command(command): """ diff --git a/unit-tests/service/test_api_routes.py b/unit-tests/service/test_api_routes.py index 758fcaec4..222d4d530 100644 --- a/unit-tests/service/test_api_routes.py +++ b/unit-tests/service/test_api_routes.py @@ -70,5 +70,5 @@ def test_v1_direct_route_keeps_legacy_execution_path(self): self.assertEqual(response.status_code, 200) self.assertEqual(response.headers.get('X-API-Legacy'), 'true') self.assertEqual(response.get_json(), {"status": "success", "data": {"command": "ls"}}) - mock_execute.assert_called_once_with('ls') + mock_execute.assert_called_once_with('ls', temp_files=[]) mock_submit.assert_not_called() diff --git a/unit-tests/service/test_auth_security.py b/unit-tests/service/test_auth_security.py index 68913db28..9a8d3c567 100644 --- a/unit-tests/service/test_auth_security.py +++ b/unit-tests/service/test_auth_security.py @@ -169,3 +169,12 @@ def test_validate_service_mode_restrictions_attachments(self): ) self.assertIsNone(check(['record-add', '--title', 't', '-rt', 'login', 'login=user'])) self.assertIsNone(check(['record-add', '--title', 't', '-rt', 'login', 'my.file=x'])) + + def test_validate_service_mode_restrictions_host_filesystem(self): + check = Verifycommand.validate_service_mode_restrictions + ban = 'Local filesystem access' + self.assertIn(ban, check(['run-batch', '--dry-run', '/etc/passwd'])) + self.assertIn(ban, check(['export', '--format=json', '/tmp/out.json'])) + self.assertIn(ban, check(['audit-report', '--format=json', '--output=/tmp/r.json'])) + self.assertIsNone(check(['audit-report', '--format=json'])) + self.assertIsNone(check(['clipboard-copy', 'uid', '--output=stdout'])) diff --git a/unit-tests/service/test_queue_concurrency.py b/unit-tests/service/test_queue_concurrency.py index 5a2800e30..25cf04e60 100644 --- a/unit-tests/service/test_queue_concurrency.py +++ b/unit-tests/service/test_queue_concurrency.py @@ -52,7 +52,7 @@ def test_queue_manager_serializes_concurrent_submissions(self): inflight = {"count": 0, "max": 0} results = {} - def fake_execute(command): + def fake_execute(command, **kwargs): with state_lock: inflight["count"] += 1 inflight["max"] = max(inflight["max"], inflight["count"]) @@ -91,7 +91,7 @@ def test_v1_and_v2_share_single_queue_worker(self): outputs = {} start_barrier = threading.Barrier(3) - def fake_execute(command): + def fake_execute(command, **kwargs): with state_lock: inflight["count"] += 1 inflight["max"] = max(inflight["max"], inflight["count"]) @@ -148,7 +148,7 @@ def test_timed_out_v1_request_does_not_execute_after_expiration(self): executed_commands = [] executed_lock = threading.Lock() - def fake_execute(command): + def fake_execute(command, **kwargs): with executed_lock: executed_commands.append(command) @@ -189,7 +189,7 @@ def test_processing_v1_request_waits_past_queue_timeout(self): started_processing = threading.Event() - def fake_execute(command): + def fake_execute(command, **kwargs): started_processing.set() time.sleep(request_timeout + 0.15) return {"status": "success", "data": {"command": command}}, 200 diff --git a/unit-tests/service/test_service_mode_pam_tunnel.py b/unit-tests/service/test_service_mode_pam_tunnel.py index 257ca41dd..083280bff 100644 --- a/unit-tests/service/test_service_mode_pam_tunnel.py +++ b/unit-tests/service/test_service_mode_pam_tunnel.py @@ -1,5 +1,8 @@ from unittest import TestCase from html import unescape +import os +import shutil +import tempfile import shlex @@ -90,6 +93,203 @@ def test_attachment_commands_blocked_for_remote_api(self): ) self.assertIsNone(check(_tokens('record-add --title t -rt login login=user'))) + def test_host_filesystem_commands_blocked(self): + check = Verifycommand.validate_service_mode_restrictions + ban = 'Local filesystem access' + for cmd in ( + 'run-batch --dry-run /etc/passwd', + 'run --dry-run ~/.keeper/config.json', + 'export --format=json /tmp/out.json', + 'download-membership --source=keeper /tmp/m.json', + 'download-record-types --source=keeper /tmp/rt.json', + 'apply-membership /tmp/m.json', + 'load-record-types /tmp/rt.json', + ): + with self.subTest(cmd=cmd): + err = check(_tokens(cmd)) + self.assertIsNotNone(err) + self.assertIn(ban, err) + + def test_host_path_output_args_blocked(self): + check = Verifycommand.validate_service_mode_restrictions + ban = 'Local filesystem access' + for cmd in ( + 'audit-report --format=json --output=/tmp/report.json', + 'share-report --format=csv --output=out.csv', + 'generate --output /tmp/passwords.txt', + 'generate -o /tmp/passwords.txt', + 'pam project export --project-uid UID -o /tmp/proj.json', + 'ls --format=pdf --output=/tmp/x.pdf', + 'audit-report --format=pdf --output=report.pdf', + ): + with self.subTest(cmd=cmd): + err = check(_tokens(cmd)) + self.assertIsNotNone(err) + self.assertIn(ban, err) + + # Non-path --output modes remain allowed + self.assertIsNone(check(_tokens('clipboard-copy UID --output=stdout'))) + self.assertIsNone(check(_tokens('credential-provision --config-base64 dGVzdA== --output json'))) + self.assertIsNone(check(_tokens('audit-report --format=json'))) + + def test_filename_allows_temp_filedata_paths_only(self): + check = Verifycommand.validate_service_mode_restrictions + ban = 'Local filesystem access' + self.assertIsNotNone(check(_tokens('pam project import --filename=/etc/passwd'))) + self.assertIsNotNone(check(_tokens('pam project import -f /etc/passwd'))) + # Use a path outside any OS temp dir -- on Linux, tempfile.gettempdir() + # often *is* /tmp, so a literal /tmp path would be misclassified as safe. + self.assertIn(ban, check(_tokens('import --format=json /etc/vault.json'))) + + request_temp_dir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, request_temp_dir, ignore_errors=True) + temp_path = os.path.join(request_temp_dir, 'service_filedata_test.json') + + self.assertIsNone( + check(_tokens(f'pam project import --filename={temp_path}'), request_temp_dir) + ) + self.assertIsNone(check(_tokens(f'pam project import -f {temp_path}'), request_temp_dir)) + self.assertIsNone(check(_tokens(f'import --format=json {temp_path}'), request_temp_dir)) + self.assertIsNone( + check(_tokens(f'enterprise-push {temp_path} --email user@example.com'), request_temp_dir) + ) + # PAM --config is a vault UID, not a host path + self.assertIsNone( + check(_tokens('pam project extend --config=SOME_UID -f ' + temp_path), request_temp_dir) + ) + + def test_command_aliases_do_not_bypass_host_path_checks(self): + check = Verifycommand.validate_service_mode_restrictions + ban = 'Local filesystem access' + for cmd in ( + 'gen -o /tmp/passwords.txt', + 'gen --output /tmp/passwords.txt', + 'pam p x --project-uid UID -o /tmp/proj.json', + 'pam p i -f /etc/passwd', + 'pam p i --filename=/etc/passwd', + 'pam p e -f /etc/passwd', + ): + with self.subTest(cmd=cmd): + err = check(_tokens(cmd)) + self.assertIsNotNone(err) + self.assertIn(ban, err) + + # Aliased forms of the safe cases (non-path --output, temp-path filename) stay allowed. + request_temp_dir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, request_temp_dir, ignore_errors=True) + temp_path = os.path.join(request_temp_dir, 'service_filedata_alias_test.json') + self.assertIsNone(check(_tokens('gen --output stdout'))) + self.assertIsNone(check(_tokens(f'pam p i -f {temp_path}'), request_temp_dir)) + self.assertIsNone(check(_tokens(f'pam p e --filename={temp_path}'), request_temp_dir)) + + def test_non_request_temp_path_still_rejected(self): + """A path under a DIFFERENT request's temp dir is not automatically safe, + even though it's still somewhere under the shared OS temp root.""" + check = Verifycommand.validate_service_mode_restrictions + ban = 'Local filesystem access' + + request_temp_dir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, request_temp_dir, ignore_errors=True) + other_request_dir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, other_request_dir, ignore_errors=True) + other_path = os.path.join(other_request_dir, 'someone_elses_file.json') + + self.assertIn( + ban, check(_tokens(f'pam project import -f {other_path}'), request_temp_dir) + ) + # No request_temp_dir at all (e.g. a request with no FILEDATA) trusts nothing. + self.assertIn( + ban, check(_tokens(f'pam project import -f {other_path}')) + ) + + def test_flag_abbreviations_do_not_bypass_host_path_checks(self): + """argparse's default allow_abbrev means '--out' really does mean + '--output' to the real command -- our checker has to agree.""" + check = Verifycommand.validate_service_mode_restrictions + ban = 'Local filesystem access' + for cmd in ( + 'generate --out /etc/evil', + 'audit-report --form pdf --out /etc/evil.pdf', + 'pam project import --filenam=/etc/passwd', + 'pam project import --fil /etc/passwd', + ): + with self.subTest(cmd=cmd): + err = check(_tokens(cmd)) + self.assertIsNotNone(err) + self.assertIn(ban, err) + + # A complete, distinct flag must not be misread as an abbreviation of + # a different one just because it's a literal prefix of it. + self.assertIsNone(check(_tokens('clipboard-copy UID --output=stdout'))) + + def test_import_bare_name_positional_requires_format_awareness(self): + """A plain filename with no slash/extension must still be checked -- + unless the format means `name` isn't a file at all (account/URL).""" + check = Verifycommand.validate_service_mode_restrictions + ban = 'Local filesystem access' + + # 'data' has no slash and no known extension, but json/csv always + # read it as a local file -- must not slip through on shape alone. + self.assertIn(ban, check(_tokens('import --format=json data'))) + self.assertIn(ban, check(_tokens('import --format=csv data'))) + + # lastpass/manageengine/thycotic/cyberark/cyberark_portal treat `name` + # as an account/URL, not a file -- must stay allowed even though it's + # a bare, non-path-looking value. + self.assertIsNone(check(_tokens('import --format=lastpass my-lastpass-account'))) + self.assertIsNone(check(_tokens('import --format=manageengine https://me.example.com'))) + self.assertIsNone(check(_tokens('import --format=thycotic https://thycotic.example.com'))) + self.assertIsNone(check(_tokens('import --format=cyberark pvwa.example.com'))) + self.assertIsNone(check(_tokens('import --format=cyberark_portal example-tenant'))) + + # A real per-request temp path still works normally for file-based formats. + request_temp_dir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, request_temp_dir, ignore_errors=True) + temp_path = os.path.join(request_temp_dir, 'import_data.json') + self.assertIsNone(check(_tokens(f'import --format=json {temp_path}'), request_temp_dir)) + + def test_filedata_substitution_happens_before_validation(self): + """process_file_data must run before validate_service_mode_restrictions, + so --filename=FILEDATA resolves to a real per-request temp path by the + time the host-path checks run (see CommandExecutor.execute ordering).""" + from keepercommander.service.util.request_validation import RequestValidator + + request_data = {'filedata': {'some': 'data'}} + command = 'pam project import --filename=FILEDATA' + processed_command, temp_files = RequestValidator.process_file_data(request_data, command) + self.addCleanup(RequestValidator.cleanup_temp_files, temp_files) + + self.assertTrue(temp_files, 'expected a temp file to be created') + self.assertNotIn('FILEDATA', processed_command) + + request_temp_dir = os.path.dirname(temp_files[0]) + tokens = _tokens(processed_command) + self.assertIsNone( + Verifycommand.validate_service_mode_restrictions(tokens, request_temp_dir) + ) + + def test_option_values_yields_dash_leading_values(self): + is_file = Verifycommand._option_values + self.assertEqual(list(is_file(['generate', '--output', '-'], '--output')), ['-']) + self.assertEqual( + list(is_file(['pam', 'project', 'import', '-f', '-etc/passwd'], '-f')), + ['-etc/passwd'], + ) + + check = Verifycommand.validate_service_mode_restrictions + ban = 'Local filesystem access' + self.assertIn(ban, check(_tokens('pam project import -f -etc/passwd'))) + self.assertIn(ban, check(_tokens('pam project import --filename -etc/passwd'))) + + def test_config_file_blocked_config_base64_allowed(self): + check = Verifycommand.validate_service_mode_restrictions + self.assertIsNotNone( + check(_tokens('credential-provision --config=/tmp/cfg.yaml -c PAMUID')) + ) + self.assertIsNone( + check(_tokens('credential-provision --config-base64 dGVzdA== -c PAMUID')) + ) + def test_is_record_file_attachment_arg(self): is_file = Verifycommand._is_record_file_attachment_arg self.assertTrue(is_file('file=@/tmp/x'))