diff --git a/.chock/bin/_chock_runtime.py b/.chock/bin/_chock_runtime.py index 8e146e1c..d6366d9e 100644 --- a/.chock/bin/_chock_runtime.py +++ b/.chock/bin/_chock_runtime.py @@ -14,19 +14,25 @@ import fnmatch as _chock_fnmatch import re as _chock_re import os as _chock_os +import queue as _chock_queue import shlex as _chock_shlex import shutil as _chock_shutil import subprocess as _chock_subprocess +import threading as _chock_threading import time as _chock_time from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path from pathlib import PurePosixPath as _chock_PurePosixPath, PureWindowsPath as _chock_PureWindowsPath -CHOCK_RUNTIME_DIGEST = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +CHOCK_RUNTIME_DIGEST = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' ENGINE_BUDGET_SECONDS = 30 +PATTERN_SECONDS = 10 + +PATTERN_JUDGE_SOURCE = 'import json, re, sys\nfor line in sys.stdin:\n out = []\n for pattern, texts in json.loads(line):\n search = re.compile(pattern).search\n out.append([index for index, text in enumerate(texts) if search(text)])\n print(json.dumps(out), flush=True)\n' + def engine_deadline(): """The monotonic instant this invocation's budget runs out.""" return _chock_time.monotonic() + ENGINE_BUDGET_SECONDS @@ -46,6 +52,67 @@ def allowed_seconds(expired): """The time a timed-out call was actually given, as text (what the deadline left it, not the full budget).""" return f'{max(float(expired.timeout or 0), 0.0):.1f}s' +def pattern_judge(interpreter=None): + """A pattern judge, not yet started: the interpreter to run the patterns in (this one by default).""" + return {'argv': [interpreter or sys.executable, '-I', '-c', PATTERN_JUDGE_SOURCE], 'proc': None, 'lines': None} + +def _pump_lines(proc, lines): + for line in proc.stdout: + lines.put(line) + lines.put(None) + +def _start_judge(judge): + proc = _chock_subprocess.Popen(judge['argv'], stdin=_chock_subprocess.PIPE, stdout=_chock_subprocess.PIPE, stderr=_chock_subprocess.PIPE, text=True, encoding='utf-8') + lines = _chock_queue.Queue() + _chock_threading.Thread(target=_pump_lines, args=(proc, lines), daemon=True).start() + judge['proc'], judge['lines'] = (proc, lines) + +def close_judge(judge): + """Stop the judge's child, if one runs; the last line it wrote to stderr, or "".""" + proc = judge['proc'] + judge['proc'] = judge['lines'] = None + if proc is None: + return '' + if proc.poll() is None: + proc.kill() + proc.wait() + try: + spoken = proc.stderr.read() + except (OSError, ValueError): + spoken = '' + for stream in (proc.stdin, proc.stdout, proc.stderr): + with contextlib.suppress(OSError): + stream.close() + return (spoken.strip().splitlines() or [''])[-1] + +def judge_patterns(judge, jobs, seconds): + """(hits, None) or (None, why): per job of (pattern, texts), the indices of the texts the pattern matches. + + A job that runs past `seconds` kills the child and names the time; a job with no text is never + sent. `why` is text for a refusal: the judge is undecided and an undecided gate refuses. + """ + if not any((texts for _, texts in jobs)): + return ([[] for _ in jobs], None) + if judge['proc'] is None: + try: + _start_judge(judge) + except OSError as exc: + return (None, f'the pattern judge could not start ({exc})') + proc, lines = (judge['proc'], judge['lines']) + try: + proc.stdin.write(json.dumps(jobs) + '\n') + proc.stdin.flush() + except (OSError, ValueError): + return (None, f'the pattern judge exited early ({close_judge(judge)})') + try: + line = lines.get(timeout=seconds) + except _chock_queue.Empty: + close_judge(judge) + return (None, f'the patterns gave no verdict within {seconds:g}s (a catastrophic pattern?)') + if line is None: + return (None, f'the pattern judge exited early ({close_judge(judge)})') + return (json.loads(line), None) + LOG_CHAIN_PREV, LOG_CHAIN_SEQ = ('prev', 'seq') LOG_CHAIN_UNCHAINED = 'unchained' @@ -2420,10 +2487,22 @@ def tool_call_matches(globs, tool): def _tool_call_refusal(spec, spoken): return spoken or str(spec.get('message') or '').strip() or 'blocked by a chock tool_call gate' -def _tool_call_regex(spec, tool_input): - """`deny` verdict when the input, as JSON text, matches the gate's `content_pattern`.""" +def _tool_call_regex(spec, tool_input, deadline, policy): + """`deny` verdict when the input, as JSON text, matches the gate's `content_pattern`. + + The pattern runs under the engine's pattern deadline, within what is left of `deadline`; a + pattern that reaches no verdict in time refuses, naming the gate. + """ text = json.dumps(tool_input, sort_keys=True, ensure_ascii=False) - if _chock_re.search(str((spec.get('params') or {}).get('content_pattern', '')), text): + pattern = str((spec.get('params') or {}).get('content_pattern', '')) + judge = pattern_judge() + try: + hits, why = judge_patterns(judge, [(pattern, [text])], min(PATTERN_SECONDS, engine_remaining(deadline))) + finally: + close_judge(judge) + if why: + return ('deny', f'content_regex gate {policy}: {why}{_TOOL_CALL_UNDECIDED}') + if hits[0]: return ('deny', _tool_call_refusal(spec, '')) return None @@ -2450,12 +2529,12 @@ def _tool_call_script(spec, root, payload, deadline=None): detail = f': {first}' if first else '' return ('deny', f'script gate: {script.name} exited {proc.returncode}{detail}{_TOOL_CALL_UNDECIDED}') -def _tool_call_verdict(spec, root, event, deadline): +def _tool_call_verdict(spec, root, event, deadline, policy): """None to allow, else (verdict, message) for this call; a kind a tool call cannot answer refuses.""" tool, tool_input = (str(event.tool or ''), tool_call_input(event)) kind = spec.get('kind') if kind == _TOOL_CALL_REGEX_KIND: - return _tool_call_regex(spec, tool_input) + return _tool_call_regex(spec, tool_input, deadline, policy) if kind == _TOOL_CALL_SCRIPT_KIND: payload = {'event': TOOL_CALL_EVENT, 'repo_root': str(root), 'tool': tool, 'input': tool_input, 'session': session_for(event, root)} return _tool_call_script(spec, root, payload, deadline) @@ -2498,7 +2577,8 @@ def evaluate_tool_call(argv, event): root = repo_root_for(event, gate) verdict = None if TOOL_CALL_EVENT in spec.get('on', []) and tool_call_matches((spec.get('params') or {}).get('tools', []), str(event.tool or '')): - verdict = _tool_call_capped(spec, _tool_call_verdict(spec, root, event, engine_deadline())) + policy = gate.parent.parent.name + verdict = _tool_call_capped(spec, _tool_call_verdict(spec, root, event, engine_deadline(), policy)) if spec.get('kind') in _TOOL_CALL_NEEDS_SESSION: session_record(root, event, 'pre', _TOOL_CALL_BLOCKED if verdict and verdict[0] == 'deny' else None) return verdict diff --git a/.chock/bin/claude_code.py b/.chock/bin/claude_code.py index 4309b3db..2abc7f02 100755 --- a/.chock/bin/claude_code.py +++ b/.chock/bin/claude_code.py @@ -809,7 +809,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/.chock/bin/codex_cli.py b/.chock/bin/codex_cli.py index 9db704bc..6dce9316 100755 --- a/.chock/bin/codex_cli.py +++ b/.chock/bin/codex_cli.py @@ -777,7 +777,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/.chock/bin/cursor.py b/.chock/bin/cursor.py index b49d89bc..86a81d17 100755 --- a/.chock/bin/cursor.py +++ b/.chock/bin/cursor.py @@ -670,7 +670,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/.chock/bin/devin.py b/.chock/bin/devin.py index 5f5e335d..fb43ae82 100755 --- a/.chock/bin/devin.py +++ b/.chock/bin/devin.py @@ -776,7 +776,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/.chock/bin/gate.py b/.chock/bin/gate.py index fb77a0ce..f20af770 100755 --- a/.chock/bin/gate.py +++ b/.chock/bin/gate.py @@ -4,29 +4,141 @@ from __future__ import annotations import argparse +import contextlib import difflib import fnmatch import hashlib import json import os +import queue import re import shutil import subprocess import sys +import threading import time import tomllib import unicodedata from collections import Counter -from collections.abc import Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any -#: One hook invocation's time budget, in seconds (chock.gate.budget). ENGINE_BUDGET_SECONDS = 30 +PATTERN_SECONDS = 10 + + +PATTERN_JUDGE_SOURCE = ( + "import json, re, sys\n" + "for line in sys.stdin:\n" + " out = []\n" + " for pattern, texts in json.loads(line):\n" + " search = re.compile(pattern).search\n" + " out.append([index for index, text in enumerate(texts) if search(text)])\n" + " print(json.dumps(out), flush=True)\n" +) + + +def engine_deadline(): + """The monotonic instant this invocation's budget runs out.""" + return time.monotonic() + ENGINE_BUDGET_SECONDS + + +def engine_remaining(deadline): + """Seconds left before `deadline`, never negative.""" + return max(deadline - time.monotonic(), 0.0) + + +def time_left(deadline, cmd): + """Seconds left for a subprocess; raises TimeoutExpired when none are, so the caller's timeout path runs.""" + left = deadline - time.monotonic() + if left <= 0: + raise subprocess.TimeoutExpired(cmd, 0) + return left + + +def allowed_seconds(expired): + """The time a timed-out call was actually given, as text (what the deadline left it, not the full budget).""" + return f"{max(float(expired.timeout or 0), 0.0):.1f}s" + + +def pattern_judge(interpreter=None): + """A pattern judge, not yet started: the interpreter to run the patterns in (this one by default).""" + return {"argv": [interpreter or sys.executable, "-I", "-c", PATTERN_JUDGE_SOURCE], "proc": None, "lines": None} + + +def _pump_lines(proc, lines): + for line in proc.stdout: + lines.put(line) + lines.put(None) + + +def _start_judge(judge): + proc = subprocess.Popen( # noqa: S603 -- the child is this module's own source, run by this interpreter + judge["argv"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + ) + lines = queue.Queue() + threading.Thread(target=_pump_lines, args=(proc, lines), daemon=True).start() + judge["proc"], judge["lines"] = proc, lines + + +def close_judge(judge): + """Stop the judge's child, if one runs; the last line it wrote to stderr, or "".""" + proc = judge["proc"] + judge["proc"] = judge["lines"] = None + if proc is None: + return "" + if proc.poll() is None: + proc.kill() # before closing stdout: the reader thread holds it until the child's EOF + proc.wait() + try: + spoken = proc.stderr.read() + except (OSError, ValueError): + spoken = "" + for stream in (proc.stdin, proc.stdout, proc.stderr): + with contextlib.suppress(OSError): + stream.close() + return (spoken.strip().splitlines() or [""])[-1] + + +def judge_patterns(judge, jobs, seconds): + """(hits, None) or (None, why): per job of (pattern, texts), the indices of the texts the pattern matches. + + A job that runs past `seconds` kills the child and names the time; a job with no text is never + sent. `why` is text for a refusal: the judge is undecided and an undecided gate refuses. + """ + if not any(texts for _, texts in jobs): + return [[] for _ in jobs], None + if judge["proc"] is None: + try: + _start_judge(judge) + except OSError as exc: + return None, f"the pattern judge could not start ({exc})" + proc, lines = judge["proc"], judge["lines"] + try: + proc.stdin.write(json.dumps(jobs) + "\n") + proc.stdin.flush() + except (OSError, ValueError): + return None, f"the pattern judge exited early ({close_judge(judge)})" + try: + line = lines.get(timeout=seconds) + except queue.Empty: + close_judge(judge) + return None, f"the patterns gave no verdict within {seconds:g}s (a catastrophic pattern?)" + if line is None: + return None, f"the pattern judge exited early ({close_judge(judge)})" + return json.loads(line), None + + _GIT = shutil.which("git") or "git" @@ -470,41 +582,65 @@ def _judged_event(name: str, *, agent: bool) -> str: return AGENT_COMMIT_EVENT if name == "commit" and agent else name -def _waiver_re(params: dict, event: str) -> re.Pattern[str] | None: - """The waiver regex for events that honour a waiver at all, else None.""" +def _waiver_pattern(params: dict, event: str) -> str | None: + """The waiver pattern for events that honour a waiver at all, else None.""" pragma = params.get("allowlist_pragma") if event in WAIVABLE_EVENTS | HEAD_WAIVER_EVENTS else None - return re.compile(pragma) if pragma else None + return str(pragma) if pragma else None + + +class PatternDeadlineError(Exception): + """A gate's patterns reached no verdict: the judge ran past its deadline or died, as the message says.""" -def _honoured(pragma_re: re.Pattern[str] | None, text: str, head: frozenset[str] | None) -> bool: - """A waiver on `text` counts unless `head` is given and the text is not already committed.""" - return bool(pragma_re and pragma_re.search(text) and (head is None or text in head)) +def _hits(judge: dict, jobs: list[tuple[str | None, list[str]]]) -> list[set[int]]: + """Per job of (pattern, texts), the indices of the texts the pattern matches; no pattern matches nothing.""" + found, why = judge_patterns(judge, [(pattern, texts) for pattern, texts in jobs if pattern], PATTERN_SECONDS) + if why: + raise PatternDeadlineError(why) + results = iter(found) + return [set(next(results)) if pattern else set() for pattern, _ in jobs] + + +def _lines(paths: list[str], read: Callable[[str], list[str]]) -> tuple[dict[str, range], list[str]]: + """Every path's lines as `read` gives them, in one list, with each path's span of it.""" + spans: dict[str, range] = {} + texts: list[str] = [] + for path in paths: + found = read(path) + spans[path] = range(len(texts), len(texts) + len(found)) + texts += found + return spans, texts + + +def _waived(index: int, line: str, waivers: set[int], head: frozenset[str] | None) -> bool: + """A waiver on `line` counts unless `head` is given and the line is not already committed.""" + return index in waivers and (head is None or line in head) def _kind_content_regex(ctx: GateContext, params: dict, event: str) -> GateResult: - content_re = re.compile(params["content_pattern"]) - forbidden_path_regex = params.get("forbidden_path_regex") - path_re = re.compile(forbidden_path_regex) if forbidden_path_regex else None - pragma_re = _waiver_re(params, event) + pragma = _waiver_pattern(params, event) agent = event in HEAD_WAIVER_EVENTS - scan = params.get("scan", "added_lines") - diff_filter = params.get("diff_filter", "ACMRT") - + whole = params.get("scan", "added_lines") == "staged_blob" + paths = ctx.staged_paths(params.get("diff_filter", "ACMRT")) + judge = pattern_judge() + try: + forbidden = sorted(_hits(judge, [(params.get("forbidden_path_regex"), paths)])[0]) + heads = {path: ctx.committed_blob(path) if agent else "" for path in paths} + blobs = [heads[paths[index]] if agent else ctx.staged_blob(paths[index]) for index in forbidden] + spans, texts = _lines( + paths, lambda path: ctx.staged_blob(path).splitlines() if whole else ctx.added_lines(path) + ) + jobs = [(pragma, blobs), (pragma, texts), (params["content_pattern"], texts)] + blob_waivers, line_waivers, hits = _hits(judge, jobs) + finally: + close_judge(judge) matches: list[str] = [] - for path in ctx.staged_paths(diff_filter): - head_text = ctx.committed_blob(path) if agent else "" - head = frozenset(head_text.splitlines()) if agent else None - if path_re and path_re.search(path): - blob = head_text if agent else ctx.staged_blob(path) - if not (pragma_re and pragma_re.search(blob)): - matches.append(f"{path}: forbidden path") - lines = ctx.staged_blob(path).splitlines() if scan == "staged_blob" else ctx.added_lines(path) - for line in lines: - if _honoured(pragma_re, line, head): - continue - if content_re.search(line): - matches.append(f"{path}: content pattern") - break + for index, path in enumerate(paths): + if index in forbidden and forbidden.index(index) not in blob_waivers: + matches.append(f"{path}: forbidden path") + head = frozenset(heads[path].splitlines()) if agent else None + if any(n in hits and not _waived(n, texts[n], line_waivers, head) for n in spans[path]): + matches.append(f"{path}: content pattern") return GateResult(allowed=not matches, matches=matches) @@ -619,39 +755,39 @@ def _kind_dependency_allowlist(ctx: GateContext, params: dict, _event: str) -> G return GateResult(allowed=not matches, matches=matches) -def _count( - pattern: "re.Pattern[str]", lines: list[str], pragma: re.Pattern[str] | None, head: frozenset[str] | None = None -) -> int: - return sum(1 for line in lines if pattern.search(line) and not _honoured(pragma, line, head)) +def _count(span: range, hits: set[int], texts: list[str], waivers: set[int], head: frozenset[str] | None) -> int: + return sum(1 for n in span if n in hits and not _waived(n, texts[n], waivers, head)) def _kind_test_integrity(ctx: GateContext, params: dict, event: str) -> GateResult: """Block a change that wins green CI by weakening the tests rather than fixing the code.""" - path_re = re.compile(params["test_path_regex"]) - assertion_re = re.compile(params["assertion_pattern"]) - dummy_pattern = params.get("dummy_assertion_pattern") - dummy_re = re.compile(dummy_pattern) if dummy_pattern else None - pragma_re = _waiver_re(params, event) + pragma = _waiver_pattern(params, event) agent = event in HEAD_WAIVER_EVENTS - - matches: list[str] = [] - added = removed = 0 - for path in ctx.staged_paths("D"): - if path_re.search(path): - matches.append(f"{path}: test file deleted") - for path in ctx.staged_paths("ACMRT"): - if not path_re.search(path): - continue - added_lines = ctx.net_added_lines(path) + deleted, changed = ctx.staged_paths("D"), ctx.staged_paths("ACMRT") + judge = pattern_judge() + try: + gone, tests = _hits(judge, [(params["test_path_regex"], deleted), (params["test_path_regex"], changed)]) + files = [changed[index] for index in sorted(tests)] + added_spans, added = _lines(files, ctx.net_added_lines) + removed_spans, removed = _lines(files, ctx.removed_lines) + assertion = params["assertion_pattern"] + jobs = [(assertion, added), (assertion, removed), (pragma, added), (pragma, removed)] + jobs.append((params.get("dummy_assertion_pattern"), added)) + asserted_added, asserted_removed, waived_added, waived_removed, dummy = _hits(judge, jobs) + finally: + close_judge(judge) + matches = [f"{deleted[index]}: test file deleted" for index in sorted(gone)] + added_count = removed_count = 0 + for path in files: head = frozenset(ctx.committed_blob(path).splitlines()) if agent else None - if any(_honoured(pragma_re, line, head) for line in added_lines): + if any(_waived(n, added[n], waived_added, head) for n in added_spans[path]): continue - added += _count(assertion_re, added_lines, pragma_re, head) - removed += _count(assertion_re, ctx.removed_lines(path), pragma_re, head) - if dummy_re and any(dummy_re.search(line) for line in added_lines): + added_count += _count(added_spans[path], asserted_added, added, waived_added, head) + removed_count += _count(removed_spans[path], asserted_removed, removed, waived_removed, head) + if any(n in dummy for n in added_spans[path]): matches.append(f"{path}: vacuous assertion added") - if removed > added: - matches.append(f"assertions removed across tests: {removed} removed, {added} added") + if removed_count > added_count: + matches.append(f"assertions removed across tests: {removed_count} removed, {added_count} added") return GateResult(allowed=not matches, matches=matches) @@ -985,6 +1121,9 @@ def _context( _GIT_EVENTS = ("pre-commit", "pre-push", "commit-msg") +_UNJUDGED = " -- refusing rather than allowing what it never judged." + + def _policy_id(gate_path: Path) -> str | None: """The policy a compiled gate belongs to, from `.chock/compiled///`, else None.""" parents = gate_path.resolve().parents @@ -1457,12 +1596,23 @@ def evaluate( return 2, VERDICT_ERROR signal = agent_signal(repo_root) judged = _judged_event(name, agent=signal is not None) + result = _judge_kind(kind, ctx, _params(gate_path, spec), judged, gate_path) + if isinstance(result, tuple): + return result + return Evaluation(spec, result, rollout(repo_root, event, signal, base), signal, judged) + + +def _judge_kind(kind: Any, ctx: Any, params: dict, judged: str, gate_path: Path) -> GateResult | tuple[int, str]: + """The kind's result, or (2, error) with the reason on stderr when it could not judge.""" try: - result = kind(ctx, _params(gate_path, spec), judged) + return kind(ctx, params, judged) except ListingError as exc: print(f"gate: {inert(str(exc))} -- refusing rather than judging a change it cannot read.", file=sys.stderr) - return 2, VERDICT_ERROR - return Evaluation(spec, result, rollout(repo_root, event, signal, base), signal, judged) + except PatternDeadlineError as exc: + # The reason names a time or the judge's own last stderr line, never the judged text. + reason = f"gate: {_policy_id(gate_path) or gate_path.name}: {inert(str(exc))}{_UNJUDGED}" + print(reason, file=sys.stderr) + return 2, VERDICT_ERROR def judge( diff --git a/.chock/bin/gemini_cli.py b/.chock/bin/gemini_cli.py index 0b3415c8..e2b579fd 100755 --- a/.chock/bin/gemini_cli.py +++ b/.chock/bin/gemini_cli.py @@ -776,7 +776,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/.chock/bin/grok.py b/.chock/bin/grok.py index 67ed740a..6238505f 100755 --- a/.chock/bin/grok.py +++ b/.chock/bin/grok.py @@ -776,7 +776,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/.chock/bin/tabnine.py b/.chock/bin/tabnine.py index 0fa87617..57f0b2cf 100755 --- a/.chock/bin/tabnine.py +++ b/.chock/bin/tabnine.py @@ -732,7 +732,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/.chock/bin/vscode_copilot.py b/.chock/bin/vscode_copilot.py index 1b0eb253..2abbe0a0 100755 --- a/.chock/bin/vscode_copilot.py +++ b/.chock/bin/vscode_copilot.py @@ -560,7 +560,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/.chock/bin/windsurf.py b/.chock/bin/windsurf.py index 422257f6..8c469582 100755 --- a/.chock/bin/windsurf.py +++ b/.chock/bin/windsurf.py @@ -629,7 +629,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/changelog.d/gate-pattern-deadline.md b/changelog.d/gate-pattern-deadline.md new file mode 100644 index 00000000..b99394ed --- /dev/null +++ b/changelog.d/gate-pattern-deadline.md @@ -0,0 +1,9 @@ +- **A gate pattern cannot hang a commit or fail a hook open.** A `content_regex` or `test_integrity` + pattern a backtracking engine can run in exponential time (`^(a+)+$` on a 60-character line) used to + hang `git commit` on every platform, burn the client's hook timeout in the agent (which fails open) and + wedge `chock check --history` on Windows. Now every gate pattern runs in a child interpreter the engine + kills at a deadline (`PATTERN_SECONDS`, 10 s per judgement), on every platform: the git hook exits 2 + naming the gate, the write and tool_call paths refuse naming it, and the history scan errors naming it. + `chock validate` and the compile step also refuse, naming the param and the pattern, a pattern whose + nested repeats can run into each other (a linear-time check; overlapping alternation is left to the + deadline). The vendored runtimes under `.chock/bin` are regenerated; `chock sync` rewrites them. diff --git a/spec/gate-dsl.md b/spec/gate-dsl.md index a03c10f6..8238017f 100644 --- a/spec/gate-dsl.md +++ b/spec/gate-dsl.md @@ -72,7 +72,7 @@ pre-tool credit. | param | required | type | notes | |-------|----------|------|-------| -| `content_pattern` | yes | string | regex matched against each line or the blob | +| `content_pattern` | yes | string | regex matched against each line or the blob. Every pattern param runs in a child the engine kills after `PATTERN_SECONDS` (10 s per judgement), and the gate then refuses naming itself; `chock validate` and the compile step refuse a pattern whose nested repeats can run into each other (`^(a+)+$`) | | `scan` | no | string | `added_lines` (default) or `staged_blob` | | `forbidden_path_regex` | no | string | regex applied to staged file paths | | `allowlist_pragma` | no | string | regex matched on lines/blobs; matching content is ignored at commit, push and ci. On an agent commit (see [Actions](#actions)), and at `tool_use`, only a waiver already in HEAD is honoured | diff --git a/src/chock/gate/assemble.py b/src/chock/gate/assemble.py index 27e7f297..efa53c28 100644 --- a/src/chock/gate/assemble.py +++ b/src/chock/gate/assemble.py @@ -2,12 +2,11 @@ from __future__ import annotations +import ast import re from chock.resources import package_data_dir -from .budget import ENGINE_BUDGET_SECONDS - RUNNER_PACKAGE = "chock.gate.runner" #: Every module gate.py is built from. Each must exist and contribute a fragment, so a module lost @@ -40,6 +39,14 @@ def _prelude() -> str: return package_data_dir("chock.gate", "data").joinpath("gate_prelude.py.tmpl").read_text(encoding="utf-8") +def budget_fragment() -> str: + """chock.gate.budget's definitions, verbatim: the runner's deadlines and pattern judge are the engine's own.""" + source = package_data_dir("chock.gate").joinpath("budget.py").read_text(encoding="utf-8") + lines = source.splitlines(keepends=True) + kept = [node for node in ast.parse(source).body if isinstance(node, ast.FunctionDef | ast.Assign)] + return _SEPARATOR.join("".join(lines[node.lineno - 1 : node.end_lineno]).rstrip("\n") for node in kept) + + def _fragments() -> dict[int, str]: found: dict[int, str] = {} package = package_data_dir(RUNNER_PACKAGE) @@ -62,6 +69,5 @@ def runner_source() -> str: if sorted(found) != list(range(1, len(found) + 1)): msg = f"gate.py fragments must run 1..N without a gap; found {sorted(found)}" raise ValueError(msg) - budget = f"#: One hook invocation's time budget, in seconds (chock.gate.budget).\nENGINE_BUDGET_SECONDS = {ENGINE_BUDGET_SECONDS}" - body = _SEPARATOR.join([budget, *(found[number] for number in sorted(found))]) + body = _SEPARATOR.join([budget_fragment(), *(found[number] for number in sorted(found))]) return _prelude() + body + "\n" diff --git a/src/chock/gate/budget.py b/src/chock/gate/budget.py index 8169a6e2..ac8f1b40 100644 --- a/src/chock/gate/budget.py +++ b/src/chock/gate/budget.py @@ -3,17 +3,44 @@ A client's hook timeout fails OPEN, while the engine's own timers ask a person or deny, so the engine must always run out first. One hook invocation spends ONE budget across every subprocess it starts (git, a shell probe, the guard, the gate runner), each getting what is left of it. + +A gate's patterns get a deadline of their own: `re` holds the GIL and no platform can interrupt +it in-process (SIGALRM reaches the main thread on POSIX only), so the patterns run in a child +interpreter the parent can kill. Functions only, no classes: the shared runtime extracts this +module's definitions (chock.gate.runtime_shared), and so does the vendored runner (chock.gate.assemble). """ from __future__ import annotations +import contextlib +import json +import queue import subprocess +import sys +import threading import time #: Seconds one hook invocation may spend judging, after interpreter start-up. The emitted client #: timeout is this plus a start-up margin (chock.compile.emitters.in_agent_hooks). ENGINE_BUDGET_SECONDS = 30 +#: Seconds a gate's patterns may spend on one judgement (one commit's lines, one write, one blob) +#: before the gate refuses naming itself: a repo-supplied pattern can be catastrophic. +PATTERN_SECONDS = 10 + +#: The child that runs a gate's patterns. Per request: one JSON line `[[pattern, [text, ...]], ...]` +#: in, one JSON line `[[index of each text the pattern matches], ...]` out. Python source as text +#: because every file this ships in is self-contained; `-I` keeps a planted re.py off its path. +PATTERN_JUDGE_SOURCE = ( + "import json, re, sys\n" + "for line in sys.stdin:\n" + " out = []\n" + " for pattern, texts in json.loads(line):\n" + " search = re.compile(pattern).search\n" + " out.append([index for index, text in enumerate(texts) if search(text)])\n" + " print(json.dumps(out), flush=True)\n" +) + def engine_deadline(): """The monotonic instant this invocation's budget runs out.""" @@ -36,3 +63,76 @@ def time_left(deadline, cmd): def allowed_seconds(expired): """The time a timed-out call was actually given, as text (what the deadline left it, not the full budget).""" return f"{max(float(expired.timeout or 0), 0.0):.1f}s" + + +def pattern_judge(interpreter=None): + """A pattern judge, not yet started: the interpreter to run the patterns in (this one by default).""" + return {"argv": [interpreter or sys.executable, "-I", "-c", PATTERN_JUDGE_SOURCE], "proc": None, "lines": None} + + +def _pump_lines(proc, lines): + for line in proc.stdout: + lines.put(line) + lines.put(None) + + +def _start_judge(judge): + proc = subprocess.Popen( # noqa: S603 -- the child is this module's own source, run by this interpreter + judge["argv"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + ) + lines = queue.Queue() + threading.Thread(target=_pump_lines, args=(proc, lines), daemon=True).start() + judge["proc"], judge["lines"] = proc, lines + + +def close_judge(judge): + """Stop the judge's child, if one runs; the last line it wrote to stderr, or "".""" + proc = judge["proc"] + judge["proc"] = judge["lines"] = None + if proc is None: + return "" + if proc.poll() is None: + proc.kill() # before closing stdout: the reader thread holds it until the child's EOF + proc.wait() + try: + spoken = proc.stderr.read() + except (OSError, ValueError): + spoken = "" + for stream in (proc.stdin, proc.stdout, proc.stderr): + with contextlib.suppress(OSError): + stream.close() + return (spoken.strip().splitlines() or [""])[-1] + + +def judge_patterns(judge, jobs, seconds): + """(hits, None) or (None, why): per job of (pattern, texts), the indices of the texts the pattern matches. + + A job that runs past `seconds` kills the child and names the time; a job with no text is never + sent. `why` is text for a refusal: the judge is undecided and an undecided gate refuses. + """ + if not any(texts for _, texts in jobs): + return [[] for _ in jobs], None + if judge["proc"] is None: + try: + _start_judge(judge) + except OSError as exc: + return None, f"the pattern judge could not start ({exc})" + proc, lines = judge["proc"], judge["lines"] + try: + proc.stdin.write(json.dumps(jobs) + "\n") + proc.stdin.flush() + except (OSError, ValueError): + return None, f"the pattern judge exited early ({close_judge(judge)})" + try: + line = lines.get(timeout=seconds) + except queue.Empty: + close_judge(judge) + return None, f"the patterns gave no verdict within {seconds:g}s (a catastrophic pattern?)" + if line is None: + return None, f"the pattern judge exited early ({close_judge(judge)})" + return json.loads(line), None diff --git a/src/chock/gate/build.py b/src/chock/gate/build.py index d662d356..b5c45d8a 100644 --- a/src/chock/gate/build.py +++ b/src/chock/gate/build.py @@ -10,9 +10,21 @@ from chock.config import load_config from chock.emit import write_generated from chock.gate.assemble import runner_source +from chock.gate.pattern_safety import REGEX_PARAM_SUFFIXES, catastrophic, catastrophic_message from chock.manifest import load_manifest +class CatastrophicPatternError(ValueError): + """A gate pattern a backtracking engine can run in exponential time: refused at compile, never installed.""" + + +def _refuse_catastrophic(policy_dir: Path, params: dict[str, Any]) -> None: + for name, value in params.items(): + if isinstance(value, str) and name.endswith(REGEX_PARAM_SUFFIXES) and (why := catastrophic(value)): + msg = f"{policy_dir.name}: hook.gate.params.{catastrophic_message(name, value, why)}; nothing was compiled" + raise CatastrophicPatternError(msg) + + def _resolve_dotted(config: dict[str, Any], key: str) -> Any: """Resolve a dot-separated key like 'chock.defaults.refs'.""" value = config @@ -41,6 +53,7 @@ def build_gate_json(policy_dir: Path, repo_root: Path) -> dict[str, Any] | None: "message": str(gate.get("message", "")).strip(), "params": dict(gate.get("params") or {}), } + _refuse_catastrophic(policy_dir, spec["params"]) # applies_to.paths bounds which files this gate may judge. It rides beside params rather # than inside them: params are what the policy declared about its check, and this is what diff --git a/src/chock/gate/data/gate_prelude.py.tmpl b/src/chock/gate/data/gate_prelude.py.tmpl index 0c77e2d8..320bc14c 100644 --- a/src/chock/gate/data/gate_prelude.py.tmpl +++ b/src/chock/gate/data/gate_prelude.py.tmpl @@ -4,20 +4,23 @@ from __future__ import annotations import argparse +import contextlib import difflib import fnmatch import hashlib import json import os +import queue import re import shutil import subprocess import sys +import threading import time import tomllib import unicodedata from collections import Counter -from collections.abc import Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path diff --git a/src/chock/gate/data/imports.py.tmpl b/src/chock/gate/data/imports.py.tmpl index f9ea03b3..35c33d16 100644 --- a/src/chock/gate/data/imports.py.tmpl +++ b/src/chock/gate/data/imports.py.tmpl @@ -2,9 +2,11 @@ import hashlib as _chock_hashlib import fnmatch as _chock_fnmatch import re as _chock_re import os as _chock_os +import queue as _chock_queue import shlex as _chock_shlex import shutil as _chock_shutil import subprocess as _chock_subprocess +import threading as _chock_threading import time as _chock_time from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path diff --git a/src/chock/gate/pattern_safety.py b/src/chock/gate/pattern_safety.py new file mode 100644 index 00000000..e4d4e89c --- /dev/null +++ b/src/chock/gate/pattern_safety.py @@ -0,0 +1,192 @@ +"""A linear-time check for gate patterns that can take exponential time. Framework-side only. + +A backtracking engine runs a repeat nested in a repeat in exponential time when the inner +repeat and what follows it inside the outer loop accept the same character: `(a+)+`, +`(\\w+\\s?)+`, `(?:\\.[\\w.-]+)+`. The parse tree is walked once; where that shape is found the +pattern is refused before it reaches a gate. Overlapping alternation (`(a|aa)+`) is not +detected here; the runtime deadline (`chock.gate.budget`) is the backstop for every shape. +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from dataclasses import dataclass +from re import _constants as sre_constants +from re import _parser as sre_parser +from re import error as re_error + +#: The gate params that hold a regular expression. +REGEX_PARAM_SUFFIXES = ("_pattern", "_regex", "_pragma") +#: How much of a refused pattern a message shows. +_SHOWN = 120 +#: One character of each class a gate pattern commonly names, to probe a class predicate with. +_SAMPLES = ("a", "Z", "5", "_", " ", "!", ".", "-", "/") +#: A repeat that can run twice is a loop. +_TWICE = 2 +_REPEATS = (sre_constants.MAX_REPEAT, sre_constants.MIN_REPEAT, sre_constants.POSSESSIVE_REPEAT) +_ENDS = (sre_constants.AT_END, sre_constants.AT_END_STRING) +_SINGLE = ( + sre_constants.LITERAL, + sre_constants.NOT_LITERAL, + sre_constants.ANY, + sre_constants.IN, + sre_constants.CATEGORY, +) +_CATEGORY_TESTS: dict[object, Callable[[str], bool]] = { + sre_constants.CATEGORY_DIGIT: str.isdecimal, + sre_constants.CATEGORY_NOT_DIGIT: lambda ch: not ch.isdecimal(), + sre_constants.CATEGORY_SPACE: str.isspace, + sre_constants.CATEGORY_NOT_SPACE: lambda ch: not ch.isspace(), + sre_constants.CATEGORY_WORD: lambda ch: ch.isalnum() or ch == "_", + sre_constants.CATEGORY_NOT_WORD: lambda ch: not (ch.isalnum() or ch == "_"), + sre_constants.CATEGORY_LINEBREAK: lambda ch: ch == "\n", + sre_constants.CATEGORY_NOT_LINEBREAK: lambda ch: ch != "\n", +} + + +@dataclass(frozen=True) +class _Chars: + """One single-character node: a membership test and some characters known to pass it.""" + + contains: Callable[[str], bool] + samples: frozenset[str] + + +_ANY = _Chars(lambda ch: ch != "\n", frozenset(_SAMPLES)) + + +def _member(item: tuple) -> Callable[[str], bool]: + """The membership test of one item of a character class.""" + op, av = item + if op == sre_constants.LITERAL: + return lambda ch: ord(ch) == av + if op == sre_constants.RANGE: + return lambda ch: av[0] <= ord(ch) <= av[1] + if op == sre_constants.CATEGORY: + return _CATEGORY_TESTS.get(av, _ANY.contains) + return _ANY.contains + + +def _chars(op: object, av: object) -> _Chars: + """The character set of a single-character node.""" + if op == sre_constants.LITERAL: + return _Chars(lambda ch: ord(ch) == av, frozenset(chr(av))) + if op == sre_constants.NOT_LITERAL: + return _Chars(lambda ch: ord(ch) != av, frozenset(_SAMPLES) - {chr(av)}) + if op == sre_constants.CATEGORY: + items: list[tuple] = [(op, av)] + elif op == sre_constants.IN: + items = list(av) + else: + return _ANY + negate = bool(items) and items[0][0] == sre_constants.NEGATE + tests = [_member(item) for item in items[1:] if negate] or [_member(item) for item in items] + contains = (lambda ch: not any(t(ch) for t in tests)) if negate else (lambda ch: any(t(ch) for t in tests)) + named = {chr(av) for op, av in items if op == sre_constants.LITERAL} + named |= {chr(av[0]) for op, av in items if op == sre_constants.RANGE} + named |= {chr(av[1]) for op, av in items if op == sre_constants.RANGE} + return _Chars(contains, frozenset(ch for ch in named | set(_SAMPLES) if contains(ch))) + + +def _overlap(left: Sequence[_Chars], right: Sequence[_Chars]) -> str | None: + """A character both sets accept, or None.""" + for a in left: + for b in right: + hit = next((ch for ch in sorted(a.samples) if b.contains(ch)), None) or next( + (ch for ch in sorted(b.samples) if a.contains(ch)), None + ) + if hit is not None: + return hit + return None + + +def _first_node(op: object, av: object) -> tuple[list[_Chars], bool]: # noqa: PLR0911 -- one return per node shape + """(characters a node can start with, whether it can match nothing).""" + if op in _SINGLE: + return [_chars(op, av)], False + if op in _REPEATS: + chars, nullable = _first(av[2]) + return chars, nullable or av[0] == 0 + if op in (sre_constants.SUBPATTERN, sre_constants.ATOMIC_GROUP): + return _first(av[-1] if op == sre_constants.SUBPATTERN else av) + if op == sre_constants.BRANCH: + parts = [_first(branch) for branch in av[1]] + return [c for chars, _ in parts for c in chars], any(nullable for _, nullable in parts) + if op == sre_constants.GROUPREF_EXISTS: + parts = [_first(branch) for branch in av[1:] if branch is not None] + return [c for chars, _ in parts for c in chars], any(nullable for _, nullable in parts) + if op == sre_constants.AT: + return [], av not in _ENDS # nothing can follow the end of the text + if op == sre_constants.ASSERT and av[0] > 0: + return _first(av[1])[0], False # a lookahead names what the next character must be + if op in (sre_constants.ASSERT, sre_constants.ASSERT_NOT): + return [], True + return [_ANY], True + + +def _first(seq: Sequence[tuple]) -> tuple[list[_Chars], bool]: + """(characters a sequence can start with, whether it can match nothing).""" + chars: list[_Chars] = [] + for op, av in seq: + found, nullable = _first_node(op, av) + chars += found + if not nullable: + return chars, False + return chars, True + + +def _children(op: object, av: object) -> list[Sequence[tuple]]: + """The sequences a node holds, for the walk to descend into.""" + if op == sre_constants.SUBPATTERN: + return [av[-1]] + if op in (sre_constants.ATOMIC_GROUP, sre_constants.ASSERT, sre_constants.ASSERT_NOT): + return [av if op == sre_constants.ATOMIC_GROUP else av[1]] + if op == sre_constants.BRANCH: + return list(av[1]) + if op == sre_constants.GROUPREF_EXISTS: + return [branch for branch in av[1:] if branch is not None] + return [] + + +def _walk(seq: Sequence[tuple], after: list[_Chars], in_loop: bool) -> str | None: # noqa: FBT001 -- internal recursion flag + """The character that makes a nested repeat ambiguous, or None. + + `after` is what the enclosing loop's next iteration starts with, when nothing that must + consume text stands between this sequence's end and it; `in_loop` says such a loop + encloses this sequence. A repeat is ambiguous when it can run into that next iteration. + """ + for index, (op, av) in enumerate(seq): + follow = after if _first(seq[index + 1 :])[1] else [] + if op in _REPEATS: + low, high, body = av + loops = high >= _TWICE + if in_loop and loops and high > low and (hit := _overlap(_first(body)[0], follow)) is not None: + return hit + found = _walk(body, _first(body)[0] if loops else follow, in_loop or loops) + else: + found = next((hit for child in _children(op, av) if (hit := _walk(child, follow, in_loop))), None) + if found is not None: + return found + return None + + +def catastrophic(pattern: str) -> str | None: + """Why `pattern` can take exponential time on a crafted line, or None; a pattern that does not compile is None too.""" + try: + parsed = sre_parser.parse(pattern) + except (re_error, ValueError, OverflowError, RecursionError): + return None + hit = _walk(list(parsed), [], in_loop=False) + if hit is None: + return None + return f"a repeat nested in a repeat, where the inner repeat and what follows it both accept {hit!r}" + + +def catastrophic_message(name: str, pattern: str, why: str) -> str: + """The refusal for param `name` holding `pattern`, naming both and the shape found.""" + shown = pattern if len(pattern) <= _SHOWN else pattern[:_SHOWN] + "..." + return ( + f"{name} can take exponential time on a crafted line ({why}): {shown!r}. " + "The gate would run out its deadline and refuse rather than judge; rewrite the pattern so the inner " + "repeat cannot run into the next iteration (anchor it, or separate the repeats with a character neither accepts)" + ) diff --git a/src/chock/gate/runner/actor.py b/src/chock/gate/runner/actor.py index d2fb96c9..2e088993 100644 --- a/src/chock/gate/runner/actor.py +++ b/src/chock/gate/runner/actor.py @@ -3,7 +3,6 @@ from __future__ import annotations import os -import re import subprocess from pathlib import Path @@ -164,15 +163,10 @@ def _judged_event(name: str, *, agent: bool) -> str: return AGENT_COMMIT_EVENT if name == "commit" and agent else name -def _waiver_re(params: dict, event: str) -> re.Pattern[str] | None: - """The waiver regex for events that honour a waiver at all, else None.""" +def _waiver_pattern(params: dict, event: str) -> str | None: + """The waiver pattern for events that honour a waiver at all, else None.""" pragma = params.get("allowlist_pragma") if event in WAIVABLE_EVENTS | HEAD_WAIVER_EVENTS else None - return re.compile(pragma) if pragma else None - - -def _honoured(pragma_re: re.Pattern[str] | None, text: str, head: frozenset[str] | None) -> bool: - """A waiver on `text` counts unless `head` is given and the text is not already committed.""" - return bool(pragma_re and pragma_re.search(text) and (head is None or text in head)) + return str(pragma) if pragma else None # >>> gate.py 16 diff --git a/src/chock/gate/runner/kinds.py b/src/chock/gate/runner/kinds.py index e1bc462c..280adb5b 100644 --- a/src/chock/gate/runner/kinds.py +++ b/src/chock/gate/runner/kinds.py @@ -6,38 +6,69 @@ import json import re import tomllib +from collections.abc import Callable -from .actor import _honoured, _waiver_re +from ..budget import PATTERN_SECONDS, close_judge, judge_patterns, pattern_judge +from .actor import _waiver_pattern from .constants import _DEPENDENCY_KIND, HEAD_WAIVER_EVENTS from .context import GateContext, GateResult from .script import _kind_script # >>> gate.py 06 +class PatternDeadlineError(Exception): + """A gate's patterns reached no verdict: the judge ran past its deadline or died, as the message says.""" + + +def _hits(judge: dict, jobs: list[tuple[str | None, list[str]]]) -> list[set[int]]: + """Per job of (pattern, texts), the indices of the texts the pattern matches; no pattern matches nothing.""" + found, why = judge_patterns(judge, [(pattern, texts) for pattern, texts in jobs if pattern], PATTERN_SECONDS) + if why: + raise PatternDeadlineError(why) + results = iter(found) + return [set(next(results)) if pattern else set() for pattern, _ in jobs] + + +def _lines(paths: list[str], read: Callable[[str], list[str]]) -> tuple[dict[str, range], list[str]]: + """Every path's lines as `read` gives them, in one list, with each path's span of it.""" + spans: dict[str, range] = {} + texts: list[str] = [] + for path in paths: + found = read(path) + spans[path] = range(len(texts), len(texts) + len(found)) + texts += found + return spans, texts + + +def _waived(index: int, line: str, waivers: set[int], head: frozenset[str] | None) -> bool: + """A waiver on `line` counts unless `head` is given and the line is not already committed.""" + return index in waivers and (head is None or line in head) + + def _kind_content_regex(ctx: GateContext, params: dict, event: str) -> GateResult: - content_re = re.compile(params["content_pattern"]) - forbidden_path_regex = params.get("forbidden_path_regex") - path_re = re.compile(forbidden_path_regex) if forbidden_path_regex else None - pragma_re = _waiver_re(params, event) + pragma = _waiver_pattern(params, event) agent = event in HEAD_WAIVER_EVENTS - scan = params.get("scan", "added_lines") - diff_filter = params.get("diff_filter", "ACMRT") - + whole = params.get("scan", "added_lines") == "staged_blob" + paths = ctx.staged_paths(params.get("diff_filter", "ACMRT")) + judge = pattern_judge() + try: + forbidden = sorted(_hits(judge, [(params.get("forbidden_path_regex"), paths)])[0]) + heads = {path: ctx.committed_blob(path) if agent else "" for path in paths} + blobs = [heads[paths[index]] if agent else ctx.staged_blob(paths[index]) for index in forbidden] + spans, texts = _lines( + paths, lambda path: ctx.staged_blob(path).splitlines() if whole else ctx.added_lines(path) + ) + jobs = [(pragma, blobs), (pragma, texts), (params["content_pattern"], texts)] + blob_waivers, line_waivers, hits = _hits(judge, jobs) + finally: + close_judge(judge) matches: list[str] = [] - for path in ctx.staged_paths(diff_filter): - head_text = ctx.committed_blob(path) if agent else "" - head = frozenset(head_text.splitlines()) if agent else None - if path_re and path_re.search(path): - blob = head_text if agent else ctx.staged_blob(path) - if not (pragma_re and pragma_re.search(blob)): - matches.append(f"{path}: forbidden path") - lines = ctx.staged_blob(path).splitlines() if scan == "staged_blob" else ctx.added_lines(path) - for line in lines: - if _honoured(pragma_re, line, head): - continue - if content_re.search(line): - matches.append(f"{path}: content pattern") - break + for index, path in enumerate(paths): + if index in forbidden and forbidden.index(index) not in blob_waivers: + matches.append(f"{path}: forbidden path") + head = frozenset(heads[path].splitlines()) if agent else None + if any(n in hits and not _waived(n, texts[n], line_waivers, head) for n in spans[path]): + matches.append(f"{path}: content pattern") return GateResult(allowed=not matches, matches=matches) @@ -152,39 +183,39 @@ def _kind_dependency_allowlist(ctx: GateContext, params: dict, _event: str) -> G return GateResult(allowed=not matches, matches=matches) -def _count( - pattern: "re.Pattern[str]", lines: list[str], pragma: re.Pattern[str] | None, head: frozenset[str] | None = None -) -> int: - return sum(1 for line in lines if pattern.search(line) and not _honoured(pragma, line, head)) +def _count(span: range, hits: set[int], texts: list[str], waivers: set[int], head: frozenset[str] | None) -> int: + return sum(1 for n in span if n in hits and not _waived(n, texts[n], waivers, head)) def _kind_test_integrity(ctx: GateContext, params: dict, event: str) -> GateResult: """Block a change that wins green CI by weakening the tests rather than fixing the code.""" - path_re = re.compile(params["test_path_regex"]) - assertion_re = re.compile(params["assertion_pattern"]) - dummy_pattern = params.get("dummy_assertion_pattern") - dummy_re = re.compile(dummy_pattern) if dummy_pattern else None - pragma_re = _waiver_re(params, event) + pragma = _waiver_pattern(params, event) agent = event in HEAD_WAIVER_EVENTS - - matches: list[str] = [] - added = removed = 0 - for path in ctx.staged_paths("D"): - if path_re.search(path): - matches.append(f"{path}: test file deleted") - for path in ctx.staged_paths("ACMRT"): - if not path_re.search(path): - continue - added_lines = ctx.net_added_lines(path) + deleted, changed = ctx.staged_paths("D"), ctx.staged_paths("ACMRT") + judge = pattern_judge() + try: + gone, tests = _hits(judge, [(params["test_path_regex"], deleted), (params["test_path_regex"], changed)]) + files = [changed[index] for index in sorted(tests)] + added_spans, added = _lines(files, ctx.net_added_lines) + removed_spans, removed = _lines(files, ctx.removed_lines) + assertion = params["assertion_pattern"] + jobs = [(assertion, added), (assertion, removed), (pragma, added), (pragma, removed)] + jobs.append((params.get("dummy_assertion_pattern"), added)) + asserted_added, asserted_removed, waived_added, waived_removed, dummy = _hits(judge, jobs) + finally: + close_judge(judge) + matches = [f"{deleted[index]}: test file deleted" for index in sorted(gone)] + added_count = removed_count = 0 + for path in files: head = frozenset(ctx.committed_blob(path).splitlines()) if agent else None - if any(_honoured(pragma_re, line, head) for line in added_lines): + if any(_waived(n, added[n], waived_added, head) for n in added_spans[path]): continue - added += _count(assertion_re, added_lines, pragma_re, head) - removed += _count(assertion_re, ctx.removed_lines(path), pragma_re, head) - if dummy_re and any(dummy_re.search(line) for line in added_lines): + added_count += _count(added_spans[path], asserted_added, added, waived_added, head) + removed_count += _count(removed_spans[path], asserted_removed, removed, waived_removed, head) + if any(n in dummy for n in added_spans[path]): matches.append(f"{path}: vacuous assertion added") - if removed > added: - matches.append(f"assertions removed across tests: {removed} removed, {added} added") + if removed_count > added_count: + matches.append(f"assertions removed across tests: {removed_count} removed, {added_count} added") return GateResult(allowed=not matches, matches=matches) diff --git a/src/chock/gate/runner/verdict.py b/src/chock/gate/runner/verdict.py index 8d1a5bfe..5bdeb93d 100644 --- a/src/chock/gate/runner/verdict.py +++ b/src/chock/gate/runner/verdict.py @@ -33,7 +33,7 @@ ) from .context import GateResult, ListingError from .halt import HALT_POLICY, HALT_REASON, halt_everywhere_read, halt_find, halt_git_roots, halt_message -from .kinds import KINDS +from .kinds import KINDS, PatternDeadlineError from .log import _actor, _held_record, _log_halt, _log_outcome, _log_script_hook from .material import _context, _params, own_paths from .report import ( @@ -47,8 +47,10 @@ inert, ) - # >>> gate.py 14 +_UNJUDGED = " -- refusing rather than allowing what it never judged." + + def _policy_id(gate_path: Path) -> str | None: """The policy a compiled gate belongs to, from `.chock/compiled///`, else None.""" parents = gate_path.resolve().parents @@ -194,12 +196,23 @@ def evaluate( return 2, VERDICT_ERROR signal = agent_signal(repo_root) judged = _judged_event(name, agent=signal is not None) + result = _judge_kind(kind, ctx, _params(gate_path, spec), judged, gate_path) + if isinstance(result, tuple): + return result + return Evaluation(spec, result, rollout(repo_root, event, signal, base), signal, judged) + + +def _judge_kind(kind: Any, ctx: Any, params: dict, judged: str, gate_path: Path) -> GateResult | tuple[int, str]: + """The kind's result, or (2, error) with the reason on stderr when it could not judge.""" try: - result = kind(ctx, _params(gate_path, spec), judged) + return kind(ctx, params, judged) except ListingError as exc: print(f"gate: {inert(str(exc))} -- refusing rather than judging a change it cannot read.", file=sys.stderr) - return 2, VERDICT_ERROR - return Evaluation(spec, result, rollout(repo_root, event, signal, base), signal, judged) + except PatternDeadlineError as exc: + # The reason names a time or the judge's own last stderr line, never the judged text. + reason = f"gate: {_policy_id(gate_path) or gate_path.name}: {inert(str(exc))}{_UNJUDGED}" + print(reason, file=sys.stderr) + return 2, VERDICT_ERROR def judge( diff --git a/src/chock/gate/runtime_shared.py b/src/chock/gate/runtime_shared.py index 7826a85d..22fc3e8b 100644 --- a/src/chock/gate/runtime_shared.py +++ b/src/chock/gate/runtime_shared.py @@ -39,9 +39,11 @@ "hashlib": "_chock_hashlib", "re": "_chock_re", "os": "_chock_os", + "queue": "_chock_queue", "shlex": "_chock_shlex", "shutil": "_chock_shutil", "subprocess": "_chock_subprocess", + "threading": "_chock_threading", "time": "_chock_time", "datetime": "_chock_datetime", "timezone": "_chock_timezone", diff --git a/src/chock/gate/tool_call_gate.py b/src/chock/gate/tool_call_gate.py index 28485150..b9941c23 100644 --- a/src/chock/gate/tool_call_gate.py +++ b/src/chock/gate/tool_call_gate.py @@ -11,12 +11,20 @@ import fnmatch import json -import re import subprocess import sys from pathlib import Path -from .budget import allowed_seconds, engine_deadline, time_left +from .budget import ( + PATTERN_SECONDS, + allowed_seconds, + close_judge, + engine_deadline, + engine_remaining, + judge_patterns, + pattern_judge, + time_left, +) from .session_log import ( session_for, session_outcome, @@ -60,10 +68,22 @@ def _tool_call_refusal(spec, spoken): return spoken or str(spec.get("message") or "").strip() or "blocked by a chock tool_call gate" -def _tool_call_regex(spec, tool_input): - """`deny` verdict when the input, as JSON text, matches the gate's `content_pattern`.""" +def _tool_call_regex(spec, tool_input, deadline, policy): + """`deny` verdict when the input, as JSON text, matches the gate's `content_pattern`. + + The pattern runs under the engine's pattern deadline, within what is left of `deadline`; a + pattern that reaches no verdict in time refuses, naming the gate. + """ text = json.dumps(tool_input, sort_keys=True, ensure_ascii=False) - if re.search(str((spec.get("params") or {}).get("content_pattern", "")), text): + pattern = str((spec.get("params") or {}).get("content_pattern", "")) + judge = pattern_judge() + try: + hits, why = judge_patterns(judge, [(pattern, [text])], min(PATTERN_SECONDS, engine_remaining(deadline))) + finally: + close_judge(judge) + if why: + return ("deny", f"content_regex gate {policy}: {why}{_TOOL_CALL_UNDECIDED}") + if hits[0]: return ("deny", _tool_call_refusal(spec, "")) return None @@ -102,12 +122,12 @@ def _tool_call_script(spec, root, payload, deadline=None): return ("deny", f"script gate: {script.name} exited {proc.returncode}{detail}{_TOOL_CALL_UNDECIDED}") -def _tool_call_verdict(spec, root, event, deadline): +def _tool_call_verdict(spec, root, event, deadline, policy): """None to allow, else (verdict, message) for this call; a kind a tool call cannot answer refuses.""" tool, tool_input = str(event.tool or ""), tool_call_input(event) kind = spec.get("kind") if kind == _TOOL_CALL_REGEX_KIND: - return _tool_call_regex(spec, tool_input) + return _tool_call_regex(spec, tool_input, deadline, policy) if kind == _TOOL_CALL_SCRIPT_KIND: payload = { "event": TOOL_CALL_EVENT, @@ -167,7 +187,8 @@ def evaluate_tool_call(argv, event): if TOOL_CALL_EVENT in spec.get("on", []) and tool_call_matches( (spec.get("params") or {}).get("tools", []), str(event.tool or "") ): - verdict = _tool_call_capped(spec, _tool_call_verdict(spec, root, event, engine_deadline())) + policy = gate.parent.parent.name + verdict = _tool_call_capped(spec, _tool_call_verdict(spec, root, event, engine_deadline(), policy)) if spec.get("kind") in _TOOL_CALL_NEEDS_SESSION: session_record(root, event, "pre", _TOOL_CALL_BLOCKED if verdict and verdict[0] == "deny" else None) return verdict diff --git a/src/chock/history/gates.py b/src/chock/history/gates.py index 0db17ce6..e39b83fd 100644 --- a/src/chock/history/gates.py +++ b/src/chock/history/gates.py @@ -26,47 +26,41 @@ @dataclass(frozen=True) class Rule: + """A gate's patterns as text: the scan runs them in the pattern judge (chock.gate.budget), never in-process.""" + policy: str - content_re: re.Pattern[str] - path_re: re.Pattern[str] | None - pragma_re: re.Pattern[str] | None + content_pattern: str + path_pattern: str | None + pragma_pattern: str | None judge: GateContext def covers(self, path: str) -> bool: return self.judge.in_scope(path) - def hits(self, text: str) -> list[int]: - """1-based line numbers whose text matches and carries no waiver pragma.""" - found = [] - for number, line in enumerate(text.splitlines(), start=1): - if self.pragma_re and self.pragma_re.search(line): - continue - if self.content_re.search(line): - found.append(number) - return found - -def _compile(policy: str, pattern: object) -> re.Pattern[str] | None: +def _checked(policy: str, pattern: object) -> str | None: + """`pattern` as text once it is known to compile; None for none.""" if not pattern: return None try: - return re.compile(str(pattern)) + re.compile(str(pattern)) except re.error as exc: raise HistoryError(f"{policy}: gate pattern does not compile ({exc}); refusing to scan without it") from exc + return str(pattern) def _rule(repo: Path, gate_path: Path, spec: dict) -> Rule: policy = gate_path.parents[1].name params = spec.get("params") or {} - content = _compile(policy, params.get("content_pattern")) + content = _checked(policy, params.get("content_pattern")) if content is None: raise HistoryError(f"{policy}: content_regex gate has no content_pattern; refusing to scan without it") judge = GateContext(repo_root=repo, scope=spec.get("paths"), own=own_paths(gate_path)) return Rule( policy, content, - _compile(policy, params.get("forbidden_path_regex")), - _compile(policy, params.get("allowlist_pragma")), + _checked(policy, params.get("forbidden_path_regex")), + _checked(policy, params.get("allowlist_pragma")), judge, ) diff --git a/src/chock/history/scan.py b/src/chock/history/scan.py index 600a6ed0..db6e3fee 100644 --- a/src/chock/history/scan.py +++ b/src/chock/history/scan.py @@ -2,24 +2,23 @@ from __future__ import annotations -import contextlib import hashlib -import signal +import shutil import sys -import threading import time -from collections.abc import Iterator from dataclasses import dataclass, field from pathlib import Path +from chock.gate.budget import PATTERN_SECONDS, close_judge, judge_patterns, pattern_judge from chock.history import gitlog from chock.history.gates import RULE_CONTENT, RULE_PATH, Rule DEFAULT_MAX_COMMITS = 5000 DEFAULT_MAX_BLOB_BYTES = 1_048_576 DEFAULT_TIME_BUDGET = 1800 -#: Seconds one gate may spend on one blob; a repo-supplied pattern can be catastrophic. -BLOB_SECONDS = 10 +#: Seconds one gate may spend on one blob (and on the paths): a repo-supplied pattern can be catastrophic, so +#: every pattern runs in the engine's pattern judge, a child the scan kills at the deadline on every platform. +BLOB_SECONDS = PATTERN_SECONDS _PROGRESS_EVERY = 500 Judged = dict[str, tuple[list[tuple[int, int, str]], set[int]]] @@ -66,36 +65,40 @@ def _digest(text: str) -> str: return hashlib.sha256(text.encode("utf-8", "replace")).hexdigest() -@contextlib.contextmanager -def _gate_budget(policy: str) -> Iterator[None]: - """Turn one gate running past BLOB_SECONDS into an error naming it (POSIX main thread; else unbounded).""" - if not hasattr(signal, "setitimer") or threading.current_thread() is not threading.main_thread(): - yield - return +def _judge() -> dict: + """A pattern judge in this Python, or the first on PATH when chock runs as a frozen binary.""" + interpreter = ( + sys.executable if not getattr(sys, "frozen", False) else shutil.which("python3") or shutil.which("python") + ) + if not interpreter: + raise gitlog.HistoryError( + "no python on PATH to run the gate patterns under a deadline; refusing to scan unbounded" + ) + return pattern_judge(interpreter) - def expired(_signum: int, _frame: object) -> None: - raise gitlog.HistoryError(f"gate {policy} ran over {BLOB_SECONDS}s on one blob (catastrophic pattern?)") - previous = signal.signal(signal.SIGALRM, expired) - signal.setitimer(signal.ITIMER_REAL, BLOB_SECONDS) - try: - yield - finally: - signal.setitimer(signal.ITIMER_REAL, 0) - signal.signal(signal.SIGALRM, previous) +def _judged(judge: dict, policy: str, jobs: list[tuple[str, list[str]]]) -> list[set[int]]: + """Per job of (pattern, texts), the texts `policy`'s pattern matches; past BLOB_SECONDS the scan errors naming it.""" + hits, why = judge_patterns(judge, jobs, BLOB_SECONDS) + if why: + raise gitlog.HistoryError(f"gate {policy}: {why}; a scan that cannot judge a blob reports nothing") + return [set(found) for found in hits] -def _judge_blob(rules: list[Rule], data: bytes) -> tuple[list[tuple[int, int, str]], set[int]]: +def _judge_blob(rules: list[Rule], data: bytes, judge: dict) -> tuple[list[tuple[int, int, str]], set[int]]: """(rule index, line number, line digest) per hit, and the rules whose waiver pragma the blob carries.""" text = data.decode("utf-8", "replace") lines = text.splitlines() hits: list[tuple[int, int, str]] = [] waived: set[int] = set() for index, rule in enumerate(rules): - with _gate_budget(rule.policy): - hits += [(index, n, _digest(lines[n - 1])) for n in rule.hits(text)] - if rule.pragma_re and rule.pragma_re.search(text): - waived.add(index) + jobs = [(rule.content_pattern, lines)] + if rule.pragma_pattern: + jobs += [(rule.pragma_pattern, lines), (rule.pragma_pattern, [text])] + matched, pragma_lines, pragma_blob = [*_judged(judge, rule.policy, jobs), set(), set()][:3] + hits += [(index, n + 1, _digest(lines[n])) for n in sorted(matched - pragma_lines)] + if pragma_blob: + waived.add(index) return hits, waived @@ -109,25 +112,44 @@ def _read_judged(repo: Path, rules: list[Rule], oids: list[str], limits: Limits, wanted = [o for o in oids if 0 < sizes[o] <= limits.max_blob_bytes] deadline = time.monotonic() + limits.time_budget judged: Judged = {} - for done, (oid, data) in enumerate(gitlog.read_blobs(repo, wanted, sizes), start=1): - if time.monotonic() > deadline: - raise gitlog.HistoryError( - f"time budget of {limits.time_budget}s exceeded; narrow with --since or --max-commits" - ) - if gitlog.is_binary(data): - report.skipped_binary += 1 - else: - judged[oid] = _judge_blob(rules, data) - report.blobs_scanned += 1 - if done % _PROGRESS_EVERY == 0: - _say(f"history: read {done}/{len(wanted)} blobs") + judge = _judge() + try: + for done, (oid, data) in enumerate(gitlog.read_blobs(repo, wanted, sizes), start=1): + if time.monotonic() > deadline: + raise gitlog.HistoryError( + f"time budget of {limits.time_budget}s exceeded; narrow with --since or --max-commits" + ) + if gitlog.is_binary(data): + report.skipped_binary += 1 + else: + judged[oid] = _judge_blob(rules, data, judge) + report.blobs_scanned += 1 + if done % _PROGRESS_EVERY == 0: + _say(f"history: read {done}/{len(wanted)} blobs") + finally: + close_judge(judge) return judged +def _forbidden_paths(rules: list[Rule], paths: list[str]) -> list[set[str]]: + """Per rule, the touched paths its forbidden-path pattern names (none for a rule without one).""" + judge = _judge() + try: + return [ + {paths[n] for n in _judged(judge, rule.policy, [(rule.path_pattern, paths)])[0]} + if rule.path_pattern + else set() + for rule in rules + ] + finally: + close_judge(judge) + + def _collect(touched: list[gitlog.Touch], rules: list[Rule], judged: Judged) -> list[Finding]: """Findings oldest commit first; a line that persists across edits is reported once, where it entered.""" found: list[Finding] = [] seen: set[tuple[str, str, str]] = set() + forbidden = _forbidden_paths(rules, sorted({item.path for item in touched})) def add(item: gitlog.Touch, rule: Rule, kind: str, line: int | None, key: str) -> None: marker = (item.path, rule.policy, key) @@ -138,7 +160,7 @@ def add(item: gitlog.Touch, rule: Rule, kind: str, line: int | None, key: str) - for item in touched: hits, waived = judged.get(item.oid, ([], set())) for index, rule in enumerate(rules): - if rule.covers(item.path) and rule.path_re and rule.path_re.search(item.path) and index not in waived: + if rule.covers(item.path) and item.path in forbidden[index] and index not in waived: add(item, rule, RULE_PATH, None, RULE_PATH) for index, number, digest in hits: if rules[index].covers(item.path): diff --git a/src/chock/toggles.py b/src/chock/toggles.py index c98d4d5e..8e54e09f 100644 --- a/src/chock/toggles.py +++ b/src/chock/toggles.py @@ -14,6 +14,7 @@ from chock.compile.surfaces import AGENTS_ARG_REQUIRED_MSG from chock.config import agents_from_config as _agents_from_config from chock.config import load_config, policy_status, set_disabled +from chock.gate.build import CatastrophicPatternError from chock.index.cli import cmd_refresh from chock.manifest import ManifestSourceError, load_manifest from chock.output import error, warn @@ -209,7 +210,7 @@ def recompile_main(argv: list[str] | None) -> int: try: coverage = recompile(repo_root, agents, skip_hooks=args.skip_hooks) - except BookkeepingError as exc: + except (BookkeepingError, CatastrophicPatternError) as exc: error(str(exc)) return 1 print(f"Recompiled {len(coverage)} policies") diff --git a/src/chock/validation/checks_gate_shape.py b/src/chock/validation/checks_gate_shape.py index 97444a2f..246ac8ed 100644 --- a/src/chock/validation/checks_gate_shape.py +++ b/src/chock/validation/checks_gate_shape.py @@ -6,6 +6,7 @@ from pathlib import Path from typing import Any +from chock.gate.pattern_safety import REGEX_PARAM_SUFFIXES, catastrophic, catastrophic_message from chock.gate.runner import KINDS, WRITE_PATH_KINDS from chock.gate.schema import GATEWAY_ONLY_KINDS, KIND_PARAM_SCHEMAS, TOOL_CALL_EVENT, TOOL_CALL_KINDS from chock.validation.loading import schema_validator @@ -114,13 +115,10 @@ def _validate_script_shipped(params: dict[str, Any], artifact_dir: Path, gate_re ) -_REGEX_PARAM_SUFFIXES = ("_pattern", "_regex", "_pragma") - - def _validate_regex_params(params: dict[str, Any], gate_ref: str, report: Report) -> None: - """A pattern that does not compile is found at validate, not by the first commit it guards.""" + """A pattern that does not compile, or that can run forever, is found at validate, not by the first commit it guards.""" for name, value in params.items(): - if not (isinstance(value, str) and name.endswith(_REGEX_PARAM_SUFFIXES)): + if not (isinstance(value, str) and name.endswith(REGEX_PARAM_SUFFIXES)): continue try: re.compile(value) @@ -128,3 +126,7 @@ def _validate_regex_params(params: dict[str, Any], gate_ref: str, report: Report report.add( Finding(f"{gate_ref} (params)", _CATEGORY, "error", f"{name} is not a valid regular expression: {exc}") ) + continue + why = catastrophic(value) + if why: + report.add(Finding(f"{gate_ref} (params)", _CATEGORY, "error", catastrophic_message(name, value, why))) diff --git a/tests/fixtures/runtime_goldens/_chock_runtime.py b/tests/fixtures/runtime_goldens/_chock_runtime.py index 8e146e1c..d6366d9e 100644 --- a/tests/fixtures/runtime_goldens/_chock_runtime.py +++ b/tests/fixtures/runtime_goldens/_chock_runtime.py @@ -14,19 +14,25 @@ import fnmatch as _chock_fnmatch import re as _chock_re import os as _chock_os +import queue as _chock_queue import shlex as _chock_shlex import shutil as _chock_shutil import subprocess as _chock_subprocess +import threading as _chock_threading import time as _chock_time from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path from pathlib import PurePosixPath as _chock_PurePosixPath, PureWindowsPath as _chock_PureWindowsPath -CHOCK_RUNTIME_DIGEST = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +CHOCK_RUNTIME_DIGEST = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' ENGINE_BUDGET_SECONDS = 30 +PATTERN_SECONDS = 10 + +PATTERN_JUDGE_SOURCE = 'import json, re, sys\nfor line in sys.stdin:\n out = []\n for pattern, texts in json.loads(line):\n search = re.compile(pattern).search\n out.append([index for index, text in enumerate(texts) if search(text)])\n print(json.dumps(out), flush=True)\n' + def engine_deadline(): """The monotonic instant this invocation's budget runs out.""" return _chock_time.monotonic() + ENGINE_BUDGET_SECONDS @@ -46,6 +52,67 @@ def allowed_seconds(expired): """The time a timed-out call was actually given, as text (what the deadline left it, not the full budget).""" return f'{max(float(expired.timeout or 0), 0.0):.1f}s' +def pattern_judge(interpreter=None): + """A pattern judge, not yet started: the interpreter to run the patterns in (this one by default).""" + return {'argv': [interpreter or sys.executable, '-I', '-c', PATTERN_JUDGE_SOURCE], 'proc': None, 'lines': None} + +def _pump_lines(proc, lines): + for line in proc.stdout: + lines.put(line) + lines.put(None) + +def _start_judge(judge): + proc = _chock_subprocess.Popen(judge['argv'], stdin=_chock_subprocess.PIPE, stdout=_chock_subprocess.PIPE, stderr=_chock_subprocess.PIPE, text=True, encoding='utf-8') + lines = _chock_queue.Queue() + _chock_threading.Thread(target=_pump_lines, args=(proc, lines), daemon=True).start() + judge['proc'], judge['lines'] = (proc, lines) + +def close_judge(judge): + """Stop the judge's child, if one runs; the last line it wrote to stderr, or "".""" + proc = judge['proc'] + judge['proc'] = judge['lines'] = None + if proc is None: + return '' + if proc.poll() is None: + proc.kill() + proc.wait() + try: + spoken = proc.stderr.read() + except (OSError, ValueError): + spoken = '' + for stream in (proc.stdin, proc.stdout, proc.stderr): + with contextlib.suppress(OSError): + stream.close() + return (spoken.strip().splitlines() or [''])[-1] + +def judge_patterns(judge, jobs, seconds): + """(hits, None) or (None, why): per job of (pattern, texts), the indices of the texts the pattern matches. + + A job that runs past `seconds` kills the child and names the time; a job with no text is never + sent. `why` is text for a refusal: the judge is undecided and an undecided gate refuses. + """ + if not any((texts for _, texts in jobs)): + return ([[] for _ in jobs], None) + if judge['proc'] is None: + try: + _start_judge(judge) + except OSError as exc: + return (None, f'the pattern judge could not start ({exc})') + proc, lines = (judge['proc'], judge['lines']) + try: + proc.stdin.write(json.dumps(jobs) + '\n') + proc.stdin.flush() + except (OSError, ValueError): + return (None, f'the pattern judge exited early ({close_judge(judge)})') + try: + line = lines.get(timeout=seconds) + except _chock_queue.Empty: + close_judge(judge) + return (None, f'the patterns gave no verdict within {seconds:g}s (a catastrophic pattern?)') + if line is None: + return (None, f'the pattern judge exited early ({close_judge(judge)})') + return (json.loads(line), None) + LOG_CHAIN_PREV, LOG_CHAIN_SEQ = ('prev', 'seq') LOG_CHAIN_UNCHAINED = 'unchained' @@ -2420,10 +2487,22 @@ def tool_call_matches(globs, tool): def _tool_call_refusal(spec, spoken): return spoken or str(spec.get('message') or '').strip() or 'blocked by a chock tool_call gate' -def _tool_call_regex(spec, tool_input): - """`deny` verdict when the input, as JSON text, matches the gate's `content_pattern`.""" +def _tool_call_regex(spec, tool_input, deadline, policy): + """`deny` verdict when the input, as JSON text, matches the gate's `content_pattern`. + + The pattern runs under the engine's pattern deadline, within what is left of `deadline`; a + pattern that reaches no verdict in time refuses, naming the gate. + """ text = json.dumps(tool_input, sort_keys=True, ensure_ascii=False) - if _chock_re.search(str((spec.get('params') or {}).get('content_pattern', '')), text): + pattern = str((spec.get('params') or {}).get('content_pattern', '')) + judge = pattern_judge() + try: + hits, why = judge_patterns(judge, [(pattern, [text])], min(PATTERN_SECONDS, engine_remaining(deadline))) + finally: + close_judge(judge) + if why: + return ('deny', f'content_regex gate {policy}: {why}{_TOOL_CALL_UNDECIDED}') + if hits[0]: return ('deny', _tool_call_refusal(spec, '')) return None @@ -2450,12 +2529,12 @@ def _tool_call_script(spec, root, payload, deadline=None): detail = f': {first}' if first else '' return ('deny', f'script gate: {script.name} exited {proc.returncode}{detail}{_TOOL_CALL_UNDECIDED}') -def _tool_call_verdict(spec, root, event, deadline): +def _tool_call_verdict(spec, root, event, deadline, policy): """None to allow, else (verdict, message) for this call; a kind a tool call cannot answer refuses.""" tool, tool_input = (str(event.tool or ''), tool_call_input(event)) kind = spec.get('kind') if kind == _TOOL_CALL_REGEX_KIND: - return _tool_call_regex(spec, tool_input) + return _tool_call_regex(spec, tool_input, deadline, policy) if kind == _TOOL_CALL_SCRIPT_KIND: payload = {'event': TOOL_CALL_EVENT, 'repo_root': str(root), 'tool': tool, 'input': tool_input, 'session': session_for(event, root)} return _tool_call_script(spec, root, payload, deadline) @@ -2498,7 +2577,8 @@ def evaluate_tool_call(argv, event): root = repo_root_for(event, gate) verdict = None if TOOL_CALL_EVENT in spec.get('on', []) and tool_call_matches((spec.get('params') or {}).get('tools', []), str(event.tool or '')): - verdict = _tool_call_capped(spec, _tool_call_verdict(spec, root, event, engine_deadline())) + policy = gate.parent.parent.name + verdict = _tool_call_capped(spec, _tool_call_verdict(spec, root, event, engine_deadline(), policy)) if spec.get('kind') in _TOOL_CALL_NEEDS_SESSION: session_record(root, event, 'pre', _TOOL_CALL_BLOCKED if verdict and verdict[0] == 'deny' else None) return verdict diff --git a/tests/fixtures/runtime_goldens/antigravity.py b/tests/fixtures/runtime_goldens/antigravity.py index 9018a886..1d28dcf4 100644 --- a/tests/fixtures/runtime_goldens/antigravity.py +++ b/tests/fixtures/runtime_goldens/antigravity.py @@ -698,7 +698,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/claude_code.py b/tests/fixtures/runtime_goldens/claude_code.py index 4309b3db..2abc7f02 100644 --- a/tests/fixtures/runtime_goldens/claude_code.py +++ b/tests/fixtures/runtime_goldens/claude_code.py @@ -809,7 +809,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/codex_cli.py b/tests/fixtures/runtime_goldens/codex_cli.py index 9db704bc..6dce9316 100644 --- a/tests/fixtures/runtime_goldens/codex_cli.py +++ b/tests/fixtures/runtime_goldens/codex_cli.py @@ -777,7 +777,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/cursor.py b/tests/fixtures/runtime_goldens/cursor.py index b49d89bc..86a81d17 100644 --- a/tests/fixtures/runtime_goldens/cursor.py +++ b/tests/fixtures/runtime_goldens/cursor.py @@ -670,7 +670,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/devin.py b/tests/fixtures/runtime_goldens/devin.py index 5f5e335d..fb43ae82 100644 --- a/tests/fixtures/runtime_goldens/devin.py +++ b/tests/fixtures/runtime_goldens/devin.py @@ -776,7 +776,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/factory_droid.py b/tests/fixtures/runtime_goldens/factory_droid.py index bb2469c8..821ebfa1 100644 --- a/tests/fixtures/runtime_goldens/factory_droid.py +++ b/tests/fixtures/runtime_goldens/factory_droid.py @@ -763,7 +763,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/gemini_cli.py b/tests/fixtures/runtime_goldens/gemini_cli.py index 0b3415c8..e2b579fd 100644 --- a/tests/fixtures/runtime_goldens/gemini_cli.py +++ b/tests/fixtures/runtime_goldens/gemini_cli.py @@ -776,7 +776,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/goose.py b/tests/fixtures/runtime_goldens/goose.py index cb101514..a59f01fd 100644 --- a/tests/fixtures/runtime_goldens/goose.py +++ b/tests/fixtures/runtime_goldens/goose.py @@ -732,7 +732,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/grok.py b/tests/fixtures/runtime_goldens/grok.py index 67ed740a..6238505f 100644 --- a/tests/fixtures/runtime_goldens/grok.py +++ b/tests/fixtures/runtime_goldens/grok.py @@ -776,7 +776,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/mistral_vibe.py b/tests/fixtures/runtime_goldens/mistral_vibe.py index 513f3352..98a6c4d2 100644 --- a/tests/fixtures/runtime_goldens/mistral_vibe.py +++ b/tests/fixtures/runtime_goldens/mistral_vibe.py @@ -759,7 +759,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/openhands.py b/tests/fixtures/runtime_goldens/openhands.py index 139fb0c3..a4cb61df 100644 --- a/tests/fixtures/runtime_goldens/openhands.py +++ b/tests/fixtures/runtime_goldens/openhands.py @@ -732,7 +732,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/tabnine.py b/tests/fixtures/runtime_goldens/tabnine.py index 0fa87617..57f0b2cf 100644 --- a/tests/fixtures/runtime_goldens/tabnine.py +++ b/tests/fixtures/runtime_goldens/tabnine.py @@ -732,7 +732,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/vscode_copilot.py b/tests/fixtures/runtime_goldens/vscode_copilot.py index 1b0eb253..2abbe0a0 100644 --- a/tests/fixtures/runtime_goldens/vscode_copilot.py +++ b/tests/fixtures/runtime_goldens/vscode_copilot.py @@ -560,7 +560,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/fixtures/runtime_goldens/windsurf.py b/tests/fixtures/runtime_goldens/windsurf.py index 422257f6..8c469582 100644 --- a/tests/fixtures/runtime_goldens/windsurf.py +++ b/tests/fixtures/runtime_goldens/windsurf.py @@ -629,7 +629,7 @@ def degrade(decision, event): # Not self-contained: chock's guard logic lives in the sibling _chock_runtime.py, loaded from this # file's own folder and pinned to the render this entry came from. Missing, unreadable, another # version, or failing to import: every call refuses (_CHOCK_RUNTIME_FAULT), never allows. -_CHOCK_RUNTIME_PIN = 'a2059af84a40b0a567de8f6d9bf7abcb7e7949159d016fdf19691fd2e0fef82a' +_CHOCK_RUNTIME_PIN = 'a0dfe128bff953bb9f650d1e8712bf3c519af60325fa39207fadbd566c886124' def _chock_load_runtime(): diff --git a/tests/test_gate_bypasses.py b/tests/test_gate_bypasses.py index a1238d2b..b3769883 100644 --- a/tests/test_gate_bypasses.py +++ b/tests/test_gate_bypasses.py @@ -18,7 +18,7 @@ AWS_KEY = 'KEY = "AKIAIOSFODNN7EXAMPLE"\n' # pragma: allowlist secret (documented AWS test fixture) -def _secret_gate(tmp_path: Path) -> Path: +def _akia_gate(tmp_path: Path) -> Path: return write_gate( tmp_path, {"kind": "content_regex", "on": ["commit"], "action": "block", "message": "secret", "params": SECRET_PARAMS}, @@ -36,13 +36,13 @@ def test_renamed_and_edited_file_is_scanned(tmp_path: Path) -> None: subprocess.run(["git", "mv", "notes.txt", "config.txt"], cwd=tmp_path, check=True) (tmp_path / "config.txt").write_text("just notes\n" + AWS_KEY, encoding="utf-8") subprocess.run(["git", "add", "config.txt"], cwd=tmp_path, check=True) - assert run(_secret_gate(tmp_path), "pre-commit", None, tmp_path) == 1 + assert run(_akia_gate(tmp_path), "pre-commit", None, tmp_path) == 1 def test_non_ascii_filename_is_scanned(tmp_path: Path) -> None: init_repo(tmp_path) stage(tmp_path, "sécrets.txt", AWS_KEY) - assert run(_secret_gate(tmp_path), "pre-commit", None, tmp_path) == 1 + assert run(_akia_gate(tmp_path), "pre-commit", None, tmp_path) == 1 def test_nested_manifest_dependency_is_scanned(tmp_path: Path) -> None: @@ -83,7 +83,7 @@ def test_ci_fails_closed_on_missing_base(tmp_path: Path, capsys) -> None: init_repo(tmp_path) stage(tmp_path, "app.py", "x = 1\n") _commit(tmp_path) - rc = run(_secret_gate(tmp_path), "ci", None, tmp_path, base="origin/does-not-exist") + rc = run(_akia_gate(tmp_path), "ci", None, tmp_path, base="origin/does-not-exist") assert rc == 2 assert "does not resolve" in capsys.readouterr().err @@ -97,7 +97,7 @@ def test_ci_scans_a_valid_base(tmp_path: Path) -> None: ).stdout.strip() stage(tmp_path, "leak.py", AWS_KEY) _commit(tmp_path, "leak") - assert run(_secret_gate(tmp_path), "ci", None, tmp_path, base=base) == 1 + assert run(_akia_gate(tmp_path), "ci", None, tmp_path, base=base) == 1 def test_merge_commit_installs_a_pre_merge_commit_dispatcher(tmp_path: Path) -> None: diff --git a/tests/test_gate_core.py b/tests/test_gate_core.py index e5ecd76f..748c640c 100644 --- a/tests/test_gate_core.py +++ b/tests/test_gate_core.py @@ -17,7 +17,7 @@ } -def _secret_gate(tmp_path: Path) -> Path: +def _akia_gate(tmp_path: Path) -> Path: return write_gate( tmp_path, { @@ -33,19 +33,19 @@ def _secret_gate(tmp_path: Path) -> Path: def test_blocks_aws_key(tmp_path: Path) -> None: init_repo(tmp_path) stage(tmp_path, "app.py", 'KEY = "AKIAIOSFODNN7EXAMPLE"\n') - assert run(_secret_gate(tmp_path), "pre-commit", None, tmp_path) == 1 + assert run(_akia_gate(tmp_path), "pre-commit", None, tmp_path) == 1 def test_allows_clean_file(tmp_path: Path) -> None: init_repo(tmp_path) stage(tmp_path, "app.py", "x = 1\n") - assert run(_secret_gate(tmp_path), "pre-commit", None, tmp_path) == 0 + assert run(_akia_gate(tmp_path), "pre-commit", None, tmp_path) == 0 def test_pragma_exempts_line(tmp_path: Path) -> None: init_repo(tmp_path) stage(tmp_path, "t.py", 'KEY = "AKIAIOSFODNN7EXAMPLE" # pragma: allowlist secret\n') - assert run(_secret_gate(tmp_path), "pre-commit", None, tmp_path) == 0 + assert run(_akia_gate(tmp_path), "pre-commit", None, tmp_path) == 0 def test_forbidden_path_blocked(tmp_path: Path) -> None: diff --git a/tests/test_pattern_deadline.py b/tests/test_pattern_deadline.py new file mode 100644 index 00000000..03d4a0f2 --- /dev/null +++ b/tests/test_pattern_deadline.py @@ -0,0 +1,172 @@ +"""A gate pattern cannot hang a hook or fail it open: every pattern runs under a deadline, on every platform. + +`re` holds the GIL and nothing can interrupt it in-process, so the patterns run in a child the +engine kills at the deadline (chock.gate.budget). The git hook, the in-agent write path, the +tool_call path and the history scan all refuse naming the gate; none waits past the deadline. +""" + +from __future__ import annotations + +import ast +import json +import subprocess +import sys +import time +from pathlib import Path +from types import SimpleNamespace + +import pytest +from conftest import init_repo, stage + +from chock.gate import budget, tool_call_gate, write_gate +from chock.gate.assemble import runner_source +from chock.gate.gate_outcome import GATE_ERRORED, gate_decision +from chock.gate.runner import kinds, run +from chock.history import scan + +POLICY = "redos-gate" +REDOS = r"^(a+)+$" +LINE = "a" * 60 + "!\n" +SECONDS = 1 +SPEC = {"kind": "content_regex", "on": ["commit", "tool_use"], "action": "block", "message": "m", "params": {}} + + +def _vendored(seconds: int = SECONDS) -> str: + """The runner as `chock sync` vendors it, with the pattern deadline cut to `seconds`.""" + source = runner_source() + assert f"\nPATTERN_SECONDS = {budget.PATTERN_SECONDS}\n" in source + return source.replace(f"\nPATTERN_SECONDS = {budget.PATTERN_SECONDS}\n", f"\nPATTERN_SECONDS = {seconds}\n", 1) + + +def _installed(repo: Path, surface: str, pattern: str = REDOS) -> Path: + """A compiled content gate on `surface` and the vendored runner, laid out as `chock sync` leaves them.""" + gate = repo / ".chock" / "compiled" / POLICY / surface / "gate.json" + gate.parent.mkdir(parents=True, exist_ok=True) + gate.write_text(json.dumps({**SPEC, "params": {"content_pattern": pattern}}), encoding="utf-8") + runner = repo / ".chock" / "bin" / "gate.py" + runner.parent.mkdir(parents=True, exist_ok=True) + runner.write_text(_vendored(), encoding="utf-8", newline="\n") + return gate + + +def test_the_git_path_exits_2_within_the_deadline_naming_the_gate(tmp_path: Path) -> None: + repo = init_repo(tmp_path) + gate = _installed(repo, "git-hook") + stage(repo, "a.txt", LINE) + started = time.monotonic() + proc = subprocess.run( + [sys.executable, str(repo / ".chock" / "bin" / "gate.py"), "run", "--gate", str(gate), "--event", "pre-commit"], + cwd=repo, + capture_output=True, + text=True, + timeout=budget.PATTERN_SECONDS, + check=False, + ) + assert time.monotonic() - started < budget.PATTERN_SECONDS + assert proc.returncode == 2, proc.stderr + assert f"gate: {POLICY}: the patterns gave no verdict within {SECONDS}s" in proc.stderr + assert "catastrophic pattern" in proc.stderr and "refusing rather than allowing" in proc.stderr + + +def test_a_sane_pattern_on_the_same_line_still_blocks(tmp_path: Path) -> None: + repo = init_repo(tmp_path) + gate = _installed(repo, "git-hook", pattern="!") + stage(repo, "a.txt", LINE) + proc = subprocess.run( + [sys.executable, str(repo / ".chock" / "bin" / "gate.py"), "run", "--gate", str(gate), "--event", "pre-commit"], + cwd=repo, + capture_output=True, + text=True, + timeout=budget.PATTERN_SECONDS, + check=False, + ) + assert proc.returncode == 1 and "a.txt" in proc.stderr + + +def test_the_in_package_runner_refuses_the_same_way(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys) -> None: + repo = init_repo(tmp_path) + gate = _installed(repo, "git-hook") + stage(repo, "a.txt", LINE) + monkeypatch.setattr(kinds, "PATTERN_SECONDS", SECONDS) + assert run(gate, "pre-commit", None, repo) == 2 + err = capsys.readouterr().err + assert f"gate: {POLICY}: the patterns gave no verdict within {SECONDS}s" in err + + +def test_the_write_path_refuses_naming_the_gate(tmp_path: Path) -> None: + repo = init_repo(tmp_path) + gate = _installed(repo, "pre-tool-use") + outcome, message = write_gate.run_gate(gate, {"a.txt": LINE}, "pre-tool-use", (repo, {}), budget.engine_deadline()) + assert outcome == GATE_ERRORED + assert f"gate: {POLICY}: the patterns gave no verdict within {SECONDS}s" in message + verdict, spoken = gate_decision(outcome, message, gate) + assert verdict == "deny" and POLICY in spoken + + +def test_the_tool_call_path_refuses_naming_the_gate(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = init_repo(tmp_path) + gate = repo / ".chock" / "compiled" / POLICY / "pre-tool-use" / "tool-call-gate.json" + gate.parent.mkdir(parents=True) + spec = { + "kind": "content_regex", + "on": ["tool_call"], + "action": "block", + "params": {"content_pattern": r'(a+)+"', "tools": ["Bash"]}, # judged as JSON text: quoted, in braces + } + gate.write_text(json.dumps(spec), encoding="utf-8") + monkeypatch.setattr(tool_call_gate, "PATTERN_SECONDS", SECONDS) + event = SimpleNamespace( + event="pre_tool", tool="Bash", raw={"tool_input": {"command": "a" * 60 + "!"}}, cwd=str(repo) + ) + started = time.monotonic() + verdict = tool_call_gate.evaluate_tool_call(["--tool-call", str(gate)], event) + assert time.monotonic() - started < budget.PATTERN_SECONDS + assert verdict is not None and verdict[0] == "deny" + assert f"content_regex gate {POLICY}: the patterns gave no verdict within {SECONDS}s" in verdict[1] + assert "refusing rather than allowing" in verdict[1] + event.raw = {"tool_input": {"command": "aaa"}} + assert tool_call_gate.evaluate_tool_call(["--tool-call", str(gate)], event) == ( + "deny", + "blocked by a chock tool_call gate", + ) + event.raw = {"tool_input": {"command": "b"}} + assert tool_call_gate.evaluate_tool_call(["--tool-call", str(gate)], event) is None + + +def test_the_history_scan_needs_no_signal() -> None: + """The scan's deadline is the judge's, not SIGALRM's: Windows is bounded the same way.""" + tree = ast.parse(Path(scan.__file__).read_text(encoding="utf-8")) + imported = {alias.name for node in tree.body if isinstance(node, ast.Import) for alias in node.names} + assert "signal" not in imported and "threading" not in imported + assert scan.BLOB_SECONDS == budget.PATTERN_SECONDS + + +def test_the_judge_matches_times_out_and_reports_a_dead_child() -> None: + judge = budget.pattern_judge() + assert budget.judge_patterns(judge, [("a", []), ("b", [])], SECONDS) == ([[], []], None) + assert judge["proc"] is None, "nothing to judge starts no child" + assert budget.judge_patterns(judge, [("a+", ["xa", "b", "aaa"]), (r"\d", ["1", "x"])], SECONDS) == ( + [[0, 2], [0]], + None, + ) + started = time.monotonic() + hits, why = budget.judge_patterns(judge, [(REDOS, [LINE])], SECONDS) + assert hits is None and why == f"the patterns gave no verdict within {SECONDS}s (a catastrophic pattern?)" + assert SECONDS <= time.monotonic() - started < budget.PATTERN_SECONDS + assert judge["proc"] is None, "a timed-out child is killed" + hits, why = budget.judge_patterns(judge, [("(", ["x"])], SECONDS) + assert hits is None and why.startswith("the pattern judge exited early (") + assert budget.judge_patterns(judge, [("x", ["x"])], SECONDS) == ([[0]], None), "a new child after a crash" + assert budget.close_judge(judge) == "" and judge["proc"] is None + + +def test_the_judge_keeps_text_whole_across_the_pipe() -> None: + judge = budget.pattern_judge() + texts = ["café ☃", "a\nb", "lone \udcff surrogate", "tab\there", '"quoted" \\ back'] + try: + assert budget.judge_patterns(judge, [("☃|\udcff|\\\\ back|^a$", texts)], budget.PATTERN_SECONDS) == ( + [[0, 2, 4]], + None, + ) + finally: + budget.close_judge(judge) diff --git a/tests/test_pattern_safety.py b/tests/test_pattern_safety.py new file mode 100644 index 00000000..e5d9eaef --- /dev/null +++ b/tests/test_pattern_safety.py @@ -0,0 +1,133 @@ +"""A gate pattern a backtracking engine can run in exponential time is refused before it reaches a gate.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml +from conftest import init_repo + +from chock.gate.build import CatastrophicPatternError, build_gate_json +from chock.gate.pattern_safety import catastrophic +from chock.toggles import recompile_main +from chock.validation.checks_gate_shape import _validate_gate +from chock.validation.report import Report + +REFUSED = { + "nested plus": r"^(a+)+$", + "optional separator": r"(\w+\s?)+$", + "separator inside the class": r"(?:\.[\w.-]+)+", + "nested star": r"(a*)*b", + "bounded inner": r"(a{2,3})+$", + "nested through a group": r"((a+)b?)+", + "two inner repeats": r"(x+x+)+y", + "overlapping branch in a loop": r"(?:a+|b)+$", + "bounded outer": r"(a+){1,50}$", + "classic email": r"([a-z]+\d*)*@", +} +ACCEPTED = { + "separated repeats": r"(\d+\.){3}\d+", + "fixed count": r"(?i)AKIA[0-9A-Z]{16}", + "literal separator": r"(a+b)+", + "literal lead": r"(ab+)+", + "adjacent repeats": r"a+a+", + "base64": r"(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==)?", + "distinct separator": r"(\w+@)+", + "optional inner": r"^(a?)+$", + "lookahead-bounded whitespace": r"(?:\s+(?=[^\s#])|x)*", + "end anchor in the loop": r"(?:[&!]\S*(?:\s+|$))*", + "literal word after the repeat": r"(?:(?:[\w./-]*/)?sudo\s+)*", + "alternation": r"(?:foo|bar)+", +} +#: Shapes the check does not see; the runtime deadline (chock.gate.budget) is their backstop. +MISSED = {"overlapping alternation": r"(a|aa)+", "separator the inner absorbs": r"(.*a)+"} + + +@pytest.mark.parametrize("name", sorted(REFUSED)) +def test_a_nested_repeat_that_can_run_into_the_next_iteration_is_refused(name: str) -> None: + assert catastrophic(REFUSED[name]), name + + +@pytest.mark.parametrize("name", sorted(ACCEPTED)) +def test_a_pattern_whose_repeats_cannot_overlap_is_accepted(name: str) -> None: + assert catastrophic(ACCEPTED[name]) is None, name + + +@pytest.mark.parametrize("name", sorted(MISSED)) +def test_the_shapes_the_check_leaves_to_the_deadline_are_pinned(name: str) -> None: + assert catastrophic(MISSED[name]) is None, name + + +def test_a_pattern_that_does_not_compile_is_not_this_checks_finding() -> None: + assert catastrophic("(") is None + + +def test_every_installed_gate_pattern_is_accepted() -> None: + """The check never fires on the patterns chock itself installs.""" + root = Path(__file__).resolve().parents[1] + seen = 0 + for manifest in sorted((root / ".agents" / "policies").glob("*/manifest.yaml")): + params = ((yaml.safe_load(manifest.read_text(encoding="utf-8")) or {}).get("hook") or {}).get("gate") or {} + for name, value in (params.get("params") or {}).items(): + if isinstance(value, str) and name.endswith(("_pattern", "_regex", "_pragma")): + seen += 1 + assert catastrophic(value) is None, f"{manifest.parent.name}: {name}" + assert seen + + +def _gate(**params: str) -> dict: + return {"kind": "content_regex", "on": ["commit"], "action": "block", "message": "m", "params": params} + + +def test_validate_names_the_param_the_pattern_and_the_shape() -> None: + report = Report() + _validate_gate(_gate(content_pattern=r"^(a+)+$", allowlist_pragma="ok"), "test", report) + messages = [f.message for f in report.errors] + assert len(messages) == 1 + assert messages[0].startswith("content_pattern can take exponential time on a crafted line (a repeat nested") + assert "'^(a+)+$'" in messages[0] and "both accept 'a'" in messages[0] + + +def test_validate_shows_a_long_pattern_cut_short() -> None: + report = Report() + _validate_gate(_gate(content_pattern="x" * 200 + r"(a+)+"), "test", report) + assert "..." in report.errors[0].message and "x" * 200 not in report.errors[0].message + + +def _policy(root: Path, pattern: str) -> Path: + policy = root / ".agents" / "policies" / "redos-gate" + policy.mkdir(parents=True) + manifest = { + "id": "redos-gate", + "name": "redos", + "version": "0.0.1", + "description": "d", + "artifact": "hook", + "enforcement": "block", + "hook": {"gate": {"kind": "content_regex", "on": ["commit"], "params": {"content_pattern": pattern}}}, + "provenance": {"author": "t"}, + "lifecycle": {"status": "draft"}, + } + (policy / "manifest.yaml").write_text(yaml.safe_dump(manifest), encoding="utf-8") + return policy + + +def test_the_compile_step_refuses_naming_policy_param_and_pattern(tmp_path: Path) -> None: + policy = _policy(tmp_path, r"^(a+)+$") + with pytest.raises( + CatastrophicPatternError, match=r"redos-gate: hook\.gate\.params\.content_pattern .*'\^\(a\+\)\+\$'" + ): + build_gate_json(policy, tmp_path) + assert build_gate_json(_policy(tmp_path / "ok", r"^a+$"), tmp_path / "ok") is not None + + +def test_sync_refuses_the_pattern_and_compiles_nothing(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + repo = init_repo(tmp_path) + _policy(repo, r"^(a+)+$") + (repo / ".chock").mkdir() + (repo / ".chock" / "config.yaml").write_text("agents: [claude]\n", encoding="utf-8") + rc = recompile_main(["--repo", str(repo), "--skip-hooks"]) + err = capsys.readouterr().err + assert rc == 1 and "redos-gate: hook.gate.params.content_pattern" in err and "nothing was compiled" in err + assert not (repo / ".chock" / "compiled" / "redos-gate" / "git-hook" / "gate.json").exists() diff --git a/tests/test_runner_assemble.py b/tests/test_runner_assemble.py index b26edc14..6d22e5b3 100644 --- a/tests/test_runner_assemble.py +++ b/tests/test_runner_assemble.py @@ -8,11 +8,11 @@ import pytest -from chock.gate import assemble -from chock.gate.assemble import runner_source -from chock.gate.budget import ENGINE_BUDGET_SECONDS +from chock.gate import assemble, budget +from chock.gate.assemble import budget_fragment, runner_source PACKAGE = Path(assemble.__file__).resolve().parent / "runner" +BUDGET = Path(budget.__file__).resolve() PRELUDE = Path(assemble.__file__).resolve().parent / "data" / "gate_prelude.py.tmpl" MARKER = re.compile(r"^# >>> gate\.py (\d+)$", re.MULTILINE) ENTRY_POINTS = {"__init__.py", "__main__.py"} @@ -82,11 +82,23 @@ def test_a_header_imports_only_the_stdlib_and_its_siblings() -> None: def test_the_prelude_binds_every_stdlib_name_a_header_imports() -> None: """A module may import what it reads; gate.py has only the prelude's imports to read it from.""" prelude = _bound(ast.parse(PRELUDE.read_text(encoding="utf-8"))) - for path in _modules(): + for path in [*_modules(), BUDGET]: missing = _bound(ast.parse(_header(path))) - prelude assert not missing, f"{path.name} imports {sorted(missing)}; add them to {PRELUDE.name}" +def test_the_budget_module_is_vendored_whole_and_class_free() -> None: + """gate.py carries chock.gate.budget's every definition, so the runner's deadlines are the engine's own.""" + tree = ast.parse(BUDGET.read_text(encoding="utf-8")) + assert not any(isinstance(node, ast.ClassDef | ast.AnnAssign) for node in tree.body), ( + "budget.py is extracted by definition: a class or annotated assignment would be dropped" + ) + defined = [node.name for node in tree.body if isinstance(node, ast.FunctionDef)] + for name in defined: + assert f"\ndef {name}(" in runner_source(), name + assert runner_source().count("def judge_patterns(") == 1 + + def test_the_assembled_runner_is_one_compilable_file() -> None: source = runner_source() compile(source, "gate.py", "exec") @@ -101,7 +113,11 @@ def _package(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, modules: dict[str, for name, text in modules.items(): (pkg / name).write_text(text, encoding="utf-8") real = assemble.package_data_dir - monkeypatch.setattr(assemble, "package_data_dir", lambda package, *sub: pkg if not sub else real(package, *sub)) + + def stand_in(package: str, *sub: str) -> Path: + return pkg if package == assemble.RUNNER_PACKAGE and not sub else real(package, *sub) + + monkeypatch.setattr(assemble, "package_data_dir", stand_in) monkeypatch.setattr(assemble, "MODULES", ("a", "b")) @@ -113,8 +129,7 @@ def test_fragments_join_in_number_order_whatever_module_holds_them(tmp_path: Pat _package( tmp_path, monkeypatch, {"a.py": "import os\n# >>> gate.py 2\nB = 2\n", "b.py": "# >>> gate.py 1\nA = 1\n\n"} ) - budget = f"#: One hook invocation's time budget, in seconds (chock.gate.budget).\nENGINE_BUDGET_SECONDS = {ENGINE_BUDGET_SECONDS}" - assert runner_source() == PRELUDE.read_text(encoding="utf-8") + budget + "\n\n\nA = 1\n\n\nB = 2\n" + assert runner_source() == PRELUDE.read_text(encoding="utf-8") + budget_fragment() + "\n\n\nA = 1\n\n\nB = 2\n" ONE = "# >>> gate.py 1\nA = 1\n"