Skip to content
30 changes: 26 additions & 4 deletions src/path_scope.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,22 +60,44 @@ 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,
'path resolves outside workspace scope',
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)
Expand Down Expand Up @@ -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)
Expand Down
35 changes: 32 additions & 3 deletions tests/test_pre_push_hook_contract.py
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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()

Expand Down
53 changes: 45 additions & 8 deletions tests/test_roadmap_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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'
Expand All @@ -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'
Expand All @@ -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'
Expand All @@ -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'
Expand Down Expand Up @@ -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',
Expand All @@ -112,23 +148,24 @@ 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',
'doctor',
'--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)
Expand All @@ -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)
Expand Down
100 changes: 97 additions & 3 deletions tests/test_security_scope.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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)
Expand Down Expand Up @@ -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'
Expand All @@ -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:
Expand Down