From f16de95e123782eb40e7e7c56d25bab59bc2a249 Mon Sep 17 00:00:00 2001 From: Saagar Date: Sat, 4 Jul 2026 04:23:36 -0700 Subject: [PATCH] Add portfolio security drift gate --- docs/security-model.md | 11 ++ src/cli.py | 69 ++++++++++- src/portfolio_security_gate.py | 160 ++++++++++++++++++++++++++ tests/test_portfolio_security_gate.py | 121 +++++++++++++++++++ 4 files changed, 359 insertions(+), 2 deletions(-) create mode 100644 src/portfolio_security_gate.py create mode 100644 tests/test_portfolio_security_gate.py diff --git a/docs/security-model.md b/docs/security-model.md index 37195bd..36a5578 100644 --- a/docs/security-model.md +++ b/docs/security-model.md @@ -59,6 +59,17 @@ Failure behavior: a 403 or 404 on any endpoint records `available: false` for th Output lands in `output/ghas-alerts--.json`. Excel and control-center surfacing is wired via S2.4. +## Portfolio Security Gate + +Run `audit security-gate --output-dir output` after generating portfolio truth with +`--portfolio-truth-include-security`. The gate reads +`output/portfolio-truth-latest.json` and exits nonzero when any scanned repo has open +high/critical Dependabot alerts. + +The gate is deliberately strict: a snapshot with no scanned security overlay is +reported as `UNKNOWN`, not clear. This prevents a missing GHAS overlay from looking +like a clean portfolio. + ## OSSF Scorecard Pass `--ossf-scorecard` to enrich each repo with pre-computed OSSF Scorecard data. diff --git a/src/cli.py b/src/cli.py index 338fa16..feec1df 100644 --- a/src/cli.py +++ b/src/cli.py @@ -81,6 +81,7 @@ audit triage --approval-center audit report --portfolio-truth audit report --campaign security-review --writeback-target github + audit security-gate --output-dir output audit serve [--port 8080] Legacy flat form (deprecated, still supported): @@ -1527,6 +1528,30 @@ def _build_security_burndown_subparser(subparsers: argparse._SubParsersAction) - ) +def _build_security_gate_subparser(subparsers: argparse._SubParsersAction) -> None: # type: ignore[type-arg] + """Subcommand: `audit security-gate` — fail on portfolio high/critical drift.""" + p = subparsers.add_parser( + "security-gate", + help="Fail if portfolio truth has open high/critical Dependabot alerts", + description=( + "Read output/portfolio-truth-latest.json and fail if any scanned repo has\n" + "open high/critical Dependabot alerts. Missing security overlay data is\n" + "reported as unknown and exits nonzero." + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + p.add_argument( + "--output-dir", + default="output", + help="Directory containing portfolio-truth-latest.json (default: output/)", + ) + p.add_argument( + "--json", + action="store_true", + help="Print machine-readable JSON instead of Markdown", + ) + + def build_subcommand_parser() -> argparse.ArgumentParser: """Return the subcommand-aware parser used by main(). @@ -1556,6 +1581,7 @@ def build_subcommand_parser() -> argparse.ArgumentParser: _build_report_subparser(subparsers) _build_serve_subparser(subparsers) _build_security_burndown_subparser(subparsers) + _build_security_gate_subparser(subparsers) return parser @@ -6974,7 +7000,7 @@ def _infer_subcommand_from_flags(args: argparse.Namespace) -> str: _KNOWN_SUBCOMMANDS: frozenset[str] = frozenset( - {"run", "triage", "report", "serve", "security-burndown"} + {"run", "triage", "report", "serve", "security-burndown", "security-gate"} ) @@ -7139,6 +7165,41 @@ def _run_security_burndown_mode(args) -> None: print_info(f"Burndown JSON written to {json_path}") +def _run_security_gate_mode(args) -> None: + """Dispatch for `audit security-gate`.""" + from src.portfolio_security_gate import ( + build_security_gate_report, + render_security_gate_markdown, + ) + + truth_path = Path(args.output_dir) / TRUTH_LATEST_FILENAME + if not truth_path.exists(): + print_info( + f"{TRUTH_LATEST_FILENAME} not found in {truth_path.parent}. " + "Run `audit report --portfolio-truth --portfolio-truth-include-security` first." + ) + raise SystemExit(1) + + try: + with truth_path.open(encoding="utf-8") as fh: + portfolio_truth = json.load(fh) + except Exception as exc: # noqa: BLE001 + print_info(f"Could not read {truth_path}: {exc}") + raise SystemExit(1) + + if not isinstance(portfolio_truth, dict): + print_info(f"{truth_path} is not a portfolio-truth object.") + raise SystemExit(1) + + report = build_security_gate_report(portfolio_truth) + if getattr(args, "json", False): + print(json.dumps(report.to_dict(), indent=2)) + else: + print(render_security_gate_markdown(report)) + if not report.passed: + raise SystemExit(1) + + # ── Main entry point ────────────────────────────────────────────────── def main() -> None: raw_argv = sys.argv[1:] @@ -7153,11 +7214,15 @@ def main() -> None: subcommand_parser = build_subcommand_parser() legacy_parser = build_parser() - # ── Subcommand: security-burndown (own parser — no legacy equivalent) ── + # ── Subcommands with no legacy equivalent ─────────────────────────────── if argv and argv[0] == "security-burndown": sb_args = subcommand_parser.parse_args(argv) _run_security_burndown_mode(sb_args) return + if argv and argv[0] == "security-gate": + sg_args = subcommand_parser.parse_args(argv) + _run_security_gate_mode(sg_args) + return if argv and argv[0] in _KNOWN_SUBCOMMANDS: # Subcommand form — detect the subcommand with the subcommand parser, diff --git a/src/portfolio_security_gate.py b/src/portfolio_security_gate.py new file mode 100644 index 0000000..23598f8 --- /dev/null +++ b/src/portfolio_security_gate.py @@ -0,0 +1,160 @@ +"""Portfolio-level security drift gate. + +The gate reads the canonical portfolio-truth snapshot and answers one narrow +operator question: did any scanned repo regain open high/critical Dependabot +alerts? Missing security overlay data is treated as unknown, not healthy. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class SecurityGateItem: + repo: str + critical: int + high: int + risk_tier: str + + @property + def total(self) -> int: + return self.critical + self.high + + def to_dict(self) -> dict[str, Any]: + return { + "repo": self.repo, + "critical": self.critical, + "high": self.high, + "risk_tier": self.risk_tier, + } + + +@dataclass(frozen=True) +class SecurityGateReport: + generated_at: str + scanned_count: int + total_open_critical: int + total_open_high: int + flagged_repos: tuple[SecurityGateItem, ...] + + @property + def repos_with_open_high_critical(self) -> int: + return len(self.flagged_repos) + + @property + def passed(self) -> bool: + return self.scanned_count > 0 and self.repos_with_open_high_critical == 0 + + @property + def status(self) -> str: + if self.scanned_count <= 0: + return "unknown" + if self.repos_with_open_high_critical > 0: + return "fail" + return "pass" + + def to_dict(self) -> dict[str, Any]: + return { + "generated_at": self.generated_at, + "status": self.status, + "passed": self.passed, + "scanned_count": self.scanned_count, + "repos_with_open_high_critical": self.repos_with_open_high_critical, + "total_open_critical": self.total_open_critical, + "total_open_high": self.total_open_high, + "flagged_repos": [item.to_dict() for item in self.flagged_repos], + } + + +def _mapping(value: Any) -> dict[str, Any]: + return value if isinstance(value, dict) else {} + + +def _text(value: Any) -> str: + return value.strip() if isinstance(value, str) else "" + + +def _int(value: Any) -> int: + try: + return int(value or 0) + except (TypeError, ValueError): + return 0 + + +def build_security_gate_report(portfolio_truth: dict[str, Any]) -> SecurityGateReport: + projects = portfolio_truth.get("projects") or [] + scanned_count = 0 + total_critical = 0 + total_high = 0 + flagged: list[SecurityGateItem] = [] + + for project in projects: + if not isinstance(project, dict): + continue + security = _mapping(project.get("security")) + if not security.get("alerts_available"): + continue + + scanned_count += 1 + critical = _int(security.get("dependabot_critical")) + high = _int(security.get("dependabot_high")) + total_critical += critical + total_high += high + if critical <= 0 and high <= 0: + continue + + identity = _mapping(project.get("identity")) + risk = _mapping(project.get("risk")) + flagged.append( + SecurityGateItem( + repo=( + _text(identity.get("display_name")) + or _text(identity.get("repo_full_name")) + or _text(identity.get("path")) + or "Repo" + ), + critical=critical, + high=high, + risk_tier=_text(risk.get("risk_tier")) or "baseline", + ) + ) + + flagged.sort(key=lambda item: (-item.critical, -item.high, item.repo.lower())) + return SecurityGateReport( + generated_at=_text(portfolio_truth.get("generated_at")) or "unknown", + scanned_count=scanned_count, + total_open_critical=total_critical, + total_open_high=total_high, + flagged_repos=tuple(flagged), + ) + + +def render_security_gate_markdown(report: SecurityGateReport) -> str: + lines = [ + "# Portfolio Security Gate", + "", + ( + f"Status: {report.status.upper()} | scanned {report.scanned_count} | " + f"repos with open high/critical {report.repos_with_open_high_critical} | " + f"critical {report.total_open_critical} | high {report.total_open_high}" + ), + f"Source freshness: {report.generated_at}", + "", + ] + + if report.status == "unknown": + lines.append( + "Security overlay was not present in the snapshot. Re-run portfolio truth with " + "`--portfolio-truth-include-security` before treating the portfolio as clear." + ) + elif report.passed: + lines.append("All scanned repos are clear of open high/critical Dependabot alerts.") + else: + lines.append("| Repo | Risk | Critical | High |") + lines.append("|---|---:|---:|---:|") + for item in report.flagged_repos: + lines.append(f"| {item.repo} | {item.risk_tier} | {item.critical} | {item.high} |") + + return "\n".join(lines) diff --git a/tests/test_portfolio_security_gate.py b/tests/test_portfolio_security_gate.py new file mode 100644 index 0000000..95705ff --- /dev/null +++ b/tests/test_portfolio_security_gate.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +import json +from types import SimpleNamespace + +import pytest + +from src.cli import _run_security_gate_mode, build_subcommand_parser +from src.portfolio_security_gate import ( + build_security_gate_report, + render_security_gate_markdown, +) + + +def _project( + name: str, + *, + alerts_available: bool = True, + critical: int = 0, + high: int = 0, + risk_tier: str = "baseline", +) -> dict: + return { + "identity": {"display_name": name, "path": name}, + "risk": {"risk_tier": risk_tier}, + "security": { + "alerts_available": alerts_available, + "dependabot_critical": critical, + "dependabot_high": high, + }, + } + + +def test_security_gate_passes_when_scanned_repos_are_clear() -> None: + report = build_security_gate_report( + { + "generated_at": "2026-07-04T11:04:28+00:00", + "projects": [ + _project("RepoA"), + _project("RepoB", critical=0, high=0), + ], + } + ) + + assert report.passed is True + assert report.status == "pass" + assert report.scanned_count == 2 + assert report.repos_with_open_high_critical == 0 + assert "All scanned repos are clear" in render_security_gate_markdown(report) + + +def test_security_gate_fails_and_ranks_open_high_critical_repos() -> None: + report = build_security_gate_report( + { + "projects": [ + _project("HighOnly", high=2, risk_tier="moderate"), + _project("Critical", critical=1, risk_tier="elevated"), + _project("Clear"), + ], + } + ) + + assert report.passed is False + assert report.status == "fail" + assert report.scanned_count == 3 + assert report.total_open_critical == 1 + assert report.total_open_high == 2 + assert [item.repo for item in report.flagged_repos] == ["Critical", "HighOnly"] + rendered = render_security_gate_markdown(report) + assert "| Critical | elevated | 1 | 0 |" in rendered + assert "| HighOnly | moderate | 0 | 2 |" in rendered + + +def test_security_gate_treats_missing_overlay_as_unknown_not_pass() -> None: + report = build_security_gate_report( + { + "projects": [ + _project("Unscanned", alerts_available=False), + {"identity": {"display_name": "NoSecurityBlock"}}, + ] + } + ) + + assert report.passed is False + assert report.status == "unknown" + assert report.scanned_count == 0 + assert "Security overlay was not present" in render_security_gate_markdown(report) + + +def test_security_gate_subcommand_parses() -> None: + parser = build_subcommand_parser() + args = parser.parse_args(["security-gate", "--output-dir", "out", "--json"]) + + assert args._subcommand == "security-gate" + assert args.output_dir == "out" + assert args.json is True + + +def test_security_gate_cli_json_exits_zero_on_clear_snapshot(tmp_path, capsys) -> None: + (tmp_path / "portfolio-truth-latest.json").write_text( + json.dumps({"projects": [_project("Clear")]}), + encoding="utf-8", + ) + + _run_security_gate_mode(SimpleNamespace(output_dir=str(tmp_path), json=True)) + + payload = json.loads(capsys.readouterr().out) + assert payload["status"] == "pass" + assert payload["scanned_count"] == 1 + + +def test_security_gate_cli_exits_nonzero_on_open_alerts(tmp_path) -> None: + (tmp_path / "portfolio-truth-latest.json").write_text( + json.dumps({"projects": [_project("Open", high=1)]}), + encoding="utf-8", + ) + + with pytest.raises(SystemExit) as exc: + _run_security_gate_mode(SimpleNamespace(output_dir=str(tmp_path), json=False)) + + assert exc.value.code == 1