diff --git a/.jules/sentinel.md b/.jules/sentinel.md index 5c671f1a5..51d6bbe5b 100644 --- a/.jules/sentinel.md +++ b/.jules/sentinel.md @@ -30,3 +30,8 @@ Explicitly defining `allow_pickle=False` is a robust defense-in-depth practice. **Vulnerability:** The functions `parse_generated_item_candidate` and `_contract_object` used `json.loads` directly on string payloads before strictly enforcing depth limits over the string itself. A maliciously nested JSON string (e.g. `{"a": {"a": ...}}`) could exceed the Python maximum recursion limit, crashing the process with a `RecursionError` and causing a Denial of Service (DoS) attack, because Python's built-in `json.loads` recurses natively while decoding. **Learning:** Checking for JSON nested depth after decoding using `json.loads` (or implicitly relying on string size constraints) is insufficient to prevent recursion crashes on deep but compact objects. Depth checking must happen by scanning the raw string stream prior to any decoding engine invocations. **Prevention:** Always implement a character-level depth limit scanner (`_validate_raw_json_depth`) and enforce it on raw strings before passing them to `json.loads`. + +## 2026-08-18 - [Prevent subprocess hang DoS] +**Vulnerability:** External `subprocess.run` calls without timeouts can hang indefinitely during GitHub CLI network or provider failures, stalling repository automation. +**Learning:** Command duration is a separate resource bound from JSON size/depth. A bounded parser cannot terminate a child process that never returns. +**Prevention:** Supply an explicit timeout for external repository-automation subprocesses and convert `subprocess.TimeoutExpired` into stable fail-closed evidence rather than hanging indefinitely. diff --git a/scripts/_bounded_subprocess.py b/scripts/_bounded_subprocess.py new file mode 100644 index 000000000..f846e7bf0 --- /dev/null +++ b/scripts/_bounded_subprocess.py @@ -0,0 +1,219 @@ +"""Bound subprocess stdout/stderr in memory while preserving a hard deadline.""" + +from __future__ import annotations + +import os +import signal +import subprocess +import threading +import time +from collections.abc import Mapping, Sequence +from pathlib import Path + +_READ_CHUNK_BYTES = 64 * 1024 +_DATA_ERROR_RETURN_CODE = 65 + + +class BoundedSubprocessOutputError(RuntimeError): + """Raised when a captured subprocess stream exceeds its configured limit.""" + + def __init__(self, stream: str, limit_bytes: int) -> None: + self.stream = stream + self.limit_bytes = limit_bytes + super().__init__(f"{stream} exceeded bounded capture limit of {limit_bytes} bytes") + + +class BoundedSubprocessDecodeError(UnicodeError): + """Describe machine-readable subprocess stdout that is not valid UTF-8.""" + + def __init__(self, stream: str) -> None: + self.stream = stream + super().__init__(f"{stream} was not valid UTF-8") + + +def _drain_bounded( + stream: object, + *, + limit_bytes: int, + buffer: bytearray, + overflow: threading.Event, +) -> None: + """Drain one binary pipe without retaining more than ``limit_bytes + 1`` bytes.""" + read = getattr(stream, "read") + while True: + try: + chunk = read(_READ_CHUNK_BYTES) + except (OSError, ValueError): + return + if not chunk: + return + remaining = (limit_bytes + 1) - len(buffer) + if remaining > 0: + buffer.extend(chunk[:remaining]) + if len(buffer) > limit_bytes: + overflow.set() + + +def _terminate_process_tree(process: subprocess.Popen[bytes]) -> None: + """Terminate the owned process tree without signalling the caller process.""" + if os.name == "posix": + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + return + return + if process.poll() is None: + try: + process.kill() + except ProcessLookupError: + return + + +def _close_capture_pipes(process: subprocess.Popen[bytes]) -> None: + """Close parent-side capture pipes to unblock any remaining daemon reader.""" + for stream in (process.stdout, process.stderr): + if stream is not None: + try: + stream.close() + except (OSError, ValueError): + pass + + +def _remaining(deadline: float) -> float: + """Return non-negative seconds remaining before one absolute deadline.""" + return max(0.0, deadline - time.monotonic()) + + +def run_bounded_capture( + command: Sequence[str], + *, + timeout_seconds: float, + max_stdout_bytes: int, + max_stderr_bytes: int, + cwd: str | Path | None = None, + env: Mapping[str, str] | None = None, +) -> subprocess.CompletedProcess[str]: + """Run ``command`` with one hard deadline and bounded output capture. + + Stdout and stderr are drained concurrently so neither pipe can deadlock the + child. POSIX commands run in a dedicated session so timeout/overflow cleanup + can terminate descendants that inherited a capture pipe. Process reaping + and reader joins share the original deadline rather than extending it. + Machine-readable stdout is decoded strictly as UTF-8. Malformed stdout + becomes a stable data-error result rather than replacement-decoded content; + diagnostic stderr alone uses replacement decoding. + """ + if timeout_seconds <= 0: + raise ValueError("timeout_seconds must be positive") + if max_stdout_bytes < 0 or max_stderr_bytes < 0: + raise ValueError("output limits must be non-negative") + if not command: + raise ValueError("command must not be empty") + + process = subprocess.Popen( + list(command), + cwd=cwd, + env=dict(env) if env is not None else None, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=False, + start_new_session=os.name == "posix", + ) + assert process.stdout is not None + assert process.stderr is not None + + stdout = bytearray() + stderr = bytearray() + stdout_overflow = threading.Event() + stderr_overflow = threading.Event() + readers = [ + threading.Thread( + target=_drain_bounded, + kwargs={ + "stream": process.stdout, + "limit_bytes": max_stdout_bytes, + "buffer": stdout, + "overflow": stdout_overflow, + }, + daemon=True, + ), + threading.Thread( + target=_drain_bounded, + kwargs={ + "stream": process.stderr, + "limit_bytes": max_stderr_bytes, + "buffer": stderr, + "overflow": stderr_overflow, + }, + daemon=True, + ), + ] + for reader in readers: + reader.start() + + deadline = time.monotonic() + timeout_seconds + timed_out = False + overflowed = False + while process.poll() is None: + if stdout_overflow.is_set() or stderr_overflow.is_set(): + overflowed = True + _terminate_process_tree(process) + break + remaining = _remaining(deadline) + if remaining <= 0.0: + timed_out = True + _terminate_process_tree(process) + break + time.sleep(min(0.01, remaining)) + + if process.poll() is None: + try: + process.wait(timeout=_remaining(deadline)) + except subprocess.TimeoutExpired: + timed_out = True + _terminate_process_tree(process) + + for reader in readers: + reader.join(timeout=_remaining(deadline)) + if reader.is_alive(): + if not overflowed: + timed_out = True + _terminate_process_tree(process) + _close_capture_pipes(process) + break + + if timed_out: + _terminate_process_tree(process) + _close_capture_pipes(process) + raise subprocess.TimeoutExpired(list(command), timeout_seconds) + if stdout_overflow.is_set(): + _terminate_process_tree(process) + _close_capture_pipes(process) + raise BoundedSubprocessOutputError("stdout", max_stdout_bytes) + if stderr_overflow.is_set(): + _terminate_process_tree(process) + _close_capture_pipes(process) + raise BoundedSubprocessOutputError("stderr", max_stderr_bytes) + + stderr_text = stderr.decode("utf-8", errors="replace") + try: + stdout_text = stdout.decode("utf-8", errors="strict") + except UnicodeDecodeError: + decode_error = BoundedSubprocessDecodeError("stdout") + diagnostic = stderr_text.strip() + if diagnostic: + diagnostic = f"{diagnostic}\n{decode_error}" + else: + diagnostic = str(decode_error) + return subprocess.CompletedProcess( + list(command), + _DATA_ERROR_RETURN_CODE, + "", + diagnostic, + ) + return subprocess.CompletedProcess( + list(command), + process.returncode, + stdout_text, + stderr_text, + ) diff --git a/scripts/build_pr_queue_governance.py b/scripts/build_pr_queue_governance.py index 65b8b3b9e..94f198d62 100644 --- a/scripts/build_pr_queue_governance.py +++ b/scripts/build_pr_queue_governance.py @@ -17,9 +17,11 @@ from urllib.parse import urlparse try: - from scripts._bounded_json import read_json_object + from scripts._bounded_json import MAX_JSON_BYTES, parse_json_bounded, read_json_object + from scripts._bounded_subprocess import BoundedSubprocessOutputError, run_bounded_capture except ModuleNotFoundError: - from _bounded_json import read_json_object + from _bounded_json import MAX_JSON_BYTES, parse_json_bounded, read_json_object + from _bounded_subprocess import BoundedSubprocessOutputError, run_bounded_capture RISK_COUNT_KEYS = [ @@ -86,6 +88,9 @@ _GH_TRANSIENT_STATUS_RE = re.compile(r"\bHTTP (?:502|503|504)\b", re.IGNORECASE) _GH_JSON_MAX_ATTEMPTS = 3 _GH_JSON_RETRY_SLEEP_SECONDS = 0.5 +_GH_COMMAND_TIMEOUT_SECONDS = 60 +_GH_STDOUT_MAX_BYTES = MAX_JSON_BYTES +_GH_STDERR_MAX_BYTES = 1024 * 1024 GIT_METADATA_TIMEOUT_SECONDS = 5 @@ -159,10 +164,10 @@ def _check( def _json_from_completed(completed: subprocess.CompletedProcess[str]) -> Any: - """Decode command stdout when the command succeeded and emitted JSON.""" + """Decode bounded command stdout when the command succeeded and emitted JSON.""" if completed.returncode != 0 or not completed.stdout.strip(): return None - return json.loads(completed.stdout) + return parse_json_bounded(completed.stdout, max_bytes=_GH_STDOUT_MAX_BYTES) def _is_transient_gh_stderr(stderr: str) -> bool: @@ -178,14 +183,43 @@ def _run_gh_json( ) -> tuple[Any, dict[str, Any] | None]: """Execute a GitHub CLI JSON command and return payload plus redacted error. - Retries only on HTTP 502/503/504. Non-transient failures fail closed on the - first response so real auth/query defects are not masked. + Retries only on HTTP 502/503/504. Non-transient, bounded-output, and JSON + decoding failures fail closed on the first response so real defects are not + masked and untrusted command output cannot grow without bound in memory. """ attempts = max(1, int(max_attempts)) last_error: dict[str, Any] | None = None for attempt in range(1, attempts + 1): - completed = subprocess.run(command, capture_output=True, text=True) - payload = _json_from_completed(completed) + try: + completed = run_bounded_capture( + command, + timeout_seconds=_GH_COMMAND_TIMEOUT_SECONDS, + max_stdout_bytes=_GH_STDOUT_MAX_BYTES, + max_stderr_bytes=_GH_STDERR_MAX_BYTES, + ) + except subprocess.TimeoutExpired: + last_error = { + "command": command[1:3], + "stderr": f"command timed out after {_GH_COMMAND_TIMEOUT_SECONDS} seconds", + "returncode": 124, + } + break + except BoundedSubprocessOutputError as exc: + last_error = { + "command": command[1:3], + "stderr": str(exc), + "returncode": 75, + } + break + try: + payload = _json_from_completed(completed) + except ValueError as exc: + last_error = { + "command": command[1:3], + "stderr": str(exc), + "returncode": 65, + } + break if completed.returncode == 0: return payload, None stderr = completed.stderr.strip() diff --git a/scripts/build_procurement_due_diligence.py b/scripts/build_procurement_due_diligence.py index 0caa408e9..87cc378f3 100644 --- a/scripts/build_procurement_due_diligence.py +++ b/scripts/build_procurement_due_diligence.py @@ -16,11 +16,17 @@ from typing import Any GIT_METADATA_TIMEOUT_SECONDS = 5 +_GH_COMMAND_TIMEOUT_SECONDS = 60 try: - from scripts._bounded_json import read_json_object + from scripts._bounded_json import MAX_JSON_BYTES, parse_json_bounded, read_json_object + from scripts._bounded_subprocess import BoundedSubprocessOutputError, run_bounded_capture except ModuleNotFoundError: - from _bounded_json import read_json_object + from _bounded_json import MAX_JSON_BYTES, parse_json_bounded, read_json_object + from _bounded_subprocess import BoundedSubprocessOutputError, run_bounded_capture + +_GH_STDOUT_MAX_BYTES = MAX_JSON_BYTES +_GH_STDERR_MAX_BYTES = 1024 * 1024 POLICY_FILES = [ @@ -276,6 +282,90 @@ def _commercial_checks( return payload, checks +def _bounded_json_snapshot(command: list[str]) -> dict[str, Any]: + """Run one GitHub JSON command with bounded output and stable failures.""" + try: + completed = run_bounded_capture( + command, + timeout_seconds=_GH_COMMAND_TIMEOUT_SECONDS, + max_stdout_bytes=_GH_STDOUT_MAX_BYTES, + max_stderr_bytes=_GH_STDERR_MAX_BYTES, + ) + except subprocess.TimeoutExpired: + return { + "ok": False, + "returncode": 124, + "data": None, + "stderr": f"command timed out after {_GH_COMMAND_TIMEOUT_SECONDS} seconds", + } + except BoundedSubprocessOutputError as exc: + return { + "ok": False, + "returncode": 75, + "data": None, + "stderr": str(exc), + } + if completed.returncode != 0: + return { + "ok": False, + "returncode": completed.returncode, + "data": None, + "stderr": completed.stderr.strip(), + } + try: + data = ( + parse_json_bounded(completed.stdout, max_bytes=_GH_STDOUT_MAX_BYTES) + if completed.stdout.strip() + else None + ) + except ValueError as exc: + return { + "ok": False, + "returncode": 65, + "data": None, + "stderr": str(exc), + } + return { + "ok": True, + "returncode": 0, + "data": data, + "stderr": completed.stderr.strip(), + } + + +def _bounded_lines_snapshot(command: list[str]) -> dict[str, Any]: + """Run one GitHub text command with bounded output and stable failures.""" + try: + completed = run_bounded_capture( + command, + timeout_seconds=_GH_COMMAND_TIMEOUT_SECONDS, + max_stdout_bytes=_GH_STDOUT_MAX_BYTES, + max_stderr_bytes=_GH_STDERR_MAX_BYTES, + ) + except subprocess.TimeoutExpired: + return { + "ok": False, + "returncode": 124, + "lines": [], + "stderr": f"command timed out after {_GH_COMMAND_TIMEOUT_SECONDS} seconds", + } + except BoundedSubprocessOutputError as exc: + return { + "ok": False, + "returncode": 75, + "lines": [], + "stderr": str(exc), + } + return { + "ok": completed.returncode == 0, + "returncode": completed.returncode, + "lines": [line for line in completed.stdout.splitlines() if line.strip()] + if completed.returncode == 0 + else [], + "stderr": completed.stderr.strip(), + } + + def _github_snapshot(repo: str, *, offline: bool) -> dict[str, Any]: if offline: return {"mode": "offline", "repo": repo, "checks": {"snapshot_recorded": True}} @@ -304,26 +394,10 @@ def _github_snapshot(repo: str, *, offline: bool) -> dict[str, Any]: ], } for name, command in commands.items(): - completed = subprocess.run(command, capture_output=True, text=True) - snapshot[name] = { - "ok": completed.returncode == 0, - "returncode": completed.returncode, - "data": json.loads(completed.stdout) - if completed.returncode == 0 and completed.stdout.strip() - else None, - "stderr": completed.stderr.strip(), - } - release = subprocess.run( - ["gh", "release", "list", "--repo", repo, "--limit", "20"], - capture_output=True, - text=True, + snapshot[name] = _bounded_json_snapshot(command) + snapshot["releases"] = _bounded_lines_snapshot( + ["gh", "release", "list", "--repo", repo, "--limit", "20"] ) - snapshot["releases"] = { - "ok": release.returncode == 0, - "returncode": release.returncode, - "lines": [line for line in release.stdout.splitlines() if line.strip()], - "stderr": release.stderr.strip(), - } return snapshot diff --git a/tests/test_pr_queue_governance.py b/tests/test_pr_queue_governance.py index 51d8ee9c9..1d79d8b0b 100644 --- a/tests/test_pr_queue_governance.py +++ b/tests/test_pr_queue_governance.py @@ -381,7 +381,7 @@ def test_run_gh_snapshot_records_base_sha_history_and_errors(monkeypatch): subprocess.CompletedProcess([], 0, json.dumps({"sha": "c" * 40}), ""), ] ) - monkeypatch.setattr(module.subprocess, "run", lambda *args, **kwargs: next(responses)) + monkeypatch.setattr(module, "run_bounded_capture", lambda *args, **kwargs: next(responses)) snapshot = module._run_gh_snapshot("ContextualWisdomLab/fast-mlsirm") @@ -401,7 +401,7 @@ def test_run_gh_snapshot_fails_closed_on_command_errors(monkeypatch): subprocess.CompletedProcess([], 1, "", "history failed"), ] ) - monkeypatch.setattr(module.subprocess, "run", lambda *args, **kwargs: next(responses)) + monkeypatch.setattr(module, "run_bounded_capture", lambda *args, **kwargs: next(responses)) snapshot = module._run_gh_snapshot("ContextualWisdomLab/fast-mlsirm") @@ -462,8 +462,8 @@ def test_run_gh_json_retries_only_transient_gateway_statuses(monkeypatch): ] ) monkeypatch.setattr( - module.subprocess, - "run", + module, + "run_bounded_capture", lambda *args, **kwargs: next(transient), ) payload, error = module._run_gh_json( @@ -482,8 +482,8 @@ def test_run_gh_json_retries_only_transient_gateway_statuses(monkeypatch): ] ) monkeypatch.setattr( - module.subprocess, - "run", + module, + "run_bounded_capture", lambda *args, **kwargs: next(permanent), ) payload, error = module._run_gh_json( @@ -506,7 +506,7 @@ def test_run_gh_snapshot_uses_light_history_fields_and_recovers_from_502( recorded: list[list[str]] = [] history_attempts = {"n": 0} - def fake_run(command, capture_output=True, text=True): + def fake_run(command, **kwargs): recorded.append(list(command)) if command[1:3] == ["repo", "view"]: return subprocess.CompletedProcess( @@ -557,7 +557,7 @@ def fake_run(command, capture_output=True, text=True): ) raise AssertionError(f"unexpected command: {command}") - monkeypatch.setattr(module.subprocess, "run", fake_run) + monkeypatch.setattr(module, "run_bounded_capture", fake_run) monkeypatch.setattr(module.time, "sleep", lambda seconds: None) snapshot = module._run_gh_snapshot("ContextualWisdomLab/fast-mlsirm") @@ -583,7 +583,7 @@ def test_run_gh_snapshot_records_exhausted_history_502_without_dropping_open_prs """When history stays 502 after retries, open PRs remain and the error is kept.""" module = _load_governance() - def fake_run(command, capture_output=True, text=True): + def fake_run(command, **kwargs): if command[1:3] == ["repo", "view"]: return subprocess.CompletedProcess( command, @@ -615,7 +615,7 @@ def fake_run(command, capture_output=True, text=True): ) raise AssertionError(command) - monkeypatch.setattr(module.subprocess, "run", fake_run) + monkeypatch.setattr(module, "run_bounded_capture", fake_run) monkeypatch.setattr(module.time, "sleep", lambda seconds: None) snapshot = module._run_gh_snapshot("ContextualWisdomLab/fast-mlsirm") @@ -763,8 +763,8 @@ def test_run_gh_snapshot_handles_missing_branch_and_nonobject_base(monkeypatch): ] ) monkeypatch.setattr( - module.subprocess, - "run", + module, + "run_bounded_capture", lambda *args, **kwargs: next(no_branch_responses), ) snapshot = module._run_gh_snapshot("ContextualWisdomLab/fast-mlsirm") @@ -785,8 +785,8 @@ def test_run_gh_snapshot_handles_missing_branch_and_nonobject_base(monkeypatch): ] ) monkeypatch.setattr( - module.subprocess, - "run", + module, + "run_bounded_capture", lambda *args, **kwargs: next(base_responses), ) snapshot = module._run_gh_snapshot("ContextualWisdomLab/fast-mlsirm") diff --git a/tests/test_pr_queue_governance_review_contract.py b/tests/test_pr_queue_governance_review_contract.py index 1f7c9e5bb..97c9924b5 100644 --- a/tests/test_pr_queue_governance_review_contract.py +++ b/tests/test_pr_queue_governance_review_contract.py @@ -60,7 +60,11 @@ def test_run_gh_json_recovers_from_each_approved_transient_status( ), ] ) - monkeypatch.setattr(module.subprocess, "run", lambda *args, **kwargs: next(responses)) + monkeypatch.setattr( + module, + "run_bounded_capture", + lambda *args, **kwargs: next(responses), + ) monkeypatch.setattr(module.time, "sleep", lambda seconds: sleeps.append(seconds)) payload, error = module._run_gh_json( @@ -94,7 +98,7 @@ def fail_transiently(*args, **kwargs): f"HTTP {status}: transient gateway failure", ) - monkeypatch.setattr(module.subprocess, "run", fail_transiently) + monkeypatch.setattr(module, "run_bounded_capture", fail_transiently) monkeypatch.setattr(module.time, "sleep", lambda seconds: sleeps.append(seconds)) payload, error = module._run_gh_json( diff --git a/tests/test_pr_queue_governance_timeout.py b/tests/test_pr_queue_governance_timeout.py new file mode 100644 index 000000000..1dadf2746 --- /dev/null +++ b/tests/test_pr_queue_governance_timeout.py @@ -0,0 +1,51 @@ +"""Regression coverage for bounded GitHub CLI command duration.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path +import subprocess + + +def _load_governance(): + """Load the governance script without requiring package installation.""" + script = Path(__file__).resolve().parents[1] / "scripts" / "build_pr_queue_governance.py" + spec = importlib.util.spec_from_file_location("build_pr_queue_governance_timeout", script) + assert spec is not None + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +def test_run_gh_json_timeout_fails_closed_without_retrying(monkeypatch) -> None: + """A hung GitHub CLI call becomes stable error evidence after one attempt.""" + module = _load_governance() + observed: list[tuple[list[str], float | None]] = [] + + def fake_run( + command, + *, + timeout_seconds, + max_stdout_bytes, + max_stderr_bytes, + **kwargs, + ): + observed.append((list(command), timeout_seconds)) + raise subprocess.TimeoutExpired(command, timeout_seconds) + + monkeypatch.setattr(module, "run_bounded_capture", fake_run) + + payload, error = module._run_gh_json( + ["gh", "api", "/rate_limit"], + max_attempts=3, + retry_sleep_seconds=0.0, + ) + + assert payload is None + assert error == { + "command": ["api", "/rate_limit"], + "stderr": "command timed out after 60 seconds", + "returncode": 124, + } + assert observed == [(["gh", "api", "/rate_limit"], 60)] diff --git a/tests/test_subprocess_output_bounds.py b/tests/test_subprocess_output_bounds.py new file mode 100644 index 000000000..56b0ec830 --- /dev/null +++ b/tests/test_subprocess_output_bounds.py @@ -0,0 +1,208 @@ +"""Regression coverage for bounded subprocess capture in operator scripts.""" + +from __future__ import annotations + +import os +import subprocess +import sys +import time + +import pytest + +from scripts import build_pr_queue_governance as governance +from scripts import build_procurement_due_diligence as procurement +from scripts._bounded_subprocess import BoundedSubprocessOutputError + + +def test_bounded_capture_rejects_stdout_overflow() -> None: + """The shared runner must cap stdout while the child is still running.""" + from scripts._bounded_subprocess import run_bounded_capture + + with pytest.raises(BoundedSubprocessOutputError, match="stdout"): + run_bounded_capture( + [sys.executable, "-c", "import sys; sys.stdout.write('x' * 4096)"], + timeout_seconds=5, + max_stdout_bytes=64, + max_stderr_bytes=64, + ) + + +def test_bounded_capture_rejects_stderr_overflow() -> None: + """The shared runner must cap stderr independently from stdout.""" + from scripts._bounded_subprocess import run_bounded_capture + + with pytest.raises(BoundedSubprocessOutputError, match="stderr"): + run_bounded_capture( + [sys.executable, "-c", "import sys; sys.stderr.write('x' * 4096)"], + timeout_seconds=5, + max_stdout_bytes=64, + max_stderr_bytes=64, + ) + + +def test_bounded_capture_invalid_utf8_is_data_error() -> None: + """Machine-readable stdout must never be silently replacement-decoded.""" + from scripts._bounded_subprocess import run_bounded_capture + + command = [ + sys.executable, + "-c", + "import sys; sys.stdout.buffer.write(b'{\"record\":\"ok' + bytes([255]) + b'\"}')", + ] + completed = run_bounded_capture( + command, + timeout_seconds=5, + max_stdout_bytes=1024, + max_stderr_bytes=1024, + ) + assert completed.returncode == 65 + assert completed.stdout == "" + assert "not valid UTF-8" in completed.stderr + assert "�" not in completed.stderr + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX process-group ownership contract") +def test_bounded_capture_deadline_kills_pipe_inheriting_descendants() -> None: + """A descendant holding captured pipes cannot extend the configured deadline.""" + from scripts._bounded_subprocess import run_bounded_capture + + grandchild = "import time; time.sleep(2)" + child = ( + "import subprocess, sys; " + f"subprocess.Popen([sys.executable, '-c', {grandchild!r}]); " + "sys.exit(0)" + ) + started = time.monotonic() + with pytest.raises(subprocess.TimeoutExpired): + run_bounded_capture( + [sys.executable, "-c", child], + timeout_seconds=0.2, + max_stdout_bytes=1024, + max_stderr_bytes=1024, + ) + assert time.monotonic() - started < 1.0 + + +def test_governance_parse_failure_is_stable_error(monkeypatch: pytest.MonkeyPatch) -> None: + """Malformed successful gh output must not crash the governance builder.""" + monkeypatch.setattr( + governance, + "run_bounded_capture", + lambda *args, **kwargs: subprocess.CompletedProcess(args[0], 0, "{", ""), + ) + + payload, error = governance._run_gh_json( + ["gh", "api", "repos/example/project"], + max_attempts=1, + retry_sleep_seconds=0, + ) + + assert payload is None + assert error is not None + assert error["returncode"] == 65 + assert "JSON" in error["stderr"] + + +def test_governance_decode_failure_is_stable_error(monkeypatch: pytest.MonkeyPatch) -> None: + """Invalid UTF-8 from gh must retain the helper's data-error status.""" + monkeypatch.setattr( + governance, + "run_bounded_capture", + lambda *args, **kwargs: subprocess.CompletedProcess( + args[0], 65, "", "stdout was not valid UTF-8" + ), + ) + payload, error = governance._run_gh_json( + ["gh", "api", "repos/example/project"], + max_attempts=1, + retry_sleep_seconds=0, + ) + assert payload is None + assert error is not None + assert error["returncode"] == 65 + assert "UTF-8" in error["stderr"] + + +def test_governance_timeout_remains_fail_closed(monkeypatch: pytest.MonkeyPatch) -> None: + """The bounded-output successor must preserve the parent timeout contract.""" + def raise_timeout(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: + raise subprocess.TimeoutExpired(command, 1) + + monkeypatch.setattr(governance, "run_bounded_capture", raise_timeout) + + payload, error = governance._run_gh_json( + ["gh", "api", "repos/example/project"], + max_attempts=1, + retry_sleep_seconds=0, + ) + + assert payload is None + assert error is not None + assert error["returncode"] == 124 + assert "timed out" in error["stderr"] + + +def test_governance_overflow_is_stable_error(monkeypatch: pytest.MonkeyPatch) -> None: + """Oversized gh output must fail closed through the governance error schema.""" + def raise_overflow(*args: object, **kwargs: object) -> subprocess.CompletedProcess[str]: + raise BoundedSubprocessOutputError("stdout", 64) + + monkeypatch.setattr(governance, "run_bounded_capture", raise_overflow) + + payload, error = governance._run_gh_json( + ["gh", "api", "repos/example/project"], + max_attempts=1, + retry_sleep_seconds=0, + ) + + assert payload is None + assert error is not None + assert error["returncode"] == 75 + assert "stdout" in error["stderr"] + + +def test_procurement_parse_failure_is_snapshot_error(monkeypatch: pytest.MonkeyPatch) -> None: + """Malformed gh JSON must be recorded as evidence failure, not raised.""" + monkeypatch.setattr( + procurement, + "run_bounded_capture", + lambda *args, **kwargs: subprocess.CompletedProcess(args[0], 0, "{", ""), + ) + + snapshot = procurement._github_snapshot("example/project", offline=False) + + assert snapshot["repo"]["ok"] is False + assert snapshot["repo"]["returncode"] == 65 + assert snapshot["repo"]["data"] is None + assert "JSON" in snapshot["repo"]["stderr"] + + +def test_procurement_decode_failure_is_snapshot_error(monkeypatch: pytest.MonkeyPatch) -> None: + """Invalid UTF-8 must remain a stable procurement evidence failure.""" + monkeypatch.setattr( + procurement, + "run_bounded_capture", + lambda *args, **kwargs: subprocess.CompletedProcess( + args[0], 65, "", "stdout was not valid UTF-8" + ), + ) + snapshot = procurement._github_snapshot("example/project", offline=False) + assert snapshot["repo"]["ok"] is False + assert snapshot["repo"]["returncode"] == 65 + assert snapshot["repo"]["data"] is None + assert "UTF-8" in snapshot["repo"]["stderr"] + + +def test_procurement_overflow_is_snapshot_error(monkeypatch: pytest.MonkeyPatch) -> None: + """Oversized gh output must be recorded in procurement evidence, not raised.""" + def raise_overflow(*args: object, **kwargs: object) -> subprocess.CompletedProcess[str]: + raise BoundedSubprocessOutputError("stdout", 64) + + monkeypatch.setattr(procurement, "run_bounded_capture", raise_overflow) + + snapshot = procurement._github_snapshot("example/project", offline=False) + + assert snapshot["repo"]["ok"] is False + assert snapshot["repo"]["returncode"] == 75 + assert snapshot["repo"]["data"] is None + assert "stdout" in snapshot["repo"]["stderr"]