Merge remote-tracking branch 'gitea-ssh/main' into codex/p0-product-manifest-standard-20260629
This commit is contained in:
Your Name
@@ -9,6 +9,7 @@ touches hosts.
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
@@ -18,7 +19,50 @@ 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"
|
||||
_VALIDATION_SCHEMA_VERSION = "credential_escrow_evidence_owner_response_validation_v1"
|
||||
_OWNER_REQUEST_SCOPE = "credential_escrow_evidence_owner_request"
|
||||
_OWNER_RESPONSE_SCHEMA = "awoooi_post_reboot_next_gate_owner_response_v1"
|
||||
_GATE_ID = "credential_escrow_evidence"
|
||||
_REQUIRED_ITEM_IDS = {
|
||||
"restic_repository_password",
|
||||
"offsite_provider_credentials",
|
||||
"break_glass_admin_credentials",
|
||||
"dns_registrar_recovery",
|
||||
"oauth_ai_provider_recovery",
|
||||
}
|
||||
_FORBIDDEN_BOOLEAN_FIELDS = {
|
||||
"runtime_action_requested",
|
||||
"runtime_action_authorized",
|
||||
"host_write_requested",
|
||||
"host_write_authorized",
|
||||
"secret_value_included",
|
||||
"secret_value_collection_allowed",
|
||||
"credential_marker_write_requested",
|
||||
"credential_marker_write_authorized",
|
||||
}
|
||||
_PLACEHOLDER_VALUES = {
|
||||
"",
|
||||
"pending",
|
||||
"todo",
|
||||
"tbd",
|
||||
"n/a",
|
||||
"na",
|
||||
"owner_role_here",
|
||||
"owner_team_here",
|
||||
"decision_reason_here",
|
||||
"redacted_evidence_ref_here",
|
||||
"non_secret_evidence_ref_here",
|
||||
"followup_owner_here",
|
||||
"reviewer_here",
|
||||
}
|
||||
_SECRET_VALUE_PATTERNS = {
|
||||
"authorization_header": re.compile(r"Authorization\s*:", re.IGNORECASE),
|
||||
"bearer_token": re.compile(r"Bearer\s+[A-Za-z0-9._~+/=-]{12,}", re.IGNORECASE),
|
||||
"client_keys": re.compile(r"\bclient\.keys\b", 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),
|
||||
}
|
||||
|
||||
|
||||
def load_latest_credential_escrow_evidence_intake_readiness(
|
||||
@@ -45,6 +89,149 @@ def load_latest_credential_escrow_evidence_intake_readiness(
|
||||
return payload
|
||||
|
||||
|
||||
def validate_credential_escrow_evidence_owner_response(
|
||||
owner_response: dict[str, Any],
|
||||
readiness: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate one redacted P0-005 owner response without persisting it."""
|
||||
current = readiness or load_latest_credential_escrow_evidence_intake_readiness()
|
||||
blockers: list[str] = []
|
||||
sensitive_hits = _find_sensitive_strings(owner_response)
|
||||
forbidden_boolean_hits = _find_forbidden_booleans(owner_response)
|
||||
if owner_response.get("schema_version") != _OWNER_RESPONSE_SCHEMA:
|
||||
blockers.append(f"schema_version={owner_response.get('schema_version')!r}")
|
||||
|
||||
responses = [
|
||||
item
|
||||
for item in _list(owner_response.get("responses"))
|
||||
if isinstance(item, dict) and str(item.get("gate_id") or "") == _GATE_ID
|
||||
]
|
||||
response = responses[0] if responses else {}
|
||||
if not response:
|
||||
blockers.append(f"{_GATE_ID}.response_missing")
|
||||
|
||||
for key in (
|
||||
"owner_role",
|
||||
"owner_team",
|
||||
"decision",
|
||||
"decision_reason",
|
||||
"affected_scope",
|
||||
"followup_owner",
|
||||
):
|
||||
if _is_placeholder(response.get(key)):
|
||||
blockers.append(f"{_GATE_ID}.{key}_missing")
|
||||
decision = str(response.get("decision") or "").strip().lower()
|
||||
if decision not in {"accepted", "needs_supplement", "rejected"}:
|
||||
blockers.append(f"{_GATE_ID}.decision_invalid={decision!r}")
|
||||
elif decision != "accepted":
|
||||
blockers.append(f"{_GATE_ID}.decision_not_accepted={decision!r}")
|
||||
|
||||
evidence_refs = [
|
||||
ref
|
||||
for ref in _list(response.get("redacted_evidence_refs"))
|
||||
if not _is_placeholder(ref)
|
||||
]
|
||||
if not evidence_refs:
|
||||
blockers.append(f"{_GATE_ID}.redacted_evidence_refs_missing")
|
||||
|
||||
escrow_items = [
|
||||
item for item in _list(response.get("escrow_items")) if isinstance(item, dict)
|
||||
]
|
||||
seen_item_ids = {
|
||||
str(item.get("item_id") or "").strip()
|
||||
for item in escrow_items
|
||||
if str(item.get("item_id") or "").strip()
|
||||
}
|
||||
missing_item_ids = sorted(_REQUIRED_ITEM_IDS - seen_item_ids)
|
||||
unknown_item_ids = sorted(seen_item_ids - _REQUIRED_ITEM_IDS)
|
||||
if missing_item_ids:
|
||||
blockers.append(f"{_GATE_ID}.missing_items={missing_item_ids}")
|
||||
for item_id in unknown_item_ids:
|
||||
blockers.append(f"{_GATE_ID}.unknown_item={item_id!r}")
|
||||
accepted_item_ids: list[str] = []
|
||||
for item in escrow_items:
|
||||
item_id = str(item.get("item_id") or "").strip()
|
||||
if item_id not in _REQUIRED_ITEM_IDS:
|
||||
continue
|
||||
item_blocked = False
|
||||
for key in (
|
||||
"non_secret_evidence_ref",
|
||||
"recovery_owner",
|
||||
"reviewer",
|
||||
"last_reviewed_at",
|
||||
):
|
||||
if _is_placeholder(item.get(key)):
|
||||
blockers.append(f"{_GATE_ID}.{item_id}.{key}_missing")
|
||||
item_blocked = True
|
||||
if item.get("contains_secret_value") is not False:
|
||||
blockers.append(f"{_GATE_ID}.{item_id}.contains_secret_value_not_false")
|
||||
item_blocked = True
|
||||
if not item_blocked:
|
||||
accepted_item_ids.append(item_id)
|
||||
|
||||
if forbidden_boolean_hits:
|
||||
status = "rejected_runtime_or_marker_action"
|
||||
elif sensitive_hits:
|
||||
status = "quarantined_sensitive_payload"
|
||||
elif blockers:
|
||||
status = "needs_supplement"
|
||||
else:
|
||||
status = "accepted_for_independent_reviewer_readiness_only"
|
||||
|
||||
owner_response_received_count = 1 if status in {
|
||||
"accepted_for_independent_reviewer_readiness_only",
|
||||
"needs_supplement",
|
||||
} and response else 0
|
||||
owner_response_accepted_count = (
|
||||
1 if status == "accepted_for_independent_reviewer_readiness_only" else 0
|
||||
)
|
||||
return {
|
||||
"schema_version": _VALIDATION_SCHEMA_VERSION,
|
||||
"status": status,
|
||||
"priority": "P0-005",
|
||||
"scope": "credential_escrow_evidence_intake",
|
||||
"source_readiness_status": current.get("status"),
|
||||
"result": {
|
||||
"owner_response_received_count": owner_response_received_count,
|
||||
"owner_response_accepted_count": owner_response_accepted_count,
|
||||
"required_item_count": len(_REQUIRED_ITEM_IDS),
|
||||
"provided_item_count": len(seen_item_ids & _REQUIRED_ITEM_IDS),
|
||||
"accepted_item_count": len(set(accepted_item_ids)),
|
||||
"missing_item_ids": missing_item_ids,
|
||||
"unknown_item_ids": unknown_item_ids,
|
||||
"redacted_evidence_ref_count": len(evidence_refs),
|
||||
"blocker_count": len(blockers),
|
||||
"sensitive_payload_hit_count": len(sensitive_hits),
|
||||
"forbidden_true_field_count": len(forbidden_boolean_hits),
|
||||
"runtime_gate_count": 0,
|
||||
"credential_marker_write_authorized_count": 0,
|
||||
"secret_value_collection_allowed": False,
|
||||
},
|
||||
"blockers": blockers,
|
||||
"sensitive_payload_hits": sensitive_hits,
|
||||
"forbidden_boolean_hits": forbidden_boolean_hits,
|
||||
"operation_boundaries": {
|
||||
"payload_persisted": False,
|
||||
"backup_execution_performed": False,
|
||||
"restore_execution_performed": False,
|
||||
"offsite_sync_execution_performed": False,
|
||||
"credential_marker_write_performed": False,
|
||||
"credential_read_performed": False,
|
||||
"secret_plaintext_read": False,
|
||||
"secret_value_collection_allowed": False,
|
||||
"workflow_trigger_performed": False,
|
||||
"host_or_k8s_write_performed": False,
|
||||
"raw_session_or_sqlite_read_performed": False,
|
||||
"runtime_action_performed": False,
|
||||
},
|
||||
"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],
|
||||
@@ -230,3 +417,67 @@ def _int(value: Any) -> int:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return 0
|
||||
|
||||
|
||||
def _normalized(value: Any) -> str:
|
||||
return "" if value is None else str(value).strip()
|
||||
|
||||
|
||||
def _is_placeholder(value: Any) -> bool:
|
||||
text = _normalized(value)
|
||||
lower = text.lower()
|
||||
if lower in _PLACEHOLDER_VALUES:
|
||||
return True
|
||||
if re.fullmatch(r"<[^<>]+>", text):
|
||||
return True
|
||||
if lower in {"yyyy-mm-dd", "yyyy/mm/dd"}:
|
||||
return True
|
||||
return lower.startswith(
|
||||
(
|
||||
"vault-item-id-for-",
|
||||
"sealed-envelope-id-for-",
|
||||
"recovery-checklist-id-for-",
|
||||
"ticket-id-for-",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _collect_strings(value: Any, path: str = "$") -> list[tuple[str, str]]:
|
||||
strings: list[tuple[str, str]] = []
|
||||
if isinstance(value, str):
|
||||
strings.append((path, value))
|
||||
elif isinstance(value, dict):
|
||||
for key, child in value.items():
|
||||
strings.extend(_collect_strings(child, f"{path}.{key}"))
|
||||
elif isinstance(value, list):
|
||||
for index, child in enumerate(value):
|
||||
strings.extend(_collect_strings(child, f"{path}[{index}]"))
|
||||
return strings
|
||||
|
||||
|
||||
def _find_sensitive_strings(payload: dict[str, Any]) -> list[str]:
|
||||
hits: list[str] = []
|
||||
for path, value in _collect_strings(payload):
|
||||
if path.endswith(".item_id") and value in _REQUIRED_ITEM_IDS:
|
||||
continue
|
||||
if path.endswith(".gate_id") and value == _GATE_ID:
|
||||
continue
|
||||
for label, pattern in _SECRET_VALUE_PATTERNS.items():
|
||||
if pattern.search(value):
|
||||
hits.append(f"{label}_at={path}")
|
||||
break
|
||||
return hits
|
||||
|
||||
|
||||
def _find_forbidden_booleans(value: Any, path: str = "$") -> list[str]:
|
||||
hits: list[str] = []
|
||||
if isinstance(value, dict):
|
||||
for key, child in value.items():
|
||||
child_path = f"{path}.{key}"
|
||||
if key in _FORBIDDEN_BOOLEAN_FIELDS and child is not False:
|
||||
hits.append(f"{child_path}={child!r}")
|
||||
hits.extend(_find_forbidden_booleans(child, child_path))
|
||||
elif isinstance(value, list):
|
||||
for index, child in enumerate(value):
|
||||
hits.extend(_find_forbidden_booleans(child, f"{path}[{index}]"))
|
||||
return hits
|
||||
|
||||
Reference in New Issue
Block a user