feat(api): validate gitea private inventory payloads

apps/api/src/services/gitea_authenticated_inventory_payload_validation.py

@@ -0,0 +1,328 @@
+"""P0-003 Gitea authenticated/admin inventory payload validation.
+
+This service validates one redacted Gitea inventory payload for reviewer
+readiness. It never calls Gitea or GitHub, never stores token values, and never
+writes repos, refs, secrets, workflow state, or runtime state.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+from urllib.parse import parse_qsl, urlsplit
+
+from src.services.gitea_private_inventory_p0_scorecard import (
+    load_latest_gitea_private_inventory_p0_scorecard,
+)
+
+_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"}
+_BLOCKERS_CLEARED_BY_ACCEPTED_PAYLOAD = {
+    "gitea_repo_inventory_status_not_ok",
+    "gitea_visibility_scope_public_only_or_unknown",
+    "gitea_authenticated_inventory_payload_not_accepted",
+}
+
+
+def validate_gitea_authenticated_inventory_payload(
+    inventory_payload: dict[str, Any],
+    scorecard: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Validate one redacted P0-003 inventory payload without persisting it."""
+    current = scorecard or load_latest_gitea_private_inventory_p0_scorecard()
+    blockers: list[str] = []
+    sensitive_hits = _find_sensitive_strings(inventory_payload)
+    forbidden_true_fields = _find_forbidden_true_fields(inventory_payload)
+
+    if inventory_payload.get("schema_version") != _PAYLOAD_SCHEMA_VERSION:
+        blockers.append(f"schema_version_not_{_PAYLOAD_SCHEMA_VERSION}")
+    if inventory_payload.get("status") != "ok":
+        blockers.append("status_not_ok")
+    visibility_scope = str(inventory_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(inventory_payload.get("repos")) if isinstance(repo, dict)
+    ]
+    repo_count = _as_int(inventory_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(inventory_payload.get("coverage_gap_explanation")):
+        blockers.append("coverage_gap_explanation_missing")
+    blockers.extend(
+        _validate_redaction_attestation(inventory_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"
+
+    accepted = status == "accepted_for_private_inventory_review_only"
+    current_rollups = _as_dict(current.get("rollups"))
+    current_blockers = _strings(current.get("active_blockers"))
+    projected_blockers = (
+        [
+            blocker
+            for blocker in current_blockers
+            if blocker not in _BLOCKERS_CLEARED_BY_ACCEPTED_PAYLOAD
+        ]
+        if accepted
+        else current_blockers
+    )
+    current_accepted_payload_count = _as_int(
+        current_rollups.get("accepted_inventory_payload_count")
+    )
+    projected_accepted_payload_count = (
+        max(current_accepted_payload_count, 1) if accepted else current_accepted_payload_count
+    )
+    owner_attestation_accepted_count = _as_int(
+        current_rollups.get("owner_coverage_attestation_accepted_count")
+    )
+
+    return {
+        "schema_version": _SCHEMA_VERSION,
+        "status": status,
+        "priority": "P0-003",
+        "scope": "gitea_authenticated_inventory_payload_validation",
+        "source_scorecard_status": current.get("status"),
+        "result": {
+            "accepted_payload_count": 1 if accepted 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),
+            "current_active_blocker_count": len(current_blockers),
+            "projected_active_blocker_count": len(projected_blockers),
+            "current_accepted_inventory_payload_count": current_accepted_payload_count,
+            "projected_accepted_inventory_payload_count": projected_accepted_payload_count,
+            "projected_gitea_repo_inventory_status": (
+                "ok" if accepted else current_rollups.get("gitea_repo_inventory_status")
+            ),
+            "projected_gitea_visibility_scope": (
+                visibility_scope
+                if accepted
+                else current_rollups.get("gitea_visibility_scope")
+            ),
+            "owner_coverage_attestation_still_required_count": (
+                0 if owner_attestation_accepted_count > 0 else 1
+            ),
+            "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,
+            "github_cli_used": False,
+            "secret_plaintext_read": False,
+            "token_value_collection_allowed": False,
+            "runtime_action_performed": False,
+            "raw_session_or_sqlite_read_performed": False,
+        },
+        "reviewer_readiness": {
+            "schema_version": "gitea_private_inventory_reviewer_readiness_v1",
+            "status": (
+                "ready_for_owner_coverage_attestation_review"
+                if accepted
+                else "not_ready_for_owner_coverage_attestation_review"
+            ),
+            "accepted_repo_count": repo_count if accepted else 0,
+            "projected_active_blocker_count": len(projected_blockers),
+            "projected_remaining_blockers": projected_blockers,
+            "redacted_inventory_receipt_writeback_ready_count": 1 if accepted else 0,
+            "owner_coverage_attestation_still_required_count": (
+                0 if owner_attestation_accepted_count > 0 else 1
+            ),
+            "repo_write_authorized_count": 0,
+            "refs_sync_authorized_count": 0,
+            "github_primary_switch_authorized_count": 0,
+            "runtime_gate_count": 0,
+            "token_value_collection_allowed": False,
+            "payload_persisted": False,
+            "safe_next_step": (
+                "write_redacted_inventory_review_receipt_then_collect_owner_coverage_attestation"
+                if accepted
+                else "supplement_authenticated_or_admin_export_redacted_inventory_payload"
+            ),
+            "blocked_operations": [
+                "store_token_value",
+                "gitea_api_write",
+                "repo_write",
+                "refs_sync",
+                "github_api",
+                "github_primary_switch",
+                "workflow_dispatch",
+                "raw_session_or_sqlite_read",
+            ],
+        },
+        "safe_next_step": (
+            "review_redacted_inventory_payload_then_collect_owner_coverage_attestation"
+            if accepted
+            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")
+        owner = repo.get("owner")
+        if _is_placeholder(owner) and _is_placeholder(_as_dict(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"]
+    return [
+        f"redaction_attestation.{key}_not_true"
+        for key in sorted(_REQUIRED_ATTESTATIONS)
+        if attestation.get(key) is not True
+    ]
+
+
+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 _strings(value: Any) -> list[str]:
+    if not isinstance(value, list):
+        return []
+    return [str(item) for item in value if item is not None]
+
+
+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

apps/api/src/services/gitea_private_inventory_p0_scorecard.py

@@ -40,6 +40,7 @@ def load_latest_gitea_private_inventory_p0_scorecard(
 def _build_payload(scorecard: dict[str, Any], path: Path) -> dict[str, Any]:
     gitea_inventory = _dict(scorecard.get("gitea_inventory"))
     import_acceptance = _dict(scorecard.get("authenticated_import_acceptance"))
+    payload_validation = _dict(scorecard.get("authenticated_payload_validation"))
     coverage_attestation = _dict(scorecard.get("coverage_attestation"))
     owner_response_validation = _dict(scorecard.get("owner_response_validation"))
     product_coverage = _dict(scorecard.get("product_row_coverage"))
@@ -70,6 +71,7 @@ def _build_payload(scorecard: dict[str, Any], path: Path) -> dict[str, Any]:
         },
         "gitea_inventory": gitea_inventory,
         "authenticated_import_acceptance": import_acceptance,
+        "authenticated_payload_validation": payload_validation,
         "coverage_attestation": coverage_attestation,
         "owner_response_validation": owner_response_validation,
         "product_row_coverage": product_coverage,
@@ -89,6 +91,15 @@ def _build_payload(scorecard: dict[str, Any], path: Path) -> dict[str, Any]:
             "accepted_inventory_payload_count": _int(
                 import_acceptance.get("accepted_payload_count")
             ),
+            "authenticated_payload_validation_status": str(
+                payload_validation.get("status") or "unknown"
+            ),
+            "authenticated_payload_validation_accepted_count": _int(
+                payload_validation.get("accepted_payload_count")
+            ),
+            "authenticated_payload_validation_blocker_count": _int(
+                payload_validation.get("blocker_count")
+            ),
             "owner_coverage_attestation_received_count": _int(
                 coverage_attestation.get("received_attestation_count")
             ),