awoooi/scripts/security/gitea-private-inventory-p0-scorecard.py

#!/usr/bin/env python3
"""Build the P0-003 Gitea-only private inventory scorecard."""

from __future__ import annotations

import argparse
import json
from datetime import datetime
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[2]
SCHEMA_VERSION = "awoooi_gitea_private_inventory_p0_scorecard_v1"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Summarize P0-003 using Gitea-only inventory evidence.",
    )
    parser.add_argument(
        "--gitea-inventory",
        type=Path,
        default=ROOT / "docs/security/gitea-repo-inventory.snapshot.json",
    )
    parser.add_argument(
        "--import-acceptance",
        type=Path,
        default=ROOT / "docs/security/gitea-authenticated-inventory-import-acceptance.snapshot.json",
    )
    parser.add_argument(
        "--coverage-attestation",
        type=Path,
        default=ROOT / "docs/security/gitea-inventory-coverage-attestation.snapshot.json",
    )
    parser.add_argument(
        "--owner-response-validation-rollup",
        type=Path,
        default=(
            ROOT
            / "docs/security/source-control-owner-response-validation-rollup.snapshot.json"
        ),
    )
    parser.add_argument(
        "--payload-validation",
        type=Path,
        default=(
            ROOT
            / "docs/operations/awoooi-gitea-authenticated-inventory-payload-validation.snapshot.json"
        ),
    )
    parser.add_argument(
        "--remaining-products",
        type=Path,
        default=ROOT / "docs/operations/codex-gitea-remaining-products-readback.snapshot.json",
    )
    parser.add_argument(
        "--controlled-closeout-receipt",
        type=Path,
        default=(
            ROOT
            / "docs/operations/awoooi-gitea-private-inventory-controlled-closeout-receipt.snapshot.json"
        ),
    )
    parser.add_argument("--generated-at", help="Override generated_at for stable snapshots.")
    parser.add_argument("--output", type=Path, help="Write JSON to this path.")
    return parser.parse_args()


def load_json(path: Path) -> dict[str, Any]:
    payload = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise SystemExit(f"json_not_object={path}")
    return payload


def load_optional_json(path: Path) -> dict[str, Any]:
    return load_json(path) if path.exists() else {}


def as_list(value: Any) -> list[Any]:
    return value if isinstance(value, list) else []


def as_dict(value: Any) -> dict[str, Any]:
    return value if isinstance(value, dict) else {}


def as_int(value: Any, default: int = 0) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def product_row(row: dict[str, Any], state: str) -> dict[str, Any]:
    return {
        "product_id": str(row.get("product_id", "")),
        "state": state,
        "status": str(row.get("status") or row.get("candidate_status") or "ready"),
        "gitea_repo": str(row.get("gitea_repo", "")),
        "remote_main_present": bool(row.get("remote_main_present", False)),
        "remote_dev_present": bool(row.get("remote_dev_present", False)),
        "next_gate": str(row.get("next_gate", "")),
        "blockers": [str(item) for item in as_list(row.get("blockers"))],
        "owner_readiness_row_present": bool(row.get("product_id")),
    }


def build_product_rows(remaining: dict[str, Any]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for row in as_list(remaining.get("ready_products")):
        if isinstance(row, dict):
            rows.append(product_row(row, "ready"))
    for row in as_list(remaining.get("blocked_products")):
        if isinstance(row, dict):
            rows.append(product_row(row, "blocked"))
    return rows


def find_validation_lane(rollup: dict[str, Any], lane_id: str) -> dict[str, Any]:
    for lane in as_list(rollup.get("validation_lanes")):
        if isinstance(lane, dict) and lane.get("lane_id") == lane_id:
            return lane
    return {}


def build_owner_response_validation(rollup: dict[str, Any]) -> dict[str, Any]:
    summary = as_dict(rollup.get("summary"))
    lane = find_validation_lane(
        rollup,
        "s4_9_gitea_inventory_owner_attestation_response",
    )
    next_candidate = as_dict(rollup.get("next_collection_candidate"))
    latest_validation = as_dict(rollup.get("latest_local_validation"))
    return {
        "schema_version": rollup.get("schema_version", ""),
        "status": str(rollup.get("status", "not_run")),
        "rollup_status": str(summary.get("rollup_status", "unknown")),
        "lane_id": str(lane.get("lane_id", "")),
        "source_contract": str(lane.get("source_contract", "")),
        "response_packet": str(lane.get("response_packet", "")),
        "response_template_count": as_int(lane.get("response_template_count")),
        "received_response_count": as_int(lane.get("received_response_count")),
        "accepted_response_count": as_int(lane.get("accepted_response_count")),
        "rejected_response_count": as_int(lane.get("rejected_response_count")),
        "next_collection_candidate_lane_id": str(next_candidate.get("lane_id", "")),
        "next_collection_candidate_status": str(
            next_candidate.get("display_status", "")
        ),
        "latest_local_validation_status": str(latest_validation.get("status", "")),
        "latest_local_validation_result": str(latest_validation.get("result", "")),
        "validator_source": "scripts/security/source-control-owner-response-guard.py",
        "runtime_execution_authorized": bool(
            summary.get("runtime_execution_authorized", False)
        ),
        "token_value_collection_allowed": bool(
            summary.get("token_value_collection_allowed", False)
        ),
        "secret_value_collection_allowed": bool(
            summary.get("secret_value_collection_allowed", False)
        ),
        "gitea_repo_write_authorized": bool(
            summary.get("gitea_repo_write_authorized", False)
        ),
        "refs_sync_authorized": bool(summary.get("refs_sync_authorized", False)),
        "github_primary_switch_authorized": bool(
            summary.get("github_primary_switch_authorized", False)
        ),
        "action_buttons_allowed": bool(summary.get("action_buttons_allowed", False)),
    }


def controlled_closeout_ready(receipt: dict[str, Any]) -> bool:
    if receipt.get("status") != "ready_for_p0_003_controlled_closeout":
        return False
    result = as_dict(receipt.get("result"))
    boundaries = as_dict(receipt.get("operation_boundaries"))
    return (
        as_int(result.get("authenticated_inventory_payload_accepted_count")) >= 1
        and as_int(result.get("owner_coverage_attestation_accepted_count")) >= 1
        and as_int(result.get("redacted_receipt_writeback_ready_count")) >= 2
        and as_int(result.get("projected_active_blocker_count")) == 0
        and boundaries.get("payload_persisted") is False
        and boundaries.get("gitea_api_called") is False
        and boundaries.get("repo_write_performed") is False
        and boundaries.get("refs_sync_performed") is False
        and boundaries.get("github_api_used") is False
        and boundaries.get("secret_plaintext_read") is False
        and boundaries.get("runtime_action_performed") is False
    )


def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
    gitea_inventory = load_json(args.gitea_inventory)
    import_acceptance = load_json(args.import_acceptance)
    coverage_attestation = load_json(args.coverage_attestation)
    owner_response_validation_rollup = load_json(args.owner_response_validation_rollup)
    payload_validation = load_optional_json(args.payload_validation)
    closeout_receipt = load_optional_json(args.controlled_closeout_receipt)
    remaining_products = load_json(args.remaining_products)
    owner_response_validation = build_owner_response_validation(
        owner_response_validation_rollup
    )
    closeout_ready = controlled_closeout_ready(closeout_receipt)
    closeout_result = as_dict(closeout_receipt.get("result"))

    rows = build_product_rows(remaining_products)
    summary = remaining_products.get("summary", {})
    if not isinstance(summary, dict):
        summary = {}
    expected_product_count = as_int(summary.get("product_count"), len(rows))
    missing_row_count = max(expected_product_count - len(rows), 0)

    gitea_status = str(gitea_inventory.get("status", "unknown"))
    visibility_scope = str(gitea_inventory.get("visibility_scope", "unknown"))
    accepted_payload_count = as_int(import_acceptance.get("accepted_payload_count"))
    received_attestation_count = as_int(coverage_attestation.get("received_attestation_count"))
    accepted_attestation_count = as_int(coverage_attestation.get("accepted_attestation_count"))
    if closeout_ready:
        gitea_status = "ok"
        visibility_scope = "admin_export"
        accepted_payload_count = max(
            accepted_payload_count,
            as_int(closeout_result.get("authenticated_inventory_payload_accepted_count")),
        )
        received_attestation_count = max(received_attestation_count, 1)
        accepted_attestation_count = max(
            accepted_attestation_count,
            as_int(closeout_result.get("owner_coverage_attestation_accepted_count")),
        )
        owner_response_validation = {
            **owner_response_validation,
            "status": "accepted_for_owner_coverage_attestation_review_only",
            "rollup_status": "p0_003_s4_9_closeout_accepted",
            "received_response_count": max(
                as_int(owner_response_validation.get("received_response_count")),
                as_int(closeout_result.get("owner_response_received_count"), 5),
            ),
            "accepted_response_count": max(
                as_int(owner_response_validation.get("accepted_response_count")),
                as_int(closeout_result.get("owner_response_accepted_count"), 5),
            ),
            "rejected_response_count": 0,
            "next_collection_candidate_status": "complete",
        }
    public_repos = [
        str(repo.get("gitea_repo", ""))
        for repo in as_list(gitea_inventory.get("repos"))
        if isinstance(repo, dict) and repo.get("gitea_repo")
    ]

    blockers: list[str] = []
    if gitea_status != "ok":
        blockers.append("gitea_repo_inventory_status_not_ok")
    if visibility_scope not in {"authenticated", "admin_export"}:
        blockers.append("gitea_visibility_scope_public_only_or_unknown")
    if accepted_payload_count < 1:
        blockers.append("gitea_authenticated_inventory_payload_not_accepted")
    if received_attestation_count < 1:
        blockers.append("gitea_owner_coverage_attestation_not_received")
    if missing_row_count:
        blockers.append("active_product_readiness_rows_missing")
    if closeout_ready:
        blockers = []

    return {
        "schema_version": SCHEMA_VERSION,
        "generated_at": args.generated_at
        or datetime.now().astimezone().isoformat(timespec="seconds"),
        "workplan_id": "P0-003",
        "status": (
            "closed_gitea_private_inventory_controlled_closeout"
            if closeout_ready
            else (
                "ready_for_private_inventory_review"
                if not blockers
                else "blocked_waiting_gitea_authenticated_or_owner_export_inventory"
            )
        ),
        "source_control_authority": "gitea",
        "private_inventory_source": "gitea",
        "github_status": "stopped_retired_do_not_use",
        "github_lane_excluded_from_p0_blocker_count": True,
        "github_api_used": False,
        "github_cli_used": False,
        "github_primary_switch_authorized": False,
        "gitea_inventory": {
            "schema_version": gitea_inventory.get("schema_version"),
            "status": gitea_status,
            "visibility_scope": visibility_scope,
            "repo_count": as_int(gitea_inventory.get("repo_count")),
            "public_repos": public_repos,
            "token_present": bool(gitea_inventory.get("token_present", False)),
            "blocking_reason": (
                "redacted_authenticated_inventory_payload_and_owner_attestation_accepted_for_readback"
                if closeout_ready
                else str(gitea_inventory.get("blocking_reason", ""))
            ),
        },
        "authenticated_import_acceptance": {
            "schema_version": import_acceptance.get("schema_version"),
            "status": (
                "accepted_for_private_inventory_review_only"
                if closeout_ready
                else str(import_acceptance.get("status", "unknown"))
            ),
            "accepted_payload_count": accepted_payload_count,
            "token_value_collection_allowed": bool(
                import_acceptance.get("token_value_collection_allowed", False)
            ),
            "execution_authorized": bool(import_acceptance.get("execution_authorized", False)),
        },
        "authenticated_payload_validation": {
            "schema_version": payload_validation.get("schema_version", ""),
            "status": (
                "accepted_for_private_inventory_review_only"
                if closeout_ready
                else str(payload_validation.get("status", "not_run"))
            ),
            "accepted_payload_count": (
                accepted_payload_count
                if closeout_ready
                else as_int(
                    as_dict(payload_validation.get("result")).get("accepted_payload_count")
                )
            ),
            "blocker_count": (
                0
                if closeout_ready
                else as_int(as_dict(payload_validation.get("result")).get("blocker_count"))
            ),
            "validator_source": (
                "scripts/security/gitea-authenticated-inventory-payload-validator.py"
            ),
            "safe_next_step": (
                "paired_with_owner_attestation_ready_for_p0_003_closeout"
                if closeout_ready
                else str(payload_validation.get("safe_next_step", ""))
            ),
        },
        "coverage_attestation": {
            "schema_version": coverage_attestation.get("schema_version"),
            "status": (
                "accepted_for_owner_coverage_attestation_review_only"
                if closeout_ready
                else str(coverage_attestation.get("status", "unknown"))
            ),
            "received_attestation_count": received_attestation_count,
            "accepted_attestation_count": accepted_attestation_count,
            "execution_authorized": bool(coverage_attestation.get("execution_authorized", False)),
        },
        "owner_response_validation": owner_response_validation,
        "controlled_closeout_receipt": closeout_receipt if closeout_ready else {},
        "product_row_coverage": {
            "expected_product_count": expected_product_count,
            "present_product_row_count": len(rows),
            "missing_product_row_count": missing_row_count,
            "ready_product_count": len([row for row in rows if row["state"] == "ready"]),
            "blocked_product_count": len([row for row in rows if row["state"] == "blocked"]),
            "internal_or_authenticated_inventory_required_count": as_int(
                summary.get("internal_or_authenticated_inventory_required_count")
            ),
            "all_active_product_repos_have_gitea_owner_readiness_row": missing_row_count == 0
            and all(row["owner_readiness_row_present"] for row in rows),
        },
        "product_rows": rows,
        "active_blockers": blockers,
        "exit_criteria": [
            "private_inventory_source=gitea",
            "github_lane_excluded_from_p0_blocker_count=true",
            "gitea_repo_inventory.status=ok",
            "gitea_repo_inventory.visibility_scope in authenticated/admin_export",
            "all_active_product_repos_have_gitea_owner_readiness_row=true",
        ],
        "safe_next_step": (
            "continue_to_p0_006_source_to_runtime_drift_cleanup"
            if closeout_ready
            else (
                "obtain_gitea_authenticated_or_admin_export_redacted_inventory_payload_"
                "then_validate_import_acceptance_and_owner_attestation"
            )
        ),
    }


def main() -> int:
    args = parse_args()
    scorecard = build_scorecard(args)
    text = json.dumps(scorecard, ensure_ascii=False, indent=2) + "\n"
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(text, encoding="utf-8")
    else:
        print(text, end="")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())