feat(security): validate gitea inventory payloads

scripts/security/gitea-authenticated-inventory-payload-validator.py

@@ -0,0 +1,271 @@
+#!/usr/bin/env python3
+"""Validate a redacted Gitea authenticated/admin inventory payload.
+
+This is a preflight only. It never calls Gitea, never stores token values, and
+never writes repos, refs, secrets, or runtime state.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+from pathlib import Path
+from typing import Any
+from urllib.parse import parse_qsl, urlsplit
+
+
+SCHEMA_VERSION = "gitea_authenticated_inventory_payload_validation_v1"
+PAYLOAD_SCHEMA_VERSION = "gitea_repo_inventory_v1"
+ACCEPTED_VISIBILITY_SCOPES = {"authenticated", "admin_export"}
+REQUIRED_ATTESTATIONS = {
+    "no_token_value",
+    "no_write_token",
+    "no_webhook_secret",
+    "no_deploy_key_private_key",
+    "no_runner_registration_token",
+    "no_cookie_or_session",
+    "no_gitea_db_dump",
+    "no_git_object_pack",
+}
+FORBIDDEN_TRUE_FIELDS = {
+    "repo_write_allowed",
+    "refs_sync_allowed",
+    "github_primary_switch_authorized",
+    "runtime_execution_authorized",
+    "write_to_gitea",
+    "create_gitea_repo",
+    "delete_or_archive_gitea_repo",
+    "sync_git_refs",
+    "force_push",
+}
+SECRET_PATTERNS = {
+    "authorization_header": re.compile(r"Authorization\s*:", re.IGNORECASE),
+    "bearer_token": re.compile(r"Bearer\s+[A-Za-z0-9._~+/=-]{12,}", re.IGNORECASE),
+    "cookie_header": re.compile(r"\bCookie\s*:", re.IGNORECASE),
+    "password_assignment": re.compile(r"\bpassword\s*[:=]\s*[^,\s]+", re.IGNORECASE),
+    "private_key": re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
+    "token_assignment": re.compile(r"\btoken\s*[:=]\s*[^,\s]+", re.IGNORECASE),
+}
+SECRET_QUERY_KEYS = {"access_token", "auth", "key", "password", "secret", "token"}
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Validate redacted Gitea authenticated/admin inventory payload.",
+    )
+    parser.add_argument(
+        "--input",
+        type=Path,
+        default=Path("docs/security/gitea-repo-inventory.snapshot.json"),
+        help="Payload JSON to validate.",
+    )
+    parser.add_argument("--output", type=Path, help="Write validation JSON here.")
+    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 validate_payload(payload: dict[str, Any]) -> dict[str, Any]:
+    blockers: list[str] = []
+    sensitive_hits = find_sensitive_strings(payload)
+    forbidden_true_fields = find_forbidden_true_fields(payload)
+
+    if payload.get("schema_version") != PAYLOAD_SCHEMA_VERSION:
+        blockers.append(f"schema_version_not_{PAYLOAD_SCHEMA_VERSION}")
+    if payload.get("status") != "ok":
+        blockers.append("status_not_ok")
+    visibility_scope = str(payload.get("visibility_scope") or "")
+    if visibility_scope not in ACCEPTED_VISIBILITY_SCOPES:
+        blockers.append("visibility_scope_not_authenticated_or_admin_export")
+
+    repos = [repo for repo in as_list(payload.get("repos")) if isinstance(repo, dict)]
+    repo_count = as_int(payload.get("repo_count"))
+    if repo_count != len(repos):
+        blockers.append("repo_count_mismatch")
+    if repo_count < 4:
+        blockers.append("repo_count_below_current_public_floor")
+    blockers.extend(validate_repos(repos))
+
+    if is_placeholder(payload.get("coverage_gap_explanation")):
+        blockers.append("coverage_gap_explanation_missing")
+    blockers.extend(validate_redaction_attestation(payload.get("redaction_attestation")))
+
+    if forbidden_true_fields:
+        status = "rejected_execution_request"
+    elif sensitive_hits:
+        status = "quarantined_sensitive_payload"
+    elif blockers:
+        status = "needs_supplement"
+    else:
+        status = "accepted_for_private_inventory_review_only"
+
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "status": status,
+        "priority": "P0-003",
+        "scope": "gitea_authenticated_inventory_payload_validation",
+        "result": {
+            "accepted_payload_count": (
+                1 if status == "accepted_for_private_inventory_review_only" else 0
+            ),
+            "repo_count": repo_count,
+            "visible_repo_count": len(repos),
+            "blocker_count": len(blockers),
+            "sensitive_payload_hit_count": len(sensitive_hits),
+            "forbidden_true_field_count": len(forbidden_true_fields),
+            "token_value_collection_allowed": False,
+            "repo_write_allowed": False,
+            "refs_sync_allowed": False,
+            "github_primary_switch_authorized": False,
+            "runtime_gate_count": 0,
+        },
+        "blockers": blockers,
+        "sensitive_payload_hits": sensitive_hits,
+        "forbidden_true_fields": forbidden_true_fields,
+        "operation_boundaries": {
+            "payload_persisted": False,
+            "gitea_api_called": False,
+            "gitea_write_performed": False,
+            "repo_write_performed": False,
+            "refs_sync_performed": False,
+            "github_api_used": False,
+            "secret_plaintext_read": False,
+            "token_value_collection_allowed": False,
+            "runtime_action_performed": False,
+            "raw_session_or_sqlite_read_performed": False,
+        },
+        "safe_next_step": (
+            "review_redacted_inventory_payload_then_update_gitea_inventory_snapshot"
+            if status == "accepted_for_private_inventory_review_only"
+            else "supplement_authenticated_or_admin_export_redacted_inventory_payload"
+        ),
+    }
+
+
+def validate_repos(repos: list[dict[str, Any]]) -> list[str]:
+    blockers: list[str] = []
+    seen: set[str] = set()
+    for index, repo in enumerate(repos):
+        identity = str(repo.get("full_name") or repo.get("gitea_repo") or "")
+        if not identity:
+            blockers.append(f"repos[{index}].identity_missing")
+        elif identity in seen:
+            blockers.append(f"repos[{index}].identity_duplicate")
+        seen.add(identity)
+        for key in ("name", "default_branch", "clone_url_redacted", "ssh_url_redacted"):
+            if is_placeholder(repo.get(key)):
+                blockers.append(f"repos[{index}].{key}_missing")
+        if is_placeholder(repo.get("owner")) and is_placeholder(as_dict(repo.get("owner")).get("login")):
+            blockers.append(f"repos[{index}].owner_missing")
+        for key in ("private", "archived", "empty"):
+            if not isinstance(repo.get(key), bool):
+                blockers.append(f"repos[{index}].{key}_not_boolean")
+        for key in ("clone_url_redacted", "ssh_url_redacted"):
+            value = str(repo.get(key) or "")
+            if url_has_secret(value):
+                blockers.append(f"repos[{index}].{key}_not_redacted")
+    return blockers
+
+
+def validate_redaction_attestation(value: Any) -> list[str]:
+    attestation = as_dict(value)
+    if not attestation:
+        return ["redaction_attestation_missing"]
+    blockers: list[str] = []
+    for key in sorted(REQUIRED_ATTESTATIONS):
+        if attestation.get(key) is not True:
+            blockers.append(f"redaction_attestation.{key}_not_true")
+    return blockers
+
+
+def find_sensitive_strings(value: Any) -> list[str]:
+    hits: list[str] = []
+
+    def walk(node: Any, path: str) -> None:
+        if isinstance(node, dict):
+            for key, item in node.items():
+                walk(item, f"{path}.{key}" if path else str(key))
+        elif isinstance(node, list):
+            for index, item in enumerate(node):
+                walk(item, f"{path}[{index}]")
+        elif isinstance(node, str):
+            for name, pattern in SECRET_PATTERNS.items():
+                if pattern.search(node):
+                    hits.append(f"{path}:{name}")
+            if url_has_secret(node):
+                hits.append(f"{path}:url_contains_secret_material")
+
+    walk(value, "")
+    return sorted(set(hits))
+
+
+def find_forbidden_true_fields(value: Any) -> list[str]:
+    hits: list[str] = []
+
+    def walk(node: Any, path: str) -> None:
+        if isinstance(node, dict):
+            for key, item in node.items():
+                next_path = f"{path}.{key}" if path else str(key)
+                if key in FORBIDDEN_TRUE_FIELDS and item is True:
+                    hits.append(next_path)
+                walk(item, next_path)
+        elif isinstance(node, list):
+            for index, item in enumerate(node):
+                walk(item, f"{path}[{index}]")
+
+    walk(value, "")
+    return sorted(hits)
+
+
+def url_has_secret(value: str) -> bool:
+    if "://" not in value:
+        return False
+    parsed = urlsplit(value)
+    if parsed.username or parsed.password:
+        return True
+    return any(key.lower() in SECRET_QUERY_KEYS for key, _ in parse_qsl(parsed.query))
+
+
+def is_placeholder(value: Any) -> bool:
+    if value is None:
+        return True
+    if isinstance(value, str):
+        return value.strip().lower() in {"", "pending", "todo", "tbd", "n/a", "na"}
+    return False
+
+
+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) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return 0
+
+
+def main() -> int:
+    args = parse_args()
+    validation = validate_payload(load_json(args.input))
+    text = json.dumps(validation, 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())

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

@@ -33,6 +33,14 @@ def parse_args() -> argparse.Namespace:
         type=Path,
         default=ROOT / "docs/security/gitea-inventory-coverage-attestation.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,
@@ -50,10 +58,18 @@ def load_json(path: Path) -> dict[str, Any]:
     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)
@@ -90,6 +106,7 @@ 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)
+    payload_validation = load_optional_json(args.payload_validation)
     remaining_products = load_json(args.remaining_products)
 
     rows = build_product_rows(remaining_products)
@@ -152,6 +169,20 @@ def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
             ),
             "execution_authorized": bool(import_acceptance.get("execution_authorized", False)),
         },
+        "authenticated_payload_validation": {
+            "schema_version": payload_validation.get("schema_version", ""),
+            "status": str(payload_validation.get("status", "not_run")),
+            "accepted_payload_count": as_int(
+                as_dict(payload_validation.get("result")).get("accepted_payload_count")
+            ),
+            "blocker_count": as_int(
+                as_dict(payload_validation.get("result")).get("blocker_count")
+            ),
+            "validator_source": (
+                "scripts/security/gitea-authenticated-inventory-payload-validator.py"
+            ),
+            "safe_next_step": str(payload_validation.get("safe_next_step", "")),
+        },
         "coverage_attestation": {
             "schema_version": coverage_attestation.get("schema_version"),
             "status": str(coverage_attestation.get("status", "unknown")),

scripts/security/tests/test_gitea_authenticated_inventory_payload_validator.py

@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[3]
+SCRIPT = ROOT / "scripts" / "security" / "gitea-authenticated-inventory-payload-validator.py"
+
+
+def run_validator(path: Path | None = None) -> dict:
+    command = [sys.executable, str(SCRIPT)]
+    if path:
+        command.extend(["--input", str(path)])
+    result = subprocess.run(command, text=True, capture_output=True, check=True)
+    return json.loads(result.stdout)
+
+
+def test_current_public_inventory_stays_needs_supplement() -> None:
+    validation = run_validator()
+
+    assert validation["schema_version"] == "gitea_authenticated_inventory_payload_validation_v1"
+    assert validation["priority"] == "P0-003"
+    assert validation["status"] == "needs_supplement"
+    assert validation["result"]["accepted_payload_count"] == 0
+    assert validation["result"]["token_value_collection_allowed"] is False
+    assert validation["operation_boundaries"]["gitea_write_performed"] is False
+    assert "visibility_scope_not_authenticated_or_admin_export" in validation["blockers"]
+    assert "redaction_attestation_missing" in validation["blockers"]
+
+
+def test_accepts_redacted_admin_export_payload(tmp_path: Path) -> None:
+    payload_path = tmp_path / "gitea-admin-export-redacted.json"
+    payload_path.write_text(json.dumps(valid_payload()), encoding="utf-8")
+
+    validation = run_validator(payload_path)
+
+    assert validation["status"] == "accepted_for_private_inventory_review_only"
+    assert validation["result"]["accepted_payload_count"] == 1
+    assert validation["result"]["repo_count"] == 4
+    assert validation["result"]["runtime_gate_count"] == 0
+    assert validation["operation_boundaries"]["payload_persisted"] is False
+    assert validation["operation_boundaries"]["repo_write_performed"] is False
+
+
+def test_quarantines_secret_material(tmp_path: Path) -> None:
+    payload = valid_payload()
+    payload["repos"][0]["clone_url_redacted"] = "https://user:password@example.test/repo.git"
+    payload_path = tmp_path / "secret-payload.json"
+    payload_path.write_text(json.dumps(payload), encoding="utf-8")
+
+    validation = run_validator(payload_path)
+
+    assert validation["status"] == "quarantined_sensitive_payload"
+    assert validation["result"]["accepted_payload_count"] == 0
+    assert validation["result"]["sensitive_payload_hit_count"] >= 1
+
+
+def test_rejects_execution_request(tmp_path: Path) -> None:
+    payload = valid_payload()
+    payload["repo_write_allowed"] = True
+    payload_path = tmp_path / "execution-request.json"
+    payload_path.write_text(json.dumps(payload), encoding="utf-8")
+
+    validation = run_validator(payload_path)
+
+    assert validation["status"] == "rejected_execution_request"
+    assert validation["result"]["accepted_payload_count"] == 0
+    assert validation["result"]["forbidden_true_field_count"] == 1
+    assert validation["operation_boundaries"]["gitea_write_performed"] is False
+
+
+def valid_payload() -> dict:
+    repos = [
+        repo("wooo/awoooi"),
+        repo("wooo/ewoooc"),
+        repo("wooo/agent-bounty-protocol"),
+        repo("wooo/2026FIFAWorldCup"),
+    ]
+    return {
+        "schema_version": "gitea_repo_inventory_v1",
+        "base_url": "https://gitea.wooo.work",
+        "org": "wooo",
+        "visibility_scope": "admin_export",
+        "token_present": False,
+        "status": "ok",
+        "repo_count": len(repos),
+        "repos": repos,
+        "coverage_gap_explanation": {
+            "public_only_vs_admin_export": "admin export includes all in-scope repos",
+            "internal_110_adjacent_scope": "covered by owner scope decision",
+            "org_user_endpoint_identity": "wooo namespace owner confirmed",
+        },
+        "redaction_attestation": {
+            "no_token_value": True,
+            "no_write_token": True,
+            "no_webhook_secret": True,
+            "no_deploy_key_private_key": True,
+            "no_runner_registration_token": True,
+            "no_cookie_or_session": True,
+            "no_gitea_db_dump": True,
+            "no_git_object_pack": True,
+        },
+    }
+
+
+def repo(full_name: str) -> dict:
+    _, name = full_name.split("/", 1)
+    return {
+        "gitea_repo": full_name,
+        "name": name,
+        "owner": "wooo",
+        "private": False,
+        "empty": False,
+        "archived": False,
+        "default_branch": "main",
+        "clone_url_redacted": f"https://gitea.wooo.work/{full_name}.git",
+        "ssh_url_redacted": f"ssh://gitea.wooo.work/{full_name}.git",
+        "github_repo_candidate": "",
+    }

scripts/security/tests/test_gitea_private_inventory_p0_scorecard.py

@@ -49,6 +49,12 @@ def test_scorecard_preserves_current_gitea_inventory_blocker() -> None:
         "wooo/2026FIFAWorldCup",
     } <= set(scorecard["gitea_inventory"]["public_repos"])
     assert scorecard["authenticated_import_acceptance"]["accepted_payload_count"] == 0
+    assert scorecard["authenticated_payload_validation"]["status"] == "needs_supplement"
+    assert scorecard["authenticated_payload_validation"]["accepted_payload_count"] == 0
+    assert (
+        scorecard["authenticated_payload_validation"]["validator_source"]
+        == "scripts/security/gitea-authenticated-inventory-payload-validator.py"
+    )
     assert scorecard["coverage_attestation"]["received_attestation_count"] == 0
     assert "gitea_repo_inventory_status_not_ok" in scorecard["active_blockers"]
     assert "gitea_authenticated_inventory_payload_not_accepted" in scorecard["active_blockers"]