diff --git a/.chock/bin/antigravity.py b/.chock/bin/antigravity.py index e08fd43..a0d4f13 100755 --- a/.chock/bin/antigravity.py +++ b/.chock/bin/antigravity.py @@ -15,6 +15,7 @@ import os as _chock_os import shlex as _chock_shlex +import shutil as _chock_shutil import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path @@ -608,7 +609,7 @@ def find_bash(guard: _chock_Path) -> str | None: """First interpreter that can actually see `guard`, or None.""" for candidate in _BASH_CANDIDATES: try: - proc = _chock_subprocess.run([candidate, '-c', f'test -f "{guard.as_posix()}"'], capture_output=True, timeout=10) + proc = _chock_subprocess.run([candidate, '-c', f'test -f "{guard.as_posix()}"'], capture_output=True, timeout=10, check=False) except (OSError, _chock_subprocess.SubprocessError): continue if proc.returncode == 0: @@ -630,7 +631,7 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS) + proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED @@ -649,7 +650,7 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_ERRORED return GUARD_CLEAN -def log_outcome(guard: _chock_Path, tool: str, blocked: bool) -> None: +def log_outcome(guard: _chock_Path, tool: str, *, blocked: bool) -> None: """Append one outcome record. Best effort: never raises, never changes the verdict.""" try: if _chock_os.environ.get(GATE_LOG_ENV) == '0': @@ -669,7 +670,6 @@ def log_outcome(guard: _chock_Path, tool: str, blocked: bool) -> None: log_path = log_dir / 'gate-events.jsonl' if log_path.exists() and log_path.stat().st_size > _LOG_MAX_BYTES: log_path.replace(log_dir / 'gate-events.1.jsonl') - import json record = {'ts': _chock_datetime.now(_chock_timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), 'policy_id': guard.parent.parent.name, 'surface': 'pre-tool-use', 'event': 'tool_use', 'kind': guard.stem, 'tool': tool, 'verdict': 'block' if blocked else 'allow'} with log_path.open('a', encoding='utf-8') as fh: fh.write(json.dumps(record, ensure_ascii=False) + '\n') @@ -683,7 +683,7 @@ def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | N return None verdict = run_guard(guard, command) if verdict in (GUARD_BLOCKED, GUARD_CLEAN): - log_outcome(guard, tool, verdict == GUARD_BLOCKED) + log_outcome(guard, tool, blocked=verdict == GUARD_BLOCKED) if verdict == GUARD_BLOCKED: return (VERDICT_DENY, f'Blocked by chock policy: {guard.stem}') if verdict == GUARD_ERRORED: diff --git a/.chock/bin/claude_code.py b/.chock/bin/claude_code.py index 5211c41..3e94a8b 100755 --- a/.chock/bin/claude_code.py +++ b/.chock/bin/claude_code.py @@ -15,6 +15,7 @@ import os as _chock_os import shlex as _chock_shlex +import shutil as _chock_shutil import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path @@ -663,7 +664,7 @@ def find_bash(guard: _chock_Path) -> str | None: """First interpreter that can actually see `guard`, or None.""" for candidate in _BASH_CANDIDATES: try: - proc = _chock_subprocess.run([candidate, '-c', f'test -f "{guard.as_posix()}"'], capture_output=True, timeout=10) + proc = _chock_subprocess.run([candidate, '-c', f'test -f "{guard.as_posix()}"'], capture_output=True, timeout=10, check=False) except (OSError, _chock_subprocess.SubprocessError): continue if proc.returncode == 0: @@ -685,7 +686,7 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS) + proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED @@ -704,7 +705,7 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_ERRORED return GUARD_CLEAN -def log_outcome(guard: _chock_Path, tool: str, blocked: bool) -> None: +def log_outcome(guard: _chock_Path, tool: str, *, blocked: bool) -> None: """Append one outcome record. Best effort: never raises, never changes the verdict.""" try: if _chock_os.environ.get(GATE_LOG_ENV) == '0': @@ -724,7 +725,6 @@ def log_outcome(guard: _chock_Path, tool: str, blocked: bool) -> None: log_path = log_dir / 'gate-events.jsonl' if log_path.exists() and log_path.stat().st_size > _LOG_MAX_BYTES: log_path.replace(log_dir / 'gate-events.1.jsonl') - import json record = {'ts': _chock_datetime.now(_chock_timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), 'policy_id': guard.parent.parent.name, 'surface': 'pre-tool-use', 'event': 'tool_use', 'kind': guard.stem, 'tool': tool, 'verdict': 'block' if blocked else 'allow'} with log_path.open('a', encoding='utf-8') as fh: fh.write(json.dumps(record, ensure_ascii=False) + '\n') @@ -738,13 +738,15 @@ def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | N return None verdict = run_guard(guard, command) if verdict in (GUARD_BLOCKED, GUARD_CLEAN): - log_outcome(guard, tool, verdict == GUARD_BLOCKED) + log_outcome(guard, tool, blocked=verdict == GUARD_BLOCKED) if verdict == GUARD_BLOCKED: return (VERDICT_DENY, f'Blocked by chock policy: {guard.stem}') if verdict == GUARD_ERRORED: return (VERDICT_ESCALATE, f"chock could not check this command: the {guard.stem} guard did not complete (see this hook's stderr). Approving runs it unchecked.") return None +_GIT = _chock_shutil.which('git') or 'git' + _INSTRUCTION = "Chock: this clone's git hooks are NOT installed -- git never clones hooks, so commit-time gates will not run locally until someone runs:\n pip install chock && chock sync --repo .\nRun that before the first commit. (The repo's CI gate, where wired, enforces regardless.)" def _repo_root() -> _chock_Path: @@ -754,7 +756,7 @@ def _repo_root() -> _chock_Path: def _hooks_pre_commit(repo_root: _chock_Path) -> _chock_Path | None: """The active pre-commit hook path, honouring core.hooksPath. None when git is absent.""" try: - proc = _chock_subprocess.run(['git', 'rev-parse', '--git-path', 'hooks'], cwd=repo_root, capture_output=True, text=True, timeout=15) + proc = _chock_subprocess.run([_GIT, 'rev-parse', '--git-path', 'hooks'], cwd=repo_root, capture_output=True, text=True, timeout=15, check=False) except (OSError, _chock_subprocess.TimeoutExpired): return None if proc.returncode != 0: diff --git a/.chock/bin/codex_cli.py b/.chock/bin/codex_cli.py index 3b5a068..3d67533 100755 --- a/.chock/bin/codex_cli.py +++ b/.chock/bin/codex_cli.py @@ -15,6 +15,7 @@ import os as _chock_os import shlex as _chock_shlex +import shutil as _chock_shutil import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path @@ -650,7 +651,7 @@ def find_bash(guard: _chock_Path) -> str | None: """First interpreter that can actually see `guard`, or None.""" for candidate in _BASH_CANDIDATES: try: - proc = _chock_subprocess.run([candidate, '-c', f'test -f "{guard.as_posix()}"'], capture_output=True, timeout=10) + proc = _chock_subprocess.run([candidate, '-c', f'test -f "{guard.as_posix()}"'], capture_output=True, timeout=10, check=False) except (OSError, _chock_subprocess.SubprocessError): continue if proc.returncode == 0: @@ -672,7 +673,7 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS) + proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED @@ -691,7 +692,7 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_ERRORED return GUARD_CLEAN -def log_outcome(guard: _chock_Path, tool: str, blocked: bool) -> None: +def log_outcome(guard: _chock_Path, tool: str, *, blocked: bool) -> None: """Append one outcome record. Best effort: never raises, never changes the verdict.""" try: if _chock_os.environ.get(GATE_LOG_ENV) == '0': @@ -711,7 +712,6 @@ def log_outcome(guard: _chock_Path, tool: str, blocked: bool) -> None: log_path = log_dir / 'gate-events.jsonl' if log_path.exists() and log_path.stat().st_size > _LOG_MAX_BYTES: log_path.replace(log_dir / 'gate-events.1.jsonl') - import json record = {'ts': _chock_datetime.now(_chock_timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), 'policy_id': guard.parent.parent.name, 'surface': 'pre-tool-use', 'event': 'tool_use', 'kind': guard.stem, 'tool': tool, 'verdict': 'block' if blocked else 'allow'} with log_path.open('a', encoding='utf-8') as fh: fh.write(json.dumps(record, ensure_ascii=False) + '\n') @@ -725,7 +725,7 @@ def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | N return None verdict = run_guard(guard, command) if verdict in (GUARD_BLOCKED, GUARD_CLEAN): - log_outcome(guard, tool, verdict == GUARD_BLOCKED) + log_outcome(guard, tool, blocked=verdict == GUARD_BLOCKED) if verdict == GUARD_BLOCKED: return (VERDICT_DENY, f'Blocked by chock policy: {guard.stem}') if verdict == GUARD_ERRORED: diff --git a/.chock/bin/cursor.py b/.chock/bin/cursor.py index c1a44ed..767e653 100755 --- a/.chock/bin/cursor.py +++ b/.chock/bin/cursor.py @@ -15,6 +15,7 @@ import os as _chock_os import shlex as _chock_shlex +import shutil as _chock_shutil import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path @@ -573,7 +574,7 @@ def find_bash(guard: _chock_Path) -> str | None: """First interpreter that can actually see `guard`, or None.""" for candidate in _BASH_CANDIDATES: try: - proc = _chock_subprocess.run([candidate, '-c', f'test -f "{guard.as_posix()}"'], capture_output=True, timeout=10) + proc = _chock_subprocess.run([candidate, '-c', f'test -f "{guard.as_posix()}"'], capture_output=True, timeout=10, check=False) except (OSError, _chock_subprocess.SubprocessError): continue if proc.returncode == 0: @@ -595,7 +596,7 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS) + proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED @@ -614,7 +615,7 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_ERRORED return GUARD_CLEAN -def log_outcome(guard: _chock_Path, tool: str, blocked: bool) -> None: +def log_outcome(guard: _chock_Path, tool: str, *, blocked: bool) -> None: """Append one outcome record. Best effort: never raises, never changes the verdict.""" try: if _chock_os.environ.get(GATE_LOG_ENV) == '0': @@ -634,7 +635,6 @@ def log_outcome(guard: _chock_Path, tool: str, blocked: bool) -> None: log_path = log_dir / 'gate-events.jsonl' if log_path.exists() and log_path.stat().st_size > _LOG_MAX_BYTES: log_path.replace(log_dir / 'gate-events.1.jsonl') - import json record = {'ts': _chock_datetime.now(_chock_timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), 'policy_id': guard.parent.parent.name, 'surface': 'pre-tool-use', 'event': 'tool_use', 'kind': guard.stem, 'tool': tool, 'verdict': 'block' if blocked else 'allow'} with log_path.open('a', encoding='utf-8') as fh: fh.write(json.dumps(record, ensure_ascii=False) + '\n') @@ -648,7 +648,7 @@ def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | N return None verdict = run_guard(guard, command) if verdict in (GUARD_BLOCKED, GUARD_CLEAN): - log_outcome(guard, tool, verdict == GUARD_BLOCKED) + log_outcome(guard, tool, blocked=verdict == GUARD_BLOCKED) if verdict == GUARD_BLOCKED: return (VERDICT_DENY, f'Blocked by chock policy: {guard.stem}') if verdict == GUARD_ERRORED: diff --git a/.chock/bin/devin.py b/.chock/bin/devin.py index 02cee60..24a7ffd 100755 --- a/.chock/bin/devin.py +++ b/.chock/bin/devin.py @@ -15,6 +15,7 @@ import os as _chock_os import shlex as _chock_shlex +import shutil as _chock_shutil import subprocess as _chock_subprocess from datetime import datetime as _chock_datetime, timezone as _chock_timezone from pathlib import Path as _chock_Path @@ -634,7 +635,7 @@ def find_bash(guard: _chock_Path) -> str | None: """First interpreter that can actually see `guard`, or None.""" for candidate in _BASH_CANDIDATES: try: - proc = _chock_subprocess.run([candidate, '-c', f'test -f "{guard.as_posix()}"'], capture_output=True, timeout=10) + proc = _chock_subprocess.run([candidate, '-c', f'test -f "{guard.as_posix()}"'], capture_output=True, timeout=10, check=False) except (OSError, _chock_subprocess.SubprocessError): continue if proc.returncode == 0: @@ -656,7 +657,7 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS) + proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED @@ -675,7 +676,7 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_ERRORED return GUARD_CLEAN -def log_outcome(guard: _chock_Path, tool: str, blocked: bool) -> None: +def log_outcome(guard: _chock_Path, tool: str, *, blocked: bool) -> None: """Append one outcome record. Best effort: never raises, never changes the verdict.""" try: if _chock_os.environ.get(GATE_LOG_ENV) == '0': @@ -695,7 +696,6 @@ def log_outcome(guard: _chock_Path, tool: str, blocked: bool) -> None: log_path = log_dir / 'gate-events.jsonl' if log_path.exists() and log_path.stat().st_size > _LOG_MAX_BYTES: log_path.replace(log_dir / 'gate-events.1.jsonl') - import json record = {'ts': _chock_datetime.now(_chock_timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), 'policy_id': guard.parent.parent.name, 'surface': 'pre-tool-use', 'event': 'tool_use', 'kind': guard.stem, 'tool': tool, 'verdict': 'block' if blocked else 'allow'} with log_path.open('a', encoding='utf-8') as fh: fh.write(json.dumps(record, ensure_ascii=False) + '\n') @@ -709,7 +709,7 @@ def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | N return None verdict = run_guard(guard, command) if verdict in (GUARD_BLOCKED, GUARD_CLEAN): - log_outcome(guard, tool, verdict == GUARD_BLOCKED) + log_outcome(guard, tool, blocked=verdict == GUARD_BLOCKED) if verdict == GUARD_BLOCKED: return (VERDICT_DENY, f'Blocked by chock policy: {guard.stem}') if verdict == GUARD_ERRORED: diff --git a/.chock/bin/gate.py b/.chock/bin/gate.py index 32073d4..95b7ecd 100755 --- a/.chock/bin/gate.py +++ b/.chock/bin/gate.py @@ -8,6 +8,7 @@ import json import os import re +import shutil import subprocess import sys import tomllib @@ -15,6 +16,8 @@ from datetime import datetime, timezone from pathlib import Path +_GIT = shutil.which("git") or "git" + @dataclass class GateResult: @@ -44,8 +47,8 @@ def _range(self) -> list[str]: def _git(self, *args: str) -> str: try: - proc = subprocess.run( - ["git", "-c", "core.quotePath=false", *args], + proc = subprocess.run( # noqa: S603 -- reading repo facts via git is this class's whole job + [_GIT, "-c", "core.quotePath=false", *args], cwd=str(self.repo_root), capture_output=True, text=True, @@ -53,9 +56,10 @@ def _git(self, *args: str) -> str: errors="replace", check=True, ) - return proc.stdout or "" except (subprocess.CalledProcessError, FileNotFoundError, UnicodeError): return "" + else: + return proc.stdout or "" def rev_exists(self, ref: str) -> bool: """True when `ref` resolves to a commit. Used to fail CI closed on a missing base.""" @@ -91,14 +95,15 @@ def push_refs(self) -> list[str]: refs: list[str] = [] for line in self._push_stdin.splitlines(): parts = line.split() - if len(parts) >= 3: + if len(parts) >= _PUSH_LINE_MIN_PARTS: refs.append(parts[2]) return refs -def _kind_content_regex(ctx: GateContext, params: dict, event: str) -> GateResult: +def _kind_content_regex(ctx: GateContext, params: dict, _event: str) -> GateResult: content_re = re.compile(params["content_pattern"]) - path_re = re.compile(params["forbidden_path_regex"]) if params.get("forbidden_path_regex") else None + forbidden_path_regex = params.get("forbidden_path_regex") + path_re = re.compile(forbidden_path_regex) if forbidden_path_regex else None pragma_re = re.compile(params["allowlist_pragma"]) if params.get("allowlist_pragma") else None scan = params.get("scan", "added_lines") diff_filter = params.get("diff_filter", "ACMRT") @@ -140,7 +145,7 @@ def _deps_requirements(text: str) -> set[str]: names: set[str] = set() for line in text.splitlines(): s = line.strip() - if not s or s.startswith("#") or s.startswith("-"): + if not s or s.startswith(("#", "-")): continue m = _REQ_RE.match(line) if m: @@ -206,11 +211,11 @@ def _extract(path: str, text: str) -> set[str]: return set() try: return fn(text) - except Exception: + except Exception: # noqa: BLE001 -- untrusted, possibly-malformed manifest content; never crash the gate on it return set() -def _kind_dependency_allowlist(ctx: GateContext, params: dict, event: str) -> GateResult: +def _kind_dependency_allowlist(ctx: GateContext, params: dict, _event: str) -> GateResult: watched = set(params.get("manifests", [])) allow: set[str] = set() allow_path = ctx.repo_root / params["allowlist_file"] @@ -241,6 +246,15 @@ def _kind_dependency_allowlist(ctx: GateContext, params: dict, event: str) -> Ga _LOG_MAX_BYTES = 1_048_576 _LOG_MATCH_CAP = 20 +#: A pre-push stdin line is ` `; +#: at least 3 whitespace-separated parts to reach the remote ref at index 2. +_PUSH_LINE_MIN_PARTS = 3 + +#: `/.chock/compiled//git-hook/