feat(governance): expose credential escrow intake readiness
This commit is contained in:
Your Name
@@ -0,0 +1,232 @@
|
||||
"""P0-005 credential escrow evidence intake readiness.
|
||||
|
||||
This read-only projection turns the Backup / DR escrow blocker into a direct
|
||||
API contract. It only reads committed, redacted metadata snapshots; it never
|
||||
reads credential values, writes escrow markers, triggers backups/restores, or
|
||||
touches hosts.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from src.services.backup_dr_readiness_matrix import load_latest_backup_dr_readiness_matrix
|
||||
from src.services.snapshot_paths import default_security_dir
|
||||
|
||||
_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"
|
||||
_OWNER_REQUEST_SCOPE = "credential_escrow_evidence_owner_request"
|
||||
|
||||
|
||||
def load_latest_credential_escrow_evidence_intake_readiness(
|
||||
security_dir: Path | None = None,
|
||||
backup_dr_readiness_matrix: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Load P0-005 credential escrow evidence intake readiness."""
|
||||
directory = security_dir or _DEFAULT_SECURITY_DIR
|
||||
path = directory / _OWNER_REQUEST_FILE
|
||||
with path.open(encoding="utf-8") as handle:
|
||||
owner_request = json.load(handle)
|
||||
|
||||
if not isinstance(owner_request, dict):
|
||||
raise ValueError(f"{path}: expected JSON object")
|
||||
matrix = backup_dr_readiness_matrix or load_latest_backup_dr_readiness_matrix()
|
||||
if not isinstance(matrix, dict):
|
||||
raise ValueError("backup_dr_readiness_matrix: expected JSON object")
|
||||
|
||||
_require_owner_request(owner_request, str(path))
|
||||
_require_backup_rollups(matrix, "backup_dr_readiness_matrix")
|
||||
payload = _build_payload(owner_request, matrix, path)
|
||||
_require_operation_boundaries(payload, str(path))
|
||||
_require_rollup_consistency(payload, str(path))
|
||||
return payload
|
||||
|
||||
|
||||
def _build_payload(
|
||||
owner_request: dict[str, Any],
|
||||
matrix: dict[str, Any],
|
||||
owner_request_path: Path,
|
||||
) -> dict[str, Any]:
|
||||
rollups = _dict(matrix.get("rollups"))
|
||||
missing_items = [_normalize_item(item) for item in _list(owner_request.get("missing_items"))]
|
||||
blocked_items = [
|
||||
item["item_id"]
|
||||
for item in missing_items
|
||||
if item.get("status") != "verified"
|
||||
]
|
||||
effective_missing = _int(rollups.get("credential_escrow_effective_missing_count"))
|
||||
forbidden_values = _strings(owner_request.get("forbidden_values"))
|
||||
status = str(
|
||||
rollups.get("credential_escrow_intake_status")
|
||||
or "blocked_waiting_non_secret_credential_escrow_evidence"
|
||||
)
|
||||
if effective_missing == 0 and not blocked_items:
|
||||
status = "ready_for_escrow_marker_review"
|
||||
|
||||
payload = {
|
||||
"schema_version": _SCHEMA_VERSION,
|
||||
"generated_at": owner_request.get("generated_at"),
|
||||
"priority": "P0-005",
|
||||
"scope": "credential_escrow_evidence_intake",
|
||||
"status": status,
|
||||
"readback": {
|
||||
"workplan_id": "P0-005",
|
||||
"workplan_title": "產品資料與備份 contract",
|
||||
"source_owner_request_ref": (
|
||||
f"docs/security/{owner_request_path.name}"
|
||||
),
|
||||
"source_backup_dr_readiness_matrix_ref": (
|
||||
"docs/evaluations/backup_dr_readiness_matrix_2026-06-04.json"
|
||||
),
|
||||
"safe_next_step": (
|
||||
"collect_redacted_non_secret_evidence_refs_then_rerun_preflight"
|
||||
),
|
||||
},
|
||||
"required_evidence_items": missing_items,
|
||||
"forbidden_values": forbidden_values,
|
||||
"next_actions": [
|
||||
"collect_redacted_non_secret_evidence_refs_for_all_missing_items",
|
||||
"rerun_owner_response_preflight_without_secret_values",
|
||||
"keep_credential_marker_write_closed_until_preflight_accepts_all_items",
|
||||
],
|
||||
"blocked_operations": [
|
||||
"credential_marker_write",
|
||||
"credential_read",
|
||||
"secret_plaintext_export",
|
||||
"restore_execution",
|
||||
"offsite_sync_execution",
|
||||
"backup_execution",
|
||||
"workflow_dispatch",
|
||||
"host_or_k8s_write",
|
||||
],
|
||||
"rollups": {
|
||||
"required_item_count": len(missing_items),
|
||||
"missing_item_count": len(blocked_items),
|
||||
"blocked_item_ids": blocked_items,
|
||||
"effective_escrow_missing_count": effective_missing,
|
||||
"owner_response_received_count": _int(
|
||||
rollups.get("credential_escrow_owner_response_received_count")
|
||||
),
|
||||
"owner_response_accepted_count": _int(
|
||||
rollups.get("credential_escrow_owner_response_accepted_count")
|
||||
),
|
||||
"runtime_gate_count": _int(rollups.get("credential_escrow_runtime_gate_count")),
|
||||
"credential_marker_write_authorized_count": _int(
|
||||
rollups.get("credential_marker_write_authorized_count")
|
||||
),
|
||||
"secret_value_collection_allowed": bool(
|
||||
rollups.get("credential_escrow_secret_value_collection_allowed")
|
||||
),
|
||||
"forbidden_true_field_count": _int(
|
||||
rollups.get("credential_escrow_forbidden_true_field_count")
|
||||
),
|
||||
"forbidden_value_count": len(forbidden_values),
|
||||
},
|
||||
"operation_boundaries": {
|
||||
"read_only_api_allowed": True,
|
||||
"backup_execution_allowed": False,
|
||||
"restore_execution_allowed": False,
|
||||
"offsite_sync_execution_allowed": False,
|
||||
"credential_marker_write_allowed": False,
|
||||
"credential_read_allowed": False,
|
||||
"secret_plaintext_allowed": False,
|
||||
"secret_value_collection_allowed": False,
|
||||
"workflow_trigger_allowed": False,
|
||||
"host_or_k8s_write_allowed": False,
|
||||
"raw_session_or_sqlite_read_allowed": False,
|
||||
},
|
||||
}
|
||||
return payload
|
||||
|
||||
|
||||
def _normalize_item(item: Any) -> dict[str, Any]:
|
||||
data = _dict(item)
|
||||
return {
|
||||
"item_id": str(data.get("item") or data.get("item_id") or ""),
|
||||
"status": str(data.get("status") or "missing"),
|
||||
"allowed_evidence_id_types": _strings(data.get("allowed_evidence_id_types")),
|
||||
"contains_secret_value_allowed": False,
|
||||
"required_non_secret_evidence_ref": True,
|
||||
}
|
||||
|
||||
|
||||
def _require_owner_request(payload: dict[str, Any], label: str) -> None:
|
||||
if payload.get("scope") != _OWNER_REQUEST_SCOPE:
|
||||
raise ValueError(f"{label}: scope must be {_OWNER_REQUEST_SCOPE}")
|
||||
if payload.get("gates", {}).get("secret_value_collection_authorized") is not False:
|
||||
raise ValueError(f"{label}: secret value collection must remain false")
|
||||
if payload.get("gates", {}).get("marker_write_completed") is not False:
|
||||
raise ValueError(f"{label}: marker_write_completed must remain false")
|
||||
if not _list(payload.get("missing_items")):
|
||||
raise ValueError(f"{label}: missing_items must be present")
|
||||
|
||||
|
||||
def _require_backup_rollups(payload: dict[str, Any], label: str) -> None:
|
||||
rollups = _dict(payload.get("rollups"))
|
||||
if rollups.get("credential_escrow_secret_value_collection_allowed") is not False:
|
||||
raise ValueError(f"{label}: secret value collection must remain false")
|
||||
if _int(rollups.get("credential_marker_write_authorized_count")) != 0:
|
||||
raise ValueError(f"{label}: credential marker write must remain closed")
|
||||
|
||||
|
||||
def _require_operation_boundaries(payload: dict[str, Any], label: str) -> None:
|
||||
boundaries = _dict(payload.get("operation_boundaries"))
|
||||
if boundaries.get("read_only_api_allowed") is not True:
|
||||
raise ValueError(f"{label}: read_only_api_allowed must be true")
|
||||
blocked_flags = {
|
||||
"backup_execution_allowed",
|
||||
"restore_execution_allowed",
|
||||
"offsite_sync_execution_allowed",
|
||||
"credential_marker_write_allowed",
|
||||
"credential_read_allowed",
|
||||
"secret_plaintext_allowed",
|
||||
"secret_value_collection_allowed",
|
||||
"workflow_trigger_allowed",
|
||||
"host_or_k8s_write_allowed",
|
||||
"raw_session_or_sqlite_read_allowed",
|
||||
}
|
||||
allowed = sorted(flag for flag in blocked_flags if boundaries.get(flag) is not False)
|
||||
if allowed:
|
||||
raise ValueError(f"{label}: operation boundaries must remain false: {allowed}")
|
||||
|
||||
|
||||
def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
|
||||
items = _list(payload.get("required_evidence_items"))
|
||||
blocked = [
|
||||
str(_dict(item).get("item_id") or "")
|
||||
for item in items
|
||||
if _dict(item).get("status") != "verified"
|
||||
]
|
||||
rollups = _dict(payload.get("rollups"))
|
||||
if rollups.get("required_item_count") != len(items):
|
||||
raise ValueError(f"{label}: required_item_count mismatch")
|
||||
if rollups.get("missing_item_count") != len(blocked):
|
||||
raise ValueError(f"{label}: missing_item_count mismatch")
|
||||
if rollups.get("blocked_item_ids") != blocked:
|
||||
raise ValueError(f"{label}: blocked_item_ids mismatch")
|
||||
if rollups.get("secret_value_collection_allowed") is not False:
|
||||
raise ValueError(f"{label}: secret_value_collection_allowed must be false")
|
||||
if rollups.get("credential_marker_write_authorized_count") != 0:
|
||||
raise ValueError(f"{label}: marker write must remain closed")
|
||||
|
||||
|
||||
def _dict(value: Any) -> dict[str, Any]:
|
||||
return value if isinstance(value, dict) else {}
|
||||
|
||||
|
||||
def _list(value: Any) -> list[Any]:
|
||||
return value if isinstance(value, list) else []
|
||||
|
||||
|
||||
def _strings(value: Any) -> list[str]:
|
||||
return [str(item) for item in _list(value) if str(item)]
|
||||
|
||||
|
||||
def _int(value: Any) -> int:
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return 0
|
||||
Reference in New Issue
Block a user