From 93f206a14f13506f4da8f8aa48fabaa31721e9d6 Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Wed, 14 Oct 2026 18:46:54 +0530 Subject: [PATCH 1/7] Fix Service Mode filesystem bypasses --- keepercommander/service/util/command_util.py | 3 +- .../service/util/request_validation.py | 38 +++- .../service/util/verified_command.py | 66 ++++++- unit-tests/service/test_verified_command.py | 179 +++++++++++++++++- 4 files changed, 269 insertions(+), 17 deletions(-) diff --git a/keepercommander/service/util/command_util.py b/keepercommander/service/util/command_util.py index d38ca9d2b..0d8976614 100644 --- a/keepercommander/service/util/command_util.py +++ b/keepercommander/service/util/command_util.py @@ -284,7 +284,8 @@ def blocked(error): ) return {"status": "error", "error": error}, 403 - # Same tokens the CLI will run — do not use raw HTTP split(" ") + # Apply every Service Mode restriction to the same tokens the + # CLI's Command.execute_args parser will receive. service_mode_error = Verifycommand.validate_service_mode_restrictions( command_tokens, request_temp_dir ) diff --git a/keepercommander/service/util/request_validation.py b/keepercommander/service/util/request_validation.py index c5d9cbf32..6e3210464 100644 --- a/keepercommander/service/util/request_validation.py +++ b/keepercommander/service/util/request_validation.py @@ -16,6 +16,7 @@ import tempfile import os import json +import shlex import shutil from ..decorators.logging import logger, sanitize_command_fields @@ -82,19 +83,40 @@ def process_file_data(request_data: Dict[str, Any], command: str) -> Tuple[str, temp_file_path = temp_file.name temp_files.append(temp_file_path) - # Replace FILEDATA placeholder with temporary file path, but only when it's used as a file parameter - # Supports: PAM import (--filename), Import (positional), Enterprise-push (positional) - # Callable replacement -- a plain string would let a Windows path's backslashes be parsed as regex backreferences. - file_param_pattern = r'(--filename[=\s]+["\']?)FILEDATA(["\']?)' + # Replace FILEDATA in PAM --filename forms. Quote an + # unquoted path so command execution's shlex parser keeps + # Windows separators and paths containing spaces intact. + file_param_pattern = r'(--filename[=\s]+)(["\']?)FILEDATA(["\']?)' + + def replace_filename_placeholder(match): + prefix, opening_quote, closing_quote = match.groups() + if opening_quote and opening_quote == closing_quote: + replacement = temp_file_path + else: + replacement = shlex.quote(temp_file_path) + return f'{prefix}{opening_quote}{replacement}{closing_quote}' + processed_command = re.sub( file_param_pattern, - lambda m: f'{m.group(1)}{temp_file_path}{m.group(2)}', + replace_filename_placeholder, processed_command, ) - # Also handle standalone FILEDATA (not in quotes) - standalone_pattern = r'\bFILEDATA\b(?!["\'])' - processed_command = re.sub(standalone_pattern, lambda _: temp_file_path, processed_command) + # Also handle positional FILEDATA (including a quoted + # placeholder). Keep existing quotes, otherwise shell-quote + # the generated path as one argument. + standalone_pattern = r'(?P["\']?)\bFILEDATA\b(?P=quote)' + + def replace_standalone_placeholder(match): + quote = match.group('quote') + replacement = temp_file_path if quote else shlex.quote(temp_file_path) + return f'{quote}{replacement}{quote}' + + processed_command = re.sub( + standalone_pattern, + replace_standalone_placeholder, + processed_command, + ) logger.debug(f"Created temporary file {temp_file_path} for filedata") diff --git a/keepercommander/service/util/verified_command.py b/keepercommander/service/util/verified_command.py index 71c77e033..1528ed250 100644 --- a/keepercommander/service/util/verified_command.py +++ b/keepercommander/service/util/verified_command.py @@ -99,16 +99,20 @@ class Verifycommand: _HOST_FS_MSG = ( 'Local filesystem access is not permitted through Service Mode' ) + _AUTOMATOR_DATA_EXPANSION_MSG = ( + 'Automator file and environment expansions are not permitted through Service Mode' + ) @staticmethod def validate_service_mode_restrictions(command_tokens, request_temp_dir=None): - """Run Service Mode bans on executor tokens (shlex); error string or None.""" + """Run Service Mode bans over the exact tokens used by the command parser.""" if not command_tokens: return None for validator in ( Verifycommand.validate_service_mode_double_dash, Verifycommand.validate_service_mode_env_var_expansion_command, + Verifycommand.validate_service_mode_automator_file_access_command, Verifycommand.validate_service_mode_legacy_command, Verifycommand.validate_nsf_convert_command, Verifycommand.validate_service_mode_pam_tunnel_command, @@ -128,8 +132,8 @@ def validate_service_mode_restrictions(command_tokens, request_temp_dir=None): @staticmethod def tokenize_service_command(command): - """Tokenize a command using the executor's backslash handling.""" - return shlex.split(command.replace('\\', '\\\\')) + """Tokenize with the same shlex semantics used by command execution.""" + return shlex.split(command) @staticmethod def validate_nsf_convert_command(command_tokens, request_temp_dir=None): @@ -186,6 +190,60 @@ def validate_service_mode_env_var_expansion_command(command_tokens, request_temp return Verifycommand._ENV_VAR_EXPANSION_MSG return None + @staticmethod + def validate_service_mode_automator_file_access_command(command_tokens, request_temp_dir=None): + """Block Automator operations that read local files or environment values.""" + if not command_tokens: + return None + + command = command_tokens[0].lower() + if command == 'automator-certificate': + return Verifycommand._HOST_FS_MSG + + if command != 'automator' or len(command_tokens) < 2: + return None + + action = command_tokens[1].lower() + if action == 'certificate': + # This is the registered group form in current Commander releases. + return Verifycommand._HOST_FS_MSG + + if action != 'edit': + return None + + # --set supports both ``--set KEY=VALUE`` and ``--set=KEY=VALUE``; + # argparse also accepts the unambiguous ``--se`` abbreviation. + # Extract only those settings instead of matching unrelated Automator + # arguments such as --url. + settings = [] + i = 2 + while i < len(command_tokens): + token = command_tokens[i] + option, separator, inline_value = token.partition('=') + if option in ('--set', '--se'): + if separator: + settings.append(inline_value) + elif i + 1 < len(command_tokens): + settings.append(command_tokens[i + 1]) + i += 1 + i += 1 + + for setting in settings: + separator = setting.find('=') + if separator <= 1: + continue + value = setting[separator + 1:].strip() + if not (value.startswith('${') and value.endswith('}')): + continue + expansion_type, colon, source = value[2:-1].partition(':') + if ( + colon + and expansion_type.lower() in ('file', 'base64', 'env') + and source + and '}' not in source): + return Verifycommand._AUTOMATOR_DATA_EXPANSION_MSG + return None + @staticmethod def validate_service_mode_legacy_command(command_tokens, request_temp_dir=None): """Block legacy commands in Service Mode; error or None.""" @@ -254,7 +312,7 @@ def validate_service_mode_record_file_attachment_command(command_tokens, request @staticmethod def _is_record_file_attachment_arg(token): - """True when parse_field would treat this token as a file attachment field.""" + """True when parse_field would treat this executor token as a file field.""" if not token or '=' not in token: return False # Mirror RecordEditMixin.parse_field field-name normalization. diff --git a/unit-tests/service/test_verified_command.py b/unit-tests/service/test_verified_command.py index 2796a3cb5..e78ca111d 100644 --- a/unit-tests/service/test_verified_command.py +++ b/unit-tests/service/test_verified_command.py @@ -12,10 +12,7 @@ def _tokens(command: str): """Same normalization CommandExecutor uses before policy checks.""" command = unescape(command) - try: - return shlex.split(command.replace('\\', '\\\\')) - except ValueError: - return command.split() + return Verifycommand.tokenize_service_command(command) class TestServiceModeCommandPolicy(TestCase): @@ -108,6 +105,94 @@ def test_attachment_commands_blocked_for_remote_api(self): ) self.assertIsNone(check(_tokens('record-add --title t -rt login login=user'))) + def test_backslash_escaped_record_file_field_is_blocked(self): + """Escaped field names are checked after the same parsing as execution.""" + check = Verifycommand.validate_service_mode_restrictions + + # The HTTP JSON representation of this argument contains ``file\\=``; + # after JSON decoding there is one backslash, which shlex consumes and + # turns into the actual ``file=`` field name seen by record-add. + for field in ( + r'file\=/etc/passwd', + r'f.file\=/etc/passwd', + r'fi\le=/etc/passwd', + r'f.fi\le.doc\=/etc/passwd', + ): + command = f'record-add -t f01 -rt login --self-destruct=1d {field}' + with self.subTest(field=field): + self.assertEqual(_tokens(command), shlex.split(command)) + self.assertIsNotNone(check(_tokens(command))) + + # The ordinary spelling remains blocked as before. + self.assertIsNotNone(check(_tokens( + 'record-add -t f01 -rt login --self-destruct=1d file=/etc/passwd' + ))) + self.assertIsNone(check(_tokens( + r'record-add -t f01 -rt login lo\gin=user' + ))) + + def test_escaped_host_path_options_are_blocked(self): + """Filesystem flags and parser-driven file inputs use execution tokens.""" + check = Verifycommand.validate_service_mode_restrictions + ban = 'Local filesystem access' + for cmd in ( + r'generate --out\put=/tmp/sm-write-canary.txt', + r'generate --out\put /tmp/sm-write-canary.txt', + r'import --for\mat=json /etc/passwd', + r'import --file\-cache=/tmp/sm-cache', + r'pam project import --filen\ame=/etc/passwd', + r'audit-report --form\at=pdf', + ): + with self.subTest(cmd=cmd): + self.assertIn(ban, check(_tokens(cmd))) + + def test_automator_local_file_access_is_blocked(self): + check = Verifycommand.validate_service_mode_restrictions + host_fs_msg = 'Local filesystem access' + for cmd in ( + 'automator-certificate /etc/passwd', + 'automator certificate /etc/passwd', + ): + with self.subTest(cmd=cmd): + self.assertIn(host_fs_msg, check(_tokens(cmd))) + + def test_automator_local_data_expansions_are_blocked(self): + check = Verifycommand.validate_service_mode_restrictions + expansion_msg = 'Automator file and environment expansions' + for cmd in ( + 'automator edit TestAutomator --set debugUrl=${file:/etc/passwd}', + 'automator edit TestAutomator --set key=${base64:/etc/shadow}', + 'automator edit TestAutomator --set key=${env:AWS_SECRET_ACCESS_KEY}', + 'automator edit TestAutomator --set key=${FiLe:/etc/passwd}', + 'automator edit TestAutomator --set "debugUrl = ${file:/etc/passwd}"', + 'automator edit TestAutomator --set "debugUrl= ${file:/etc/passwd}"', + 'automator edit TestAutomator --set "debugUrl=${file:/etc/passwd} "', + 'automator edit TestAutomator --set="debugUrl = ${base64:/etc/shadow}"', + 'automator edit TestAutomator --se "debugUrl = ${file:/etc/passwd}"', + ): + with self.subTest(cmd=cmd): + self.assertIn(expansion_msg, check(_tokens(cmd))) + + # Ordinary Automator edits remain usable through Service Mode. + self.assertIsNone(check(_tokens( + 'automator edit TestAutomator --set debugUrl=https://example.invalid' + ))) + # The expansion syntax is only interpreted for --set values, not URL. + self.assertIsNone(check(_tokens( + 'automator edit TestAutomator --url=${file:/tmp/canary.txt}' + ))) + + def test_quoted_backslash_path_tokens_match_executor(self): + # A quoted Windows-style path remains one argument with its + # backslashes intact under the CLI's shlex parser. + path = r'C:\service\files\payload.txt' + command = f'generate --output "{path}"' + self.assertEqual( + Verifycommand.tokenize_service_command(command), + shlex.split(command), + ) + self.assertIn(path, _tokens(command)) + def test_host_filesystem_commands_blocked(self): check = Verifycommand.validate_service_mode_restrictions ban = 'Local filesystem access' @@ -336,6 +421,7 @@ def test_filedata_substitution_happens_before_validation(self): self.assertTrue(temp_files, 'expected a temp file to be created') self.assertNotIn('FILEDATA', processed_command) + self.assertTrue(any(token.endswith(temp_files[0]) for token in _tokens(processed_command))) request_temp_dir = os.path.dirname(temp_files[0]) tokens = _tokens(processed_command) @@ -343,6 +429,91 @@ def test_filedata_substitution_happens_before_validation(self): Verifycommand.validate_service_mode_restrictions(tokens, request_temp_dir) ) + def test_filedata_positional_placeholder_is_shell_quoted(self): + """A generated path remains one executor token, including with spaces.""" + from keepercommander.service.util.request_validation import RequestValidator + from unittest.mock import patch + + request_data = {'filedata': {'some': 'data'}} + command = 'import --format=json FILEDATA' + request_temp_dir = tempfile.mkdtemp(prefix='service mode ') + self.addCleanup(shutil.rmtree, request_temp_dir, ignore_errors=True) + with patch( + 'keepercommander.service.util.request_validation.tempfile.mkdtemp', + return_value=request_temp_dir): + processed_command, temp_files = RequestValidator.process_file_data( + request_data, command + ) + self.addCleanup(RequestValidator.cleanup_temp_files, temp_files) + + self.assertEqual( + shlex.split(processed_command), + ['import', '--format=json', temp_files[0]], + ) + + def test_filedata_filename_placeholder_is_shell_quoted(self): + from keepercommander.service.util.request_validation import RequestValidator + from unittest.mock import patch + + request_data = {'filedata': {'some': 'data'}} + command = 'pam project import --filename=FILEDATA' + request_temp_dir = tempfile.mkdtemp(prefix='service mode ') + self.addCleanup(shutil.rmtree, request_temp_dir, ignore_errors=True) + with patch( + 'keepercommander.service.util.request_validation.tempfile.mkdtemp', + return_value=request_temp_dir): + processed_command, temp_files = RequestValidator.process_file_data( + request_data, command + ) + self.addCleanup(RequestValidator.cleanup_temp_files, temp_files) + + tokens = shlex.split(processed_command) + self.assertEqual(tokens[-1], f'--filename={temp_files[0]}') + self.assertIsNone( + Verifycommand.validate_service_mode_restrictions( + tokens, request_temp_dir + ) + ) + + def test_filedata_windows_path_survives_executor_tokenization(self): + from keepercommander.service.util.request_validation import RequestValidator + from unittest.mock import MagicMock, patch + + request_data = {'filedata': {'some': 'data'}} + command = 'import --format=json FILEDATA' + temp_file_path = r'C:\service mode\payload.json' + temp_file = MagicMock() + temp_file.name = temp_file_path + with patch( + 'keepercommander.service.util.request_validation.tempfile.mkdtemp', + return_value=r'C:\service mode'), patch( + 'keepercommander.service.util.request_validation.tempfile.NamedTemporaryFile') as make_temp_file: + make_temp_file.return_value.__enter__.return_value = temp_file + processed_command, temp_files = RequestValidator.process_file_data( + request_data, command + ) + + self.assertEqual(temp_files, [temp_file_path]) + self.assertEqual( + shlex.split(processed_command), + ['import', '--format=json', temp_file_path], + ) + + def test_filedata_quoted_positional_placeholder_is_replaced(self): + from keepercommander.service.util.request_validation import RequestValidator + + request_data = {'filedata': {'some': 'data'}} + command = 'import --format=json "FILEDATA"' + processed_command, temp_files = RequestValidator.process_file_data( + request_data, command + ) + self.addCleanup(RequestValidator.cleanup_temp_files, temp_files) + + self.assertEqual( + shlex.split(processed_command), + ['import', '--format=json', temp_files[0]], + ) + def test_option_values_yields_dash_leading_values(self): is_file = Verifycommand._option_values self.assertEqual(list(is_file(['generate', '--output', '-'], '--output')), ['-']) From b13a298d0e037fc0f4bb5c84dc6913698ef394be Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Thu, 15 Oct 2026 11:24:26 +0530 Subject: [PATCH 2/7] Add github actions for running windows unit-tests --- .github/workflows/test-with-pytest.yml | 8 +++++--- keepercommander/service/util/verified_command.py | 2 ++ unit-tests/service/test_verified_command.py | 14 ++++++++++++++ 3 files changed, 21 insertions(+), 3 deletions(-) diff --git a/.github/workflows/test-with-pytest.yml b/.github/workflows/test-with-pytest.yml index 395c12fda..a5294163d 100644 --- a/.github/workflows/test-with-pytest.yml +++ b/.github/workflows/test-with-pytest.yml @@ -7,11 +7,13 @@ env: jobs: test-with-pytest: + name: Unit tests (${{ matrix.os }}, Python ${{ matrix.python-version }}) strategy: matrix: + os: ['ubuntu-24.04', 'windows-latest'] python-version: ['3.9', '3.14'] - runs-on: ubuntu-24.04 + runs-on: ${{ matrix.os }} steps: - name: Checkout branch @@ -23,7 +25,7 @@ jobs: python-version: ${{ matrix.python-version }} - name: Install package with test dependencies - run: pip install .[test] + run: python -m pip install ".[test]" - name: Run unit tests - run: pytest unit-tests/ + run: python -m pytest unit-tests/ diff --git a/keepercommander/service/util/verified_command.py b/keepercommander/service/util/verified_command.py index 1528ed250..c5894f28a 100644 --- a/keepercommander/service/util/verified_command.py +++ b/keepercommander/service/util/verified_command.py @@ -231,6 +231,8 @@ def validate_service_mode_automator_file_access_command(command_tokens, request_ for setting in settings: separator = setting.find('=') if separator <= 1: + # AutomatorEditCommand uses the same pos > 1 condition and + # returns before resolving expansions for malformed settings. continue value = setting[separator + 1:].strip() if not (value.startswith('${') and value.endswith('}')): diff --git a/unit-tests/service/test_verified_command.py b/unit-tests/service/test_verified_command.py index e78ca111d..048a7cd4f 100644 --- a/unit-tests/service/test_verified_command.py +++ b/unit-tests/service/test_verified_command.py @@ -514,6 +514,20 @@ def test_filedata_quoted_positional_placeholder_is_replaced(self): ['import', '--format=json', temp_files[0]], ) + def test_mismatched_filedata_quotes_remain_invalid_command_syntax(self): + """Malformed placeholder quoting must fail closed, not be repaired.""" + 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) + + with self.assertRaises(ValueError): + Verifycommand.tokenize_service_command(processed_command) + def test_option_values_yields_dash_leading_values(self): is_file = Verifycommand._option_values self.assertEqual(list(is_file(['generate', '--output', '-'], '--output')), ['-']) From 874aa19513127c352c4ac81376ae6ff92117ed67 Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Thu, 15 Oct 2026 15:08:18 +0530 Subject: [PATCH 3/7] Fix windows test cases failing --- .github/workflows/test-with-pytest.yml | 5 +-- keepercommander/utils.py | 34 +++++++++++---- unit-tests/service/test_verified_command.py | 46 ++++++++++++++++----- unit-tests/test_windows_file_permissions.py | 39 +++++++++++++++++ 4 files changed, 104 insertions(+), 20 deletions(-) diff --git a/.github/workflows/test-with-pytest.yml b/.github/workflows/test-with-pytest.yml index a5294163d..47fa47fa9 100644 --- a/.github/workflows/test-with-pytest.yml +++ b/.github/workflows/test-with-pytest.yml @@ -7,13 +7,12 @@ env: jobs: test-with-pytest: - name: Unit tests (${{ matrix.os }}, Python ${{ matrix.python-version }}) + name: Unit tests (Python ${{ matrix.python-version }}) strategy: matrix: - os: ['ubuntu-24.04', 'windows-latest'] python-version: ['3.9', '3.14'] - runs-on: ${{ matrix.os }} + runs-on: ubuntu-24.04 steps: - name: Checkout branch diff --git a/keepercommander/utils.py b/keepercommander/utils.py index 07278516a..32c794ae3 100644 --- a/keepercommander/utils.py +++ b/keepercommander/utils.py @@ -103,19 +103,39 @@ def _windows_icacls_principal(): # type: () -> str return username +def _path_has_link_component(file_path): # type: (str) -> bool + """Return whether a path or one of its parent components is a link.""" + drive, tail = os.path.splitdrive(os.path.abspath(file_path)) + current_path = drive + os.sep if drive else os.sep + for component in tail.split(os.sep): + if not component: + continue + current_path = os.path.join(current_path, component) + file_stat = os.lstat(current_path) + if stat.S_ISLNK(file_stat.st_mode): + return True + + # Windows junctions and other reparse points are not always reported + # as symbolic links. Avoid changing permissions through those paths too. + if os.name == 'nt': + reparse_point = getattr(stat, 'FILE_ATTRIBUTE_REPARSE_POINT', 0x400) + if getattr(file_stat, 'st_file_attributes', 0) & reparse_point: + return True + return False + + def set_file_permissions(file_path): # type: (str) -> None """ Set secure file permissions (600) for configuration files containing sensitive data. This ensures only the owner can read and write the file. """ file_path = os.path.abspath(file_path) - real_path = os.path.realpath(file_path) - if file_path != real_path: - return - + try: - if os.path.islink(file_path): - logging.warning(f'Skipping permission setting on symbolic link: {file_path}') + # Check each component explicitly: realpath() can canonicalize valid + # Windows short paths, while following a symlink/junction is unsafe. + if _path_has_link_component(file_path): + logging.warning(f'Skipping permission setting on linked path: {file_path}') return if platform.system() != 'Windows': @@ -631,4 +651,4 @@ def is_windows_11(): if sys.platform != "win32": return False version = sys.getwindowsversion() - return version.major >= 10 and version.build >= 22000 \ No newline at end of file + return version.major >= 10 and version.build >= 22000 diff --git a/unit-tests/service/test_verified_command.py b/unit-tests/service/test_verified_command.py index 048a7cd4f..ab8f9704c 100644 --- a/unit-tests/service/test_verified_command.py +++ b/unit-tests/service/test_verified_command.py @@ -306,16 +306,29 @@ def test_filename_allows_temp_filedata_paths_only(self): 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) + check( + _tokens(f'pam project import --filename={shlex.quote(temp_path)}'), + request_temp_dir, + ) + ) + self.assertIsNone( + check(_tokens(f'pam project import -f {shlex.quote(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) + check(_tokens(f'import --format=json {shlex.quote(temp_path)}'), request_temp_dir) + ) + self.assertIsNone( + check( + _tokens(f'enterprise-push {shlex.quote(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) + check( + _tokens(f'pam project extend --config=SOME_UID -f {shlex.quote(temp_path)}'), + request_temp_dir, + ) ) def test_command_aliases_do_not_bypass_host_path_checks(self): @@ -339,8 +352,15 @@ def test_command_aliases_do_not_bypass_host_path_checks(self): 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)) + self.assertIsNone( + check(_tokens(f'pam p i -f {shlex.quote(temp_path)}'), request_temp_dir) + ) + self.assertIsNone( + check( + _tokens(f'pam p e --filename={shlex.quote(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, @@ -355,11 +375,15 @@ def test_non_request_temp_path_still_rejected(self): 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) + ban, + check( + _tokens(f'pam project import -f {shlex.quote(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}')) + ban, check(_tokens(f'pam project import -f {shlex.quote(other_path)}')) ) def test_flag_abbreviations_do_not_bypass_host_path_checks(self): @@ -406,7 +430,9 @@ def test_import_bare_name_positional_requires_format_awareness(self): 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)) + self.assertIsNone( + check(_tokens(f'import --format=json {shlex.quote(temp_path)}'), request_temp_dir) + ) def test_filedata_substitution_happens_before_validation(self): """process_file_data must run before validate_service_mode_restrictions, diff --git a/unit-tests/test_windows_file_permissions.py b/unit-tests/test_windows_file_permissions.py index 27a54baed..cc9de3144 100644 --- a/unit-tests/test_windows_file_permissions.py +++ b/unit-tests/test_windows_file_permissions.py @@ -1,10 +1,26 @@ import os +import stat import tempfile +from types import SimpleNamespace from unittest import TestCase, mock from keepercommander import utils +class TestPathLinkDetection(TestCase): + def test_detects_link_in_parent_component(self): + with tempfile.TemporaryDirectory() as tmp_dir: + link_dir = os.path.join(tmp_dir, 'linked-dir') + file_path = os.path.join(link_dir, 'config.json') + + def fake_lstat(path): + mode = stat.S_IFLNK if path == link_dir else stat.S_IFDIR + return SimpleNamespace(st_mode=mode) + + with mock.patch('keepercommander.utils.os.lstat', side_effect=fake_lstat): + self.assertTrue(utils._path_has_link_component(file_path)) + + class TestWindowsIcaclsPrincipal(TestCase): def test_userdomain_and_username(self): with mock.patch.dict(os.environ, {'USERNAME': 'ivan', 'USERDOMAIN': 'IVAN'}, clear=False): @@ -66,3 +82,26 @@ def test_grant_uses_domain_principal(self, _islink, _system, mock_run): self.assertEqual(self._grant_principal(mock_run), 'CORP\\jdoe:RW') finally: os.unlink(path) + + @mock.patch('subprocess.run') + @mock.patch('platform.system', return_value='Windows') + @mock.patch('os.path.realpath', return_value=r'C:\Users\LongName\Temp\file.json') + @mock.patch( + 'keepercommander.utils.os.lstat', + return_value=SimpleNamespace(st_mode=stat.S_IFREG), + ) + def test_canonical_path_alias_does_not_skip_permissions( + self, _lstat, _realpath, _system, mock_run): + """Windows short-path canonicalization must not skip ACL hardening.""" + with tempfile.NamedTemporaryFile(delete=False) as tmp: + path = os.path.abspath(tmp.name) + try: + with mock.patch.dict( + os.environ, {'USERNAME': 'jdoe', 'USERDOMAIN': 'CORP'}, clear=False): + utils.set_file_permissions(path) + self.assertTrue(any( + '/inheritance:r' in call.args[0] + for call in mock_run.call_args_list + )) + finally: + os.unlink(path) From f21b1a01aa2492cb3793202afe21db0ba03d6cc8 Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Fri, 16 Oct 2026 11:07:26 +0530 Subject: [PATCH 4/7] Fix claude review comments --- keepercommander/commands/base.py | 15 ++++-- keepercommander/service/util/command_util.py | 28 +++++++++-- .../service/util/request_validation.py | 25 ++++------ .../service/util/verified_command.py | 8 +-- keepercommander/utils.py | 49 ++++++++++--------- unit-tests/service/test_verified_command.py | 41 +++++++++++++++- unit-tests/test_windows_file_permissions.py | 42 ++++++++++++---- 7 files changed, 146 insertions(+), 62 deletions(-) diff --git a/keepercommander/commands/base.py b/keepercommander/commands/base.py index 572e0f8c0..a3190f666 100644 --- a/keepercommander/commands/base.py +++ b/keepercommander/commands/base.py @@ -709,6 +709,14 @@ def normalize_output_param(args: str) -> str: return args +def tokenize_command_args(args, envvars=None): + """Tokenize command arguments using the same transforms as Command.execute_args.""" + args = '' if args is None else args + if envvars is not None: + args = expand_cmd_args(args, envvars) + return shlex.split(normalize_output_param(args)) + + def as_boolean(value, default=None): # type: (Any, Optional[bool]) -> bool if isinstance(value, bool): return value @@ -765,14 +773,13 @@ def execute_args(self, params, args, **kwargs): alias_cmd = d.get('command') if alias_cmd and alias_cmd != parser.prog: parser.prog = alias_cmd - args = expand_cmd_args(args, envvars) - args = normalize_output_param(args) + command_tokens = tokenize_command_args(args, envvars) if self.support_extra_parameters(): - opts, extra_args = parser.parse_known_args(shlex.split(args)) + opts, extra_args = parser.parse_known_args(command_tokens) if extra_args: self.extra_parameters = ' '.join(extra_args) else: - opts = parser.parse_args(shlex.split(args)) + opts = parser.parse_args(command_tokens) d.update(opts.__dict__) return self.execute(params, **d) diff --git a/keepercommander/service/util/command_util.py b/keepercommander/service/util/command_util.py index 0d8976614..ef7162d11 100644 --- a/keepercommander/service/util/command_util.py +++ b/keepercommander/service/util/command_util.py @@ -270,20 +270,38 @@ def execute(cls, command: str, temp_files: Optional[list] = None) -> Tuple[Any, command = ensure_record_add_json_format(html.unescape(command)) try: - command_tokens = Verifycommand.tokenize_service_command(command) + # Check the raw command before environment expansion: Service + # Mode explicitly forbids this syntax, and expansion must not + # erase evidence of it before the restriction check. + raw_command_tokens = Verifycommand.tokenize_service_command(command) except ValueError: return {"status": "error", "error": "Invalid command syntax"}, 400 - # 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 + blocked_command = raw_command_tokens[0] if raw_command_tokens else '' def blocked(error): logger.warning( - f"Service Mode blocked command '{command_tokens[0] if command_tokens else ''}': {error}" + f"Service Mode blocked command '{blocked_command}': {error}" ) return {"status": "error", "error": error}, 403 + expansion_error = Verifycommand.validate_service_mode_env_var_expansion_command( + raw_command_tokens + ) + if expansion_error: + return blocked(expansion_error) + + try: + command_tokens = Verifycommand.tokenize_service_command( + command, params.environment_variables + ) + except ValueError: + return {"status": "error", "error": "Invalid command syntax"}, 400 + + # 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 + # Apply every Service Mode restriction to the same tokens the # CLI's Command.execute_args parser will receive. service_mode_error = Verifycommand.validate_service_mode_restrictions( diff --git a/keepercommander/service/util/request_validation.py b/keepercommander/service/util/request_validation.py index 6e3210464..7bb6b939c 100644 --- a/keepercommander/service/util/request_validation.py +++ b/keepercommander/service/util/request_validation.py @@ -83,18 +83,16 @@ def process_file_data(request_data: Dict[str, Any], command: str) -> Tuple[str, temp_file_path = temp_file.name temp_files.append(temp_file_path) - # Replace FILEDATA in PAM --filename forms. Quote an - # unquoted path so command execution's shlex parser keeps - # Windows separators and paths containing spaces intact. - file_param_pattern = r'(--filename[=\s]+)(["\']?)FILEDATA(["\']?)' + # Replace the entire FILEDATA token with a quoted path. + # Reusing the caller's quotes would break when the + # generated path itself contains a single quote. + file_param_pattern = ( + r'(--filename[=\s]+)(?P["\']?)FILEDATA(?P=quote)' + ) def replace_filename_placeholder(match): - prefix, opening_quote, closing_quote = match.groups() - if opening_quote and opening_quote == closing_quote: - replacement = temp_file_path - else: - replacement = shlex.quote(temp_file_path) - return f'{prefix}{opening_quote}{replacement}{closing_quote}' + prefix = match.group(1) + return f'{prefix}{shlex.quote(temp_file_path)}' processed_command = re.sub( file_param_pattern, @@ -103,14 +101,11 @@ def replace_filename_placeholder(match): ) # Also handle positional FILEDATA (including a quoted - # placeholder). Keep existing quotes, otherwise shell-quote - # the generated path as one argument. + # placeholder), replacing its original quoting entirely. standalone_pattern = r'(?P["\']?)\bFILEDATA\b(?P=quote)' def replace_standalone_placeholder(match): - quote = match.group('quote') - replacement = temp_file_path if quote else shlex.quote(temp_file_path) - return f'{quote}{replacement}{quote}' + return shlex.quote(temp_file_path) processed_command = re.sub( standalone_pattern, diff --git a/keepercommander/service/util/verified_command.py b/keepercommander/service/util/verified_command.py index c5894f28a..ba4165227 100644 --- a/keepercommander/service/util/verified_command.py +++ b/keepercommander/service/util/verified_command.py @@ -2,7 +2,6 @@ import contextlib import io import os -import shlex class Verifycommand: @@ -131,9 +130,10 @@ def validate_service_mode_restrictions(command_tokens, request_temp_dir=None): return None @staticmethod - def tokenize_service_command(command): - """Tokenize with the same shlex semantics used by command execution.""" - return shlex.split(command) + def tokenize_service_command(command, envvars=None): + """Tokenize with the same argument transforms used by Command.execute_args.""" + from ...commands.base import tokenize_command_args + return tokenize_command_args(command, envvars) @staticmethod def validate_nsf_convert_command(command_tokens, request_temp_dir=None): diff --git a/keepercommander/utils.py b/keepercommander/utils.py index 32c794ae3..7304ea22d 100644 --- a/keepercommander/utils.py +++ b/keepercommander/utils.py @@ -103,24 +103,13 @@ def _windows_icacls_principal(): # type: () -> str return username -def _path_has_link_component(file_path): # type: (str) -> bool - """Return whether a path or one of its parent components is a link.""" - drive, tail = os.path.splitdrive(os.path.abspath(file_path)) - current_path = drive + os.sep if drive else os.sep - for component in tail.split(os.sep): - if not component: - continue - current_path = os.path.join(current_path, component) - file_stat = os.lstat(current_path) - if stat.S_ISLNK(file_stat.st_mode): - return True - - # Windows junctions and other reparse points are not always reported - # as symbolic links. Avoid changing permissions through those paths too. - if os.name == 'nt': - reparse_point = getattr(stat, 'FILE_ATTRIBUTE_REPARSE_POINT', 0x400) - if getattr(file_stat, 'st_file_attributes', 0) & reparse_point: - return True +def _is_link_or_reparse_point(file_stat): # type: (os.stat_result) -> bool + """Return whether the final path component is a link or Windows reparse point.""" + if stat.S_ISLNK(file_stat.st_mode): + return True + if os.name == 'nt': + reparse_point = getattr(stat, 'FILE_ATTRIBUTE_REPARSE_POINT', 0x400) + return bool(getattr(file_stat, 'st_file_attributes', 0) & reparse_point) return False @@ -132,19 +121,31 @@ def set_file_permissions(file_path): # type: (str) -> None file_path = os.path.abspath(file_path) try: - # Check each component explicitly: realpath() can canonicalize valid - # Windows short paths, while following a symlink/junction is unsafe. - if _path_has_link_component(file_path): - logging.warning(f'Skipping permission setting on linked path: {file_path}') + # Inspect only the file itself. Parent-directory aliases should not + # prevent permission hardening of an otherwise regular file. + file_stat = os.lstat(file_path) + if _is_link_or_reparse_point(file_stat): + logging.warning(f'Skipping permission setting on linked file: {file_path}') return if platform.system() != 'Windows': - file_stat = os.stat(file_path) if file_stat.st_uid != os.getuid(): logging.warning(f'Skipping permission setting on file not owned by current user: {file_path}') return - os.chmod(file_path, stat.S_IRUSR | stat.S_IWUSR) + permissions = stat.S_IRUSR | stat.S_IWUSR + if os.chmod in os.supports_follow_symlinks: + os.chmod(file_path, permissions, follow_symlinks=False) + elif hasattr(os, 'O_NOFOLLOW') and hasattr(os, 'fchmod'): + fd = os.open(file_path, os.O_RDONLY | os.O_NOFOLLOW) + try: + os.fchmod(fd, permissions) + finally: + os.close(fd) + else: + raise NotImplementedError( + 'Platform does not support safe permission changes without following symlinks' + ) logging.debug(f'Set secure permissions (600) for file: {file_path}') else: principal = _windows_icacls_principal() diff --git a/unit-tests/service/test_verified_command.py b/unit-tests/service/test_verified_command.py index ab8f9704c..9ad2eeabe 100644 --- a/unit-tests/service/test_verified_command.py +++ b/unit-tests/service/test_verified_command.py @@ -1,4 +1,4 @@ -from unittest import TestCase +from unittest import TestCase, mock from html import unescape import os import shutil @@ -434,6 +434,19 @@ def test_import_bare_name_positional_requires_format_awareness(self): check(_tokens(f'import --format=json {shlex.quote(temp_path)}'), request_temp_dir) ) + def test_windows_output_normalization_is_checked_as_a_file_path(self): + check = Verifycommand.validate_service_mode_restrictions + normalized_cases = [] + with mock.patch('sys.platform', 'win32'): + for output_value in Verifycommand._NON_PATH_OUTPUT_VALUES: + command = rf'public-api-key list --format json --output \{output_value}' + normalized_cases.append((output_value, _tokens(command))) + + for output_value, tokens in normalized_cases: + with self.subTest(output_value=output_value): + self.assertEqual(tokens[-1], f'/{output_value}') + self.assertIn('Local filesystem access', check(tokens)) + 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 @@ -540,6 +553,32 @@ def test_filedata_quoted_positional_placeholder_is_replaced(self): ['import', '--format=json', temp_files[0]], ) + def test_quoted_filedata_placeholders_support_apostrophes_in_path(self): + from keepercommander.service.util.request_validation import RequestValidator + request_data = {'filedata': {'some': 'data'}} + commands = ( + "import --format=json 'FILEDATA'", + "pam project import --filename='FILEDATA'", + ) + + for command in commands: + with self.subTest(command=command), tempfile.TemporaryDirectory() as parent_dir: + request_temp_dir = os.path.join(parent_dir, "service user's files") + os.mkdir(request_temp_dir) + with mock.patch( + 'keepercommander.service.util.request_validation.tempfile.mkdtemp', + return_value=request_temp_dir): + processed_command, temp_files = RequestValidator.process_file_data( + request_data, command + ) + self.addCleanup(RequestValidator.cleanup_temp_files, temp_files) + + expected_last_token = ( + f'--filename={temp_files[0]}' + if '--filename' in command else temp_files[0] + ) + self.assertEqual(_tokens(processed_command)[-1], expected_last_token) + def test_mismatched_filedata_quotes_remain_invalid_command_syntax(self): """Malformed placeholder quoting must fail closed, not be repaired.""" from keepercommander.service.util.request_validation import RequestValidator diff --git a/unit-tests/test_windows_file_permissions.py b/unit-tests/test_windows_file_permissions.py index cc9de3144..57fbee42c 100644 --- a/unit-tests/test_windows_file_permissions.py +++ b/unit-tests/test_windows_file_permissions.py @@ -7,18 +7,42 @@ from keepercommander import utils -class TestPathLinkDetection(TestCase): - def test_detects_link_in_parent_component(self): +class TestSetFilePermissionsSymlinks(TestCase): + def test_hardens_file_through_symlinked_parent(self): with tempfile.TemporaryDirectory() as tmp_dir: - link_dir = os.path.join(tmp_dir, 'linked-dir') - file_path = os.path.join(link_dir, 'config.json') + if os.name == 'nt': + self.skipTest('Creating a Windows symlink may require elevated privileges') - def fake_lstat(path): - mode = stat.S_IFLNK if path == link_dir else stat.S_IFDIR - return SimpleNamespace(st_mode=mode) + target_dir = os.path.join(tmp_dir, 'target-dir') + os.mkdir(target_dir) + real_path = os.path.join(target_dir, 'config.json') + with open(real_path, 'w') as config_file: + config_file.write('secret') + os.chmod(real_path, 0o644) - with mock.patch('keepercommander.utils.os.lstat', side_effect=fake_lstat): - self.assertTrue(utils._path_has_link_component(file_path)) + alias_dir = os.path.join(tmp_dir, 'linked-dir') + os.symlink(target_dir, alias_dir) + alias_path = os.path.join(alias_dir, 'config.json') + + utils.set_file_permissions(alias_path) + + self.assertEqual(os.stat(real_path).st_mode & 0o777, 0o600) + + def test_skips_final_symlink(self): + with tempfile.TemporaryDirectory() as tmp_dir: + if os.name == 'nt': + self.skipTest('Creating a Windows symlink may require elevated privileges') + + target_path = os.path.join(tmp_dir, 'config.json') + with open(target_path, 'w') as config_file: + config_file.write('secret') + os.chmod(target_path, 0o644) + symlink_path = os.path.join(tmp_dir, 'config-link.json') + os.symlink(target_path, symlink_path) + + utils.set_file_permissions(symlink_path) + + self.assertEqual(os.stat(target_path).st_mode & 0o777, 0o644) class TestWindowsIcaclsPrincipal(TestCase): From 00b9a8896fe8f47c8c99b8e98bae73df3456ebbd Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Fri, 16 Oct 2026 11:33:39 +0530 Subject: [PATCH 5/7] Fix failing test case in windows --- .github/workflows/test-with-pytest.yml | 5 ++--- unit-tests/service/test_verified_command.py | 17 ++++++++++------- 2 files changed, 12 insertions(+), 10 deletions(-) diff --git a/.github/workflows/test-with-pytest.yml b/.github/workflows/test-with-pytest.yml index 47fa47fa9..395c12fda 100644 --- a/.github/workflows/test-with-pytest.yml +++ b/.github/workflows/test-with-pytest.yml @@ -7,7 +7,6 @@ env: jobs: test-with-pytest: - name: Unit tests (Python ${{ matrix.python-version }}) strategy: matrix: python-version: ['3.9', '3.14'] @@ -24,7 +23,7 @@ jobs: python-version: ${{ matrix.python-version }} - name: Install package with test dependencies - run: python -m pip install ".[test]" + run: pip install .[test] - name: Run unit tests - run: python -m pytest unit-tests/ + run: pytest unit-tests/ diff --git a/unit-tests/service/test_verified_command.py b/unit-tests/service/test_verified_command.py index 9ad2eeabe..f70a6ec75 100644 --- a/unit-tests/service/test_verified_command.py +++ b/unit-tests/service/test_verified_command.py @@ -6,6 +6,7 @@ import shlex +from keepercommander.commands.base import normalize_output_param from keepercommander.service.util.verified_command import Verifycommand @@ -183,15 +184,17 @@ def test_automator_local_data_expansions_are_blocked(self): ))) def test_quoted_backslash_path_tokens_match_executor(self): - # A quoted Windows-style path remains one argument with its - # backslashes intact under the CLI's shlex parser. + # The CLI normalizes --output before shlex on Windows; Service Mode + # must apply the same sequence and keep the path as one argument. path = r'C:\service\files\payload.txt' command = f'generate --output "{path}"' - self.assertEqual( - Verifycommand.tokenize_service_command(command), - shlex.split(command), - ) - self.assertIn(path, _tokens(command)) + with mock.patch('sys.platform', 'win32'): + expected_tokens = shlex.split(normalize_output_param(command)) + actual_tokens = Verifycommand.tokenize_service_command(command) + + self.assertEqual(actual_tokens, expected_tokens) + self.assertEqual(actual_tokens[-1], r'C:/service/files/payload.txt') + self.assertEqual(len(actual_tokens), 3) def test_host_filesystem_commands_blocked(self): check = Verifycommand.validate_service_mode_restrictions From eb24321e5da4d327ed8d56e3cace032148278cf7 Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Mon, 19 Oct 2026 14:41:11 +0530 Subject: [PATCH 6/7] Fix failing test cases --- keepercommander/utils.py | 35 +++--------- unit-tests/test_windows_file_permissions.py | 63 --------------------- 2 files changed, 7 insertions(+), 91 deletions(-) diff --git a/keepercommander/utils.py b/keepercommander/utils.py index 7304ea22d..8d8f9f600 100644 --- a/keepercommander/utils.py +++ b/keepercommander/utils.py @@ -103,49 +103,28 @@ def _windows_icacls_principal(): # type: () -> str return username -def _is_link_or_reparse_point(file_stat): # type: (os.stat_result) -> bool - """Return whether the final path component is a link or Windows reparse point.""" - if stat.S_ISLNK(file_stat.st_mode): - return True - if os.name == 'nt': - reparse_point = getattr(stat, 'FILE_ATTRIBUTE_REPARSE_POINT', 0x400) - return bool(getattr(file_stat, 'st_file_attributes', 0) & reparse_point) - return False - - def set_file_permissions(file_path): # type: (str) -> None """ Set secure file permissions (600) for configuration files containing sensitive data. This ensures only the owner can read and write the file. """ file_path = os.path.abspath(file_path) + real_path = os.path.realpath(file_path) + if file_path != real_path: + return try: - # Inspect only the file itself. Parent-directory aliases should not - # prevent permission hardening of an otherwise regular file. - file_stat = os.lstat(file_path) - if _is_link_or_reparse_point(file_stat): - logging.warning(f'Skipping permission setting on linked file: {file_path}') + if os.path.islink(file_path): + logging.warning(f'Skipping permission setting on symbolic link: {file_path}') return if platform.system() != 'Windows': + file_stat = os.stat(file_path) if file_stat.st_uid != os.getuid(): logging.warning(f'Skipping permission setting on file not owned by current user: {file_path}') return - permissions = stat.S_IRUSR | stat.S_IWUSR - if os.chmod in os.supports_follow_symlinks: - os.chmod(file_path, permissions, follow_symlinks=False) - elif hasattr(os, 'O_NOFOLLOW') and hasattr(os, 'fchmod'): - fd = os.open(file_path, os.O_RDONLY | os.O_NOFOLLOW) - try: - os.fchmod(fd, permissions) - finally: - os.close(fd) - else: - raise NotImplementedError( - 'Platform does not support safe permission changes without following symlinks' - ) + os.chmod(file_path, stat.S_IRUSR | stat.S_IWUSR) logging.debug(f'Set secure permissions (600) for file: {file_path}') else: principal = _windows_icacls_principal() diff --git a/unit-tests/test_windows_file_permissions.py b/unit-tests/test_windows_file_permissions.py index 57fbee42c..27a54baed 100644 --- a/unit-tests/test_windows_file_permissions.py +++ b/unit-tests/test_windows_file_permissions.py @@ -1,50 +1,10 @@ import os -import stat import tempfile -from types import SimpleNamespace from unittest import TestCase, mock from keepercommander import utils -class TestSetFilePermissionsSymlinks(TestCase): - def test_hardens_file_through_symlinked_parent(self): - with tempfile.TemporaryDirectory() as tmp_dir: - if os.name == 'nt': - self.skipTest('Creating a Windows symlink may require elevated privileges') - - target_dir = os.path.join(tmp_dir, 'target-dir') - os.mkdir(target_dir) - real_path = os.path.join(target_dir, 'config.json') - with open(real_path, 'w') as config_file: - config_file.write('secret') - os.chmod(real_path, 0o644) - - alias_dir = os.path.join(tmp_dir, 'linked-dir') - os.symlink(target_dir, alias_dir) - alias_path = os.path.join(alias_dir, 'config.json') - - utils.set_file_permissions(alias_path) - - self.assertEqual(os.stat(real_path).st_mode & 0o777, 0o600) - - def test_skips_final_symlink(self): - with tempfile.TemporaryDirectory() as tmp_dir: - if os.name == 'nt': - self.skipTest('Creating a Windows symlink may require elevated privileges') - - target_path = os.path.join(tmp_dir, 'config.json') - with open(target_path, 'w') as config_file: - config_file.write('secret') - os.chmod(target_path, 0o644) - symlink_path = os.path.join(tmp_dir, 'config-link.json') - os.symlink(target_path, symlink_path) - - utils.set_file_permissions(symlink_path) - - self.assertEqual(os.stat(target_path).st_mode & 0o777, 0o644) - - class TestWindowsIcaclsPrincipal(TestCase): def test_userdomain_and_username(self): with mock.patch.dict(os.environ, {'USERNAME': 'ivan', 'USERDOMAIN': 'IVAN'}, clear=False): @@ -106,26 +66,3 @@ def test_grant_uses_domain_principal(self, _islink, _system, mock_run): self.assertEqual(self._grant_principal(mock_run), 'CORP\\jdoe:RW') finally: os.unlink(path) - - @mock.patch('subprocess.run') - @mock.patch('platform.system', return_value='Windows') - @mock.patch('os.path.realpath', return_value=r'C:\Users\LongName\Temp\file.json') - @mock.patch( - 'keepercommander.utils.os.lstat', - return_value=SimpleNamespace(st_mode=stat.S_IFREG), - ) - def test_canonical_path_alias_does_not_skip_permissions( - self, _lstat, _realpath, _system, mock_run): - """Windows short-path canonicalization must not skip ACL hardening.""" - with tempfile.NamedTemporaryFile(delete=False) as tmp: - path = os.path.abspath(tmp.name) - try: - with mock.patch.dict( - os.environ, {'USERNAME': 'jdoe', 'USERDOMAIN': 'CORP'}, clear=False): - utils.set_file_permissions(path) - self.assertTrue(any( - '/inheritance:r' in call.args[0] - for call in mock_run.call_args_list - )) - finally: - os.unlink(path) From ea1e6900113bbea68aeabfc9f70a09c07263d6f5 Mon Sep 17 00:00:00 2001 From: amangalampalli-ks Date: Mon, 19 Oct 2026 14:45:23 +0530 Subject: [PATCH 7/7] Undo test case changes --- keepercommander/utils.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/keepercommander/utils.py b/keepercommander/utils.py index 8d8f9f600..07278516a 100644 --- a/keepercommander/utils.py +++ b/keepercommander/utils.py @@ -112,7 +112,7 @@ def set_file_permissions(file_path): # type: (str) -> None real_path = os.path.realpath(file_path) if file_path != real_path: return - + try: if os.path.islink(file_path): logging.warning(f'Skipping permission setting on symbolic link: {file_path}') @@ -631,4 +631,4 @@ def is_windows_11(): if sys.platform != "win32": return False version = sys.getwindowsversion() - return version.major >= 10 and version.build >= 22000 + return version.major >= 10 and version.build >= 22000 \ No newline at end of file