fix(ops): replay bounded subprocess integrity on current review workflow

.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.

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.

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,
+    )

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()