wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

fix(automation): bind backup approvals to full owner

Browse Source
This commit is contained in:
ogt
2026-07-14 11:24:50 +08:00
parent bbd479a36c
commit 8bee637466
3 changed files with 875 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

375
apps/api/src/services/backup_restore_alertmanager_ingress.py
View File

@@ -9,6 +9,7 @@ write escrow markers, or read secrets.
from __future__ import annotations
import hashlib
import json
from typing import Any
from uuid import NAMESPACE_URL, UUID, uuid5
@@ -38,10 +39,14 @@ logger = get_logger("awoooi.backup_restore_alertmanager_ingress")
BACKUP_RESTORE_AGENT99_ROUTE_ID = "agent99:backup_health:BackupCheck"
BACKUP_RESTORE_DISPATCH_OWNER = "alertmanager_raw_signal"
BACKUP_RESTORE_OWNER_SCHEMA_VERSION = "backup_restore_ingress_owner_v1"
BACKUP_RESTORE_LEGACY_OWNER_SCHEMA_VERSION = "backup_restore_ingress_owner_v1"
BACKUP_RESTORE_OWNER_SCHEMA_VERSION = "backup_restore_ingress_owner_v2"
BACKUP_RESTORE_OWNER_KEY_ENCODING = "canonical_json_v1"
BACKUP_RESTORE_LEGACY_BACKFILL_PREFIX = "legacy-backup-occurrence:"
APPROVAL_FINGERPRINT_MAX_LENGTH = 64
APPROVAL_FINGERPRINT_HASH_DOMAIN = (
"awoooi.backup_restore.approval_fingerprint.v2"
APPROVAL_FINGERPRINT_HASH_DOMAIN = "awoooi.backup_restore.approval_fingerprint.v2"
APPROVAL_OWNER_FINGERPRINT_HASH_DOMAIN = (
"awoooi.backup_restore.approval_owner_fingerprint.v3"
)
BACKUP_RESTORE_PROHIBITED_ACTIONS = (
"run_backup",
@@ -59,13 +64,72 @@ def _stable_digest(*parts: object) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
def _approval_storage_fingerprint(source_fingerprint: str) -> str:
"""Return the domain-separated durable identity for every source value.
def _canonical_json(value: Any) -> str:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
Hashing only values longer than the PostgreSQL column made a long source
collide with a raw 64-character source equal to that source's SHA-256.
Hash every value in the lane with a domain tag so the two source domains
remain distinct while still fitting ``approval_records.fingerprint``.
def _canonical_source_occurrence(
*,
source_fingerprint: str,
source_event_fingerprint: str,
source_started_at: str,
) -> dict[str, str]:
event_fingerprint = str(source_event_fingerprint or "").strip()
started_at = str(source_started_at or "").strip()
return {
"source_event_fingerprint": event_fingerprint,
"source_started_at": started_at,
"fallback_source_fingerprint": (
""
if event_fingerprint or started_at
else str(source_fingerprint or "").strip()
),
}
def _legacy_backup_restore_ingress_owner(
*,
project_id: str,
source_fingerprint: str,
source_event_fingerprint: str,
source_started_at: str,
) -> dict[str, str]:
"""Rebuild the exact v1 IDs solely for bounded compatibility lookup."""
source_occurrence = (
"|".join(
value for value in (source_event_fingerprint, source_started_at) if value
)
or source_fingerprint
)
owner_key = (
f"{BACKUP_RESTORE_LEGACY_OWNER_SCHEMA_VERSION}:{project_id}:"
f"{source_fingerprint}:{source_occurrence}"
)
digest = _stable_digest(owner_key)
return {
"schema_version": BACKUP_RESTORE_LEGACY_OWNER_SCHEMA_VERSION,
"project_id": project_id,
"source_fingerprint": source_fingerprint,
"source_occurrence": source_occurrence,
"source_occurrence_hash": digest,
"approval_id": str(uuid5(NAMESPACE_URL, owner_key)),
"incident_id": f"INC-BRR-{digest[:16].upper()}",
"work_item_id": f"backupcheck:{project_id}:{digest[:32]}",
}
def _approval_storage_fingerprint(source_fingerprint: str) -> str:
"""Return the transitional v2 source-only storage fingerprint.
This representation remains a lookup alias for rows written by the first
bounded-fingerprint rollout. New rows use a v3 fingerprint over the full
canonical owner identity instead of only the source fingerprint.
"""
normalized = str(source_fingerprint or "").strip()
@@ -73,59 +137,153 @@ def _approval_storage_fingerprint(source_fingerprint: str) -> str:
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def _approval_fingerprint_lookup_candidates(
source_fingerprint: str,
) -> tuple[str, ...]:
"""Return v2 first, followed by the one compatible legacy encoding.
def _approval_owner_storage_fingerprint(owner: dict[str, Any]) -> str:
identity = owner.get("owner_identity")
if not isinstance(identity, dict):
raise ValueError("backup_restore_canonical_owner_identity_required")
payload = f"{APPROVAL_OWNER_FINGERPRINT_HASH_DOMAIN}\x1f{_canonical_json(identity)}"
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
The v1 implementation stored short values verbatim and unscoped-hashed
long values. Both representations remain read-compatible during the
rollout, but new rows always use the domain-separated v2 fingerprint.
def _approval_fingerprint_lookup_candidates(
owner: dict[str, Any],
) -> tuple[str, ...]:
"""Return full-owner v3, source-only v2, then the v1 source encoding.
Every source-only candidate is only an index probe. A row is reusable
solely after its complete deterministic owner metadata also matches.
"""
normalized = str(source_fingerprint or "").strip()
canonical = _approval_storage_fingerprint(normalized)
normalized = str(owner.get("source_fingerprint") or "").strip()
canonical = _approval_owner_storage_fingerprint(owner)
source_only_v2 = _approval_storage_fingerprint(normalized)
legacy = (
normalized
if len(normalized) <= APPROVAL_FINGERPRINT_MAX_LENGTH
else hashlib.sha256(normalized.encode("utf-8")).hexdigest()
)
return tuple(dict.fromkeys((canonical, legacy)))
return tuple(dict.fromkeys((canonical, source_only_v2, legacy)))
def _approval_matches_source_fingerprint(
def _legacy_owner_reuse_is_unambiguous(owner: dict[str, Any]) -> bool:
"""Reject v1 occurrence shapes that cannot be losslessly decoded."""
event_fingerprint = str(owner.get("source_event_fingerprint") or "")
started_at = str(owner.get("source_started_at") or "")
if "|" in event_fingerprint or "|" in started_at:
return False
# With exactly one component, v1 cannot tell whether the value was the
# event fingerprint or startsAt because it discarded empty fields.
return bool(event_fingerprint) == bool(started_at)
def _approval_matches_owner_identity(
approval: Any,
source_fingerprint: str,
*,
owner: dict[str, Any],
legacy_owner: dict[str, str],
) -> bool:
"""Verify a lookup row before reusing its canonical Incident binding."""
"""Require the complete deterministic owner before approval reuse."""
metadata = getattr(approval, "metadata", None)
if not isinstance(metadata, dict):
return False
persisted_source = str(metadata.get("source_fingerprint") or "").strip()
return persisted_source == str(source_fingerprint or "").strip()
approval_id = _value(getattr(approval, "id", ""))
allowed_ids = {
_value(owner.get("canonical_approval_id")),
_value(owner.get("legacy_approval_id")),
}
if approval_id not in allowed_ids:
return False
persisted_schema = _value(metadata.get("schema_version"))
if persisted_schema == BACKUP_RESTORE_OWNER_SCHEMA_VERSION:
persisted_identity = metadata.get("owner_identity")
return bool(
isinstance(persisted_identity, dict)
and persisted_identity == owner.get("owner_identity")
and _value(metadata.get("owner_key_encoding"))
== BACKUP_RESTORE_OWNER_KEY_ENCODING
and _value(metadata.get("owner_key_sha256"))
== _value(owner.get("owner_key_sha256"))
and _value(metadata.get("approval_id")) == approval_id
)
if persisted_schema != BACKUP_RESTORE_LEGACY_OWNER_SCHEMA_VERSION:
return False
if not _legacy_owner_reuse_is_unambiguous(owner):
return False
return bool(
approval_id == legacy_owner["approval_id"]
and _value(metadata.get("approval_id")) == legacy_owner["approval_id"]
and _value(metadata.get("project_id")) == legacy_owner["project_id"]
and _value(metadata.get("source_fingerprint"))
== legacy_owner["source_fingerprint"]
and _value(metadata.get("source_occurrence"))
== legacy_owner["source_occurrence"]
and _value(metadata.get("source_occurrence_hash"))
== legacy_owner["source_occurrence_hash"]
and _value(metadata.get("incident_id")) == legacy_owner["incident_id"]
and _value(metadata.get("work_item_id")) == legacy_owner["work_item_id"]
)
async def _find_compatible_source_approval(
def _owner_with_id_scheme(
owner: dict[str, Any],
*,
scheme: str,
) -> dict[str, Any]:
if scheme not in {"canonical_json_v2", "legacy_colon_v1"}:
raise ValueError("backup_restore_owner_id_scheme_invalid")
prefix = "canonical" if scheme == "canonical_json_v2" else "legacy"
selected = dict(owner)
selected.update(
{
"approval_id": selected[f"{prefix}_approval_id"],
"incident_id": selected[f"{prefix}_incident_id"],
"work_item_id": selected[f"{prefix}_work_item_id"],
"identity_id_scheme": scheme,
"legacy_identity_compatibility": scheme == "legacy_colon_v1",
}
)
return selected
def _owner_for_approval(
owner: dict[str, Any],
approval: Any,
) -> dict[str, Any]:
approval_id = _value(getattr(approval, "id", ""))
if approval_id == _value(owner.get("canonical_approval_id")):
return _owner_with_id_scheme(owner, scheme="canonical_json_v2")
if approval_id == _value(owner.get("legacy_approval_id")):
return _owner_with_id_scheme(owner, scheme="legacy_colon_v1")
raise ValueError("backup_restore_approval_owner_id_not_deterministic")
async def _find_compatible_owner_approval(
approval_service: Any,
*,
source_fingerprint: str,
owner: dict[str, Any],
legacy_owner: dict[str, str],
) -> Any | None:
"""Read v2/v1 identities without accepting a cross-domain collision."""
"""Probe all storage generations without accepting an owner collision."""
for storage_fingerprint in _approval_fingerprint_lookup_candidates(
source_fingerprint
):
for storage_fingerprint in _approval_fingerprint_lookup_candidates(owner):
recent = await approval_service.find_by_fingerprint(
fingerprint=storage_fingerprint,
debounce_minutes=30,
)
if recent is None:
continue
if _approval_matches_source_fingerprint(recent, source_fingerprint):
if _approval_matches_owner_identity(
recent,
owner=owner,
legacy_owner=legacy_owner,
):
return recent
logger.warning(
"backup_restore_approval_fingerprint_identity_mismatch",
"backup_restore_approval_owner_identity_mismatch",
storage_fingerprint=storage_fingerprint,
approval_id=_value(getattr(recent, "id", "")),
)
@@ -145,29 +303,41 @@ def build_backup_restore_ingress_owner(
fingerprint = str(source_fingerprint or "").strip()
if not fingerprint:
raise ValueError("backup_restore_source_fingerprint_required")
source_occurrence = "|".join(
value
for value in (
str(source_event_fingerprint or "").strip(),
str(source_started_at or "").strip(),
)
if value
) or fingerprint
owner_key = (
f"{BACKUP_RESTORE_OWNER_SCHEMA_VERSION}:{project}:"
f"{fingerprint}:{source_occurrence}"
event_fingerprint = str(source_event_fingerprint or "").strip()
started_at = str(source_started_at or "").strip()
source_occurrence = _canonical_source_occurrence(
source_fingerprint=fingerprint,
source_event_fingerprint=event_fingerprint,
source_started_at=started_at,
)
digest = _stable_digest(owner_key)
approval_id = uuid5(NAMESPACE_URL, owner_key)
return {
owner_identity = {
"schema_version": BACKUP_RESTORE_OWNER_SCHEMA_VERSION,
"project_id": project,
"source_fingerprint": fingerprint,
"source_occurrence": source_occurrence,
}
owner_key = _canonical_json(owner_identity)
digest = hashlib.sha256(owner_key.encode("utf-8")).hexdigest()
legacy_owner = _legacy_backup_restore_ingress_owner(
project_id=project,
source_fingerprint=fingerprint,
source_event_fingerprint=event_fingerprint,
source_started_at=started_at,
)
owner = {
**owner_identity,
"owner_identity": owner_identity,
"owner_key_encoding": BACKUP_RESTORE_OWNER_KEY_ENCODING,
"owner_key_sha256": digest,
"source_event_fingerprint": event_fingerprint,
"source_started_at": started_at,
"source_occurrence_hash": digest,
"approval_id": str(approval_id),
"incident_id": f"INC-BRR-{digest[:16].upper()}",
"work_item_id": f"backupcheck:{project}:{digest[:32]}",
"canonical_approval_id": str(uuid5(NAMESPACE_URL, owner_key)),
"canonical_incident_id": f"INC-BRR-{digest[:16].upper()}",
"canonical_work_item_id": f"backupcheck:{project}:{digest[:32]}",
"legacy_approval_id": legacy_owner["approval_id"],
"legacy_incident_id": legacy_owner["incident_id"],
"legacy_work_item_id": legacy_owner["work_item_id"],
"route_id": BACKUP_RESTORE_AGENT99_ROUTE_ID,
"kind": "backup_health",
"mode": "BackupCheck",
@@ -177,6 +347,15 @@ def build_backup_restore_ingress_owner(
"read_only": True,
"prohibited_actions": list(BACKUP_RESTORE_PROHIBITED_ACTIONS),
}
# Production legacy-backfill claims already persist v1 IDs in their item
# payload. Preserve those IDs while upgrading their metadata and storage
# fingerprint; all other new episodes use collision-safe canonical IDs.
scheme = (
"legacy_colon_v1"
if fingerprint.startswith(BACKUP_RESTORE_LEGACY_BACKFILL_PREFIX)
else "canonical_json_v2"
)
return _owner_with_id_scheme(owner, scheme=scheme)
def _backup_restore_approval_request(
@@ -310,31 +489,83 @@ async def process_backup_restore_alertmanager_signal(
source_event_fingerprint=source_event_fingerprint,
source_started_at=source_started_at,
)
approval_storage_fingerprint = _approval_storage_fingerprint(
source_fingerprint
legacy_owner = _legacy_backup_restore_ingress_owner(
project_id=str(owner["project_id"]),
source_fingerprint=str(owner["source_fingerprint"]),
source_event_fingerprint=str(owner["source_event_fingerprint"]),
source_started_at=str(owner["source_started_at"]),
)
approval_service = get_approval_service()
owner_approval_id = UUID(str(owner["approval_id"]))
approval = await approval_service.get_approval(owner_approval_id)
created_owner = approval is None
legacy_incident_id = ""
if approval is None:
recent = await _find_compatible_source_approval(
approval_service,
source_fingerprint=source_fingerprint,
approval = None
mismatched_owner_ids: set[str] = set()
owner_candidate_ids = tuple(
dict.fromkeys(
(
_value(owner.get("approval_id")),
_value(owner.get("canonical_approval_id")),
_value(owner.get("legacy_approval_id")),
)
)
legacy_incident_id = _value(getattr(recent, "incident_id", ""))
)
for candidate_id in owner_candidate_ids:
candidate = await approval_service.get_approval(UUID(candidate_id))
if candidate is None:
continue
if _approval_matches_owner_identity(
candidate,
owner=owner,
legacy_owner=legacy_owner,
):
approval = candidate
owner = _owner_for_approval(owner, candidate)
break
mismatched_owner_ids.add(candidate_id)
logger.warning(
"backup_restore_deterministic_owner_identity_mismatch",
approval_id=candidate_id,
)
if approval is None:
recent = await _find_compatible_owner_approval(
approval_service,
owner=owner,
legacy_owner=legacy_owner,
)
if recent is not None:
approval = recent
owner = _owner_for_approval(owner, recent)
created_owner = False
if approval is None:
# A v1 delimiter collision may occupy the legacy UUID. New work uses
# the canonical JSON UUID; a mismatch on that UUID itself fails closed.
canonical_id = _value(owner.get("canonical_approval_id"))
if canonical_id in mismatched_owner_ids:
return {
"schema_version": "backup_restore_ingress_result_v1",
"status": "durable_owner_identity_mismatch_no_dispatch",
"owner": owner,
"dispatch_performed": False,
"runtime_execution_authorized": False,
"runtime_closure_verified": False,
}
if _value(owner.get("approval_id")) in mismatched_owner_ids:
owner = _owner_with_id_scheme(
owner,
scheme="canonical_json_v2",
)
owner_approval_id = UUID(str(owner["approval_id"]))
request = _backup_restore_approval_request(
owner=owner,
target_resource=target_resource,
message=message,
incident_id=legacy_incident_id,
)
try:
approval = await approval_service.create_approval_with_fingerprint(
request=request,
fingerprint=approval_storage_fingerprint,
fingerprint=_approval_owner_storage_fingerprint(owner),
)
created_owner = True
except Exception:
# A concurrent producer may have inserted the deterministic owner.
# Read it back; never create another random owner or dispatch twice.
@@ -342,11 +573,18 @@ async def process_backup_restore_alertmanager_signal(
if approval is None:
raise
else:
owner_approval_id = UUID(str(owner["approval_id"]))
incremented = await approval_service.increment_hit_count(owner_approval_id)
if incremented is not None:
approval = incremented
if _value(getattr(approval, "id", "")) != str(owner_approval_id):
if _value(getattr(approval, "id", "")) != str(
owner_approval_id
) or not _approval_matches_owner_identity(
approval,
owner=owner,
legacy_owner=legacy_owner,
):
return {
"schema_version": "backup_restore_ingress_result_v1",
"status": "durable_owner_identity_mismatch_no_dispatch",
@@ -384,6 +622,11 @@ async def process_backup_restore_alertmanager_signal(
if (
bound_approval is None
or _value(getattr(bound_approval, "incident_id", "")) != incident_id
or not _approval_matches_owner_identity(
bound_approval,
owner=owner,
legacy_owner=legacy_owner,
)
):
return {
"schema_version": "backup_restore_ingress_result_v1",
@@ -479,8 +722,7 @@ async def process_backup_restore_alertmanager_signal(
dispatch_identity
and readback_identity
and all(
_value(dispatch_identity.get(field))
== _value(readback_identity.get(field))
_value(dispatch_identity.get(field)) == _value(readback_identity.get(field))
for field in ("incident_id", "run_id", "trace_id", "work_item_id")
)
)
@@ -491,8 +733,7 @@ async def process_backup_restore_alertmanager_signal(
receipt_persisted = bool(readback and identity_matches)
verifier = (
dict(readback.get("verifier"))
if isinstance(readback, dict)
and isinstance(readback.get("verifier"), dict)
if isinstance(readback, dict) and isinstance(readback.get("verifier"), dict)
else {}
)
verifier_status = _value(verifier.get("status")) or "not_started"