a8fc1a3438
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11538-05ea-7b63-b3c7-a2df5f38898d
114 lines
4.2 KiB
Python
114 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Review primary archive generations; no provider or local deletion capability."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from primary_application_archive import PROFILES, require_version, validate_key
|
|
|
|
|
|
def identity(receipt):
|
|
return {
|
|
"object_key": receipt["object_key"],
|
|
"version_id": receipt["version_id"],
|
|
"ciphertext_sha256": receipt["ciphertext_sha256"],
|
|
"ciphertext_bytes": receipt["ciphertext_bytes"],
|
|
}
|
|
|
|
|
|
def plan(application, inventory, recovery, now=None):
|
|
now = now or datetime.now(UTC)
|
|
if (
|
|
recovery.get("status") != "application_restored_verified"
|
|
or recovery.get("application") != application
|
|
or recovery.get("application_restore_verified") is not True
|
|
):
|
|
raise ValueError("attended_application_recovery_proof_required")
|
|
protected = recovery.get("object_key", "")
|
|
validate_key(application, protected)
|
|
require_version(recovery.get("version_id"))
|
|
items, ignored = {}, []
|
|
for receipt in inventory:
|
|
key = receipt.get("object_key", "")
|
|
if (
|
|
receipt.get("application") != application
|
|
or receipt.get("status") != "offsite_fetched_verified"
|
|
or receipt.get("remote_retrieval_verified") is not True
|
|
or not isinstance(key, str)
|
|
or not key.startswith(PROFILES[application])
|
|
):
|
|
ignored.append(key)
|
|
continue
|
|
validate_key(application, key)
|
|
require_version(receipt.get("version_id"))
|
|
captured = datetime.fromisoformat(receipt["captured_at"].replace("Z", "+00:00"))
|
|
if (
|
|
captured.tzinfo is None
|
|
or captured > now
|
|
or key in items
|
|
or not re.fullmatch(r"[a-f0-9]{64}", receipt.get("ciphertext_sha256", ""))
|
|
or not isinstance(receipt.get("ciphertext_bytes"), int)
|
|
or receipt["ciphertext_bytes"] <= 0
|
|
):
|
|
raise ValueError("unverified_archive_inventory")
|
|
items[key] = (captured.astimezone(UTC), receipt)
|
|
if (
|
|
protected not in items
|
|
or items[protected][1]["version_id"] != recovery["version_id"]
|
|
or items[protected][1]["ciphertext_sha256"] != recovery.get("ciphertext_sha256")
|
|
):
|
|
raise ValueError("protected_recovery_point_missing_or_changed")
|
|
ordered = sorted(items, key=lambda key: (items[key][0], key), reverse=True)
|
|
daily, weekly = {}, {}
|
|
for key in ordered:
|
|
stamp = items[key][0]
|
|
daily.setdefault(stamp.date(), key)
|
|
weekly.setdefault(stamp.isocalendar()[:2], key)
|
|
keep = {protected, ordered[0]} | set(list(daily.values())[:14]) | set(list(weekly.values())[:4])
|
|
return {
|
|
"schema": "platform.application-retention-review.v1",
|
|
"status": "reviewed",
|
|
"application": application,
|
|
"daily_generations": 14,
|
|
"weekly_generations": 4,
|
|
"protected": identity(items[protected][1]),
|
|
"keep": [identity(items[key][1]) for key in ordered if key in keep],
|
|
"expiration_candidates": [identity(items[key][1]) for key in ordered if key not in keep],
|
|
"ignored": ignored,
|
|
"provider_mutations": False,
|
|
"local_mutations": False,
|
|
"activation_gate": "Separate owner must admit lifecycle/capacity, revalidate protected "
|
|
"recovery and exact versions before any expiration.",
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--application", choices=PROFILES, required=True)
|
|
parser.add_argument("--inventory", type=Path, required=True)
|
|
parser.add_argument("--recovery-receipt", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
print(
|
|
json.dumps(
|
|
plan(
|
|
args.application,
|
|
json.loads(args.inventory.read_text()),
|
|
json.loads(args.recovery_receipt.read_text()),
|
|
),
|
|
indent=2,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except Exception as error:
|
|
print(json.dumps({"status": "failed", "failure_type": type(error).__name__}))
|
|
raise SystemExit(1)
|