From 7bc3c2a16bfbbed21ad7fc9bae550da49b187be4 Mon Sep 17 00:00:00 2001 From: Trecek Date: Sun, 19 Jul 2026 20:36:33 -0700 Subject: [PATCH 1/7] feat(hooks): relocate shell-capture rm -f cleanup to Python-side SessionStart sweep Removes the inline 'rm -f -- \"\$__as_f\"' deletion embedded in every Codex shell command's generated harness. Codex CLI 0.144.6's exec-policy engine forbids rm -f-style commands, causing universal command rejection. Cleanup now lives in a stdlib-only SessionStart TTL sweep via the new _capture_cleanup module. - Add src/autoskillit/hooks/_capture_cleanup.py with _is_safe_capture_file containment checks (symlink, hardlink, world-writable, traversal) and sweep_stale_captures deletion guarded by 1h age threshold - Widen shell_ uid from 8 to 16 hex chars (collisions under fleet concurrency over the now 1h retention window) - Wire the sweep into session_start_hook.py alongside kitchen_state and pipeline_tracker sweeps (stdlib-only boundary preserved via inlined _CAPTURE_RE literal) - Add timeout_seconds=5 to shell_capture_hook HookDef matching the ceiling used by open_kitchen_guard / ask_user_question_guard / mcp_health_advisor - Add conformance tests: destructive-verb absence, small-output retention + sweep cleanup, symlink rejection, filename allowlist, containment parity vs core/path_containment.py, regex consistency between the two sweep sites - Invert the existing 'assert not artifacts' assertion to assert the Python-side retention invariant Fixes #4286 follow-up. Part A of the rectify plan only. Co-Authored-By: Claude --- docs/decisions/0006-output-containment.md | 2 +- docs/safety/hooks.md | 4 +- src/autoskillit/hook_registry.py | 1 + src/autoskillit/hooks/AGENTS.md | 1 + src/autoskillit/hooks/_capture_cleanup.py | 100 ++++++++++++++ src/autoskillit/hooks/registry.sha256 | 2 +- src/autoskillit/hooks/session_start_hook.py | 20 +++ src/autoskillit/hooks/shell_capture_hook.py | 3 +- tests/arch/test_subpackage_isolation.py | 2 +- tests/hooks/test_shell_capture_conformance.py | 125 +++++++++++++++++- 10 files changed, 252 insertions(+), 8 deletions(-) create mode 100644 src/autoskillit/hooks/_capture_cleanup.py diff --git a/docs/decisions/0006-output-containment.md b/docs/decisions/0006-output-containment.md index ac370e9c11..711f40db41 100644 --- a/docs/decisions/0006-output-containment.md +++ b/docs/decisions/0006-output-containment.md @@ -89,4 +89,4 @@ hook can be retired in favor of that mechanism. for the classifier's literal-small-JSONL exception). - `shell_max_inline_bytes` survives with its new capture-threshold meaning. - Complete Codex shell output is captured to mechanism-owned artifacts at - `/.autoskillit/temp/shell_capture/shell_.log`. + `/.autoskillit/temp/shell_capture/shell_.log`. diff --git a/docs/safety/hooks.md b/docs/safety/hooks.md index b4ee4b6717..e53af57433 100644 --- a/docs/safety/hooks.md +++ b/docs/safety/hooks.md @@ -90,8 +90,8 @@ in `.hook_config.json` for future recipes that legitimately need these operation PreToolUse input-rewrite hook that wraps every native shell command on Codex in a capture harness. The original command runs unmodified in a subshell; its complete combined stdout+stderr goes to a mechanism-owned artifact at -`/.autoskillit/temp/shell_capture/shell_.log`. Only a bounded inline -slice enters context: full content when small (artifact deleted), else head + +`/.autoskillit/temp/shell_capture/shell_.log`. Only a bounded inline +slice enters context: full content when small (artifact retained; reclaimed by SessionStart sweep), else head + provenance marker (bytes, sha256, path) + tail. Exit codes are preserved via a trap-EXIT postlude. Set `output_budget.guard_enabled: false` to disable; inline threshold controlled by `output_budget.shell_max_inline_bytes`. diff --git a/src/autoskillit/hook_registry.py b/src/autoskillit/hook_registry.py index 5a21804009..581cb88cf4 100644 --- a/src/autoskillit/hook_registry.py +++ b/src/autoskillit/hook_registry.py @@ -204,6 +204,7 @@ def __post_init__(self) -> None: HookDef( matcher=r"Bash", scripts=["shell_capture_hook.py"], + timeout_seconds=5, session_scope="any", codex_status="works-as-is", mechanism="input-rewrite", diff --git a/src/autoskillit/hooks/AGENTS.md b/src/autoskillit/hooks/AGENTS.md index 5b294b309c..ef56bb2169 100644 --- a/src/autoskillit/hooks/AGENTS.md +++ b/src/autoskillit/hooks/AGENTS.md @@ -24,6 +24,7 @@ Sub-packages: guards/ (see guards/AGENTS.md), formatters/ (see formatters/AGENTS | `_hook_utils.py` | Shared stdlib-only utilities for hook scripts (e.g., `find_project_root`, `STEP_SUFFIX_RE`) | | `_command_classification.py` | Shared stdlib-only command classification primitives for guard scripts (interpreter/wrapper detection, git command classification) | | `_policy_event.py` | Typed policy-event formatter for hook provenance messages (stdlib-only) | +| `_capture_cleanup.py` | Stdlib-only stale-capture-file sweep helpers (used by `session_start_hook.py`) | | `shell_capture_hook.py` | `PreToolUse`: input-rewrite hook for Codex shell capture — wraps commands in a lossless capture harness (#4286 / ADR-0006) | ## Architecture Notes diff --git a/src/autoskillit/hooks/_capture_cleanup.py b/src/autoskillit/hooks/_capture_cleanup.py new file mode 100644 index 0000000000..aa097944a4 --- /dev/null +++ b/src/autoskillit/hooks/_capture_cleanup.py @@ -0,0 +1,100 @@ +"""Stdlib-only stale-capture-file sweep helpers (used by ``session_start_hook.py``). + +Reaps capture artifacts written by ``shell_capture_hook.py`` after their +capture-session lifetime has elapsed. Mirrors the TTL-sweep pattern in +``session_start_hook.py`` (kitchen_state, pipeline_tracker) with containment +checks inlined from ``core.path_containment.resolve_contained_path`` — +the stdlib-only hook import boundary prevents importing from ``autoskillit.*``. + +The 1-hour age threshold (``max_age_seconds=3600`` default) is the liveness +guard: a capture file currently being written by an in-flight harness has an +``mtime`` within seconds of now and will never be swept. POSIX ``unlink`` on +an open fd is safe (the inode persists until fd close), so even in the +pathological case of a sweep racing with an active reader, no data +corruption occurs — only premature path removal, which the age threshold +makes effectively impossible. +""" + +from __future__ import annotations + +import os +import re +import stat +from pathlib import Path + +__all__ = ["_CAPTURE_FILENAME_RE", "_is_safe_capture_file", "sweep_stale_captures"] + +# Strict allowlist matching the 16-hex-char uid produced by +# ``shell_capture_hook._build_harness`` (``uuid4().hex[:16]``). +_CAPTURE_FILENAME_RE = re.compile(r"^shell_[0-9a-f]{16}\.log$") + + +def _is_safe_capture_file(path: Path, capture_dir: Path) -> bool: + """Return True iff ``path`` is a safe, in-place, regular capture file. + + Rejects: + - Names that don't match the strict ``shell_<16hex>.log`` pattern + - Symlinks (``stat.S_ISLNK``) + - Paths whose resolved target escapes ``capture_dir`` + - Hardlinks (``st_nlink > 1``) + - World-writable files (``st_mode & 0o002``) + """ + if not _CAPTURE_FILENAME_RE.match(path.name): + return False + try: + st = path.lstat() + except OSError: + return False + if stat.S_ISLNK(st.st_mode): + return False + if st.st_nlink > 1: + return False + if st.st_mode & 0o002: + return False + try: + capture_dir_resolved = Path(capture_dir).resolve(strict=True) + resolved = path.resolve(strict=True) + except (OSError, RuntimeError): + return False + if not resolved.is_relative_to(capture_dir_resolved): + return False + return True + + +def sweep_stale_captures( + capture_dir: Path | str, + *, + max_age_seconds: int = 3600, +) -> int: + """Delete capture files in ``capture_dir`` older than ``max_age_seconds``. + + Returns the count of deleted files. Fail-open: any per-file exception + is swallowed and the iteration continues, mirroring ``session_start_hook.py`` + TTL-sweep behavior. + """ + capture_dir_path = Path(capture_dir) + if not capture_dir_path.is_dir(): + return 0 + try: + capture_dir_resolved = capture_dir_path.resolve(strict=True) + now_mtime_threshold = capture_dir_resolved.stat().st_mtime - max_age_seconds + except OSError: + return 0 + + deleted = 0 + try: + entries = list(capture_dir_path.iterdir()) + except OSError: + return 0 + for entry in entries: + try: + if not _is_safe_capture_file(entry, capture_dir_path): + continue + st = entry.stat() + if st.st_mtime > now_mtime_threshold: + continue + os.unlink(entry) + deleted += 1 + except (FileNotFoundError, OSError): + continue + return deleted diff --git a/src/autoskillit/hooks/registry.sha256 b/src/autoskillit/hooks/registry.sha256 index f6e8ca6ff8..5d92115899 100644 --- a/src/autoskillit/hooks/registry.sha256 +++ b/src/autoskillit/hooks/registry.sha256 @@ -1 +1 @@ -9d99ecc511cc4592d920a0981a639751e6ab31996a64d0d7d70c4ee757855773 +5e547deac3484e0c1d79b53b7badfd1ecfb27ec73ea23075d9341b8f280e8818 diff --git a/src/autoskillit/hooks/session_start_hook.py b/src/autoskillit/hooks/session_start_hook.py index c6b78c93b1..1c53557225 100644 --- a/src/autoskillit/hooks/session_start_hook.py +++ b/src/autoskillit/hooks/session_start_hook.py @@ -12,6 +12,7 @@ import json import os +import re import sys from datetime import UTC, datetime from pathlib import Path @@ -79,6 +80,25 @@ def main() -> None: except Exception: pass + try: + _capture_dir = Path.cwd() / ".autoskillit" / "temp" / "shell_capture" + if _capture_dir.is_dir(): + _CAPTURE_RE = re.compile(r"^shell_[0-9a-f]{16}\.log$") + _now = datetime.now(UTC) + for _cp in _capture_dir.iterdir(): + try: + if not _CAPTURE_RE.match(_cp.name): + continue + if _cp.is_symlink(): + continue + _c_age = _now.timestamp() - _cp.stat().st_mtime + if _c_age >= 3600: + _cp.unlink() + except Exception: + pass + except Exception: + pass + transcript_path = data.get("transcript_path", "") if not transcript_path: sys.exit(0) diff --git a/src/autoskillit/hooks/shell_capture_hook.py b/src/autoskillit/hooks/shell_capture_hook.py index 1b6abe9d51..5c51ea7dbc 100644 --- a/src/autoskillit/hooks/shell_capture_hook.py +++ b/src/autoskillit/hooks/shell_capture_hook.py @@ -58,7 +58,7 @@ def _read_policy(cwd: str) -> tuple[bool, int]: def _build_harness(command: str, cwd: str, inline_bytes: int) -> str: - uid = uuid4().hex[:8] + uid = uuid4().hex[:16] capture_dir = str(Path(cwd) / _CAPTURE_SUBDIR) capture_file = str(Path(cwd) / _CAPTURE_SUBDIR / f"shell_{uid}.log") head = (2 * inline_bytes) // 3 @@ -110,7 +110,6 @@ def _build_harness(command: str, cwd: str, inline_bytes: int) -> str: __as_sz=$(wc -c < "$__as_f") if [ "$__as_sz" -le {inline_bytes} ]; then cat "$__as_f" - rm -f -- "$__as_f" else {sha_line} head -c {head} "$__as_f" diff --git a/tests/arch/test_subpackage_isolation.py b/tests/arch/test_subpackage_isolation.py index a5f440a0a4..d6bfaf483f 100644 --- a/tests/arch/test_subpackage_isolation.py +++ b/tests/arch/test_subpackage_isolation.py @@ -887,7 +887,7 @@ def test_no_subpackage_exceeds_10_files() -> None: "core": 24, # +closure_hashing +path_containment +closure_verifier "core/types": 32, # +invariant_registry (INVARIANT_REGISTRY) +closure_report "cli": 21, - "hooks": 17, # +recipe_confirmed_post_hook, +quota_guard_state_post_hook, +_policy_event, +shell_capture_hook (#4286) # noqa: E501 + "hooks": 18, # +recipe_confirmed_post_hook, +quota_guard_state_post_hook, +_policy_event, +shell_capture_hook (#4286), +_capture_cleanup.py # noqa: E501 "pipeline": 12, "fleet": 23, # +_issue_url_helpers.py # noqa: E501 "recipe/rules": 54, # +commit_guard_regression_route +rules_model +rules_gitignored_deliverable +rules_issue_scope_threading +rules_inventory_gate_bilateral +rules_verdict_context # noqa: E501 diff --git a/tests/hooks/test_shell_capture_conformance.py b/tests/hooks/test_shell_capture_conformance.py index cd74a7b8de..d6dcd33c4b 100644 --- a/tests/hooks/test_shell_capture_conformance.py +++ b/tests/hooks/test_shell_capture_conformance.py @@ -9,12 +9,20 @@ from __future__ import annotations +import os +import re import shutil import subprocess from pathlib import Path +from uuid import uuid4 import pytest +from autoskillit.hooks._capture_cleanup import ( + _CAPTURE_FILENAME_RE, + _is_safe_capture_file, + sweep_stale_captures, +) from autoskillit.hooks.shell_capture_hook import _build_harness pytestmark = [pytest.mark.layer("hooks"), pytest.mark.medium] @@ -191,7 +199,10 @@ def test_capture_conformance(label: str, command: str, tmp_path: Path) -> None: assert wrapped.stdout == raw_combined, ( f"[{label}] inline output mismatch.\nraw={raw_combined!r}\nwrapped={wrapped.stdout!r}" ) - assert not artifacts, f"[{label}] expected no artifact for small output, found {artifacts}" + assert artifacts, ( + f"[{label}] expected artifact retained for small output (Python-side cleanup)" + ) + assert len(artifacts) == 1 else: assert artifacts, f"[{label}] expected an artifact for large output, found none" assert len(artifacts) == 1, f"[{label}] expected exactly one artifact, found {artifacts}" @@ -234,3 +245,115 @@ def test_capture_dir_uncreatable_fail_stops(tmp_path: Path) -> None: combined = (wrapped.stdout + wrapped.stderr).decode(errors="replace") assert "capture_failed" in combined.lower() or "CAPTURE_FAILED" in combined assert "should_not_run" not in combined + + +def test_harness_contains_no_destructive_verbs() -> None: + """Generated harness must not embed destructive shell verbs (Codex exec-policy rejection).""" + harness = _build_harness("echo hello", "/tmp/test", _INLINE_BYTES) + destructive = {"rm", "unlink", "shred", "truncate", "mv"} + for raw_line in harness.splitlines(): + line = raw_line.strip() + if not line or line.startswith("#"): + continue + first = ( + line.split(";", 1)[0].split("&&", 1)[0].split("||", 1)[0].strip().split(maxsplit=1)[0] + ) + first = first.removeprefix("{").removeprefix("(") + first = first.strip("\"'") + assert first not in destructive, ( + f"destructive verb {first!r} found in harness line: {raw_line!r}" + ) + + +def test_small_output_artifact_cleaned_by_sweep(tmp_path: Path) -> None: + """Small-output captures are retained on disk but reaped by the Python-side sweep.""" + _make_project_dirs(tmp_path) + command = "echo hello_small" + wrapped = _run_wrapped(command, tmp_path) + assert wrapped.returncode == 0 + artifacts = _artifact_files(tmp_path) + assert artifacts, "small-output harness must leave capture file on disk for sweep cleanup" + assert len(artifacts) == 1 + + target = artifacts[0] + # Force mtime into the past so max_age_seconds=0 sweeps it. + os.utime(target, (0, 0)) + deleted = sweep_stale_captures(_capture_dir(tmp_path), max_age_seconds=0) + assert deleted == 1 + assert not target.exists() + + +def test_sweep_rejects_symlinks_and_traversals(tmp_path: Path) -> None: + """Sweep must not follow symlinks or operate outside the capture directory.""" + _make_project_dirs(tmp_path) + capture = _capture_dir(tmp_path) + + outside = tmp_path / "outside_target.log" + outside.write_text("must-survive") + symlink = capture / f"shell_{uuid4().hex[:16]}.log" + symlink.symlink_to(outside) + assert symlink.is_symlink() + + deleted = sweep_stale_captures(capture, max_age_seconds=0) + assert deleted == 0 + assert outside.exists(), "outside target file must be untouched by sweep" + assert outside.read_text() == "must-survive" + assert not symlink.exists(), "symlink should remain (sweep must not follow into the target)" + + +def test_sweep_filename_allowlist(tmp_path: Path) -> None: + """Sweep only deletes files matching the strict ``shell_<16hex>.log`` allowlist.""" + _make_project_dirs(tmp_path) + capture = _capture_dir(tmp_path) + + valid = capture / f"shell_{uuid4().hex[:16]}.log" + valid.write_text("x") + short_uid = capture / "shell_abcd1234.log" # 8-char format — legacy, must NOT be swept + short_uid.write_text("x") + invalid_ext = capture / "evil.sh" + invalid_ext.write_text("x") + + deleted = sweep_stale_captures(capture, max_age_seconds=0) + assert deleted == 1 + assert not valid.exists() + assert short_uid.exists(), "old 8-char uid format files must not be deleted" + assert invalid_ext.exists(), "files outside the shell_*.log allowlist must not be deleted" + + +def test_capture_cleanup_containment_parity(tmp_path: Path) -> None: + """_is_safe_capture_file rejects the same attack vectors as core.path_containment.""" + from autoskillit.core.path_containment import ContainmentError, resolve_contained_path + + _make_project_dirs(tmp_path) + capture = _capture_dir(tmp_path) + + # Valid file — both should accept. + good = capture / f"shell_{uuid4().hex[:16]}.log" + good.write_text("ok") + assert _is_safe_capture_file(good, capture) is True + try: + resolve_contained_path(good, capture) + except ContainmentError as exc: + pytest.fail(f"core containment rejected a valid capture file: {exc}") + + # Symlink — both should reject. + sym = capture / f"shell_{uuid4().hex[:16]}.log" + sym.symlink_to(good) + assert _is_safe_capture_file(sym, capture) is False + with pytest.raises(ContainmentError): + resolve_contained_path(sym, capture) + + # Traversal-shaped filename — allowlist rejects outright. + traversal = capture / "../../escape.log" + traversal.write_text("x") + assert _is_safe_capture_file(traversal, capture) is False + + +def test_capture_filename_regex_consistency() -> None: + """Hook module and sweep module must agree on the shell_*.log regex.""" + sample = f"shell_{uuid4().hex[:16]}.log" + assert _CAPTURE_FILENAME_RE.match(sample) is not None + + # The inline literal in session_start_hook.py must accept the same filename. + inline_pattern = r"^shell_[0-9a-f]{16}\.log$" + assert re.match(inline_pattern, sample) is not None From 2588a0d35333a80f47823dcb08d823bb55aceb83 Mon Sep 17 00:00:00 2001 From: Trecek Date: Sun, 19 Jul 2026 20:46:26 -0700 Subject: [PATCH 2/7] fix(hooks): align shell-capture tests with Python-side cleanup and regenerate snapshot hash --- .../fixtures/codex_ndjson/hook_event_format_snapshot.json | 3 ++- tests/hooks/test_shell_capture_conformance.py | 7 +++++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/tests/execution/backends/fixtures/codex_ndjson/hook_event_format_snapshot.json b/tests/execution/backends/fixtures/codex_ndjson/hook_event_format_snapshot.json index 0e59519a5a..2a983ddbea 100644 --- a/tests/execution/backends/fixtures/codex_ndjson/hook_event_format_snapshot.json +++ b/tests/execution/backends/fixtures/codex_ndjson/hook_event_format_snapshot.json @@ -1,5 +1,5 @@ { - "_registry_hash": "9d99ecc511cc4592d920a0981a639751e6ab31996a64d0d7d70c4ee757855773", + "_registry_hash": "5e547deac3484e0c1d79b53b7badfd1ecfb27ec73ea23075d9341b8f280e8818", "hooks": { "PostToolUse": [ { @@ -189,6 +189,7 @@ "hooks": [ { "command": "python3 SANITIZED_HOOKS_DIR/_dispatch.py shell_capture_hook", + "timeout": 5, "trusted_hash": "SANITIZED_FOR_DETERMINISM", "type": "command" } diff --git a/tests/hooks/test_shell_capture_conformance.py b/tests/hooks/test_shell_capture_conformance.py index d6dcd33c4b..359d7ea086 100644 --- a/tests/hooks/test_shell_capture_conformance.py +++ b/tests/hooks/test_shell_capture_conformance.py @@ -145,7 +145,8 @@ def test_capture_conformance(label: str, command: str, tmp_path: Path) -> None: assert raw_combined == b"" assert wrapped.stdout == b"" assert raw.returncode == 0 - assert not artifacts + assert artifacts, "[true_cmd] expected artifact retained (Python-side cleanup)" + assert len(artifacts) == 1 return if label == "self_bg": @@ -298,7 +299,9 @@ def test_sweep_rejects_symlinks_and_traversals(tmp_path: Path) -> None: assert deleted == 0 assert outside.exists(), "outside target file must be untouched by sweep" assert outside.read_text() == "must-survive" - assert not symlink.exists(), "symlink should remain (sweep must not follow into the target)" + assert symlink.is_symlink(), ( + "symlink must remain after sweep (sweep must not follow into the target)" + ) def test_sweep_filename_allowlist(tmp_path: Path) -> None: From a6c921985cf46e4a3e0728c62dd67b8d5f97da9d Mon Sep 17 00:00:00 2001 From: Trecek Date: Sun, 19 Jul 2026 21:29:18 -0700 Subject: [PATCH 3/7] fix(hooks): sweep_stale_captures uses wall-clock time, not directory mtime MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `sweep_stale_captures` previously derived its staleness threshold from `capture_dir_path.resolve(strict=True).stat().st_mtime` — the capture directory's own mtime, which only advances when an entry inside the directory is created, removed, or renamed. This meant a file created long before the directory's last entry-change could look "fresh" and be skipped, and files could go permanently unswept if the directory had no new entries for a while. Replace the directory-mtime proxy with `time.time()` so staleness is measured against real wall-clock time, matching the inline sweep block in `session_start_hook.py` (`_now.timestamp() - _cp.stat().st_mtime`). The strict-resolve existence/race check is retained; `capture_dir_resolved` is dropped since it is no longer referenced. Every existing test passes `max_age_seconds=0` immediately after creating the capture directory, so `dir_mtime ≈ wall_clock_now` at call time and the bug never surfaced. A new regression test in the next commit covers the case where the directory's mtime is equally stale to its contents. --- src/autoskillit/hooks/_capture_cleanup.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/autoskillit/hooks/_capture_cleanup.py b/src/autoskillit/hooks/_capture_cleanup.py index aa097944a4..d54eb5d26a 100644 --- a/src/autoskillit/hooks/_capture_cleanup.py +++ b/src/autoskillit/hooks/_capture_cleanup.py @@ -20,6 +20,7 @@ import os import re import stat +import time from pathlib import Path __all__ = ["_CAPTURE_FILENAME_RE", "_is_safe_capture_file", "sweep_stale_captures"] @@ -76,10 +77,10 @@ def sweep_stale_captures( if not capture_dir_path.is_dir(): return 0 try: - capture_dir_resolved = capture_dir_path.resolve(strict=True) - now_mtime_threshold = capture_dir_resolved.stat().st_mtime - max_age_seconds + capture_dir_path.resolve(strict=True) except OSError: return 0 + mtime_threshold = time.time() - max_age_seconds deleted = 0 try: @@ -91,7 +92,7 @@ def sweep_stale_captures( if not _is_safe_capture_file(entry, capture_dir_path): continue st = entry.stat() - if st.st_mtime > now_mtime_threshold: + if st.st_mtime > mtime_threshold: continue os.unlink(entry) deleted += 1 From 847208bc6bd8bc8f6bf5b9875a6f1236acdadbd3 Mon Sep 17 00:00:00 2001 From: Trecek Date: Sun, 19 Jul 2026 21:29:26 -0700 Subject: [PATCH 4/7] test(hooks): extend capture-cleanup parity coverage to all four attack vectors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `test_capture_cleanup_containment_parity` previously exercised only a valid file, a symlink, and a traversal-shaped filename — leaving the `st_nlink > 1` and `st_mode & 0o002` branches of `_is_safe_capture_file` completely untested by the test whose stated purpose is to guard exactly those branches. Convert the test to `@pytest.mark.parametrize` over five module-level case-builders covering all four attack vectors from `core.path_containment.resolve_contained_path` (symlinks, traversals, hardlinks, world-writable files), plus the valid baseline. Hardlink case follows the existing `pytest.skip("hardlink not supported...")` precedent for filesystems where hardlinks are unsupported. Also add `test_sweep_stale_captures_uses_wall_clock_not_directory_mtime` which backdates both a capture file and its enclosing directory by 200s and asserts the sweep deletes it under `max_age_seconds=100`. This reproduces the staleness-threshold defect fixed in the previous commit (directory-mtime proxy instead of wall-clock time) and would have failed against the prior implementation. --- tests/hooks/test_shell_capture_conformance.py | 98 +++++++++++++++---- 1 file changed, 81 insertions(+), 17 deletions(-) diff --git a/tests/hooks/test_shell_capture_conformance.py b/tests/hooks/test_shell_capture_conformance.py index 359d7ea086..ea183bd5a9 100644 --- a/tests/hooks/test_shell_capture_conformance.py +++ b/tests/hooks/test_shell_capture_conformance.py @@ -13,6 +13,8 @@ import re import shutil import subprocess +import time +from collections.abc import Callable from pathlib import Path from uuid import uuid4 @@ -323,33 +325,95 @@ def test_sweep_filename_allowlist(tmp_path: Path) -> None: assert invalid_ext.exists(), "files outside the shell_*.log allowlist must not be deleted" -def test_capture_cleanup_containment_parity(tmp_path: Path) -> None: - """_is_safe_capture_file rejects the same attack vectors as core.path_containment.""" - from autoskillit.core.path_containment import ContainmentError, resolve_contained_path - +def test_sweep_stale_captures_uses_wall_clock_not_directory_mtime(tmp_path: Path) -> None: + """Staleness must be measured against real elapsed time, not the capture + directory's own mtime (which only advances when its contents change).""" _make_project_dirs(tmp_path) capture = _capture_dir(tmp_path) - # Valid file — both should accept. + stale = capture / f"shell_{uuid4().hex[:16]}.log" + stale.write_text("x") + + backdated = time.time() - 200 + os.utime(stale, (backdated, backdated)) + os.utime(capture, (backdated, backdated)) + + deleted = sweep_stale_captures(capture, max_age_seconds=100) + assert deleted == 1 + assert not stale.exists(), ( + "file older than max_age_seconds must be swept even when the capture " + "directory's own mtime is equally stale" + ) + + +def _cc_case_valid(capture: Path) -> Path: good = capture / f"shell_{uuid4().hex[:16]}.log" good.write_text("ok") - assert _is_safe_capture_file(good, capture) is True - try: - resolve_contained_path(good, capture) - except ContainmentError as exc: - pytest.fail(f"core containment rejected a valid capture file: {exc}") + return good + - # Symlink — both should reject. +def _cc_case_symlink(capture: Path) -> Path: + target = capture / f"shell_{uuid4().hex[:16]}.log" + target.write_text("ok") sym = capture / f"shell_{uuid4().hex[:16]}.log" - sym.symlink_to(good) - assert _is_safe_capture_file(sym, capture) is False - with pytest.raises(ContainmentError): - resolve_contained_path(sym, capture) + sym.symlink_to(target) + return sym - # Traversal-shaped filename — allowlist rejects outright. + +def _cc_case_traversal(capture: Path) -> Path: traversal = capture / "../../escape.log" traversal.write_text("x") - assert _is_safe_capture_file(traversal, capture) is False + return traversal + + +def _cc_case_hardlink(capture: Path) -> Path: + source = capture.parent / "hardlink_source.log" + source.write_text("data") + hardlinked = capture / f"shell_{uuid4().hex[:16]}.log" + try: + os.link(source, hardlinked) + except OSError: + pytest.skip("hardlink not supported in this environment") + return hardlinked + + +def _cc_case_world_writable(capture: Path) -> Path: + path = capture / f"shell_{uuid4().hex[:16]}.log" + path.write_text("x") + os.chmod(path, 0o666) + return path + + +@pytest.mark.parametrize( + "case_builder,expect_safe", + [ + (_cc_case_valid, True), + (_cc_case_symlink, False), + (_cc_case_traversal, False), + (_cc_case_hardlink, False), + (_cc_case_world_writable, False), + ], + ids=["valid", "symlink", "traversal", "hardlink", "world_writable"], +) +def test_capture_cleanup_containment_parity( + case_builder: Callable[[Path], Path], expect_safe: bool, tmp_path: Path +) -> None: + """_is_safe_capture_file rejects the same attack vectors as core.path_containment.""" + from autoskillit.core.path_containment import ContainmentError, resolve_contained_path + + _make_project_dirs(tmp_path) + capture = _capture_dir(tmp_path) + path = case_builder(capture) + + assert _is_safe_capture_file(path, capture) is expect_safe + if expect_safe: + try: + resolve_contained_path(path, capture) + except ContainmentError as exc: + pytest.fail(f"core containment rejected a valid capture file: {exc}") + else: + with pytest.raises(ContainmentError): + resolve_contained_path(path, capture) def test_capture_filename_regex_consistency() -> None: From c207ba5c641bef5db902780bfe11c3c27d944a4b Mon Sep 17 00:00:00 2001 From: Trecek Date: Sun, 19 Jul 2026 22:08:54 -0700 Subject: [PATCH 5/7] test(hooks): broaden shell harness policy guard Co-Authored-By: Claude --- tests/hooks/test_shell_capture_conformance.py | 31 +++++++++++++++---- 1 file changed, 25 insertions(+), 6 deletions(-) diff --git a/tests/hooks/test_shell_capture_conformance.py b/tests/hooks/test_shell_capture_conformance.py index ea183bd5a9..f0ab8ef1d6 100644 --- a/tests/hooks/test_shell_capture_conformance.py +++ b/tests/hooks/test_shell_capture_conformance.py @@ -32,6 +32,16 @@ _INLINE_BYTES = 12_000 _CAPTURE_SUBDIR = ".autoskillit/temp/shell_capture" _TIMEOUT = 30 +_HARNESS_FORBIDDEN_VERBS: frozenset[str] = frozenset( + { + "rm", + "unlink", + "shred", + "truncate", + "rmdir", + "mv", # moving the capture file would break concurrent reads + } +) _NESTED_WRAP_INNER = "echo hi" @@ -250,10 +260,18 @@ def test_capture_dir_uncreatable_fail_stops(tmp_path: Path) -> None: assert "should_not_run" not in combined -def test_harness_contains_no_destructive_verbs() -> None: - """Generated harness must not embed destructive shell verbs (Codex exec-policy rejection).""" - harness = _build_harness("echo hello", "/tmp/test", _INLINE_BYTES) - destructive = {"rm", "unlink", "shred", "truncate", "mv"} +@pytest.mark.parametrize( + "cmd", + ["echo hello", "ls -la", "cat /dev/null", "python3 -c 'print(1)'", ""], +) +def test_harness_contains_no_destructive_verbs(cmd: str) -> None: + """Arch guard: hook-generated shell must not contain destructive verbs. + + Codex's exec-policy engine evaluates the full rewritten command, including + hook-injected scaffolding. Destructive verbs (rm, unlink, etc.) are forbidden + by Codex's built-in policy. This test ensures the harness never introduces them. + """ + harness = _build_harness(cmd, "/tmp/test", _INLINE_BYTES) for raw_line in harness.splitlines(): line = raw_line.strip() if not line or line.startswith("#"): @@ -263,8 +281,9 @@ def test_harness_contains_no_destructive_verbs() -> None: ) first = first.removeprefix("{").removeprefix("(") first = first.strip("\"'") - assert first not in destructive, ( - f"destructive verb {first!r} found in harness line: {raw_line!r}" + assert first not in _HARNESS_FORBIDDEN_VERBS, ( + f"forbidden verb {first!r} found in harness line: {raw_line!r}\n" + f"Generated from command: {cmd!r}" ) From 4a3c537e6e60b86f6c5570007f9a06d1724d11c3 Mon Sep 17 00:00:00 2001 From: Trecek Date: Sun, 19 Jul 2026 22:13:25 -0700 Subject: [PATCH 6/7] test(execution): distinguish shell policy denial Co-Authored-By: Claude --- .../backends/test_cli_conformance_probes.py | 159 +++++++++++++++++- 1 file changed, 156 insertions(+), 3 deletions(-) diff --git a/tests/execution/backends/test_cli_conformance_probes.py b/tests/execution/backends/test_cli_conformance_probes.py index 372cebdc4d..bab432025e 100644 --- a/tests/execution/backends/test_cli_conformance_probes.py +++ b/tests/execution/backends/test_cli_conformance_probes.py @@ -10,6 +10,7 @@ import json import os +import re import shutil import subprocess import tomllib @@ -330,6 +331,13 @@ def _assert(output: _CodexProbeOutput) -> None: "After the tool is denied, repeat the complete denial reason verbatim and stop. " "Do not run any other tool." ) +_POLICY_DENIAL_MARKERS = ( + "blocked by policy", + "not permitted", + "policy_violation", + "rejected:", +) +_CAPTURE_ARTIFACT_PATH_RE = re.compile(r"\.autoskillit/temp/shell_capture/shell_[0-9a-f]{16}\.log") _skip_unless_codex_output_budget_smoke = pytest.mark.skipif( not os.environ.get("CODEX_SMOKE_TEST") @@ -577,6 +585,14 @@ def test_generated_codex_child_receives_output_discipline( ) +def _policy_denial_reason(transcript: str) -> str | None: + for line in transcript.splitlines(): + lowered = line.lower() + if any(marker in lowered for marker in _POLICY_DENIAL_MARKERS): + return line.strip() + return None + + def _run_shell_capture_probe(backend: str, tmp_path: Path) -> _DenyRoundTripOutput: tmp_path.mkdir(parents=True, exist_ok=True) workspace = tmp_path / "workspace" @@ -618,7 +634,7 @@ def _run_shell_capture_probe(backend: str, tmp_path: Path) -> _DenyRoundTripOutp timeout=timeout, ) transcript = result.stdout + "\n" + result.stderr - if result.returncode != 0: + if result.returncode != 0 and _policy_denial_reason(transcript) is None: raise OSError( f"{backend} shell-capture probe failed with rc={result.returncode}: {transcript}" ) @@ -629,8 +645,145 @@ def _run_shell_capture_probe(backend: str, tmp_path: Path) -> _DenyRoundTripOutp def _assert_shell_capture_round_trip(output: _DenyRoundTripOutput) -> None: - assert _OUTPUT_BUDGET_CANARY_COMMAND in output.transcript - assert "autoskillit-shell-capture" in output.transcript + denial_reason = _policy_denial_reason(output.transcript) + assert denial_reason is None, ( + "Policy denial detected in shell-capture transcript. " + f"The generated harness was rejected by Codex's exec-policy engine: {denial_reason}" + ) + assert "autoskillit-shell-capture" in output.transcript, ( + "Harness sentinel missing — hook may not have fired" + ) + assert _OUTPUT_BUDGET_CANARY_COMMAND in output.transcript, ( + "Canary command text missing from transcript" + ) + + completed_commands: list[str] = [] + for line in output.transcript.splitlines(): + try: + event = json.loads(line) + except (json.JSONDecodeError, TypeError): + continue + if not isinstance(event, dict) or event.get("type") != "item.completed": + continue + item = event.get("item") + if not isinstance(item, dict) or item.get("type") != "command_execution": + continue + if item.get("status") == "completed": + completed_commands.append(str(item.get("command", ""))) + + assert completed_commands, ( + "No completed command_execution event found — the rewritten command did not execute" + ) + assert any( + _OUTPUT_BUDGET_CANARY_COMMAND in command + and "autoskillit-shell-capture" in command + and _CAPTURE_ARTIFACT_PATH_RE.search(command) + for command in completed_commands + ), "No completed rewritten command referenced the shell-capture artifact path" + + +def test_shell_capture_assertion_requires_completed_rewritten_command() -> None: + capture_path = "/tmp/workspace/.autoskillit/temp/shell_capture/shell_0123456789abcdef.log" + rewritten_command = ( + f"# autoskillit-shell-capture v1\n__as_f={capture_path}\n{_OUTPUT_BUDGET_CANARY_COMMAND}\n" + ) + + def _output(*, status: str, command: str = rewritten_command) -> _DenyRoundTripOutput: + event = { + "type": "item.completed", + "item": { + "type": "command_execution", + "command": command, + "status": status, + }, + } + return _DenyRoundTripOutput( + transcript=json.dumps(event), + cli_version="codex-cli test", + ) + + _assert_shell_capture_round_trip(_output(status="completed")) + + for noncompleted_status in ("denied", "failed"): + with pytest.raises(AssertionError, match="No completed command_execution"): + _assert_shell_capture_round_trip(_output(status=noncompleted_status)) + + command_without_capture_path = rewritten_command.replace(capture_path, "/tmp/missing.log") + with pytest.raises(AssertionError, match="shell-capture artifact path"): + _assert_shell_capture_round_trip( + _output(status="completed", command=command_without_capture_path) + ) + + +def test_probe_distinguishes_policy_denial_from_transport_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + exec_result = subprocess.CompletedProcess( + args=["codex", "exec"], + returncode=1, + stdout=("blocked by policy\nrm -f style commands are not permitted\n"), + stderr="", + ) + + def _fake_run( + command: list[str], + **_kwargs: object, + ) -> subprocess.CompletedProcess[str]: + if command[:2] == ["git", "init"]: + return subprocess.CompletedProcess(command, 0, "", "") + if command == ["codex", "--version"]: + return subprocess.CompletedProcess(command, 0, "codex-cli test\n", "") + if command[:2] == ["codex", "exec"]: + return exec_result + raise AssertionError(f"unexpected subprocess command: {command}") + + monkeypatch.setattr(subprocess, "run", _fake_run) + + policy_output = _run_shell_capture_probe("codex", tmp_path / "policy") + with pytest.raises(AssertionError, match="Policy denial detected"): + _assert_shell_capture_round_trip(policy_output) + + failures: list[tuple[ErrorKind, str, str, str]] = [] + with pytest.raises(AssertionError, match="Policy denial detected"): + _run_probe_with_discrimination( + "shell_capture_policy_denial", + "codex-cli test", + lambda: policy_output, + _assert_shell_capture_round_trip, + record_success=lambda _version: pytest.fail( + "policy denial must not record probe success" + ), + record_failure=lambda kind, name, version, detail: failures.append( + (kind, name, version, detail) + ), + ) + assert failures[0][0] is ErrorKind.SCHEMA + + exec_result = subprocess.CompletedProcess( + args=["codex", "exec"], + returncode=1, + stdout="request timed out before any event was emitted", + stderr="codex process crashed", + ) + with pytest.raises(OSError, match="shell-capture probe failed"): + _run_shell_capture_probe("codex", tmp_path / "transport-direct") + + failures.clear() + with pytest.raises(OSError, match="shell-capture probe failed"): + _run_probe_with_discrimination( + "shell_capture_transport_failure", + "codex-cli test", + lambda: _run_shell_capture_probe("codex", tmp_path / "transport-dispatch"), + _assert_shell_capture_round_trip, + record_success=lambda _version: pytest.fail( + "transport failure must not record probe success" + ), + record_failure=lambda kind, name, version, detail: failures.append( + (kind, name, version, detail) + ), + ) + assert failures[0][0] is ErrorKind.NETWORK def _exercise_shell_capture_probe(backend: str, tmp_path: Path) -> None: From d248752d095f0741e06a63c00d83406ba58714d4 Mon Sep 17 00:00:00 2001 From: Trecek Date: Sun, 19 Jul 2026 22:14:17 -0700 Subject: [PATCH 7/7] docs(hooks): describe shell capture lifecycle Co-Authored-By: Claude --- docs/safety/hooks.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/docs/safety/hooks.md b/docs/safety/hooks.md index e53af57433..81f5b35618 100644 --- a/docs/safety/hooks.md +++ b/docs/safety/hooks.md @@ -96,6 +96,18 @@ provenance marker (bytes, sha256, path) + tail. Exit codes are preserved via a trap-EXIT postlude. Set `output_budget.guard_enabled: false` to disable; inline threshold controlled by `output_budget.shell_max_inline_bytes`. +#### Capture Artifact Lifecycle + +| Phase | Behavior | +|-------|----------| +| Creation | Harness creates `/.autoskillit/temp/shell_capture/shell_.log` via `mkdir -p` + redirect | +| Retention | Artifact is always retained after command execution (both inline and spill branches) | +| Ownership | The hook subprocess; no downstream consumer reads these files | +| Cleanup — session lifecycle | `session_start_hook.py` sweeps stale captures (older than 1 hour, measured against wall-clock time) at every SessionStart | +| Naming contract | `shell_[0-9a-f]{16}.log` — files not matching this pattern are never deleted | +| Safety | Sweep validates the filename allowlist and rejects symlinks via `lstat`-backed checks | +| Failure mode | Sweep errors are swallowed (fail-open); cleanup failure never blocks command execution | + ### `generated_file_write_guard.py` **Guarded tools:** `Write`, `Edit` Denies writes to generated files (`hooks.json`, `settings.json`). The hooks