diff --git a/.jules/sentinel.md b/.jules/sentinel.md index 5c671f1a5..07d1a24ae 100644 --- a/.jules/sentinel.md +++ b/.jules/sentinel.md @@ -30,3 +30,23 @@ 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. + +## 2026-08-21 - [Bounded Capture Pipe and Process-Tree Cleanup] +**Vulnerability:** Repository automation can deadlock or retain descendants when +stdout/stderr pipes are inherited by a child process after the direct command +exits. Unbounded diagnostics can also exhaust memory or hide the original +fail-closed error when cleanup signalling fails. + +**Learning:** A subprocess boundary needs independent byte limits, one absolute +deadline, concurrent pipe draining, strict machine-output decoding, and cleanup +that reaps the owned child without assuming signal delivery always succeeds. + +**Prevention:** Keep stdout and stderr bounded, terminate the POSIX process +group when a reader proves a descendant owns a capture pipe, bounded-reap the +direct child, catch cleanup `OSError`, and preserve stable timeout/overflow/data +errors for governance and procurement evidence. diff --git a/docs/changelog.d/1015-bounded-subprocess-integrity.md b/docs/changelog.d/1015-bounded-subprocess-integrity.md new file mode 100644 index 000000000..cc34eca11 --- /dev/null +++ b/docs/changelog.d/1015-bounded-subprocess-integrity.md @@ -0,0 +1,5 @@ +# Harden bounded subprocess cleanup + +## Fixed + +- Keep governance and procurement subprocess capture bounded across stdout, stderr, execution time, decoding, and JSON parsing. POSIX cleanup now avoids re-signalling an already reaped process group, successful capture closes parent-side pipe descriptors deterministically, and timeout/overflow paths retain fail-closed evidence without weakening repository gates. diff --git a/scripts/_bounded_subprocess.py b/scripts/_bounded_subprocess.py new file mode 100644 index 000000000..ebcd02684 --- /dev/null +++ b/scripts/_bounded_subprocess.py @@ -0,0 +1,248 @@ +"""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 +_PROCESS_REAP_TIMEOUT_SECONDS = 5.0 + + +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. + + ``BufferedReader.read`` may wait for the requested chunk size even when a + smaller payload is already available. ``read1`` observes the pipe promptly, + which keeps overflow detection independent from a descendant holding the + write end open. + """ + if hasattr(stream, "read1"): + read = stream.read1 # type: ignore[attr-defined] + else: + read = stream.read # type: ignore[attr-defined] + 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], + *, + terminate_descendants: bool = False, +) -> None: + """Terminate and bounded-reap the owned process tree. + + A live process group is terminated after the direct child exits only when + a capture reader still proves that a descendant owns the pipe. The default + keeps repeated cleanup from signalling a reaped process group. + """ + if process.poll() is None or (terminate_descendants and os.name == "posix"): + if os.name == "posix": + try: + os.killpg(process.pid, signal.SIGKILL) + except OSError: + pass + else: + try: + process.kill() + except OSError: + pass + try: + process.wait(timeout=_PROCESS_REAP_TIMEOUT_SECONDS) + except subprocess.TimeoutExpired: + # The caller still has a hard deadline; leave pipe cleanup to the + # existing close path if a hostile child ignores the bounded reap. + pass + + +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", + ) + if process.stdout is None or process.stderr is None: + process.kill() + process.wait() + raise RuntimeError("bounded capture requires stdout and stderr pipes") + + 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(): + overflowed = overflowed or stdout_overflow.is_set() or stderr_overflow.is_set() + if not overflowed: + timed_out = True + _terminate_process_tree(process, terminate_descendants=True) + _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) + + _close_capture_pipes(process) + 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_bounded_subprocess_pipe_cleanup.py b/tests/test_bounded_subprocess_pipe_cleanup.py new file mode 100644 index 000000000..6abc790bc --- /dev/null +++ b/tests/test_bounded_subprocess_pipe_cleanup.py @@ -0,0 +1,37 @@ +"""Regression coverage for successful bounded subprocess pipe cleanup.""" + +from __future__ import annotations + +import sys + +import pytest + +from scripts import _bounded_subprocess as bounded + + +def test_successful_bounded_capture_closes_parent_pipes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Normal completion must close the parent's stdout and stderr descriptors.""" + closed_processes: list[object] = [] + real_close = bounded._close_capture_pipes + + def record_close(process: object) -> None: + closed_processes.append(process) + real_close(process) # type: ignore[arg-type] + + monkeypatch.setattr(bounded, "_close_capture_pipes", record_close) + + completed = bounded.run_bounded_capture( + [sys.executable, "-c", "print('ok')"], + timeout_seconds=5, + max_stdout_bytes=1024, + max_stderr_bytes=1024, + ) + + assert completed.returncode == 0 + assert completed.stdout.strip() == "ok" + assert len(closed_processes) == 1 + process = closed_processes[0] + assert process.stdout.closed # type: ignore[attr-defined] + assert process.stderr.closed # type: ignore[attr-defined] 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..db0fcb603 --- /dev/null +++ b/tests/test_subprocess_output_bounds.py @@ -0,0 +1,382 @@ +"""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_missing_capture_pipes(monkeypatch: pytest.MonkeyPatch) -> None: + """Pipe setup failures remain explicit when Python assertions are optimized away.""" + class MissingPipes: + stdout = None + stderr = None + + def kill(self) -> None: + """Record the required cleanup operation.""" + + def wait(self) -> None: + """Record the required reap operation.""" + + monkeypatch.setattr( + "scripts._bounded_subprocess.subprocess.Popen", + lambda *args, **kwargs: MissingPipes(), + ) + + from scripts._bounded_subprocess import run_bounded_capture + + with pytest.raises(RuntimeError, match="requires stdout and stderr pipes"): + run_bounded_capture( + [sys.executable, "-c", ""], + 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 + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX process-group ownership contract") +def test_bounded_capture_reports_descendant_overflow_before_timeout() -> None: + """A descendant-held pipe must preserve overflow precedence over timeout.""" + from scripts._bounded_subprocess import run_bounded_capture + + grandchild = "import sys, time; sys.stdout.write('x' * 4096); sys.stdout.flush(); time.sleep(2)" + child = ( + "import subprocess, sys; " + f"subprocess.Popen([sys.executable, '-c', {grandchild!r}]); " + "sys.exit(0)" + ) + with pytest.raises(BoundedSubprocessOutputError, match="stdout"): + run_bounded_capture( + [sys.executable, "-c", child], + timeout_seconds=1.0, + max_stdout_bytes=64, + max_stderr_bytes=1024, + ) + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX process-group ownership contract") +def test_process_tree_termination_reaps_the_owned_child(monkeypatch: pytest.MonkeyPatch) -> None: + """The kill path must reap the direct child within its bounded cleanup window.""" + from scripts import _bounded_subprocess as bounded + + class FakeProcess: + pid = 4242 + + def __init__(self) -> None: + self.wait_timeouts: list[float | None] = [] + + def poll(self) -> None: + return None + + def wait(self, timeout: float | None = None) -> None: + self.wait_timeouts.append(timeout) + + process = FakeProcess() + monkeypatch.setattr(bounded.os, "killpg", lambda *_args: None) + + bounded._terminate_process_tree(process) # type: ignore[arg-type] + + assert process.wait_timeouts == [bounded._PROCESS_REAP_TIMEOUT_SECONDS] + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX process-group ownership contract") +def test_process_tree_termination_does_not_resignal_reaped_child( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Repeated cleanup must never signal a process group after the child is reaped.""" + from scripts import _bounded_subprocess as bounded + + class ReapedProcess: + pid = 4242 + + def __init__(self) -> None: + self.wait_timeouts: list[float | None] = [] + + def poll(self) -> int: + return 0 + + def wait(self, timeout: float | None = None) -> None: + self.wait_timeouts.append(timeout) + + process = ReapedProcess() + + def fail_if_signalled(*_args: object) -> None: + raise AssertionError("reaped process group must not be signalled") + + monkeypatch.setattr(bounded.os, "killpg", fail_if_signalled) + + bounded._terminate_process_tree(process) # type: ignore[arg-type] + + assert process.wait_timeouts == [bounded._PROCESS_REAP_TIMEOUT_SECONDS] + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX process-group ownership contract") +def test_process_tree_cleanup_ignores_signal_permission_errors( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Cleanup signal failures must not replace the bounded subprocess error.""" + from scripts import _bounded_subprocess as bounded + + class LiveProcess: + pid = 4242 + + def __init__(self) -> None: + self.wait_timeouts: list[float | None] = [] + + def poll(self) -> None: + return None + + def wait(self, timeout: float | None = None) -> None: + self.wait_timeouts.append(timeout) + + process = LiveProcess() + + def deny_signal(*_args: object) -> None: + raise PermissionError("signal denied") + + monkeypatch.setattr(bounded.os, "killpg", deny_signal) + bounded._terminate_process_tree(process) # type: ignore[arg-type] + + assert process.wait_timeouts == [bounded._PROCESS_REAP_TIMEOUT_SECONDS] + + +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"] + + +def test_procurement_lines_snapshot_success_is_structured(monkeypatch: pytest.MonkeyPatch) -> None: + """Successful text output becomes non-empty, whitespace-trimmed evidence lines.""" + monkeypatch.setattr( + procurement, + "run_bounded_capture", + lambda *args, **kwargs: subprocess.CompletedProcess( + args[0], 0, "v1\n\n v2 \n", " warning \n" + ), + ) + + snapshot = procurement._bounded_lines_snapshot(["gh", "release", "list"]) + + assert snapshot == { + "ok": True, + "returncode": 0, + "lines": ["v1", " v2 "], + "stderr": "warning", + } + + +def test_procurement_lines_snapshot_timeout_is_fail_closed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Timed-out text commands produce the stable empty-lines error schema.""" + monkeypatch.setattr( + procurement, + "run_bounded_capture", + lambda *args, **kwargs: (_ for _ in ()).throw( + subprocess.TimeoutExpired(args[0], 1) + ), + ) + + snapshot = procurement._bounded_lines_snapshot(["gh", "release", "list"]) + + assert snapshot["ok"] is False + assert snapshot["returncode"] == 124 + assert snapshot["lines"] == [] + assert "timed out" in snapshot["stderr"]