feat(api): validate credential escrow evidence refs

apps/api/src/api/v1/agents.py

@@ -321,6 +321,7 @@ from src.services.backup_restore_drill_approval_package_template import (
 )
 from src.services.credential_escrow_evidence_intake_readiness import (
     load_latest_credential_escrow_evidence_intake_readiness,
+    validate_credential_escrow_evidence_refs,
     validate_credential_escrow_evidence_owner_response,
 )
 from src.services.delivery_closure_workbench import (
@@ -3970,6 +3971,39 @@ async def validate_credential_escrow_evidence_owner_response_packet(
         ) from exc
 
 
+@router.post(
+    "/credential-escrow-evidence-intake-readiness/validate-evidence-refs",
+    response_model=dict[str, Any],
+    summary="驗證 P0-005 credential escrow 脫敏 evidence refs",
+    description=(
+        "接收五個 credential escrow item 的 redacted non-secret evidence refs，"
+        "自動轉成既有 owner response preflight 結構並回傳 reviewer readiness；"
+        "此端點不保存 payload、不讀 credential、不寫 escrow marker、不執行 backup / "
+        "restore / offsite sync、不觸發 workflow、不碰 host/K8s、不讀 raw session/SQLite。"
+    ),
+)
+async def validate_credential_escrow_evidence_refs_packet(
+    evidence_refs_payload: dict[str, Any],
+) -> dict[str, Any]:
+    """Return no-persist validation for simplified P0-005 redacted refs."""
+    try:
+        return await asyncio.to_thread(
+            validate_credential_escrow_evidence_refs,
+            evidence_refs_payload,
+        )
+    except FileNotFoundError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except (json.JSONDecodeError, ValueError) as exc:
+        logger.error("credential_escrow_evidence_refs_invalid", error=str(exc))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="P0-005 credential escrow evidence refs 驗證器無效",
+        ) from exc
+
+
 @router.get(
     "/backup-notification-policy",
     response_model=dict[str, Any],

apps/api/src/services/credential_escrow_evidence_intake_readiness.py

@@ -22,6 +22,9 @@ _DEFAULT_SECURITY_DIR = default_security_dir(Path(__file__))
 _OWNER_REQUEST_FILE = "credential-escrow-evidence-owner-request.snapshot.json"
 _SCHEMA_VERSION = "credential_escrow_evidence_intake_readiness_v1"
 _VALIDATION_SCHEMA_VERSION = "credential_escrow_evidence_owner_response_validation_v1"
+_EVIDENCE_REFS_VALIDATION_SCHEMA_VERSION = (
+    "credential_escrow_evidence_refs_validation_v1"
+)
 _SINGLE_PREFLIGHT_INTAKE_SCHEMA_VERSION = "credential_escrow_single_preflight_intake_v1"
 _OWNER_PACKET_SCHEMA = "awoooi_post_reboot_next_gate_owner_packets_v1"
 _OWNER_REQUEST_SCOPE = "credential_escrow_evidence_owner_request"
@@ -286,6 +289,107 @@ def validate_credential_escrow_evidence_owner_response(
     }
 
 
+def validate_credential_escrow_evidence_refs(
+    evidence_refs_payload: dict[str, Any],
+    readiness: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Validate simplified redacted evidence refs without persistence or marker writes."""
+    current = readiness or load_latest_credential_escrow_evidence_intake_readiness()
+    source_sensitive_hits = _find_sensitive_strings(evidence_refs_payload)
+    source_forbidden_boolean_hits = _find_forbidden_booleans(evidence_refs_payload)
+    owner_response = _owner_response_from_evidence_refs(evidence_refs_payload)
+    validation = validate_credential_escrow_evidence_owner_response(
+        owner_response,
+        readiness=current,
+    )
+    rollups = _dict(current.get("rollups"))
+    submissions = _normalized_evidence_ref_submissions(evidence_refs_payload)
+    provided_item_ids = sorted(
+        {
+            str(item.get("item_id") or "").strip()
+            for item in submissions
+            if str(item.get("item_id") or "").strip() in _REQUIRED_ITEM_IDS
+        }
+    )
+
+    status = str(validation.get("status") or "")
+    if source_forbidden_boolean_hits:
+        status = "rejected_runtime_or_marker_action"
+    elif source_sensitive_hits:
+        status = "quarantined_sensitive_payload"
+
+    result = dict(_dict(validation.get("result")))
+    result.update(
+        {
+            "simplified_evidence_ref_submission_count": len(submissions),
+            "provided_required_item_ids": provided_item_ids,
+            "missing_required_item_ids": sorted(
+                _REQUIRED_ITEM_IDS - set(provided_item_ids)
+            ),
+            "source_sensitive_payload_hit_count": len(source_sensitive_hits),
+            "source_forbidden_true_field_count": len(source_forbidden_boolean_hits),
+            "payload_persisted": False,
+            "runtime_gate_count": 0,
+            "credential_marker_write_authorized_count": 0,
+            "secret_value_collection_allowed": False,
+        }
+    )
+    if status != validation.get("status"):
+        result["owner_response_accepted_count"] = 0
+        result["projected_effective_escrow_missing_count"] = _int(
+            rollups.get("effective_escrow_missing_count")
+        )
+
+    reviewer_readiness = dict(_dict(validation.get("reviewer_readiness")))
+    if status != validation.get("status"):
+        reviewer_readiness.update(
+            {
+                "status": "not_ready_for_reviewer_acceptance_writeback",
+                "redacted_receipt_writeback_ready_count": 0,
+                "projected_effective_escrow_missing_count": _int(
+                    rollups.get("effective_escrow_missing_count")
+                ),
+            }
+        )
+
+    return {
+        "schema_version": _EVIDENCE_REFS_VALIDATION_SCHEMA_VERSION,
+        "status": status,
+        "priority": "P0-005",
+        "scope": "credential_escrow_evidence_refs_intake",
+        "source_readiness_status": current.get("status"),
+        "result": result,
+        "blockers": validation.get("blockers", []),
+        "sensitive_payload_hits": sorted(
+            set(_list(validation.get("sensitive_payload_hits")) + source_sensitive_hits)
+        ),
+        "forbidden_boolean_hits": sorted(
+            set(
+                _list(validation.get("forbidden_boolean_hits"))
+                + source_forbidden_boolean_hits
+            )
+        ),
+        "operation_boundaries": _dict(validation.get("operation_boundaries")),
+        "reviewer_readiness": reviewer_readiness,
+        "source_owner_response_validation": {
+            "schema_version": validation.get("schema_version"),
+            "status": validation.get("status"),
+            "blocker_count": len(_list(validation.get("blockers"))),
+            "accepted_item_count": _dict(validation.get("result")).get(
+                "accepted_item_count"
+            ),
+            "projected_effective_escrow_missing_count": _dict(
+                validation.get("result")
+            ).get("projected_effective_escrow_missing_count"),
+        },
+        "safe_next_step": (
+            "independent_reviewer_acceptance_then_marker_dry_run"
+            if status == "accepted_for_independent_reviewer_readiness_only"
+            else "supplement_redacted_non_secret_evidence_refs_without_secret_values"
+        ),
+    }
+
+
 def _build_payload(
     owner_request: dict[str, Any],
     matrix: dict[str, Any],
@@ -412,6 +516,13 @@ def _build_single_preflight_intake(generated_at: Any) -> dict[str, Any]:
         "owner_response_skeleton": _build_single_preflight_owner_response_skeleton(
             generated_at,
         ),
+        "evidence_refs_validation_endpoint": (
+            "/api/v1/agents/credential-escrow-evidence-intake-readiness/"
+            "validate-evidence-refs"
+        ),
+        "evidence_refs_validation_mode": (
+            "validate_redacted_non_secret_refs_only_no_persist_no_marker_write"
+        ),
         "single_preflight_command": (
             "python3 scripts/reboot-recovery/post-reboot-owner-response-preflight.py "
             "--owner-packet-file <owner-packet.json> "
@@ -540,6 +651,115 @@ def _build_single_preflight_owner_response_skeleton(
     }
 
 
+def _owner_response_from_evidence_refs(payload: dict[str, Any]) -> dict[str, Any]:
+    submissions = _normalized_evidence_ref_submissions(payload)
+    defaults = _dict(payload.get("defaults"))
+    owner_role = _field(payload, defaults, "owner_role", "owner_role_here")
+    owner_team = _field(payload, defaults, "owner_team", "owner_team_here")
+    reviewer = _field(payload, defaults, "reviewer", "reviewer_here")
+    last_reviewed_at = _field(payload, defaults, "last_reviewed_at", "pending")
+    followup_owner = _field(payload, defaults, "followup_owner", owner_role)
+    recovery_owner = _field(payload, defaults, "recovery_owner", owner_role)
+    evidence_refs = _strings(payload.get("redacted_evidence_refs"))
+    if not evidence_refs:
+        evidence_refs = [
+            str(item.get("non_secret_evidence_ref") or "").strip()
+            for item in submissions
+            if not _is_placeholder(item.get("non_secret_evidence_ref"))
+        ]
+    item_by_id = {
+        str(item.get("item_id") or "").strip(): item
+        for item in submissions
+        if str(item.get("item_id") or "").strip()
+    }
+    return {
+        "schema_version": _OWNER_RESPONSE_SCHEMA,
+        "generated_at": payload.get("generated_at"),
+        "responses": [
+            {
+                "gate_id": _GATE_ID,
+                "owner_role": owner_role,
+                "owner_team": owner_team,
+                "decision": str(payload.get("decision") or "accepted"),
+                "decision_reason": str(
+                    payload.get("decision_reason")
+                    or "validated redacted non-secret evidence refs only"
+                ),
+                "affected_scope": str(
+                    payload.get("affected_scope")
+                    or "P0-005 DR credential escrow evidence"
+                ),
+                "redacted_evidence_refs": evidence_refs,
+                "followup_owner": followup_owner,
+                "runtime_action_requested": False,
+                "runtime_action_authorized": False,
+                "host_write_requested": False,
+                "host_write_authorized": False,
+                "secret_value_included": False,
+                "secret_value_collection_allowed": False,
+                "credential_marker_write_requested": False,
+                "credential_marker_write_authorized": False,
+                "escrow_items": [
+                    _owner_response_escrow_item(
+                        item_id,
+                        _dict(item_by_id.get(item_id)),
+                        recovery_owner,
+                        reviewer,
+                        last_reviewed_at,
+                    )
+                    for item_id in _REQUIRED_ITEM_ORDER
+                ],
+            }
+        ],
+    }
+
+
+def _owner_response_escrow_item(
+    item_id: str,
+    submission: dict[str, Any],
+    recovery_owner: str,
+    reviewer: str,
+    last_reviewed_at: str,
+) -> dict[str, Any]:
+    return {
+        "item_id": item_id,
+        "non_secret_evidence_ref": str(
+            submission.get("non_secret_evidence_ref") or "non_secret_evidence_ref_here"
+        ),
+        "recovery_owner": str(submission.get("recovery_owner") or recovery_owner),
+        "reviewer": str(submission.get("reviewer") or reviewer),
+        "last_reviewed_at": str(
+            submission.get("last_reviewed_at") or last_reviewed_at
+        ),
+        "contains_secret_value": submission.get("contains_secret_value") is True,
+    }
+
+
+def _normalized_evidence_ref_submissions(payload: dict[str, Any]) -> list[dict[str, Any]]:
+    raw = payload.get("evidence_refs")
+    if raw is None:
+        raw = payload.get("escrow_items")
+    if isinstance(raw, dict):
+        return [
+            {"item_id": str(item_id), "non_secret_evidence_ref": str(ref)}
+            for item_id, ref in raw.items()
+        ]
+    return [
+        _dict(item)
+        for item in _list(raw)
+        if isinstance(item, dict)
+    ]
+
+
+def _field(
+    payload: dict[str, Any],
+    defaults: dict[str, Any],
+    key: str,
+    fallback: str,
+) -> str:
+    return str(payload.get(key) or defaults.get(key) or fallback)
+
+
 def _normalize_item(item: Any) -> dict[str, Any]:
     data = _dict(item)
     return {