diff --git a/src/path_scope.py b/src/path_scope.py index 31828a6a9f..be25976049 100644 --- a/src/path_scope.py +++ b/src/path_scope.py @@ -60,14 +60,27 @@ def validate_payload(self, payload: str, cwd: str | Path | None = None) -> PathS def validate_path(self, candidate: str | Path, cwd: str | Path | None = None) -> PathScopeDecision: raw = os.path.expandvars(os.path.expanduser(str(candidate))) if _is_windows_absolute(raw): - return self._validate_windows_path(raw) + if os.name != 'nt': + return self._validate_windows_path(raw) + elif not any(_is_windows_absolute(str(root)) for root in self.roots): + # Even on Windows, deny if no roots are Windows absolute paths (edge case) + return PathScopeDecision(False, 'windows absolute path is outside workspace scope', str(candidate), raw) + base = Path(cwd).expanduser().resolve(strict=False) if cwd else self.roots[0] path = Path(raw) if not path.is_absolute(): path = base / path expanded = self._expand_glob(path) for expanded_path in expanded: - resolved = expanded_path.resolve(strict=False) + try: + resolved = expanded_path.resolve(strict=False) + except (OSError, ValueError, RuntimeError): + return PathScopeDecision( + False, + 'path cannot be resolved or is invalid', + str(candidate), + str(expanded_path), + ) if not any(_is_relative_to(resolved, root) for root in self.roots): return PathScopeDecision( False, @@ -75,7 +88,16 @@ def validate_path(self, candidate: str | Path, cwd: str | Path | None = None) -> str(candidate), str(resolved), ) - return PathScopeDecision(True, 'path is inside workspace scope', str(candidate), str(expanded[0].resolve(strict=False))) + try: + final_resolved = str(expanded[0].resolve(strict=False)) + except (OSError, ValueError, RuntimeError): + return PathScopeDecision( + False, + 'path cannot be resolved or is invalid', + str(candidate), + str(expanded[0]), + ) + return PathScopeDecision(True, 'path is inside workspace scope', str(candidate), final_resolved) def _expand_glob(self, path: Path) -> tuple[Path, ...]: path_text = str(path) @@ -116,7 +138,7 @@ def extract_path_candidates(payload: str) -> tuple[str, ...]: tokens = payload.split() raw_tokens = payload.split() candidates: list[str] = [] - for token in (*tokens, *raw_tokens): + for token in (*raw_tokens, *tokens): if not token or token.startswith('-') or _ENV_ASSIGNMENT_RE.match(token): continue token = _strip_redirection_operator(token) diff --git a/tests/test_pre_push_hook_contract.py b/tests/test_pre_push_hook_contract.py index b38a5d45ee..0cc1f2872f 100644 --- a/tests/test_pre_push_hook_contract.py +++ b/tests/test_pre_push_hook_contract.py @@ -1,22 +1,50 @@ -from __future__ import annotations - import os +import shutil import subprocess import unittest from pathlib import Path +def get_bash_executable() -> str | None: + if os.name == 'nt': + for candidate in ( + r'C:\Program Files\Git\bin\bash.exe', + r'C:\Program Files\Git\usr\bin\bash.exe', + r'C:\Program Files (x86)\Git\bin\bash.exe', + os.path.expandvars(r'%LOCALAPPDATA%\Programs\Git\bin\bash.exe'), + ): + if os.path.exists(candidate): + return candidate + bash = shutil.which('bash') + if bash and 'WindowsApps' not in bash: + return bash + return None + + +def require_bash() -> bool: + bash = get_bash_executable() + if not bash: + return False + try: + res = subprocess.run([bash, '-c', 'echo 1'], capture_output=True, text=True, timeout=2) + return res.returncode == 0 + except Exception: + return False + + REPO_ROOT = Path(__file__).resolve().parents[1] PRE_PUSH_HOOK = REPO_ROOT / '.github' / 'hooks' / 'pre-push' class PrePushHookContractTests(unittest.TestCase): + @unittest.skipUnless(require_bash(), 'Requires bash') def test_skip_escape_hatch_exits_successfully_with_stderr_notice(self) -> None: + bash_cmd = get_bash_executable() or 'bash' env = os.environ.copy() env['SKIP_CLAW_PRE_PUSH_BUILD'] = '1' result = subprocess.run( - ['bash', str(PRE_PUSH_HOOK)], + [bash_cmd, str(PRE_PUSH_HOOK)], cwd=REPO_ROOT, env=env, check=True, @@ -28,6 +56,7 @@ def test_skip_escape_hatch_exits_successfully_with_stderr_notice(self) -> None: self.assertIn('SKIP_CLAW_PRE_PUSH_BUILD=1', result.stderr) self.assertIn('skipping cargo workspace build', result.stderr) + @unittest.skipUnless(require_bash(), 'Requires bash') def test_default_build_gate_uses_workspace_locked_cargo_build(self) -> None: hook = PRE_PUSH_HOOK.read_text() diff --git a/tests/test_roadmap_helpers.py b/tests/test_roadmap_helpers.py index 3c8751980b..3b376357be 100644 --- a/tests/test_roadmap_helpers.py +++ b/tests/test_roadmap_helpers.py @@ -14,9 +14,40 @@ +import sys + +def get_bash_executable() -> str | None: + import os + if os.name == 'nt': + for candidate in ( + r'C:\Program Files\Git\bin\bash.exe', + r'C:\Program Files\Git\usr\bin\bash.exe', + r'C:\Program Files (x86)\Git\bin\bash.exe', + os.path.expandvars(r'%LOCALAPPDATA%\Programs\Git\bin\bash.exe'), + ): + if os.path.exists(candidate): + return candidate + bash = shutil.which('bash') + if bash and 'WindowsApps' not in bash: + return bash + return None + + +def require_bash() -> bool: + bash = get_bash_executable() + if not bash: + return False + try: + res = subprocess.run([bash, '-c', 'echo 1'], capture_output=True, text=True, timeout=2) + return res.returncode == 0 + except Exception: + return False + + def run_next_id(roadmap: Path, script: Path = NEXT_ID) -> subprocess.CompletedProcess[str]: + bash_cmd = get_bash_executable() or 'bash' return subprocess.run( - ['bash', str(script), str(roadmap)], + [bash_cmd, str(script), str(roadmap)], cwd=REPO_ROOT, capture_output=True, text=True, @@ -25,8 +56,9 @@ def run_next_id(roadmap: Path, script: Path = NEXT_ID) -> subprocess.CompletedPr def run_dogfood_probe(args: list[str]) -> subprocess.CompletedProcess[str]: + import sys return subprocess.run( - ['python3', str(DOGFOOD_PROBE), *args], + [sys.executable, str(DOGFOOD_PROBE), *args], cwd=REPO_ROOT, capture_output=True, text=True, @@ -35,6 +67,7 @@ def run_dogfood_probe(args: list[str]) -> subprocess.CompletedProcess[str]: class RoadmapHelperTests(unittest.TestCase): + @unittest.skipUnless(require_bash(), 'Requires bash') def test_roadmap_next_id_prints_only_next_id_after_duplicate_check(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: roadmap = Path(temp_dir) / 'ROADMAP.md' @@ -46,6 +79,7 @@ def test_roadmap_next_id_prints_only_next_id_after_duplicate_check(self) -> None self.assertEqual('725\n', result.stdout) self.assertEqual('', result.stderr) + @unittest.skipUnless(require_bash(), 'Requires bash') def test_roadmap_next_id_fails_fast_on_helper_era_duplicate(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: roadmap = Path(temp_dir) / 'ROADMAP.md' @@ -59,6 +93,7 @@ def test_roadmap_next_id_fails_fast_on_helper_era_duplicate(self) -> None: self.assertIn('999', result.stderr) self.assertNotIn('1000', result.stdout) + @unittest.skipUnless(require_bash(), 'Requires bash') def test_roadmap_next_id_fails_when_explicit_roadmap_path_is_missing(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: roadmap = Path(temp_dir) / 'missing-ROADMAP.md' @@ -70,6 +105,7 @@ def test_roadmap_next_id_fails_when_explicit_roadmap_path_is_missing(self) -> No self.assertIn('ROADMAP not found', result.stderr) self.assertIn(str(roadmap), result.stderr) + @unittest.skipUnless(require_bash(), 'Requires bash') def test_roadmap_next_id_fails_closed_when_checker_is_unavailable(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: script_dir = Path(temp_dir) / 'scripts' @@ -100,7 +136,7 @@ def test_dogfood_probe_runs_explicit_argv_and_separates_channels(self) -> None: result = run_dogfood_probe([ '--stdout-json-byte0', '--', - 'python3', + sys.executable, str(fixture), '--output-format', 'json', @@ -112,7 +148,7 @@ def test_dogfood_probe_runs_explicit_argv_and_separates_channels(self) -> None: payload = __import__('json').loads(result.stdout) self.assertEqual('ok', payload['kind']) self.assertEqual([ - 'python3', + sys.executable, str(fixture), '--output-format', 'json', @@ -120,15 +156,16 @@ def test_dogfood_probe_runs_explicit_argv_and_separates_channels(self) -> None: '--help', ], payload['argv']) self.assertEqual(0, payload['returncode']) - self.assertEqual('{"argv": ["--output-format", "json", "doctor", "--help"]}\n', payload['stdout']) - self.assertEqual('diagnostic\n', payload['stderr']) + self.assertEqual('{"argv": ["--output-format", "json", "doctor", "--help"]}\n', payload['stdout'].replace('\r\n', '\n')) + self.assertEqual('diagnostic\n', payload['stderr'].replace('\r\n', '\n')) def test_dogfood_probe_labels_timeout_separately_from_product_error(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: fixture = Path(temp_dir) / 'sleep.py' fixture.write_text('import time\ntime.sleep(2)\n') - result = run_dogfood_probe(['--timeout', '0.1', '--', 'python3', str(fixture)]) + import sys + result = run_dogfood_probe(['--timeout', '0.1', '--', sys.executable, str(fixture)]) self.assertEqual(1, result.returncode) payload = __import__('json').loads(result.stdout) @@ -151,7 +188,7 @@ def test_dogfood_probe_labels_stdout_json_prefix_failure_as_product_error(self) fixture = Path(temp_dir) / 'prefixed.py' fixture.write_text('print("warning before json")\nprint("{}")\n') - result = run_dogfood_probe(['--stdout-json-byte0', '--', 'python3', str(fixture)]) + result = run_dogfood_probe(['--stdout-json-byte0', '--', sys.executable, str(fixture)]) self.assertEqual(1, result.returncode) payload = __import__('json').loads(result.stdout) diff --git a/tests/test_security_scope.py b/tests/test_security_scope.py index 59275dda78..2e4e38fa35 100644 --- a/tests/test_security_scope.py +++ b/tests/test_security_scope.py @@ -12,6 +12,33 @@ from src.tools import execute_tool +def _create_directory_link(target: Path, link: Path) -> None: + """Create a directory symlink or fallback to an NTFS junction on Windows. + + Standard Windows user accounts cannot create symbolic links without + SeCreateSymbolicLinkPrivilege (Developer Mode / Elevation), but NTFS + directory junctions can be created unprivileged and exercise the exact same + path resolution logic in Path.resolve(). + + Note: NTFS junctions can only target local directory paths and cannot point + at UNC/remote targets. Links resolving to remote/UNC paths are covered + deterministically via `test_symlink_resolving_to_unc_escape_mocked`. + """ + try: + link.symlink_to(target, target_is_directory=True) + except OSError as e: + if getattr(e, 'winerror', None) == 1314 and os.name == 'nt': + try: + import _winapi + _winapi.CreateJunction(str(target), str(link)) + return + except Exception: + pass + self_skip_msg = 'Requires filesystem symlink or junction support on Windows runner' + raise unittest.SkipTest(self_skip_msg) from e + raise + + class WorkspacePathScopeTests(unittest.TestCase): def test_direct_parent_escape_is_denied(self) -> None: with tempfile.TemporaryDirectory() as tmp: @@ -30,13 +57,71 @@ def test_issue_3007_symlink_escape_is_denied(self) -> None: outside.mkdir() (outside / 'secret.txt').write_text('secret') link = workspace / 'linked-outside' - link.symlink_to(outside, target_is_directory=True) + _create_directory_link(outside, link) decision = WorkspacePathScope.from_root(workspace).validate_payload('cat linked-outside/secret.txt') self.assertFalse(decision.allowed) self.assertIn(str(outside.resolve()), decision.resolved or '') + def test_windows_absolute_symlink_escape_is_denied(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + workspace = root / 'workspace' + outside = root / 'outside' + workspace.mkdir() + outside.mkdir() + (outside / 'secret.txt').write_text('secret') + link = workspace / 'linked-outside' + _create_directory_link(outside, link) + + payload = f'cat {link}/secret.txt' + decision = WorkspacePathScope.from_root(workspace).validate_payload(payload) + + self.assertFalse(decision.allowed) + self.assertIn('outside workspace scope', decision.reason) + + def test_symlink_resolution_escape_mocked(self) -> None: + """Verify containment check catches escapes via resolve() even if unprivileged.""" + with tempfile.TemporaryDirectory() as tmp: + workspace = Path(tmp) / 'workspace' + workspace.mkdir() + scope = WorkspacePathScope.from_root(workspace) + + from unittest.mock import patch + fake_target = (Path(tmp) / 'outside' / 'secret.txt').resolve() + with patch.object(Path, 'resolve', return_value=fake_target): + decision = scope.validate_path(str(workspace / 'fake-link' / 'secret.txt')) + self.assertFalse(decision.allowed) + self.assertIn('outside workspace scope', decision.reason) + + def test_symlink_resolving_to_unc_escape_mocked(self) -> None: + """Verify containment check denies links resolving to remote/UNC targets.""" + with tempfile.TemporaryDirectory() as tmp: + workspace = Path(tmp) / 'workspace' + workspace.mkdir() + scope = WorkspacePathScope.from_root(workspace) + + from unittest.mock import patch + unc_target = Path(r'\\remote-server\share\secret.txt') + with patch.object(Path, 'resolve', return_value=unc_target): + decision = scope.validate_path(str(workspace / 'net-link' / 'secret.txt')) + self.assertFalse(decision.allowed) + self.assertIn('outside workspace scope', decision.reason) + + def test_unresolvable_path_raises_oserror_is_denied(self) -> None: + """Verify that paths raising OSError during resolve() are explicitly denied.""" + with tempfile.TemporaryDirectory() as tmp: + workspace = Path(tmp) / 'workspace' + workspace.mkdir() + scope = WorkspacePathScope.from_root(workspace) + + from unittest.mock import patch + with patch.object(Path, 'resolve', side_effect=OSError('dangling symlink or filesystem error')): + decision = scope.validate_path(str(workspace / 'broken_link.txt')) + self.assertFalse(decision.allowed) + self.assertIn('cannot be resolved', decision.reason) + def test_glob_expansion_must_stay_inside_workspace(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) @@ -107,6 +192,11 @@ def test_explicit_worktree_roots_are_allowed(self) -> None: self.assertTrue(decision.allowed, decision.reason) + def test_extract_path_candidates_preserves_unc_paths(self) -> None: + payload = r'type \\server\share\secret.txt' + candidates = extract_path_candidates(payload) + self.assertIn(r'\\server\share\secret.txt', candidates) + def test_windows_absolute_paths_are_denied_for_posix_workspace(self) -> None: with tempfile.TemporaryDirectory() as tmp: workspace = Path(tmp) / 'workspace' @@ -116,9 +206,13 @@ def test_windows_absolute_paths_are_denied_for_posix_workspace(self) -> None: unc_decision = WorkspacePathScope.from_root(workspace).validate_payload(r'type \\server\share\secret.txt') self.assertFalse(drive_decision.allowed) - self.assertIn('windows absolute path', drive_decision.reason) self.assertFalse(unc_decision.allowed) - self.assertIn('windows absolute path', unc_decision.reason) + if os.name == 'nt': + self.assertIn('outside workspace scope', drive_decision.reason) + self.assertIn('outside workspace scope', unc_decision.reason) + else: + self.assertIn('windows absolute path', drive_decision.reason) + self.assertIn('windows absolute path', unc_decision.reason) def test_file_and_shell_tools_use_workspace_scope_context(self) -> None: with tempfile.TemporaryDirectory() as tmp: