🛡️ Sentinel: [MEDIUM] Fix subprocess timeout DoS risk

scripts/build_benchmark_report.py

@@ -39,6 +39,7 @@ def _source_commit(repo_root: Path) -> str:
             capture_output=True,
             text=True,
             check=True,
+            timeout=60,
         )
     except Exception:
         return "unknown"

scripts/build_buyer_packet.py

@@ -65,6 +65,7 @@ def _source_commit(repo_root: Path) -> str:
             capture_output=True,
             text=True,
             check=True,
+            timeout=60,
         )
     except Exception:
         return "unknown"

scripts/build_commercial_release.py

@@ -39,6 +39,7 @@ def _source_commit(repo_root: Path) -> str:
             capture_output=True,
             text=True,
             check=True,
+            timeout=60,
         )
     except Exception:
         return "unknown"
@@ -62,7 +63,9 @@ def _content_security_policy() -> str:
 
 
 def _run_command(command: list[str], cwd: Path) -> subprocess.CompletedProcess[str]:
-    return subprocess.run(command, cwd=cwd, capture_output=True, text=True)
+    return subprocess.run(
+        command, cwd=cwd, capture_output=True, text=True, timeout=3600
+    )
 
 
 def _parse_last_json_line(stdout: str) -> dict[str, Any] | None:

scripts/build_figma_evidence_sync.py

@@ -71,6 +71,7 @@ def _source_commit(repo_root: Path) -> str:
             capture_output=True,
             text=True,
             check=True,
+            timeout=60,
         )
     except Exception:
         return "unknown"

scripts/build_pr_queue_governance.py

@@ -110,6 +110,7 @@ def _source_commit(repo_root: Path) -> str:
             capture_output=True,
             text=True,
             check=True,
+            timeout=60,
         )
     except Exception:
         return "unknown"
@@ -143,7 +144,7 @@ def _json_from_completed(completed: subprocess.CompletedProcess[str]) -> Any:
 
 def _run_gh_json(command: list[str]) -> tuple[Any, dict[str, Any] | None]:
     """Execute a GitHub CLI JSON command and return payload plus redacted error."""
-    completed = subprocess.run(command, capture_output=True, text=True)
+    completed = subprocess.run(command, capture_output=True, text=True, timeout=60)
     payload = _json_from_completed(completed)
     if completed.returncode == 0:
         return payload, None

scripts/build_procurement_due_diligence.py

@@ -174,6 +174,7 @@ def _source_commit(repo_root: Path) -> str:
             capture_output=True,
             text=True,
             check=True,
+            timeout=60,
         )
     except Exception:
         return "unknown"
@@ -298,7 +299,7 @@ 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)
+        completed = subprocess.run(command, capture_output=True, text=True, timeout=60)
         snapshot[name] = {
             "ok": completed.returncode == 0,
             "returncode": completed.returncode,
@@ -311,6 +312,7 @@ def _github_snapshot(repo: str, *, offline: bool) -> dict[str, Any]:
         ["gh", "release", "list", "--repo", repo, "--limit", "20"],
         capture_output=True,
         text=True,
+        timeout=60,
     )
     snapshot["releases"] = {
         "ok": release.returncode == 0,

scripts/build_release_evidence_index.py

@@ -52,6 +52,7 @@ def _source_commit(repo_root: Path) -> str:
             capture_output=True,
             text=True,
             check=True,
+            timeout=60,
         )
     except Exception:
         return "unknown"

scripts/release_acceptance.py

@@ -32,7 +32,9 @@ def _cli_env() -> dict[str, str]:
     return env
 
 
-def _run_cli(args: list[str], out_label: str, *, require_json: bool = True) -> dict[str, object]:
+def _run_cli(
+    args: list[str], out_label: str, *, require_json: bool = True
+) -> dict[str, object]:
     command = [sys.executable, "-m", "fast_mlsirm.cli", *args]
     if require_json and "--json" not in command:
         command.append("--json")
@@ -47,13 +49,20 @@ def _run_cli(args: list[str], out_label: str, *, require_json: bool = True) -> d
     if not raw_output:
         raise RuntimeError(f"{out_label} succeeded without JSON output")
     if not require_json:
-        return {"status": "ok", "command": out_label, "stdout": raw_output[-1], "duration_seconds": duration_seconds}
+        return {
+            "status": "ok",
+            "command": out_label,
+            "stdout": raw_output[-1],
+            "duration_seconds": duration_seconds,
+        }
     try:
         payload = json.loads(raw_output[-1])
         payload["duration_seconds"] = duration_seconds
         return payload
     except json.JSONDecodeError as exc:
-        raise RuntimeError(f"{out_label} produced non-JSON output: {raw_output[-1]}") from exc
+        raise RuntimeError(
+            f"{out_label} produced non-JSON output: {raw_output[-1]}"
+        ) from exc
 
 
 def _run_acceptance(args: argparse.Namespace) -> dict[str, object]:
@@ -121,7 +130,9 @@ def _run_acceptance(args: argparse.Namespace) -> dict[str, object]:
     )
     report["steps"].append(fit_auto_payload)
 
-    summary = json.loads((fit_auto_out / "fit_summary.json").read_text(encoding="utf-8"))
+    summary = json.loads(
+        (fit_auto_out / "fit_summary.json").read_text(encoding="utf-8")
+    )
     if summary.get("backend") not in {"numpy", "rust"}:
         raise RuntimeError("fit auto backend is not numpy or rust")
 
@@ -159,7 +170,9 @@ def _run_acceptance(args: argparse.Namespace) -> dict[str, object]:
             "fit_rust",
         )
         report["steps"].append(fit_rust_payload)
-        rust_summary = json.loads((fit_rust_out / "fit_summary.json").read_text(encoding="utf-8"))
+        rust_summary = json.loads(
+            (fit_rust_out / "fit_summary.json").read_text(encoding="utf-8")
+        )
         if rust_summary.get("backend") != "rust":
             raise RuntimeError("rust fit did not report rust backend")
 
@@ -257,22 +270,54 @@ def _run_acceptance(args: argparse.Namespace) -> dict[str, object]:
         "steps": report["steps"],
         "total_duration_seconds": round(time.perf_counter() - acceptance_started, 6),
     }
-    summary_path.write_text(json.dumps(summary_payload, indent=2, sort_keys=True), encoding="utf-8")
+    summary_path.write_text(
+        json.dumps(summary_payload, indent=2, sort_keys=True), encoding="utf-8"
+    )
     return {"status": "ok", "out": str(out_dir), "report": str(summary_path)}
 
 
 def build_parser() -> argparse.ArgumentParser:
-    parser = argparse.ArgumentParser(description="Run a fast-mlsirm release acceptance smoke test.")
-    parser.add_argument("--out", default="acceptance_check", help="Output directory for generated artifacts.")
-    parser.add_argument("--persons", type=int, default=12, help="Number of persons to simulate.")
-    parser.add_argument("--dims", type=int, default=1, help="Simulated true item dimensions.")
-    parser.add_argument("--items-per-dim", type=int, default=2, help="Items per dimension for simulation.")
-    parser.add_argument("--latent-dim", type=int, default=1, help="Latent dimension for fitting.")
+    parser = argparse.ArgumentParser(
+        description="Run a fast-mlsirm release acceptance smoke test."
+    )
+    parser.add_argument(
+        "--out",
+        default="acceptance_check",
+        help="Output directory for generated artifacts.",
+    )
+    parser.add_argument(
+        "--persons", type=int, default=12, help="Number of persons to simulate."
+    )
+    parser.add_argument(
+        "--dims", type=int, default=1, help="Simulated true item dimensions."
+    )
+    parser.add_argument(
+        "--items-per-dim",
+        type=int,
+        default=2,
+        help="Items per dimension for simulation.",
+    )
+    parser.add_argument(
+        "--latent-dim", type=int, default=1, help="Latent dimension for fitting."
+    )
     parser.add_argument("--seed", type=int, default=1, help="Simulation seed.")
-    parser.add_argument("--max-iter", type=int, default=1, help="Max optimization iterations for fitting.")
-    parser.add_argument("--n-restarts", type=int, default=1, help="Optimization restarts.")
-    parser.add_argument("--latent-dims", default="1,2", help="Comma-separated latent dims for dimension diagnostics.")
-    parser.add_argument("--folds", type=int, default=2, help="CV folds for dimensionality diagnostics.")
+    parser.add_argument(
+        "--max-iter",
+        type=int,
+        default=1,
+        help="Max optimization iterations for fitting.",
+    )
+    parser.add_argument(
+        "--n-restarts", type=int, default=1, help="Optimization restarts."
+    )
+    parser.add_argument(
+        "--latent-dims",
+        default="1,2",
+        help="Comma-separated latent dims for dimension diagnostics.",
+    )
+    parser.add_argument(
+        "--folds", type=int, default=2, help="CV folds for dimensionality diagnostics."
+    )
     parser.add_argument(
         "--require-rust",
         action="store_true",

scripts/run_ignored_rust_shard.py

@@ -217,6 +217,7 @@ def inventory_ignored_tests(targets: Sequence[CargoTarget]) -> list[IgnoredTest]
             check=True,
             capture_output=True,
             text=True,
+            timeout=1800,
         )
         for test_name in parse_test_names(completed.stdout):
             identifier = f"{target.identity}::{test_name}"
@@ -295,6 +296,7 @@ def run_shard(shard: int, shard_count: int, skipped: Sequence[str]) -> int:
         check=True,
         capture_output=True,
         text=True,
+        timeout=1800,
     )
     targets = parse_workspace_targets(metadata.stdout)
     inventory = inventory_ignored_tests(targets)
@@ -317,7 +319,7 @@ def run_shard(shard: int, shard_count: int, skipped: Sequence[str]) -> int:
         test = by_identifier[identifier]
         command = cargo_test_command(test.target, test.test_name)
         print("+ " + " ".join(command), flush=True)
-        completed = subprocess.run(command, check=False)
+        completed = subprocess.run(command, check=False, timeout=1800)
         if completed.returncode != 0:
             return completed.returncode
     return 0