🛡️ Sentinel: [MEDIUM] subprocess 호출에 대한 타임아웃 추가를 통한 DoS 취약점 해결

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

scripts/build_pr_queue_governance.py

@@ -86,6 +86,7 @@
 _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
 GIT_METADATA_TIMEOUT_SECONDS = 5
 
 
@@ -184,7 +185,17 @@ def _run_gh_json(
     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)
+        try:
+            completed = subprocess.run(
+                command, capture_output=True, text=True, timeout=_GH_COMMAND_TIMEOUT_SECONDS
+            )
+        except subprocess.TimeoutExpired:
+            last_error = {
+                "command": command[1:3],
+                "stderr": f"command timed out after {_GH_COMMAND_TIMEOUT_SECONDS} seconds",
+                "returncode": 124,
+            }
+            break
         payload = _json_from_completed(completed)
         if completed.returncode == 0:
             return payload, None

scripts/build_procurement_due_diligence.py

@@ -16,6 +16,7 @@
 from typing import Any
 
 GIT_METADATA_TIMEOUT_SECONDS = 5
+_GH_COMMAND_TIMEOUT_SECONDS = 60
 
 try:
     from scripts._bounded_json import read_json_object
@@ -304,26 +305,43 @@ 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(),
+        try:
+            completed = subprocess.run(command, capture_output=True, text=True, timeout=_GH_COMMAND_TIMEOUT_SECONDS)
+            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(),
+            }
+        except subprocess.TimeoutExpired:
+            snapshot[name] = {
+                "ok": False,
+                "returncode": 124,
+                "data": None,
+                "stderr": f"command timed out after {_GH_COMMAND_TIMEOUT_SECONDS} seconds",
+            }
+    try:
+        release = subprocess.run(
+            ["gh", "release", "list", "--repo", repo, "--limit", "20"],
+            capture_output=True,
+            text=True,
+            timeout=_GH_COMMAND_TIMEOUT_SECONDS,
+        )
+        snapshot["releases"] = {
+            "ok": release.returncode == 0,
+            "returncode": release.returncode,
+            "lines": [line for line in release.stdout.splitlines() if line.strip()],
+            "stderr": release.stderr.strip(),
+        }
+    except subprocess.TimeoutExpired:
+        snapshot["releases"] = {
+            "ok": False,
+            "returncode": 124,
+            "lines": [],
+            "stderr": f"command timed out after {_GH_COMMAND_TIMEOUT_SECONDS} seconds",
         }
-    release = subprocess.run(
-        ["gh", "release", "list", "--repo", repo, "--limit", "20"],
-        capture_output=True,
-        text=True,
-    )
-    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
 
 

tests/test_pr_queue_governance.py

@@ -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, capture_output=True, text=True, timeout=None):
         recorded.append(list(command))
         if command[1:3] == ["repo", "view"]:
             return subprocess.CompletedProcess(
@@ -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, capture_output=True, text=True, timeout=None):
         if command[1:3] == ["repo", "view"]:
             return subprocess.CompletedProcess(
                 command,

tests/test_pr_queue_governance_timeout.py

@@ -0,0 +1,44 @@
+"""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, *, capture_output=True, text=True, timeout=None):
+        observed.append((list(command), timeout))
+        raise subprocess.TimeoutExpired(command, timeout)
+
+    monkeypatch.setattr(module.subprocess, "run", 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)]