From d09ae66e1dca3581dc1bf50b32e3adf0c0292bcd Mon Sep 17 00:00:00 2001 From: saagpatel Date: Thu, 16 Jul 2026 20:32:11 -0700 Subject: [PATCH 1/6] feat: add receipt-backed security coverage --- src/app/portfolio_truth.py | 26 +- src/cli.py | 41 +- src/github_security_coverage.py | 943 +++++++++++++++++++++++++ src/portfolio_security_gate.py | 121 +++- src/portfolio_truth_publish.py | 2 + src/portfolio_truth_reconcile.py | 179 ++++- src/portfolio_truth_status.py | 36 + src/portfolio_truth_types.py | 105 ++- tests/test_cli_subcommands.py | 4 +- tests/test_github_security_coverage.py | 341 +++++++++ tests/test_portfolio_security_gate.py | 65 +- tests/test_portfolio_truth.py | 107 ++- 12 files changed, 1885 insertions(+), 85 deletions(-) create mode 100644 src/github_security_coverage.py create mode 100644 tests/test_github_security_coverage.py diff --git a/src/app/portfolio_truth.py b/src/app/portfolio_truth.py index 9f8ad050..ab3414ff 100644 --- a/src/app/portfolio_truth.py +++ b/src/app/portfolio_truth.py @@ -17,7 +17,7 @@ load_live_repo_status_by_name, load_release_count_by_name, load_repo_status_from_audit_by_name, - load_security_alerts_by_name, + load_security_coverage_by_full_name, ) from src.producer_preflight import load_producer_evidence @@ -53,11 +53,30 @@ def run_portfolio_truth_mode(args: Any) -> None: username=args.username, ) security_alerts_by_name: dict[str, dict] | None = None + security_coverage_metadata: dict[str, object] | None = None if getattr(args, "portfolio_truth_include_security", False): - security_alerts_by_name = load_security_alerts_by_name( + receipt_path_value = getattr(args, "portfolio_truth_security_receipt", None) + loaded_security = load_security_coverage_by_full_name( output_dir=output_dir, - username=args.username, + receipt_path=Path(receipt_path_value) if receipt_path_value else None, + max_age_hours=getattr( + args, "portfolio_truth_security_max_age_hours", 24 + ), ) + if loaded_security is not None: + security_alerts_by_name = loaded_security.entries_by_full_name + security_coverage_metadata = { + "source_id": "github-security-coverage-receipt", + "schema_version": loaded_security.schema_version, + "produced_at": loaded_security.produced_at, + "state": loaded_security.receipt_state, + "age_hours": loaded_security.age_hours, + "cohort_policy": loaded_security.cohort_policy, + "cohort_repository_count": len( + loaded_security.cohort_repositories + ), + "path": loaded_security.source_path, + } repo_status_by_name = load_live_repo_status_by_name( username=args.username, token=getattr(args, "token", None), @@ -80,6 +99,7 @@ def run_portfolio_truth_mode(args: Any) -> None: allow_empty_notion=getattr(args, "portfolio_truth_allow_empty_notion", False), release_count_by_name=release_count_by_name, security_alerts_by_name=security_alerts_by_name, + security_coverage_metadata=security_coverage_metadata, repo_status_by_name=repo_status_by_name, producer_evidence=producer_evidence, producer_repo_root=producer_repo_root, diff --git a/src/cli.py b/src/cli.py index 2c00a285..20e944a0 100644 --- a/src/cli.py +++ b/src/cli.py @@ -437,6 +437,28 @@ def _build_report_subparser(subparsers: argparse._SubParsersAction) -> None: # p.add_argument( "--portfolio-truth", action="store_true", help="Generate canonical portfolio truth snapshot" ) + p.add_argument( + "--portfolio-truth-include-security", + action="store_true", + help=( + "Overlay security coverage from the validated " + "github-security-coverage-latest.json receipt" + ), + ) + p.add_argument( + "--portfolio-truth-security-receipt", + type=str, + default=None, + metavar="PATH", + help="Explicit provenance-bearing GitHub security receipt path", + ) + p.add_argument( + "--portfolio-truth-security-max-age-hours", + type=int, + default=24, + metavar="HOURS", + help="Freshness window for GitHub security observations (default: 24)", + ) p.add_argument( "--portfolio-truth-allow-empty-notion", action="store_true", @@ -683,11 +705,24 @@ def build_parser() -> argparse.ArgumentParser: "--portfolio-truth-include-security", action="store_true", help=( - "Overlay the security.* GHAS alert counts on each project from the latest " - "output/ghas-alerts--*.json file, feeding the active-high-severity-alerts " - "risk factor (requires a prior `audit report --ghas-alerts` run)" + "Overlay security coverage from the validated " + "github-security-coverage-latest.json receipt" ), ) + parser.add_argument( + "--portfolio-truth-security-receipt", + type=str, + default=None, + metavar="PATH", + help="Explicit provenance-bearing GitHub security receipt path", + ) + parser.add_argument( + "--portfolio-truth-security-max-age-hours", + type=int, + default=24, + metavar="HOURS", + help="Freshness window for GitHub security observations (default: 24)", + ) parser.add_argument( "--portfolio-truth-allow-empty-notion", action="store_true", diff --git a/src/github_security_coverage.py b/src/github_security_coverage.py new file mode 100644 index 00000000..3e0b214b --- /dev/null +++ b/src/github_security_coverage.py @@ -0,0 +1,943 @@ +"""Bounded, receipt-backed GitHub security coverage collection. + +The receipt is intentionally count-only. It records enough provenance to make +PortfolioTruth coverage claims auditable without persisting credentials, raw +alerts, or secret-scanning payloads. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import subprocess +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import requests + +GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION = "GitHubSecurityCoverageReceiptV1" +GITHUB_SECURITY_RECEIPT_FILENAME = "github-security-coverage-latest.json" +GITHUB_API_VERSION = "2026-03-10" +DEFAULT_COHORT_POLICY = "portfolio-default-attention-v1" +DEFAULT_ATTENTION_STATES = frozenset( + {"active-product", "active-infra", "decision-needed"} +) +PROVIDER_NAMES = ("dependabot", "code_scanning", "secret_scanning") +PROVIDER_STATES = frozenset( + { + "observed", + "not_requested", + "credential_unavailable", + "forbidden", + "feature_unavailable", + "not_found", + "gone", + "rate_limited", + "transient_error", + "malformed", + "stale", + } +) +DEFAULT_BASE_REQUEST_LIMIT = 48 +DEFAULT_TOTAL_REQUEST_LIMIT = 75 +DEFAULT_QUOTA_RESERVE = 100 +_COMMIT_RE = re.compile(r"^[0-9a-f]{40}$") +_REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") + +_ENDPOINTS = { + "dependabot": "dependabot/alerts", + "code_scanning": "code-scanning/alerts", + "secret_scanning": "secret-scanning/alerts", +} +_COUNT_KEYS = { + "dependabot": ("critical", "high", "medium", "low"), + "code_scanning": ("critical", "high", "warning", "note"), + "secret_scanning": ("open",), +} +_CODE_SCANNING_BUCKET = { + "critical": "critical", + "high": "high", + "error": "high", + "medium": "warning", + "low": "warning", + "warning": "warning", + "note": "note", +} + + +class SecurityCoverageError(ValueError): + """Raised when a coverage receipt or bounded collection contract is invalid.""" + + +@dataclass(frozen=True) +class LoadedSecurityCoverage: + entries_by_full_name: dict[str, dict[str, Any]] + produced_at: str + schema_version: str + cohort_policy: str + cohort_repositories: tuple[str, ...] + receipt_state: str + age_hours: float + source_path: str + + +@dataclass +class _Budget: + base_limit: int + total_limit: int + quota_reserve: int + base_requests: int = 0 + total_requests: int = 0 + stop_reason: str | None = None + + def consume(self, *, pagination: bool) -> bool: + if self.stop_reason: + return False + if self.total_requests >= self.total_limit: + self.stop_reason = "total_request_limit" + return False + if not pagination and self.base_requests >= self.base_limit: + self.stop_reason = "base_request_limit" + return False + self.total_requests += 1 + if not pagination: + self.base_requests += 1 + return True + + +def _parse_datetime(value: Any, *, field_name: str) -> datetime: + if not isinstance(value, str) or not value.strip(): + raise SecurityCoverageError(f"{field_name} is required") + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise SecurityCoverageError(f"{field_name} is invalid: {value!r}") from exc + if parsed.tzinfo is None: + raise SecurityCoverageError(f"{field_name} must include a timezone") + return parsed.astimezone(timezone.utc) + + +def _text(value: Any) -> str: + return value.strip() if isinstance(value, str) else "" + + +def _mapping(value: Any) -> dict[str, Any]: + return value if isinstance(value, dict) else {} + + +def _canonical_repo(value: Any) -> str: + full_name = _text(value) + if not _REPOSITORY_RE.fullmatch(full_name): + raise SecurityCoverageError(f"invalid canonical repository name: {value!r}") + return full_name + + +def derive_default_attention_cohort( + portfolio_truth: dict[str, Any], + *, + expected_count: int = 16, +) -> tuple[str, ...]: + """Return the canonical default-attention GitHub cohort, failing on expansion.""" + repos: list[str] = [] + for project in portfolio_truth.get("projects") or []: + if not isinstance(project, dict): + continue + derived = _mapping(project.get("derived")) + if derived.get("attention_state") not in DEFAULT_ATTENTION_STATES: + continue + identity = _mapping(project.get("identity")) + repos.append(_canonical_repo(identity.get("repo_full_name"))) + if len({repo.lower() for repo in repos}) != len(repos): + raise SecurityCoverageError( + "default-attention cohort contains duplicate canonical repositories" + ) + cohort = tuple(sorted(repos, key=str.lower)) + if len(cohort) != expected_count: + raise SecurityCoverageError( + "default-attention cohort size changed: " + f"expected {expected_count}, observed {len(cohort)}" + ) + return cohort + + +def _empty_counts(provider: str) -> dict[str, int]: + return {key: 0 for key in _COUNT_KEYS[provider]} + + +def _provider_result( + provider: str, + *, + state: str, + observed_at: str | None = None, + http_status: int | None = None, + reason: str | None = None, + etag: str | None = None, + last_modified: str | None = None, + pagination_complete: bool = False, + counts: dict[str, int] | None = None, + conditional_request: bool = False, + conditional_result: str = "not_used", +) -> dict[str, Any]: + if state not in PROVIDER_STATES: + raise SecurityCoverageError(f"invalid provider state: {state}") + return { + "state": state, + "observed_at": observed_at, + "http_status": http_status, + "http_classification": ( + None + if http_status is None + else "success" + if http_status == 200 + else "not_modified" + if http_status == 304 + else reason + ), + "reason": reason, + "etag": etag, + "last_modified": last_modified, + "conditional": { + "requested": conditional_request, + "result": conditional_result, + }, + "pagination_complete": pagination_complete, + "counts": counts if state == "observed" else None, + } + + +def _response_message(response: requests.Response) -> str: + try: + payload = response.json() + except (ValueError, TypeError): + return "" + if not isinstance(payload, dict): + return "" + return _text(payload.get("message")).lower() + + +def _classify_failure(provider: str, response: requests.Response) -> tuple[str, str]: + status = response.status_code + message = _response_message(response) + remaining = response.headers.get("X-RateLimit-Remaining") + if status == 429 or ( + status == 403 + and ( + remaining == "0" + or "rate limit" in message + or "secondary rate" in message + ) + ): + return "rate_limited", "github_rate_limit" + if status == 403: + if provider == "code_scanning" and ( + "advanced security" in message + or "code scanning is not enabled" in message + or "code security" in message + ): + return "feature_unavailable", "code_scanning_not_enabled" + return "forbidden", "github_forbidden" + if status == 404: + return "not_found", "github_not_found" + if status == 410: + return "gone", "github_gone" + if status >= 500: + return "transient_error", f"github_http_{status}" + return "malformed", f"unexpected_http_{status}" + + +def _accumulate(provider: str, counts: dict[str, int], alerts: list[Any]) -> bool: + for alert in alerts: + if not isinstance(alert, dict): + return False + if provider == "dependabot": + advisory = _mapping(alert.get("security_advisory")) + vulnerability = _mapping(alert.get("security_vulnerability")) + severity = _text(advisory.get("severity") or vulnerability.get("severity")).lower() + if severity in counts: + counts[severity] += 1 + elif provider == "code_scanning": + rule = _mapping(alert.get("rule")) + severity = _text( + rule.get("security_severity_level") or rule.get("severity") + ).lower() + bucket = _CODE_SCANNING_BUCKET.get(severity) + if bucket: + counts[bucket] += 1 + else: + counts["open"] += 1 + return True + + +def _prior_provider( + prior_receipt: dict[str, Any] | None, + repo_full_name: str, + provider: str, +) -> dict[str, Any]: + repositories = _mapping(_mapping(prior_receipt).get("repositories")) + entry = _mapping(repositories.get(repo_full_name)) + return _mapping(_mapping(entry.get("providers")).get(provider)) + + +def _fetch_provider( + session: requests.Session, + *, + api_base_url: str, + repo_full_name: str, + provider: str, + now_iso: str, + budget: _Budget, + prior: dict[str, Any], +) -> tuple[dict[str, Any], bool]: + endpoint = _ENDPOINTS[provider] + next_url: str | None = f"{api_base_url}/repos/{repo_full_name}/{endpoint}" + params: dict[str, str] = {"state": "open", "per_page": "100"} + headers: dict[str, str] = {} + prior_etag = _text(prior.get("etag")) + if prior_etag: + headers["If-None-Match"] = prior_etag + counts = _empty_counts(provider) + first_response = True + etag = last_modified = None + + while next_url: + if not budget.consume(pagination=not first_response): + return ( + _provider_result( + provider, + state="not_requested", + reason=budget.stop_reason, + conditional_request=bool(prior_etag), + conditional_result="incomplete", + ), + True, + ) + try: + response = session.get( + next_url, + params=params if first_response else None, + headers=headers if first_response else None, + timeout=30, + ) + except requests.RequestException: + return ( + _provider_result( + provider, + state="transient_error", + observed_at=now_iso, + reason="network_error", + conditional_request=bool(prior_etag), + conditional_result="failed", + ), + False, + ) + + etag = response.headers.get("ETag") or etag + last_modified = response.headers.get("Last-Modified") or last_modified + reserve_reached = False + remaining = response.headers.get("X-RateLimit-Remaining") + if remaining is not None: + try: + reserve_reached = int(remaining) <= budget.quota_reserve + except ValueError: + pass + if response.status_code == 304: + if reserve_reached: + budget.stop_reason = "quota_reserve" + prior_counts = prior.get("counts") + if prior.get("state") != "observed" or not isinstance(prior_counts, dict): + return ( + _provider_result( + provider, + state="malformed", + http_status=304, + reason="conditional_response_without_observed_prior", + observed_at=now_iso, + conditional_request=True, + conditional_result="invalid_prior", + ), + False, + ) + return ( + _provider_result( + provider, + state="observed", + observed_at=now_iso, + http_status=304, + reason="not_modified", + etag=etag or prior_etag, + last_modified=last_modified or _text(prior.get("last_modified")) or None, + pagination_complete=True, + counts={ + key: int(prior_counts.get(key, 0) or 0) + for key in _COUNT_KEYS[provider] + }, + conditional_request=True, + conditional_result="not_modified", + ), + reserve_reached, + ) + if response.status_code != 200: + state, reason = _classify_failure(provider, response) + if state == "rate_limited": + budget.stop_reason = "rate_limited" + elif reserve_reached: + budget.stop_reason = "quota_reserve" + return ( + _provider_result( + provider, + state=state, + http_status=response.status_code, + reason=reason, + observed_at=now_iso, + etag=etag, + last_modified=last_modified, + conditional_request=bool(prior_etag), + conditional_result="failed", + ), + state == "rate_limited" or reserve_reached, + ) + try: + page = response.json() + except ValueError: + page = None + if not isinstance(page, list) or not _accumulate(provider, counts, page): + return ( + _provider_result( + provider, + state="malformed", + http_status=200, + reason="non_list_or_invalid_alert_payload", + observed_at=now_iso, + etag=etag, + last_modified=last_modified, + conditional_request=bool(prior_etag), + conditional_result="malformed", + ), + False, + ) + next_url = response.links.get("next", {}).get("url") + if reserve_reached: + budget.stop_reason = "quota_reserve" + if next_url: + return ( + _provider_result( + provider, + state="not_requested", + observed_at=now_iso, + reason="quota_reserve_before_pagination_complete", + conditional_request=bool(prior_etag), + conditional_result="incomplete", + ), + True, + ) + first_response = False + params = {} + headers = {} + + result = _provider_result( + provider, + state="observed", + observed_at=now_iso, + http_status=200, + etag=etag, + last_modified=last_modified, + pagination_complete=True, + counts=counts, + conditional_request=bool(prior_etag), + conditional_result="modified" if prior_etag else "not_used", + ) + return result, budget.stop_reason == "quota_reserve" + + +def collect_security_coverage( + portfolio_truth: dict[str, Any], + *, + token: str | None, + expected_cohort_count: int = 16, + base_request_limit: int = DEFAULT_BASE_REQUEST_LIMIT, + total_request_limit: int = DEFAULT_TOTAL_REQUEST_LIMIT, + quota_reserve: int = DEFAULT_QUOTA_RESERVE, + prior_receipt: dict[str, Any] | None = None, + session: requests.Session | None = None, + now: datetime | None = None, + producer_commit: str | None = None, + api_base_url: str | None = None, +) -> dict[str, Any]: + """Collect the bounded default-attention cohort into a provenance receipt.""" + if not token: + raise SecurityCoverageError( + "authorized GitHub read credential unavailable; no receipt was written" + ) + if ( + not 0 < base_request_limit <= DEFAULT_BASE_REQUEST_LIMIT + or not base_request_limit <= total_request_limit <= DEFAULT_TOTAL_REQUEST_LIMIT + or quota_reserve < 0 + ): + raise SecurityCoverageError("request budget limits exceed the bounded contract") + if prior_receipt is not None: + validate_security_coverage_receipt( + prior_receipt, + max_age_hours=24 * 365, + now=now, + ) + cohort = derive_default_attention_cohort( + portfolio_truth, expected_count=expected_cohort_count + ) + required_base_requests = len(cohort) * len(PROVIDER_NAMES) + if required_base_requests > base_request_limit: + raise SecurityCoverageError( + f"cohort requires {required_base_requests} base requests; " + f"limit is {base_request_limit}" + ) + commit = producer_commit + if not commit: + try: + commit = subprocess.run( + ["git", "rev-parse", "HEAD"], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + except (OSError, subprocess.CalledProcessError): + raise SecurityCoverageError( + "producer commit unavailable; refusing an unproven receipt" + ) + if not _COMMIT_RE.fullmatch(commit): + raise SecurityCoverageError( + "producer commit is invalid; refusing an unproven receipt" + ) + + collected_at = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + now_iso = collected_at.isoformat() + client = session or requests.Session() + client.headers.update( + { + "Accept": "application/vnd.github+json", + "Authorization": f"Bearer {token}", + "User-Agent": "github-repo-auditor/security-coverage", + "X-GitHub-Api-Version": GITHUB_API_VERSION, + } + ) + budget = _Budget( + base_limit=base_request_limit, + total_limit=total_request_limit, + quota_reserve=quota_reserve, + ) + repositories: dict[str, dict[str, Any]] = {} + halted = False + for repo_full_name in cohort: + providers: dict[str, dict[str, Any]] = {} + for provider in PROVIDER_NAMES: + if halted: + providers[provider] = _provider_result( + provider, + state="not_requested", + reason=budget.stop_reason or "collection_halted", + ) + continue + providers[provider], halted = _fetch_provider( + client, + api_base_url=api_base_url + or os.environ.get("GITHUB_API_BASE_URL", "https://api.github.com"), + repo_full_name=repo_full_name, + provider=provider, + now_iso=now_iso, + budget=budget, + prior=_prior_provider(prior_receipt, repo_full_name, provider), + ) + repositories[repo_full_name] = {"providers": providers} + + produced_at = ( + collected_at if now is not None else datetime.now(timezone.utc) + ).astimezone(timezone.utc) + return { + "schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "produced_at": produced_at.isoformat(), + "producer": { + "repository": "saagpatel/GithubRepoAuditor", + "commit": commit, + }, + "github_api_version": GITHUB_API_VERSION, + "cohort": { + "policy": DEFAULT_COHORT_POLICY, + "expected_count": expected_cohort_count, + "repository_count": len(cohort), + "repositories": list(cohort), + }, + "request_budget": { + "base_limit": base_request_limit, + "total_limit": total_request_limit, + "quota_reserve": quota_reserve, + "base_requests": budget.base_requests, + "total_requests": budget.total_requests, + "stop_reason": budget.stop_reason, + }, + "repositories": repositories, + } + + +def _validate_provider( + provider: str, + value: Any, + *, + receipt_is_stale: bool, + produced_at: datetime, + current: datetime, + max_age_hours: int, +) -> dict[str, Any]: + data = _mapping(value) + state = _text(data.get("state")) + if state not in PROVIDER_STATES: + raise SecurityCoverageError(f"{provider}.state is invalid: {state!r}") + counts = data.get("counts") + http_status = data.get("http_status") + http_classification = data.get("http_classification") + conditional = _mapping(data.get("conditional")) + if ( + not isinstance(conditional.get("requested"), bool) + or conditional.get("result") + not in { + "not_used", + "modified", + "not_modified", + "failed", + "malformed", + "incomplete", + "invalid_prior", + } + ): + raise SecurityCoverageError(f"{provider}.conditional metadata is invalid") + if http_status is not None and ( + not isinstance(http_status, int) or isinstance(http_status, bool) + ): + raise SecurityCoverageError(f"{provider}.http_status must be an integer or null") + if http_classification is not None and not isinstance(http_classification, str): + raise SecurityCoverageError( + f"{provider}.http_classification must be a string or null" + ) + observed_at_value = data.get("observed_at") + observed_at = ( + _parse_datetime(observed_at_value, field_name=f"{provider}.observed_at") + if observed_at_value is not None + else None + ) + if observed_at is not None: + if observed_at > produced_at: + raise SecurityCoverageError( + f"{provider}.observed_at is later than receipt produced_at" + ) + if (current - observed_at).total_seconds() / 3600 < -0.05: + raise SecurityCoverageError(f"{provider}.observed_at is future-dated") + if state == "observed": + if observed_at is None: + raise SecurityCoverageError(f"{provider}.observed_at is required") + provider_age_hours = (current - observed_at).total_seconds() / 3600 + if not data.get("pagination_complete"): + raise SecurityCoverageError(f"{provider} observed without complete pagination") + if not isinstance(counts, dict) or set(counts) != set(_COUNT_KEYS[provider]): + raise SecurityCoverageError(f"{provider}.counts required when observed") + if http_status not in {200, 304}: + raise SecurityCoverageError( + f"{provider}.http_status must be 200 or 304 when observed" + ) + expected_classification = ( + "success" if http_status == 200 else "not_modified" + ) + if http_classification != expected_classification: + raise SecurityCoverageError( + f"{provider}.http_classification does not match observed response" + ) + normalized_counts: dict[str, int] = {} + for key in _COUNT_KEYS[provider]: + raw = counts.get(key) + if not isinstance(raw, int) or raw < 0: + raise SecurityCoverageError(f"{provider}.counts.{key} must be non-negative") + normalized_counts[key] = raw + counts = normalized_counts + if provider_age_hours > max_age_hours: + state = "stale" + counts = None + elif counts is not None: + raise SecurityCoverageError(f"{provider}.counts must be null unless observed") + if state == "not_requested" and http_status is not None: + raise SecurityCoverageError(f"{provider} not_requested must not claim HTTP status") + if state == "forbidden" and http_status != 403: + raise SecurityCoverageError(f"{provider} forbidden requires HTTP 403") + if state == "feature_unavailable" and http_status not in {403, 404}: + raise SecurityCoverageError( + f"{provider} feature_unavailable requires HTTP 403 or 404" + ) + if state == "not_found" and http_status != 404: + raise SecurityCoverageError(f"{provider} not_found requires HTTP 404") + if state == "gone" and http_status != 410: + raise SecurityCoverageError(f"{provider} gone requires HTTP 410") + if state == "rate_limited" and http_status not in {403, 429}: + raise SecurityCoverageError( + f"{provider} rate_limited requires HTTP 403 or 429" + ) + if receipt_is_stale and state == "observed": + state = "stale" + counts = None + return { + "state": state, + "observed_at": data.get("observed_at"), + "http_status": http_status, + "http_classification": http_classification, + "reason": "receipt_stale" if state == "stale" else data.get("reason"), + "etag": data.get("etag"), + "last_modified": data.get("last_modified"), + "conditional": conditional, + "pagination_complete": bool(data.get("pagination_complete")), + "counts": counts, + } + + +def validate_security_coverage_receipt( + payload: Any, + *, + max_age_hours: int = 24, + expected_cohort_count: int = 16, + now: datetime | None = None, + source_path: str = "", +) -> LoadedSecurityCoverage: + """Validate provenance/freshness and return normalized full-name entries.""" + if max_age_hours <= 0: + raise SecurityCoverageError("max_age_hours must be positive") + if not isinstance(payload, dict): + raise SecurityCoverageError("security coverage receipt must be an object") + if payload.get("schema_version") != GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION: + raise SecurityCoverageError( + "unexpected security coverage receipt schema: " + f"{payload.get('schema_version')!r}" + ) + produced_at = _parse_datetime(payload.get("produced_at"), field_name="produced_at") + current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + age_hours = (current - produced_at).total_seconds() / 3600 + if age_hours < -0.05: + raise SecurityCoverageError("security coverage receipt is future-dated") + receipt_is_stale = age_hours > max_age_hours + + producer = _mapping(payload.get("producer")) + if producer.get("repository") != "saagpatel/GithubRepoAuditor": + raise SecurityCoverageError("security coverage receipt producer repository is invalid") + commit = _text(producer.get("commit")) + if not _COMMIT_RE.fullmatch(commit): + raise SecurityCoverageError("security coverage receipt producer commit is invalid") + if payload.get("github_api_version") != GITHUB_API_VERSION: + raise SecurityCoverageError("security coverage receipt API version is invalid") + + cohort = _mapping(payload.get("cohort")) + policy = _text(cohort.get("policy")) + if policy != DEFAULT_COHORT_POLICY: + raise SecurityCoverageError(f"unexpected cohort policy: {policy!r}") + expected_count = cohort.get("expected_count") + repository_count = cohort.get("repository_count") + repositories_list = cohort.get("repositories") + if ( + not isinstance(expected_count, int) + or not isinstance(repository_count, int) + or not isinstance(repositories_list, list) + ): + raise SecurityCoverageError("cohort counts and repositories are required") + canonical_cohort = tuple(_canonical_repo(repo) for repo in repositories_list) + if len({repo.lower() for repo in canonical_cohort}) != len(canonical_cohort): + raise SecurityCoverageError("cohort contains duplicate repositories") + if ( + repository_count != len(canonical_cohort) + or expected_count != repository_count + or repository_count != expected_cohort_count + ): + raise SecurityCoverageError("cohort count contract mismatch") + if canonical_cohort != tuple(sorted(canonical_cohort, key=str.lower)): + raise SecurityCoverageError("cohort repositories must be canonically sorted") + + request_budget = _mapping(payload.get("request_budget")) + base_limit = request_budget.get("base_limit") + total_limit = request_budget.get("total_limit") + quota_reserve = request_budget.get("quota_reserve") + base_requests = request_budget.get("base_requests") + total_requests = request_budget.get("total_requests") + if ( + not isinstance(base_limit, int) + or not 0 < base_limit <= DEFAULT_BASE_REQUEST_LIMIT + or not isinstance(total_limit, int) + or not base_limit <= total_limit <= DEFAULT_TOTAL_REQUEST_LIMIT + or not isinstance(quota_reserve, int) + or quota_reserve < 0 + or not isinstance(base_requests, int) + or not 0 <= base_requests <= base_limit + or not isinstance(total_requests, int) + or not base_requests <= total_requests <= total_limit + ): + raise SecurityCoverageError("request budget contract is invalid") + + raw_repositories = _mapping(payload.get("repositories")) + if set(raw_repositories) != set(canonical_cohort): + raise SecurityCoverageError("receipt repositories do not match cohort") + entries: dict[str, dict[str, Any]] = {} + for repo_full_name in canonical_cohort: + repo_data = _mapping(raw_repositories.get(repo_full_name)) + providers = _mapping(repo_data.get("providers")) + if set(providers) != set(PROVIDER_NAMES): + raise SecurityCoverageError( + f"{repo_full_name} does not contain exactly the required providers" + ) + entries[repo_full_name] = { + "repo_full_name": repo_full_name, + "cohort_member": True, + "cohort_policy": policy, + "receipt_schema_version": GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + "source_produced_at": produced_at.isoformat(), + "receipt_state": "stale" if receipt_is_stale else "fresh", + "providers": { + provider: _validate_provider( + provider, + providers[provider], + receipt_is_stale=receipt_is_stale, + produced_at=produced_at, + current=current, + max_age_hours=max_age_hours, + ) + for provider in PROVIDER_NAMES + }, + } + return LoadedSecurityCoverage( + entries_by_full_name=entries, + produced_at=produced_at.isoformat(), + schema_version=GITHUB_SECURITY_RECEIPT_SCHEMA_VERSION, + cohort_policy=policy, + cohort_repositories=canonical_cohort, + receipt_state="stale" if receipt_is_stale else "fresh", + age_hours=round(age_hours, 3), + source_path=source_path, + ) + + +def load_security_coverage_receipt( + path: Path, + *, + max_age_hours: int = 24, + now: datetime | None = None, +) -> LoadedSecurityCoverage: + try: + payload = json.loads(path.read_text()) + except FileNotFoundError as exc: + raise SecurityCoverageError(f"security coverage receipt not found: {path}") from exc + except (OSError, json.JSONDecodeError) as exc: + raise SecurityCoverageError( + f"could not read security coverage receipt {path}: {exc}" + ) from exc + return validate_security_coverage_receipt( + payload, + max_age_hours=max_age_hours, + now=now, + source_path=str(path), + ) + + +def write_security_coverage_receipt(payload: dict[str, Any], path: Path) -> None: + """Atomically write a validated receipt payload.""" + validate_security_coverage_receipt(payload, max_age_hours=24 * 365) + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.tmp") + temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + temporary.replace(path) + + +def _load_json_object(path: Path) -> dict[str, Any]: + try: + payload = json.loads(path.read_text()) + except (OSError, json.JSONDecodeError) as exc: + raise SecurityCoverageError(f"could not read JSON object {path}: {exc}") from exc + if not isinstance(payload, dict): + raise SecurityCoverageError(f"{path} must contain a JSON object") + return payload + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Collect bounded default-attention GitHub security coverage" + ) + parser.add_argument( + "--truth", + type=Path, + default=Path("output/portfolio-truth-latest.json"), + ) + parser.add_argument( + "--output", + type=Path, + default=Path("output") / GITHUB_SECURITY_RECEIPT_FILENAME, + ) + parser.add_argument( + "--validate-only", + action="store_true", + help="Validate the existing receipt without using a credential or GitHub API", + ) + parser.add_argument( + "--max-age-hours", + type=int, + default=24, + help="Freshness window used by --validate-only (default: 24)", + ) + parser.add_argument("--expected-cohort-count", type=int, default=16) + parser.add_argument( + "--base-request-limit", type=int, default=DEFAULT_BASE_REQUEST_LIMIT + ) + parser.add_argument( + "--total-request-limit", type=int, default=DEFAULT_TOTAL_REQUEST_LIMIT + ) + parser.add_argument("--quota-reserve", type=int, default=DEFAULT_QUOTA_RESERVE) + args = parser.parse_args() + + try: + if args.validate_only: + loaded = load_security_coverage_receipt( + args.output, + max_age_hours=args.max_age_hours, + ) + print( + json.dumps( + { + "state": "validated", + "path": str(args.output), + "receipt_state": loaded.receipt_state, + "age_hours": loaded.age_hours, + "cohort_count": len(loaded.cohort_repositories), + }, + indent=2, + ) + ) + return + truth = _load_json_object(args.truth) + prior = _load_json_object(args.output) if args.output.is_file() else None + receipt = collect_security_coverage( + truth, + token=os.environ.get("GITHUB_TOKEN"), + expected_cohort_count=args.expected_cohort_count, + base_request_limit=args.base_request_limit, + total_request_limit=args.total_request_limit, + quota_reserve=args.quota_reserve, + prior_receipt=prior, + ) + write_security_coverage_receipt(receipt, args.output) + except SecurityCoverageError as exc: + raise SystemExit(str(exc)) from exc + print( + json.dumps( + { + "state": "written", + "path": str(args.output), + "cohort_count": receipt["cohort"]["repository_count"], + "request_budget": receipt["request_budget"], + }, + indent=2, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/src/portfolio_security_gate.py b/src/portfolio_security_gate.py index d69eed40..2cc70fdf 100644 --- a/src/portfolio_security_gate.py +++ b/src/portfolio_security_gate.py @@ -1,8 +1,9 @@ """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. +operator question: did any required-cohort repo regain a high-severity +Dependabot/code-scanning alert or an open secret-scanning alert? Missing +coverage is treated as unknown, not healthy. """ from __future__ import annotations @@ -18,16 +19,18 @@ class SecurityGateItem: critical: int high: int risk_tier: str + secret_scanning_open: int = 0 @property def total(self) -> int: - return self.critical + self.high + return self.critical + self.high + self.secret_scanning_open def to_dict(self) -> dict[str, Any]: return { "repo": self.repo, "critical": self.critical, "high": self.high, + "secret_scanning_open": self.secret_scanning_open, "risk_tier": self.risk_tier, } @@ -39,6 +42,12 @@ class SecurityGateReport: total_open_critical: int total_open_high: int flagged_repos: tuple[SecurityGateItem, ...] + total_open_secrets: int = 0 + required_cohort_count: int = 0 + complete_count: int = 0 + partial_count: int = 0 + stale_count: int = 0 + unknown_count: int = 0 max_age_hours: int | None = None source_age_hours: float | None = None freshness_error: str | None = None @@ -50,7 +59,11 @@ def repos_with_open_high_critical(self) -> int: @property def passed(self) -> bool: return ( - self.scanned_count > 0 + self.required_cohort_count > 0 + and self.complete_count == self.required_cohort_count + and self.partial_count == 0 + and self.stale_count == 0 + and self.unknown_count == 0 and self.repos_with_open_high_critical == 0 and not self.is_stale ) @@ -67,12 +80,18 @@ def is_stale(self) -> bool: @property def status(self) -> str: - if self.scanned_count <= 0: - return "unknown" if self.is_stale: return "stale" if self.repos_with_open_high_critical > 0: return "fail" + if ( + self.required_cohort_count <= 0 + or self.complete_count != self.required_cohort_count + or self.partial_count + or self.stale_count + or self.unknown_count + ): + return "unknown" return "pass" def to_dict(self) -> dict[str, Any]: @@ -81,9 +100,15 @@ def to_dict(self) -> dict[str, Any]: "status": self.status, "passed": self.passed, "scanned_count": self.scanned_count, + "required_cohort_count": self.required_cohort_count, + "complete_count": self.complete_count, + "partial_count": self.partial_count, + "stale_count": self.stale_count, + "unknown_count": self.unknown_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, + "total_open_secrets": self.total_open_secrets, "max_age_hours": self.max_age_hours, "source_age_hours": self.source_age_hours, "freshness_error": self.freshness_error, @@ -127,8 +152,14 @@ def build_security_gate_report( ) -> SecurityGateReport: projects = portfolio_truth.get("projects") or [] scanned_count = 0 + required_cohort_count = 0 + complete_count = 0 + partial_count = 0 + stale_count = 0 + unknown_count = 0 total_critical = 0 total_high = 0 + total_secrets = 0 flagged: list[SecurityGateItem] = [] generated_at = _text(portfolio_truth.get("generated_at")) or "unknown" source_age_hours = freshness_error = None @@ -142,15 +173,50 @@ def build_security_gate_report( if not isinstance(project, dict): continue security = _mapping(project.get("security")) - if not security.get("alerts_available"): + if not security.get("cohort_member"): continue - - scanned_count += 1 - critical = _int(security.get("dependabot_critical")) - high = _int(security.get("dependabot_high")) + required_cohort_count += 1 + coverage_state = _text(security.get("coverage_state")) or "unknown" + if coverage_state == "complete" and security.get("alerts_available"): + complete_count += 1 + scanned_count += 1 + elif coverage_state == "partial": + partial_count += 1 + elif coverage_state == "stale": + stale_count += 1 + else: + unknown_count += 1 + providers = _mapping(security.get("providers")) + dependabot = _mapping(providers.get("dependabot")) + code_scanning = _mapping(providers.get("code_scanning")) + secret_scanning = _mapping(providers.get("secret_scanning")) + critical = ( + _int(security.get("dependabot_critical")) + if dependabot.get("state") == "observed" + else 0 + ) + ( + _int(security.get("code_scanning_critical")) + if code_scanning.get("state") == "observed" + else 0 + ) + high = ( + _int(security.get("dependabot_high")) + if dependabot.get("state") == "observed" + else 0 + ) + ( + _int(security.get("code_scanning_high")) + if code_scanning.get("state") == "observed" + else 0 + ) + secrets = ( + _int(security.get("secret_scanning_open")) + if secret_scanning.get("state") == "observed" + else 0 + ) total_critical += critical total_high += high - if critical <= 0 and high <= 0: + total_secrets += secrets + if critical <= 0 and high <= 0 and secrets <= 0: continue identity = _mapping(project.get("identity")) @@ -166,6 +232,7 @@ def build_security_gate_report( critical=critical, high=high, risk_tier=_text(risk.get("risk_tier")) or "baseline", + secret_scanning_open=secrets, ) ) @@ -176,6 +243,12 @@ def build_security_gate_report( total_open_critical=total_critical, total_open_high=total_high, flagged_repos=tuple(flagged), + total_open_secrets=total_secrets, + required_cohort_count=required_cohort_count, + complete_count=complete_count, + partial_count=partial_count, + stale_count=stale_count, + unknown_count=unknown_count, max_age_hours=max_age_hours, source_age_hours=source_age_hours, freshness_error=freshness_error, @@ -188,8 +261,10 @@ def render_security_gate_markdown(report: SecurityGateReport) -> str: "", ( f"Status: {report.status.upper()} | scanned {report.scanned_count} | " + f"required cohort {report.required_cohort_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"critical {report.total_open_critical} | high {report.total_open_high} | " + f"secrets {report.total_open_secrets}" ), f"Source freshness: {report.generated_at}", "", @@ -197,8 +272,10 @@ def render_security_gate_markdown(report: SecurityGateReport) -> str: 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." + "Required-cohort security coverage is missing or incomplete " + f"(complete {report.complete_count}, partial {report.partial_count}, " + f"stale {report.stale_count}, unknown {report.unknown_count}). " + "Do not treat the cohort as clear." ) elif report.status == "stale": if report.freshness_error: @@ -209,11 +286,17 @@ def render_security_gate_markdown(report: SecurityGateReport) -> str: f"{report.max_age_hours}h freshness threshold." ) elif report.passed: - lines.append("All scanned repos are clear of open high/critical Dependabot alerts.") + lines.append( + "All required-cohort repos are clear of open high-severity GitHub " + "security alerts." + ) else: - lines.append("| Repo | Risk | Critical | High |") - lines.append("|---|---:|---:|---:|") + lines.append("| Repo | Risk | Critical | High | Open secrets |") + lines.append("|---|---:|---:|---:|---:|") for item in report.flagged_repos: - lines.append(f"| {item.repo} | {item.risk_tier} | {item.critical} | {item.high} |") + lines.append( + f"| {item.repo} | {item.risk_tier} | {item.critical} | " + f"{item.high} | {item.secret_scanning_open} |" + ) return "\n".join(lines) diff --git a/src/portfolio_truth_publish.py b/src/portfolio_truth_publish.py index cc25c682..4cd1b50e 100644 --- a/src/portfolio_truth_publish.py +++ b/src/portfolio_truth_publish.py @@ -90,6 +90,7 @@ def publish_portfolio_truth( allow_empty_notion: bool = False, release_count_by_name: dict[str, int] | None = None, security_alerts_by_name: dict[str, dict] | None = None, + security_coverage_metadata: dict[str, object] | None = None, repo_status_by_name: dict[str, dict] | None = None, producer_evidence: ProducerEvidence | None = None, producer_repo_root: Path | None = None, @@ -127,6 +128,7 @@ def publish_portfolio_truth( notion_context_fallback=notion_context_fallback, release_count_by_name=release_count_by_name, security_alerts_by_name=security_alerts_by_name, + security_coverage_metadata=security_coverage_metadata, repo_status_by_name=repo_status_by_name, producer=producer_evidence.to_dict() if producer_evidence else {}, prior_notion_generated_at=prior_notion_generated_at, diff --git a/src/portfolio_truth_reconcile.py b/src/portfolio_truth_reconcile.py index 0ebe0240..b106b5e9 100644 --- a/src/portfolio_truth_reconcile.py +++ b/src/portfolio_truth_reconcile.py @@ -5,7 +5,7 @@ import re import hashlib from collections import Counter -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -233,6 +233,7 @@ def build_portfolio_truth_snapshot( now: datetime | None = None, release_count_by_name: dict[str, int] | None = None, security_alerts_by_name: dict[str, dict] | None = None, + security_coverage_metadata: dict[str, Any] | None = None, repo_status_by_name: dict[str, dict] | None = None, producer: dict[str, Any] | None = None, prior_notion_generated_at: str | None = None, @@ -337,6 +338,7 @@ def build_portfolio_truth_snapshot( notion_context_rows=len(notion_context), notion_context_carried_forward=notion_context_carried_forward, prior_notion_generated_at=prior_notion_generated_at, + security_coverage_metadata=security_coverage_metadata, ), coverage=_build_coverage_envelope( projects=projects, @@ -359,14 +361,62 @@ def _build_coverage_envelope( notion_context_carried_forward: bool, notion_context_rows: int, ) -> list[dict[str, Any]]: - scanned = sum(project.security.alerts_available for project in projects) + complete = sum(project.security.coverage_state == "complete" for project in projects) + partial = sum(project.security.coverage_state == "partial" for project in projects) + stale = sum(project.security.coverage_state == "stale" for project in projects) + unknown = len(projects) - complete - partial - stale + cohort_count = sum(project.security.cohort_member for project in projects) + cohort_complete = sum( + project.security.cohort_member + and project.security.coverage_state == "complete" + for project in projects + ) + cohort_partial = sum( + project.security.cohort_member + and project.security.coverage_state == "partial" + for project in projects + ) + cohort_stale = sum( + project.security.cohort_member + and project.security.coverage_state == "stale" + for project in projects + ) + cohort_unknown = cohort_count - cohort_complete - cohort_partial - cohort_stale + provider_counts = { + provider: sum( + project.security.provider_state(provider) == "observed" + for project in projects + ) + for provider in ("dependabot", "code_scanning", "secret_scanning") + } git_observed = sum( project.repository_state.get("state") == "observed" for project in projects ) return [ {"source": "workspace", "state": "observed", "project_count": len(projects)}, {"source": "git", "state": "observed" if git_observed else "unknown", "observed_count": git_observed, "project_count": len(projects)}, - {"source": "github_security", "state": "known" if scanned == len(projects) else "partial" if scanned else "unknown", "scanned_count": scanned, "project_count": len(projects)}, + { + "source": "github_security", + "state": ( + "known" + if complete == len(projects) + else "partial" + if complete or partial + else "unknown" + ), + "scanned_count": complete, + "complete_repo_count": complete, + "partial_repo_count": partial, + "stale_count": stale, + "unknown_count": unknown, + "cohort_repository_count": cohort_count, + "cohort_complete_count": cohort_complete, + "cohort_partial_count": cohort_partial, + "cohort_stale_count": cohort_stale, + "cohort_unknown_count": cohort_unknown, + "provider_observed_counts": provider_counts, + "project_count": len(projects), + }, {"source": "notion", "state": "carried_forward" if notion_context_carried_forward else "observed" if notion_context_rows else "unknown", "observed_count": notion_context_rows}, ] @@ -380,6 +430,7 @@ def _build_input_envelope( notion_context_rows: int, notion_context_carried_forward: bool, prior_notion_generated_at: str | None, + security_coverage_metadata: dict[str, Any] | None, ) -> dict[str, Any]: resolved_catalog = Path(str(catalog_data.get("path") or "")) catalog_hash = ( @@ -396,7 +447,7 @@ def _build_input_envelope( else: notion_mode = "live" notion_observed_at = now.isoformat() - return { + inputs = { "catalog": { "source_id": "portfolio-catalog", "sha256": catalog_hash, @@ -414,6 +465,9 @@ def _build_input_envelope( ), }, } + if security_coverage_metadata: + inputs["github_security"] = dict(security_coverage_metadata) + return inputs def load_prior_notion_context(latest_path: Path) -> dict[str, dict[str, str]]: @@ -491,30 +545,87 @@ def _has_path_catalog_contract(project: PortfolioTruthProject) -> bool: def _build_security_fields(ghas_entry: dict[str, Any] | None) -> SecurityFields: - """Map a per-repo GHAS alert entry (from output/ghas-alerts--*.json) - into SecurityFields. A missing/None entry yields all-zero counts with - alerts_available=False (the repo was not scanned) — distinct from a clean scan, - and keeps the security overlay strictly opt-in (no entry → no security signal).""" + """Map a validated receipt entry into provider-specific security fields. + + Legacy GHAS-shaped entries remain accepted for unit/backward compatibility, + but only a fresh observation from all three providers is complete coverage. + """ if not ghas_entry: return SecurityFields() - dependabot = ghas_entry.get("dependabot") or {} - code_scanning = ghas_entry.get("code_scanning") or {} - secret_scanning = ghas_entry.get("secret_scanning") or {} + raw_providers = ghas_entry.get("providers") + if isinstance(raw_providers, dict): + providers = { + name: dict(raw_providers.get(name) or {}) + for name in ("dependabot", "code_scanning", "secret_scanning") + } + else: + providers = {} + for name in ("dependabot", "code_scanning", "secret_scanning"): + legacy = dict(ghas_entry.get(name) or {}) + providers[name] = { + "state": "observed" if legacy.get("available") else "not_requested", + "observed_at": None, + "http_status": None, + "reason": "legacy_ghas_entry", + "etag": None, + "last_modified": None, + "pagination_complete": bool(legacy.get("available")), + "counts": ( + { + key: value + for key, value in legacy.items() + if key != "available" + and isinstance(value, int) + and value >= 0 + } + if legacy.get("available") + else None + ), + } - def _count(source: dict[str, Any], key: str) -> int: - value = source.get(key) + states = { + name: str((providers.get(name) or {}).get("state") or "not_requested") + for name in providers + } + observed_count = sum(state == "observed" for state in states.values()) + receipt_state = str(ghas_entry.get("receipt_state") or "unknown") + if receipt_state == "stale": + coverage_state = "stale" + elif observed_count == 3: + coverage_state = "complete" + elif observed_count: + coverage_state = "partial" + elif any(state == "stale" for state in states.values()): + coverage_state = "stale" + else: + coverage_state = "unknown" + + def _count(provider: str, key: str) -> int | None: + source = providers.get(provider) or {} + if source.get("state") != "observed": + return None + counts = source.get("counts") or {} + value = counts.get(key) return value if isinstance(value, int) and value >= 0 else 0 return SecurityFields( - alerts_available=bool(dependabot.get("available", False)), - coverage_state="known" if dependabot.get("available", False) else "unknown", - dependabot_critical=_count(dependabot, "critical"), - dependabot_high=_count(dependabot, "high"), - dependabot_medium=_count(dependabot, "medium"), - dependabot_low=_count(dependabot, "low"), - code_scanning_critical=_count(code_scanning, "critical"), - code_scanning_high=_count(code_scanning, "high"), - secret_scanning_open=_count(secret_scanning, "open"), + alerts_available=coverage_state == "complete", + coverage_state=coverage_state, + cohort_member=bool(ghas_entry.get("cohort_member", False)), + cohort_policy=str(ghas_entry.get("cohort_policy") or ""), + receipt_schema_version=str( + ghas_entry.get("receipt_schema_version") or "" + ), + receipt_state=receipt_state, + source_produced_at=str(ghas_entry.get("source_produced_at") or ""), + providers=providers, + dependabot_critical=_count("dependabot", "critical"), + dependabot_high=_count("dependabot", "high"), + dependabot_medium=_count("dependabot", "medium"), + dependabot_low=_count("dependabot", "low"), + code_scanning_critical=_count("code_scanning", "critical"), + code_scanning_high=_count("code_scanning", "high"), + secret_scanning_open=_count("secret_scanning", "open"), ) @@ -525,6 +636,11 @@ def _select_security_entry( name, but the local dir display_name often differs (e.g. "Signal & Noise" vs "signal-noise"), so match on the repo name from repo_full_name first and fall back to display_name only when repo_full_name is absent or unmatched.""" + exact = lookup.get(repo_full_name or "") + if exact is not None: + return exact + if any(entry.get("receipt_schema_version") for entry in lookup.values()): + return None repo_name = (repo_full_name or "").rsplit("/", 1)[-1] return lookup.get(repo_name) or lookup.get(display_name) @@ -723,16 +839,16 @@ def _build_truth_project( doctor_standard=declared_values["doctor_standard"], known_risks_present=bool(raw_project["known_risks_present"]), run_instructions_present=bool(raw_project["run_instructions_present"]), - security_high_alerts=security.dependabot_high, - security_critical_alerts=security.dependabot_critical, + security_high_alerts=security.dependabot_high or 0, + security_critical_alerts=security.dependabot_critical or 0, ) if ( - not security.alerts_available + security.coverage_state != "complete" and risk_entry.get("risk_summary") == "No elevated risk factors." ): risk_entry["risk_summary"] = ( - "No non-security risk factors detected; security posture is unknown " - "because alert coverage is unavailable." + "No non-security risk factors detected; GitHub security coverage is " + f"{security.coverage_state}." ) attention_state = _attention_state_for( activity_status=activity_status, @@ -743,6 +859,15 @@ def _build_truth_project( path_override=path_entry.get("path_override", ""), risk_entry=risk_entry, ) + if ( + not security.receipt_schema_version + and attention_state in {"active-product", "active-infra", "decision-needed"} + ): + security = replace( + security, + cohort_member=True, + cohort_policy="portfolio-default-attention-v1", + ) declared = DeclaredFields( owner=declared_values["owner"], diff --git a/src/portfolio_truth_status.py b/src/portfolio_truth_status.py index 2ea6b6fb..09359cb9 100644 --- a/src/portfolio_truth_status.py +++ b/src/portfolio_truth_status.py @@ -4,10 +4,17 @@ import json import logging +from datetime import datetime from pathlib import Path from src.cache import ResponseCache from src.github_client import GitHubClient +from src.github_security_coverage import ( + GITHUB_SECURITY_RECEIPT_FILENAME, + LoadedSecurityCoverage, + SecurityCoverageError, + load_security_coverage_receipt, +) def load_release_count_by_name(*, output_dir: Path, username: str) -> dict[str, int] | None: @@ -145,3 +152,32 @@ def load_security_alerts_by_name( ) return None return {name: entry for name, entry in data.items() if isinstance(entry, dict)} + + +def load_security_coverage_by_full_name( + *, + output_dir: Path, + receipt_path: Path | None = None, + max_age_hours: int = 24, + now: datetime | None = None, +) -> LoadedSecurityCoverage | None: + """Load the canonical provenance-bearing security receipt. + + Unlike the legacy GHAS overlay loader above, this path never discovers + candidates by filesystem mtime. The fixed pointer or explicit path must + validate its embedded schema, producer, cohort, observation timestamps, and + freshness before PortfolioTruth may consume it. + """ + selected = receipt_path or output_dir / GITHUB_SECURITY_RECEIPT_FILENAME + try: + return load_security_coverage_receipt( + selected, + max_age_hours=max_age_hours, + now=now, + ) + except SecurityCoverageError as exc: + logging.getLogger(__name__).warning( + "--portfolio-truth-include-security: %s — security coverage remains unknown", + exc, + ) + return None diff --git a/src/portfolio_truth_types.py b/src/portfolio_truth_types.py index 99293f13..c769a7e4 100644 --- a/src/portfolio_truth_types.py +++ b/src/portfolio_truth_types.py @@ -6,13 +6,15 @@ from pathlib import Path from typing import Any -SCHEMA_VERSION = "0.10.0" +SCHEMA_VERSION = "0.11.0" +# 0.11.0: provenance-bearing GitHub security receipts preserve per-provider +# states and expose complete/partial/stale/unknown coverage denominators. # 0.10.0: canonical producer receipts bind the exact checkout; coverage and # repository/worktree observation envelopes fail closed on unavailable evidence. # 0.8.0: derived.registry_status removed (was a stale->parked synonym table over # activity_status); derived.archived added as a first-class lifecycle boolean; # source_summary.registry_status_counts replaced by activity_status_counts + archived_count. -LEGACY_SCHEMA_VERSIONS = {"0.7.0", "0.8.0", "0.9.0"} +LEGACY_SCHEMA_VERSIONS = {"0.7.0", "0.8.0", "0.9.0", "0.10.0"} DERIVATION_POLICY_VERSION = "portfolio_attention.v2" # The published "latest" portfolio-truth artifact. The producer @@ -198,25 +200,38 @@ def to_dict(self) -> dict[str, Any]: @dataclass(frozen=True) class SecurityFields: - """Live GitHub Advanced Security alert counts, overlaid opt-in from the latest - output/ghas-alerts--*.json. When alerts_available is False the repo was - not scanned (no token / GHAS disabled / not fetched) — distinct from a clean scan - with zero open alerts, so consumers don't mislabel an unscanned repo as secure.""" + """Receipt-backed GitHub security coverage with compatibility count fields. + + ``alerts_available`` remains for older consumers, but now means all three + providers were freshly and completely observed. Provider-specific states in + ``providers`` are the authority for partial or unavailable coverage. + """ alerts_available: bool = False coverage_state: str = "unknown" - dependabot_critical: int = 0 - dependabot_high: int = 0 - dependabot_medium: int = 0 - dependabot_low: int = 0 - code_scanning_critical: int = 0 - code_scanning_high: int = 0 - secret_scanning_open: int = 0 + cohort_member: bool = False + cohort_policy: str = "" + receipt_schema_version: str = "" + receipt_state: str = "unknown" + source_produced_at: str = "" + providers: dict[str, dict[str, Any]] = field(default_factory=dict) + dependabot_critical: int | None = None + dependabot_high: int | None = None + dependabot_medium: int | None = None + dependabot_low: int | None = None + code_scanning_critical: int | None = None + code_scanning_high: int | None = None + secret_scanning_open: int | None = None @property def open_high_critical(self) -> int: """Dependabot high + critical — the security-risk-factor trigger surface.""" - return self.dependabot_high + self.dependabot_critical + return (self.dependabot_high or 0) + (self.dependabot_critical or 0) + + def provider_state(self, provider: str) -> str: + data = self.providers.get(provider) or {} + state = data.get("state") + return state if isinstance(state, str) else "not_requested" def to_dict(self) -> dict[str, Any]: data = dataclasses.asdict(self) @@ -275,6 +290,18 @@ def from_projects( total_open_high = 0 total_open_critical = 0 unavailable_count = 0 + complete_repo_count = 0 + partial_repo_count = 0 + stale_count = 0 + unknown_count = 0 + cohort_repository_count = 0 + cohort_complete_count = 0 + cohort_partial_count = 0 + cohort_stale_count = 0 + cohort_unknown_count = 0 + dependabot_observed_count = 0 + code_scanning_observed_count = 0 + secret_scanning_observed_count = 0 decision_needed_count = 0 default_attention_count = 0 for project in projects: @@ -282,13 +309,41 @@ def from_projects( if tier in risk_tier_counts: risk_tier_counts[tier] += 1 security = project.security - if security.alerts_available: - scanned_count += 1 + if security.cohort_member: + cohort_repository_count += 1 + provider_states = { + provider: security.provider_state(provider) + for provider in ("dependabot", "code_scanning", "secret_scanning") + } + dependabot_observed = provider_states["dependabot"] == "observed" + if dependabot_observed: + dependabot_observed_count += 1 if security.open_high_critical > 0: repos_with_open_high_critical += 1 - total_open_high += security.dependabot_high - total_open_critical += security.dependabot_critical + total_open_high += security.dependabot_high or 0 + total_open_critical += security.dependabot_critical or 0 + if provider_states["code_scanning"] == "observed": + code_scanning_observed_count += 1 + if provider_states["secret_scanning"] == "observed": + secret_scanning_observed_count += 1 + if security.coverage_state == "complete": + complete_repo_count += 1 + scanned_count += 1 + if security.cohort_member: + cohort_complete_count += 1 + elif security.coverage_state == "partial": + partial_repo_count += 1 + if security.cohort_member: + cohort_partial_count += 1 + elif security.coverage_state == "stale": + stale_count += 1 + if security.cohort_member: + cohort_stale_count += 1 else: + unknown_count += 1 + if security.cohort_member: + cohort_unknown_count += 1 + if not security.alerts_available: unavailable_count += 1 attention = project.derived.attention_state if attention == "decision-needed": @@ -301,11 +356,23 @@ def from_projects( security={ "scanned_count": scanned_count, "unavailable_count": unavailable_count, + "complete_repo_count": complete_repo_count, + "partial_repo_count": partial_repo_count, + "stale_count": stale_count, + "unknown_count": unknown_count, + "cohort_repository_count": cohort_repository_count, + "cohort_complete_count": cohort_complete_count, + "cohort_partial_count": cohort_partial_count, + "cohort_stale_count": cohort_stale_count, + "cohort_unknown_count": cohort_unknown_count, + "dependabot_observed_count": dependabot_observed_count, + "code_scanning_observed_count": code_scanning_observed_count, + "secret_scanning_observed_count": secret_scanning_observed_count, "coverage_state": ( "known" if unavailable_count == 0 else "partial" - if scanned_count + if complete_repo_count or partial_repo_count else "unknown" ), "repos_with_open_high_critical": repos_with_open_high_critical, diff --git a/tests/test_cli_subcommands.py b/tests/test_cli_subcommands.py index f15767c0..faee5f86 100644 --- a/tests/test_cli_subcommands.py +++ b/tests/test_cli_subcommands.py @@ -305,8 +305,8 @@ def test_triage_help_flag_count(self): def test_report_help_flag_count(self): text = _help_text("report") count = _count_flags_in_help(text) - assert count <= 41, ( - f"audit report --help shows {count} non-global flags (limit 41, raised in Arc D phase-3b for the 6 bounded-automation proposal flags: --propose-automation/--list-proposals/--approve-proposal/--reject-proposal/--execute-proposals/--apply).\n" + assert count <= 44, ( + f"audit report --help shows {count} non-global flags (limit 44, including the explicit security receipt path, freshness, and opt-in controls).\n" f"Flags found: {sorted(set(re.findall(r' (--[a-z][a-z0-9-]*)', text)))}" ) diff --git a/tests/test_github_security_coverage.py b/tests/test_github_security_coverage.py new file mode 100644 index 00000000..bc2cfe63 --- /dev/null +++ b/tests/test_github_security_coverage.py @@ -0,0 +1,341 @@ +from __future__ import annotations + +import json +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +import pytest + +from src.github_security_coverage import ( + DEFAULT_BASE_REQUEST_LIMIT, + GITHUB_SECURITY_RECEIPT_FILENAME, + SecurityCoverageError, + collect_security_coverage, + derive_default_attention_cohort, + load_security_coverage_receipt, + main, + validate_security_coverage_receipt, +) +from src.portfolio_truth_reconcile import _select_security_entry +from src.portfolio_truth_status import load_security_coverage_by_full_name + +NOW = datetime(2026, 7, 16, 12, tzinfo=timezone.utc) + + +def _truth(count: int = 16) -> dict[str, Any]: + attention = ("active-product", "active-infra", "decision-needed") + projects = [ + { + "identity": {"repo_full_name": f"owner/repo-{index:02d}"}, + "derived": {"attention_state": attention[index % len(attention)]}, + } + for index in range(count) + ] + projects.append( + { + "identity": {"repo_full_name": "owner/parked"}, + "derived": {"attention_state": "parked"}, + } + ) + return {"projects": projects} + + +class _Response: + def __init__( + self, + status_code: int = 200, + payload: Any = None, + *, + headers: dict[str, str] | None = None, + next_url: str | None = None, + ) -> None: + self.status_code = status_code + self._payload = [] if payload is None else payload + self.headers = headers or {"X-RateLimit-Remaining": "4000"} + self.links = {"next": {"url": next_url}} if next_url else {} + + def json(self) -> Any: + return self._payload + + +class _Session: + def __init__(self, responses: list[_Response] | None = None) -> None: + self.headers: dict[str, str] = {} + self.responses = list(responses or []) + self.calls: list[tuple[str, dict[str, Any]]] = [] + + def get(self, url: str, **kwargs: Any) -> _Response: + self.calls.append((url, kwargs)) + if self.responses: + return self.responses.pop(0) + return _Response() + + +def _collect( + *, + session: _Session | None = None, + prior: dict[str, Any] | None = None, +) -> dict[str, Any]: + return collect_security_coverage( + _truth(), + token="opaque-test-token", + session=session or _Session(), + prior_receipt=prior, + now=NOW, + producer_commit="a" * 40, + api_base_url="https://api.example.test", + ) + + +def test_default_attention_cohort_is_exact_and_fail_closed() -> None: + cohort = derive_default_attention_cohort(_truth()) + + assert len(cohort) == 16 + assert "owner/parked" not in cohort + with pytest.raises(SecurityCoverageError, match="expected 16, observed 17"): + derive_default_attention_cohort(_truth(17)) + + +def test_no_token_never_attempts_collection() -> None: + session = _Session() + + with pytest.raises(SecurityCoverageError, match="credential unavailable"): + collect_security_coverage(_truth(), token=None, session=session) + + assert session.calls == [] + + +def test_validate_only_requires_no_token_or_network( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + receipt = _collect() + path = tmp_path / GITHUB_SECURITY_RECEIPT_FILENAME + path.write_text(json.dumps(receipt)) + monkeypatch.delenv("GITHUB_TOKEN", raising=False) + monkeypatch.setattr( + sys, + "argv", + [ + "github_security_coverage", + "--validate-only", + "--output", + str(path), + "--max-age-hours", + "24", + ], + ) + + main() + + payload = json.loads(capsys.readouterr().out) + assert payload["state"] == "validated" + assert payload["cohort_count"] == 16 + + +def test_collector_is_serial_count_only_and_bounded_to_48_base_requests() -> None: + session = _Session() + + receipt = _collect(session=session) + + assert len(session.calls) == DEFAULT_BASE_REQUEST_LIMIT == 48 + assert receipt["request_budget"]["base_requests"] == 48 + assert receipt["request_budget"]["total_requests"] == 48 + assert receipt["request_budget"]["stop_reason"] is None + assert all("/alerts/" not in url for url, _ in session.calls) + for repository in receipt["repositories"].values(): + for provider in repository["providers"].values(): + assert provider["state"] == "observed" + assert provider["counts"] is not None + assert provider["pagination_complete"] is True + + +@pytest.mark.parametrize( + "limits", + [ + {"base_request_limit": 49}, + {"total_request_limit": 76}, + {"base_request_limit": 48, "total_request_limit": 47}, + {"quota_reserve": -1}, + ], +) +def test_collector_rejects_relaxed_request_bounds_before_network( + limits: dict[str, int], +) -> None: + session = _Session() + + with pytest.raises(SecurityCoverageError, match="bounded contract"): + collect_security_coverage( + _truth(), + token="opaque-test-token", + session=session, + producer_commit="a" * 40, + **limits, + ) + + assert session.calls == [] + + +def test_rate_limit_stops_immediately_and_leaves_remainder_not_requested() -> None: + session = _Session( + [ + _Response( + 403, + {"message": "API rate limit exceeded"}, + headers={"X-RateLimit-Remaining": "0"}, + ) + ] + ) + + receipt = _collect(session=session) + states = [ + provider["state"] + for repository in receipt["repositories"].values() + for provider in repository["providers"].values() + ] + + assert len(session.calls) == 1 + assert states.count("rate_limited") == 1 + assert states.count("not_requested") == 47 + assert receipt["request_budget"]["stop_reason"] == "rate_limited" + + +def test_quota_reserve_stops_before_following_request() -> None: + session = _Session( + [_Response(headers={"X-RateLimit-Remaining": "100"})] + ) + + receipt = _collect(session=session) + states = [ + provider["state"] + for repository in receipt["repositories"].values() + for provider in repository["providers"].values() + ] + + assert len(session.calls) == 1 + assert states.count("observed") == 1 + assert states.count("not_requested") == 47 + assert receipt["request_budget"]["stop_reason"] == "quota_reserve" + + +def test_total_request_ceiling_halts_incomplete_pagination() -> None: + session = _Session( + [ + _Response(next_url=f"https://api.example.test/page/{index + 1}") + for index in range(75) + ] + ) + + receipt = _collect(session=session) + first = receipt["repositories"]["owner/repo-00"]["providers"]["dependabot"] + + assert len(session.calls) == 75 + assert receipt["request_budget"]["total_requests"] == 75 + assert receipt["request_budget"]["stop_reason"] == "total_request_limit" + assert first["state"] == "not_requested" + assert first["counts"] is None + + +def test_forbidden_and_feature_unavailable_are_distinct() -> None: + session = _Session( + [ + _Response(403, {"message": "Resource not accessible by integration"}), + _Response(403, {"message": "Advanced Security must be enabled"}), + ] + ) + + receipt = _collect(session=session) + first = receipt["repositories"]["owner/repo-00"]["providers"] + + assert first["dependabot"]["state"] == "forbidden" + assert first["dependabot"]["counts"] is None + assert first["code_scanning"]["state"] == "feature_unavailable" + assert first["code_scanning"]["counts"] is None + + +def test_conditional_304_reuses_only_valid_prior_counts() -> None: + prior = _collect() + for repository in prior["repositories"].values(): + for provider in repository["providers"].values(): + provider["etag"] = '"stable"' + session = _Session( + [_Response(304, headers={"ETag": '"stable"'}) for _ in range(48)] + ) + + receipt = _collect(session=session, prior=prior) + + assert len(session.calls) == 48 + assert all( + kwargs["headers"] == {"If-None-Match": '"stable"'} + for _, kwargs in session.calls + ) + assert all( + provider["http_status"] == 304 and provider["state"] == "observed" + for repository in receipt["repositories"].values() + for provider in repository["providers"].values() + ) + + +def test_stale_provider_observation_becomes_unknown_count() -> None: + receipt = _collect() + receipt["produced_at"] = NOW.isoformat() + provider = receipt["repositories"]["owner/repo-00"]["providers"]["dependabot"] + provider["observed_at"] = (NOW - timedelta(hours=25)).isoformat() + + loaded = validate_security_coverage_receipt(receipt, now=NOW) + normalized = loaded.entries_by_full_name["owner/repo-00"]["providers"][ + "dependabot" + ] + + assert normalized["state"] == "stale" + assert normalized["counts"] is None + + +def test_receipt_loader_uses_embedded_provenance_not_newer_mtime( + tmp_path: Path, +) -> None: + receipt = _collect() + canonical = tmp_path / GITHUB_SECURITY_RECEIPT_FILENAME + canonical.write_text(json.dumps(receipt)) + decoy = tmp_path / "github-security-coverage-newer.json" + decoy.write_text(json.dumps({"schema_version": "forged"})) + decoy.touch() + + loaded = load_security_coverage_by_full_name(output_dir=tmp_path, now=NOW) + + assert loaded is not None + assert loaded.source_path == str(canonical) + assert len(loaded.entries_by_full_name) == 16 + + +def test_receipt_provenance_and_provider_timestamps_fail_closed(tmp_path: Path) -> None: + receipt = _collect() + receipt["producer"]["commit"] = "short" + path = tmp_path / GITHUB_SECURITY_RECEIPT_FILENAME + path.write_text(json.dumps(receipt)) + + with pytest.raises(SecurityCoverageError, match="producer commit"): + load_security_coverage_receipt(path, now=NOW) + + receipt = _collect() + provider = receipt["repositories"]["owner/repo-00"]["providers"]["dependabot"] + provider["observed_at"] = (NOW + timedelta(minutes=1)).isoformat() + with pytest.raises(SecurityCoverageError, match="later than receipt produced_at"): + validate_security_coverage_receipt(receipt, now=NOW) + + +def test_canonical_receipt_join_does_not_fall_back_to_repo_basename() -> None: + entry = { + "receipt_schema_version": "GitHubSecurityCoverageReceiptV1", + "providers": {}, + } + + assert _select_security_entry( + {"other/shared": entry}, "owner/shared", "shared" + ) is None + assert _select_security_entry( + {"owner/shared": entry}, "owner/shared", "different display" + ) is entry diff --git a/tests/test_portfolio_security_gate.py b/tests/test_portfolio_security_gate.py index 0ee7fc07..b3985f40 100644 --- a/tests/test_portfolio_security_gate.py +++ b/tests/test_portfolio_security_gate.py @@ -19,6 +19,9 @@ def _project( alerts_available: bool = True, critical: int = 0, high: int = 0, + code_critical: int = 0, + code_high: int = 0, + secrets: int = 0, risk_tier: str = "baseline", ) -> dict: return { @@ -26,8 +29,19 @@ def _project( "risk": {"risk_tier": risk_tier}, "security": { "alerts_available": alerts_available, + "cohort_member": True, + "coverage_state": "complete" if alerts_available else "unknown", + "providers": { + provider: { + "state": "observed" if alerts_available else "not_requested" + } + for provider in ("dependabot", "code_scanning", "secret_scanning") + }, "dependabot_critical": critical, "dependabot_high": high, + "code_scanning_critical": code_critical, + "code_scanning_high": code_high, + "secret_scanning_open": secrets, }, } @@ -47,7 +61,9 @@ def test_security_gate_passes_when_scanned_repos_are_clear() -> None: 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) + assert "All required-cohort repos are clear" in render_security_gate_markdown( + report + ) def test_security_gate_passes_when_snapshot_is_fresh_enough() -> None: @@ -116,8 +132,32 @@ def test_security_gate_fails_and_ranks_open_high_critical_repos() -> None: 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 + assert "| Critical | elevated | 1 | 0 | 0 |" in rendered + assert "| HighOnly | moderate | 0 | 2 | 0 |" in rendered + + +@pytest.mark.parametrize( + ("kwargs", "expected_critical", "expected_high", "expected_secrets"), + [ + ({"code_critical": 1}, 1, 0, 0), + ({"code_high": 2}, 0, 2, 0), + ({"secrets": 1}, 0, 0, 1), + ], +) +def test_security_gate_fails_on_each_non_dependabot_provider( + kwargs: dict[str, int], + expected_critical: int, + expected_high: int, + expected_secrets: int, +) -> None: + report = build_security_gate_report( + {"projects": [_project("ProviderAlert", **kwargs)]} + ) + + assert report.status == "fail" + assert report.total_open_critical == expected_critical + assert report.total_open_high == expected_high + assert report.total_open_secrets == expected_secrets def test_security_gate_treats_missing_overlay_as_unknown_not_pass() -> None: @@ -133,7 +173,24 @@ def test_security_gate_treats_missing_overlay_as_unknown_not_pass() -> None: 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) + assert "security coverage is missing or incomplete" in render_security_gate_markdown( + report + ) + + +@pytest.mark.parametrize("coverage_state", ["partial", "stale", "unknown"]) +def test_security_gate_fails_closed_on_incomplete_required_cohort( + coverage_state: str, +) -> None: + project = _project("Incomplete") + project["security"]["coverage_state"] = coverage_state + project["security"]["alerts_available"] = False + + report = build_security_gate_report({"projects": [project]}) + + assert report.passed is False + assert report.status in {"unknown", "stale"} + assert report.complete_count == 0 def test_security_gate_subcommand_parses() -> None: diff --git a/tests/test_portfolio_truth.py b/tests/test_portfolio_truth.py index 42802eed..c64624a8 100644 --- a/tests/test_portfolio_truth.py +++ b/tests/test_portfolio_truth.py @@ -391,7 +391,7 @@ def test_truth_snapshot_respects_declared_and_derived_fields( assert gamma.identity.section_marker == "iOS Projects" assert gamma.derived.stack == ["Swift"] - assert result.snapshot.schema_version == "0.10.0" + assert result.snapshot.schema_version == "0.11.0" assert result.snapshot.derivation_policy_version == "portfolio_attention.v2" assert result.snapshot.inputs["catalog"]["sha256"] assert result.snapshot.inputs["notion"]["mode"] == "unavailable" @@ -418,11 +418,26 @@ def test_truth_snapshot_respects_declared_and_derived_fields( assert set(rollups["security"]) == { "scanned_count", "unavailable_count", + "complete_repo_count", + "partial_repo_count", + "stale_count", + "unknown_count", + "cohort_repository_count", + "cohort_complete_count", + "cohort_partial_count", + "cohort_stale_count", + "cohort_unknown_count", + "dependabot_observed_count", + "code_scanning_observed_count", + "secret_scanning_observed_count", "coverage_state", "repos_with_open_high_critical", "total_open_high", "total_open_critical", } + assert rollups["security"]["cohort_repository_count"] == 1 + assert rollups["security"]["cohort_unknown_count"] == 1 + assert rollups["security"]["cohort_complete_count"] == 0 assert set(rollups["decision"]) == { "decision_needed_count", "default_attention_count", @@ -771,7 +786,7 @@ def test_build_security_fields_none_is_unscanned() -> None: fields = _build_security_fields(None) assert fields.alerts_available is False assert fields.open_high_critical == 0 - assert fields.dependabot_critical == 0 + assert fields.dependabot_critical is None def test_build_security_fields_unavailable_dependabot_is_not_available() -> None: @@ -784,21 +799,97 @@ def test_build_security_fields_unavailable_dependabot_is_not_available() -> None } ) assert fields.alerts_available is False - assert fields.dependabot_high == 0 + assert fields.dependabot_high is None -def test_build_security_fields_scanned_clean_is_available_with_zero_counts() -> None: - # A repo whose Dependabot scan succeeded with zero open alerts must read as - # scanned-and-clean (available=True), distinct from an unscanned repo. +def test_dependabot_only_clean_is_partial_not_combined_security_coverage() -> None: + # A clean Dependabot observation must not stand in for combined GitHub + # security coverage when code and secret scanning were not observed. from src.portfolio_truth_reconcile import _build_security_fields fields = _build_security_fields({"dependabot": {"available": True}}) - assert fields.alerts_available is True + assert fields.alerts_available is False + assert fields.coverage_state == "partial" assert fields.dependabot_high == 0 assert fields.dependabot_critical == 0 + assert fields.code_scanning_high is None + assert fields.secret_scanning_open is None assert fields.open_high_critical == 0 +def test_receipt_partial_provider_coverage_emits_explicit_denominators( + portfolio_workspace: Path, + portfolio_catalog: Path, + legacy_registry: Path, +) -> None: + now = datetime.fromtimestamp(1_700_200_000, tz=timezone.utc) + alpha_path = portfolio_workspace / "Alpha" + subprocess.run(["git", "init"], cwd=alpha_path, capture_output=True, check=True) + subprocess.run( + ["git", "remote", "add", "origin", "https://github.com/d/Alpha.git"], + cwd=alpha_path, + capture_output=True, + check=True, + ) + observed = { + "state": "observed", + "observed_at": now.isoformat(), + "http_status": 200, + "pagination_complete": True, + "counts": {"critical": 0, "high": 0, "medium": 0, "low": 0}, + } + security = { + "d/Alpha": { + "repo_full_name": "d/Alpha", + "cohort_member": True, + "cohort_policy": "portfolio-default-attention-v1", + "receipt_schema_version": "GitHubSecurityCoverageReceiptV1", + "receipt_state": "fresh", + "source_produced_at": now.isoformat(), + "providers": { + "dependabot": observed, + "code_scanning": { + "state": "forbidden", + "counts": None, + "pagination_complete": False, + }, + "secret_scanning": { + "state": "not_requested", + "counts": None, + "pagination_complete": False, + }, + }, + } + } + result = build_portfolio_truth_snapshot( + workspace_root=portfolio_workspace, + catalog_path=portfolio_catalog, + legacy_registry_path=legacy_registry, + include_notion=False, + now=now, + security_alerts_by_name=security, + ) + alpha = next( + project + for project in result.snapshot.projects + if project.identity.display_name == "Alpha" + ) + rollup = result.snapshot.to_dict()["rollups"]["security"] + + assert alpha.security.coverage_state == "partial" + assert alpha.security.dependabot_high == 0 + assert alpha.security.code_scanning_high is None + assert alpha.security.secret_scanning_open is None + assert rollup["cohort_repository_count"] == 1 + assert rollup["complete_repo_count"] == 0 + assert rollup["partial_repo_count"] == 1 + assert rollup["cohort_partial_count"] == 1 + assert rollup["cohort_unknown_count"] == 0 + assert rollup["dependabot_observed_count"] == 1 + assert rollup["code_scanning_observed_count"] == 0 + assert rollup["secret_scanning_observed_count"] == 0 + + def test_security_overlay_populates_and_force_elevates( portfolio_workspace: Path, portfolio_catalog: Path, @@ -837,7 +928,7 @@ def test_security_overlay_populates_and_force_elevates( # A repo with no security entry stays unscanned (overlay is strictly opt-in). calibrate = projects["Calibrate"] assert calibrate.security.alerts_available is False - assert calibrate.security.dependabot_critical == 0 + assert calibrate.security.dependabot_critical is None assert calibrate.risk.security_risk is False # Serialized snapshot carries the security block. From 979301382c7fca0840f2b9affd936ec5ddbd934e Mon Sep 17 00:00:00 2001 From: saagpatel Date: Thu, 16 Jul 2026 21:10:45 -0700 Subject: [PATCH 2/6] fix: block credential-shaped artifact data --- src/cache.py | 24 ++-- src/operator_control_center_artifacts.py | 10 +- tests/test_cache.py | 27 +++++ .../test_operator_control_center_artifacts.py | 109 +++++++++++++++++- 4 files changed, 155 insertions(+), 15 deletions(-) diff --git a/src/cache.py b/src/cache.py index 05d2c34a..5b67bf80 100644 --- a/src/cache.py +++ b/src/cache.py @@ -2,32 +2,40 @@ import hashlib import json +import re import time -from urllib.parse import parse_qsl, urlparse from pathlib import Path from typing import Any +from urllib.parse import parse_qsl, urlparse CACHE_DIR = Path("output/.cache") CACHE_TTL = 3600 # 1 hour _SENSITIVE_FIELD_NAMES = frozenset( { "access_token", - "api_key", - "apikey", - "authorization", + "api_key", + "apikey", + "authorization", "client_secret", "credential", - "password", - "private_key", + "password", + "private_key", "github_token", "secret", "token", } ) +_SENSITIVE_VALUE_PATTERNS = ( + re.compile(r"\bAKIA[0-9A-Z]{16}\b"), + re.compile(r"\bgh[pousr]_[A-Za-z0-9]{36,}\b"), + re.compile(r"\bgithub_pat_[A-Za-z0-9_]{20,}\b"), + re.compile(r"\bxox[bpors]-[A-Za-z0-9-]{10,}\b"), + re.compile(r"-----BEGIN (?:RSA |EC |DSA )?PRIVATE KEY-----"), +) def contains_sensitive_data(value: Any) -> bool: - """Return whether JSON-compatible data contains a credential field.""" + """Return whether JSON-compatible data contains credential data.""" if isinstance(value, dict): return any( str(key).lower() in _SENSITIVE_FIELD_NAMES or contains_sensitive_data(item) @@ -37,6 +45,8 @@ def contains_sensitive_data(value: Any) -> bool: return any(contains_sensitive_data(item) for item in value) if isinstance(value, tuple): return any(contains_sensitive_data(item) for item in value) + if isinstance(value, str): + return any(pattern.search(value) for pattern in _SENSITIVE_VALUE_PATTERNS) return False diff --git a/src/operator_control_center_artifacts.py b/src/operator_control_center_artifacts.py index 57364d32..c8d46323 100644 --- a/src/operator_control_center_artifacts.py +++ b/src/operator_control_center_artifacts.py @@ -60,6 +60,8 @@ def write_control_center_artifacts( report_reference: str, diff_dict: dict | None = None, ) -> tuple[Path, Path, Path, Path, dict]: + if contains_sensitive_data(report_data) or contains_sensitive_data(snapshot): + raise ValueError("control-center artifacts must not persist credential fields") filter_snapshot_for_default_view(snapshot) json_path, md_path = control_center_paths(output_dir, username, generated_at) snapshot.setdefault("operator_summary", {})["control_center_reference"] = str(json_path) @@ -75,20 +77,20 @@ def write_control_center_artifacts( report_reference=report_reference, generated_at=generated_at.isoformat(), ) + payload = control_center_artifact_payload(report_data, snapshot) + payload["weekly_command_center_digest_v1"] = weekly_digest + if contains_sensitive_data(payload) or contains_sensitive_data(snapshot): + raise ValueError("control-center artifacts must not persist credential fields") weekly_json, weekly_md = write_weekly_command_center_artifacts( output_dir, username=username, generated_at=generated_at, digest=weekly_digest, ) - payload = control_center_artifact_payload(report_data, snapshot) - payload["weekly_command_center_digest_v1"] = weekly_digest payload["weekly_command_center_reference"] = { "json_path": str(weekly_json), "markdown_path": str(weekly_md), } - if contains_sensitive_data(payload) or contains_sensitive_data(snapshot): - raise ValueError("control-center artifacts must not persist credential fields") json_path.write_text(json.dumps(payload, indent=2)) # codeql[py/clear-text-storage-sensitive-data] guarded above md_path.write_text(render_control_center_markdown(snapshot, username, generated_at.isoformat())) # codeql[py/clear-text-storage-sensitive-data] guarded above return json_path, md_path, weekly_json, weekly_md, payload diff --git a/tests/test_cache.py b/tests/test_cache.py index 0cb0fecd..0bfcd891 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -65,6 +65,33 @@ def test_put_skips_payloads_that_contain_credentials(self, tmp_path): assert not cache._path(url, {"access_token": "secret-param", "per_page": "10"}).exists() + def test_put_skips_credential_shaped_value_under_benign_key(self, tmp_path): + cache = ResponseCache(cache_dir=tmp_path / "cache", ttl=3600) + url = "https://api.github.com/repos/user/repo" + response = {"note": "ghp_" + ("a" * 36)} + + cache.put(url, None, response) + + assert not cache._path(url, None).exists() + + def test_put_skips_credential_inside_serialized_payload(self, tmp_path): + cache = ResponseCache(cache_dir=tmp_path / "cache", ttl=3600) + url = "https://api.github.com/repos/user/repo" + response = json.dumps({"note": "github_pat_" + ("a" * 40)}) + + cache.put(url, None, response) + + assert not cache._path(url, None).exists() + + def test_put_skips_private_key_material_under_benign_key(self, tmp_path): + cache = ResponseCache(cache_dir=tmp_path / "cache", ttl=3600) + url = "https://api.github.com/repos/user/repo" + response = {"content": "-----BEGIN PRIVATE KEY-----\nnot-a-real-key"} + + cache.put(url, None, response) + + assert not cache._path(url, None).exists() + def test_put_preserves_noncredential_secret_scanning_shape(self, tmp_path): cache = ResponseCache(cache_dir=tmp_path / "cache", ttl=3600) url = "https://api.github.com/repos/user/repo" diff --git a/tests/test_operator_control_center_artifacts.py b/tests/test_operator_control_center_artifacts.py index 33314d8c..320ce941 100644 --- a/tests/test_operator_control_center_artifacts.py +++ b/tests/test_operator_control_center_artifacts.py @@ -7,16 +7,29 @@ from src import operator_control_center_artifacts as artifacts -def test_control_center_artifacts_reject_credential_payload_before_writing(tmp_path, monkeypatch): +def _stub_artifact_dependencies(tmp_path, monkeypatch, *, weekly_digest=None): json_path = tmp_path / "control.json" md_path = tmp_path / "control.md" + weekly_writes = [] monkeypatch.setattr(artifacts, "control_center_paths", lambda *_args: (json_path, md_path)) monkeypatch.setattr(artifacts, "load_latest_portfolio_truth", lambda *_args: (None, {})) - monkeypatch.setattr(artifacts, "build_weekly_command_center_digest", lambda *_args, **_kwargs: {}) monkeypatch.setattr( artifacts, - "write_weekly_command_center_artifacts", - lambda *_args, **_kwargs: (tmp_path / "weekly.json", tmp_path / "weekly.md"), + "build_weekly_command_center_digest", + lambda *_args, **_kwargs: weekly_digest or {}, + ) + + def write_weekly(*_args, **_kwargs): + weekly_writes.append(True) + return tmp_path / "weekly.json", tmp_path / "weekly.md" + + monkeypatch.setattr(artifacts, "write_weekly_command_center_artifacts", write_weekly) + return json_path, md_path, weekly_writes + + +def test_control_center_artifacts_reject_credential_payload_before_writing(tmp_path, monkeypatch): + json_path, md_path, weekly_writes = _stub_artifact_dependencies( + tmp_path, monkeypatch ) monkeypatch.setattr(artifacts, "control_center_artifact_payload", lambda *_args: {"token": "secret"}) @@ -27,3 +40,91 @@ def test_control_center_artifacts_reject_credential_payload_before_writing(tmp_p assert not json_path.exists() assert not md_path.exists() + assert weekly_writes == [] + + +def test_control_center_artifacts_reject_credential_value_before_any_write( + tmp_path, monkeypatch +): + json_path, md_path, weekly_writes = _stub_artifact_dependencies( + tmp_path, monkeypatch + ) + token = "ghp_" + ("a" * 36) + + with pytest.raises(ValueError, match="must not persist credential fields"): + artifacts.write_control_center_artifacts( + {"summary": token}, + {}, + tmp_path, + username="user", + generated_at=datetime.now(timezone.utc), + report_reference="report", + ) + + assert not json_path.exists() + assert not md_path.exists() + assert weekly_writes == [] + + +def test_control_center_artifacts_reject_sensitive_derived_digest_before_writing( + tmp_path, monkeypatch +): + token = "github_pat_" + ("a" * 40) + json_path, md_path, weekly_writes = _stub_artifact_dependencies( + tmp_path, + monkeypatch, + weekly_digest={"summary": token}, + ) + monkeypatch.setattr( + artifacts, + "control_center_artifact_payload", + lambda *_args: {}, + ) + + with pytest.raises(ValueError, match="must not persist credential fields"): + artifacts.write_control_center_artifacts( + {}, + {}, + tmp_path, + username="user", + generated_at=datetime.now(timezone.utc), + report_reference="report", + ) + + assert not json_path.exists() + assert not md_path.exists() + assert weekly_writes == [] + + +def test_control_center_artifacts_preserve_normal_artifact_writes( + tmp_path, monkeypatch +): + json_path, md_path, weekly_writes = _stub_artifact_dependencies( + tmp_path, + monkeypatch, + weekly_digest={"status": "current"}, + ) + monkeypatch.setattr( + artifacts, + "control_center_artifact_payload", + lambda *_args: {"status": "current"}, + ) + monkeypatch.setattr( + artifacts, + "render_control_center_markdown", + lambda *_args: "# Current\n", + ) + + result = artifacts.write_control_center_artifacts( + {"status": "current"}, + {}, + tmp_path, + username="user", + generated_at=datetime.now(timezone.utc), + report_reference="report", + ) + + assert result[0:2] == (json_path, md_path) + assert weekly_writes == [True] + assert json_path.exists() + assert md_path.read_text() == "# Current\n" From e34ae15888ce5e84de7362a91b9778e78424d2e1 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Thu, 16 Jul 2026 21:16:10 -0700 Subject: [PATCH 3/6] fix: document count-only security output --- src/app/security_modes.py | 4 ++-- src/github_security_coverage.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/app/security_modes.py b/src/app/security_modes.py index add3e01d..94495340 100644 --- a/src/app/security_modes.py +++ b/src/app/security_modes.py @@ -85,8 +85,8 @@ def run_security_gate_mode(args: Any) -> None: max_age_hours=getattr(args, "max_age_hours", None), ) if getattr(args, "json", False): - print(json.dumps(report.to_dict(), indent=2)) + print(json.dumps(report.to_dict(), indent=2)) # codeql[py/clear-text-logging-sensitive-data] count-only alert summary else: - print(render_security_gate_markdown(report)) + print(render_security_gate_markdown(report)) # codeql[py/clear-text-logging-sensitive-data] count-only alert summary if not report.passed: raise SystemExit(1) diff --git a/src/github_security_coverage.py b/src/github_security_coverage.py index 3e0b214b..cb579417 100644 --- a/src/github_security_coverage.py +++ b/src/github_security_coverage.py @@ -343,7 +343,7 @@ def _fetch_provider( try: reserve_reached = int(remaining) <= budget.quota_reserve except ValueError: - pass + pass # Missing/malformed quota metadata leaves the conservative request budget in control. if response.status_code == 304: if reserve_reached: budget.stop_reason = "quota_reserve" From 574da4c35f37b30a3f53a5915f04a947e5556008 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Thu, 16 Jul 2026 21:19:04 -0700 Subject: [PATCH 4/6] fix: use effective CodeQL count suppressions --- src/app/security_modes.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/app/security_modes.py b/src/app/security_modes.py index 94495340..ec7edb93 100644 --- a/src/app/security_modes.py +++ b/src/app/security_modes.py @@ -85,8 +85,8 @@ def run_security_gate_mode(args: Any) -> None: max_age_hours=getattr(args, "max_age_hours", None), ) if getattr(args, "json", False): - print(json.dumps(report.to_dict(), indent=2)) # codeql[py/clear-text-logging-sensitive-data] count-only alert summary + print(json.dumps(report.to_dict(), indent=2)) # lgtm [py/clear-text-logging-sensitive-data] count-only alert summary else: - print(render_security_gate_markdown(report)) # codeql[py/clear-text-logging-sensitive-data] count-only alert summary + print(render_security_gate_markdown(report)) # lgtm [py/clear-text-logging-sensitive-data] count-only alert summary if not report.passed: raise SystemExit(1) From a3ab41c6797e4976e0304b624b19926720fd9d25 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Thu, 16 Jul 2026 21:23:29 -0700 Subject: [PATCH 5/6] fix: place CodeQL count suppressions correctly --- src/app/security_modes.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/app/security_modes.py b/src/app/security_modes.py index ec7edb93..4ac9a11d 100644 --- a/src/app/security_modes.py +++ b/src/app/security_modes.py @@ -85,8 +85,10 @@ def run_security_gate_mode(args: Any) -> None: max_age_hours=getattr(args, "max_age_hours", None), ) if getattr(args, "json", False): - print(json.dumps(report.to_dict(), indent=2)) # lgtm [py/clear-text-logging-sensitive-data] count-only alert summary + # codeql[py/clear-text-logging-sensitive-data] Count-only alert summary; no secret values. + print(json.dumps(report.to_dict(), indent=2)) else: - print(render_security_gate_markdown(report)) # lgtm [py/clear-text-logging-sensitive-data] count-only alert summary + # codeql[py/clear-text-logging-sensitive-data] Count-only alert summary; no secret values. + print(render_security_gate_markdown(report)) if not report.passed: raise SystemExit(1) From fcf6806a0be2137279e328564a2d3de539190af5 Mon Sep 17 00:00:00 2001 From: saagpatel Date: Thu, 16 Jul 2026 21:27:44 -0700 Subject: [PATCH 6/6] fix: suppress columned CodeQL count findings --- src/app/security_modes.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/app/security_modes.py b/src/app/security_modes.py index 4ac9a11d..cd76ecb0 100644 --- a/src/app/security_modes.py +++ b/src/app/security_modes.py @@ -85,10 +85,10 @@ def run_security_gate_mode(args: Any) -> None: max_age_hours=getattr(args, "max_age_hours", None), ) if getattr(args, "json", False): - # codeql[py/clear-text-logging-sensitive-data] Count-only alert summary; no secret values. + # lgtm[py/clear-text-logging-sensitive-data] Count-only alert summary; no secret values. print(json.dumps(report.to_dict(), indent=2)) else: - # codeql[py/clear-text-logging-sensitive-data] Count-only alert summary; no secret values. + # lgtm[py/clear-text-logging-sensitive-data] Count-only alert summary; no secret values. print(render_security_gate_markdown(report)) if not report.passed: raise SystemExit(1)