tegwick
2344b2d18b
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 8m0s
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
243 lines
12 KiB
Python
243 lines
12 KiB
Python
"""Application-side checks for the supervised creation-date executor.
|
|
|
|
The external supervisor still owns exact admission, pod/image/PVC verification,
|
|
fresh native backup/readback, and fencing all writers before and after this call.
|
|
Receipt metadata is validated here; this module never contacts the provider.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import copy
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
from contextlib import contextmanager
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from django.db import connection
|
|
|
|
from .bubble_transfer import source_timestamp
|
|
from .date_backfill import prepare_index_date_patch
|
|
from .date_backfill_journal import apply_date_patch, reconcile_date_patch
|
|
|
|
PREFIX = "platform-pg/application-archives/coulomb-social/"
|
|
|
|
|
|
def canonical(data):
|
|
return json.dumps(data, sort_keys=True, separators=(",", ":"), default=str).encode()
|
|
|
|
|
|
def sha(data):
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def validate_backup(receipt, image, recipient, *, now=None, require_fresh=True):
|
|
"""Check declarations and binding only, not the claimed provider readback."""
|
|
now = now or datetime.now(UTC)
|
|
captured = source_timestamp(receipt["captured_at"])
|
|
completed = source_timestamp(receipt["completed_at"])
|
|
if (now.tzinfo is None or captured > completed or completed > now
|
|
or (require_fresh and (now - captured).total_seconds() > 600)
|
|
or receipt.get("schema") != "platform.application-primary-archive.v1"
|
|
or receipt.get("status") != "offsite_fetched_verified"
|
|
or receipt.get("application") != "coulomb-social"
|
|
or receipt.get("image") != image or receipt.get("age_recipient") != recipient
|
|
or not re.fullmatch(r"forgejo\.coulomb\.social/coulomb/"
|
|
r"coulomb-social@sha256:[a-f0-9]{64}", image)
|
|
or not isinstance(recipient, str) or not recipient.startswith("age1")
|
|
or receipt.get("immutable_create") is not True
|
|
or receipt.get("remote_retrieval_verified") is not True
|
|
or not isinstance(receipt.get("version_id"), str)
|
|
or receipt["version_id"] in {"", "null"}
|
|
or not isinstance(receipt.get("object_key"), str)
|
|
or not receipt["object_key"].startswith(PREFIX)
|
|
or receipt.get("destination") != "s3://railiance-platform-pg-backup/"
|
|
+ receipt["object_key"]):
|
|
raise ValueError("Fresh exact application backup metadata required")
|
|
for field in ("encrypted_sha256", "ciphertext_sha256", "database_sha256", "content_sha256"):
|
|
if not re.fullmatch(r"[a-f0-9]{64}", receipt.get(field, "")):
|
|
raise ValueError("Complete paired backup hashes required")
|
|
if (receipt["encrypted_sha256"] != receipt["ciphertext_sha256"]
|
|
or type(receipt.get("encrypted_bytes")) is not int or receipt["encrypted_bytes"] <= 0
|
|
or type(receipt.get("ciphertext_bytes")) is not int
|
|
or receipt["ciphertext_bytes"] != receipt["encrypted_bytes"]):
|
|
raise ValueError("Consistent paired ciphertext binding required")
|
|
return sha(canonical(receipt))
|
|
|
|
|
|
def _snapshot(root):
|
|
"""Keep row values in memory only; return fingerprints to the supervisor."""
|
|
root = Path(root).absolute()
|
|
if any(p.is_symlink() for p in [root, *root.parents]) or not root.is_dir():
|
|
raise ValueError("Real content directory required")
|
|
if connection.vendor == "postgresql":
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("SELECT current_schema()")
|
|
if cursor.fetchone()[0] != "public":
|
|
raise ValueError("Expected public application schema required")
|
|
cursor.execute("SELECT tablename FROM pg_tables WHERE schemaname='public' ORDER BY 1")
|
|
names = [r[0] for r in cursor.fetchall()]
|
|
else:
|
|
names = sorted(connection.introspection.table_names())
|
|
rows = {}
|
|
with connection.cursor() as cursor:
|
|
for table in names:
|
|
qualified = ("public." if connection.vendor == "postgresql" else "")
|
|
qualified += connection.ops.quote_name(table)
|
|
cursor.execute("SELECT * FROM " + qualified)
|
|
columns = [c[0] for c in cursor.description]
|
|
rows[table] = [dict(zip(columns, row)) for row in cursor.fetchall()]
|
|
files = {}
|
|
for path in sorted(root.rglob("*")):
|
|
mode = path.lstat().st_mode
|
|
if stat.S_ISDIR(mode):
|
|
continue
|
|
if not stat.S_ISREG(mode) or path.stat().st_nlink != 1:
|
|
raise ValueError("Content link or special file refused")
|
|
if path.relative_to(root).parts[0] == ".recovery":
|
|
raise ValueError("Content recovery must be resolved before date correction")
|
|
with path.open("rb") as stream:
|
|
files[path.relative_to(root).as_posix()] = hashlib.file_digest(
|
|
stream, "sha256").hexdigest()
|
|
return {"rows": rows, "files": files}
|
|
|
|
|
|
def fingerprint(snapshot):
|
|
tables = {name: sorted(canonical(row).decode() for row in rows)
|
|
for name, rows in snapshot["rows"].items()}
|
|
return {"tables_sha256": sha(canonical(tables)),
|
|
"files_sha256": sha(canonical(snapshot["files"])),
|
|
"tables": len(tables), "rows": sum(len(rows) for rows in tables.values()),
|
|
"files": len(snapshot["files"])}
|
|
|
|
|
|
def source_fingerprint(root):
|
|
return fingerprint(_snapshot(root))
|
|
|
|
|
|
def _lock_tables():
|
|
if connection.vendor != "postgresql":
|
|
raise ValueError("PostgreSQL execution required")
|
|
if not connection.in_atomic_block:
|
|
raise ValueError("SQL locks require a transaction")
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("SELECT tablename FROM pg_tables WHERE schemaname='public' ORDER BY 1")
|
|
tables = [r[0] for r in cursor.fetchall()]
|
|
for table in tables:
|
|
cursor.execute("LOCK TABLE public." + connection.ops.quote_name(table)
|
|
+ " IN ACCESS EXCLUSIVE MODE NOWAIT")
|
|
|
|
|
|
@contextmanager
|
|
def _content_lock(root):
|
|
path = Path(root) / ".writer.lock"
|
|
if any(p.is_symlink() for p in [path, *path.absolute().parents]):
|
|
raise ValueError("Real existing writer lock required")
|
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
|
|
try:
|
|
if not stat.S_ISREG(os.fstat(fd).st_mode) or os.fstat(fd).st_nlink != 1:
|
|
raise ValueError("Regular writer lock required")
|
|
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
yield
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def _set_date(snapshot, plan, date):
|
|
selected = plan["selection"]
|
|
matches = [row for row in snapshot["rows"]["spaces_space"]
|
|
if str(row["id"]) == selected["space_id"]]
|
|
if (len(matches) != 1 or matches[0]["tenant_id"] != selected["tenant_id"]
|
|
or matches[0]["slug"] != selected["space_slug"]
|
|
or sum(row["slug"] == selected["space_slug"]
|
|
for row in snapshot["rows"]["spaces_space"]) != 1):
|
|
raise ValueError("Exact unambiguous SQL identity required")
|
|
matches[0]["created_at"] = source_timestamp(date)
|
|
|
|
|
|
def execute_date_operation(root, plan, journal, baseline, backup, *, image, recipient,
|
|
now=None, reconcile=False, checkpoint=lambda phase: None):
|
|
"""Verify whole-store pre/postconditions around the durable correction.
|
|
|
|
On any error the external supervisor MUST keep writers fenced. Reconciliation
|
|
uses the ORIGINAL bound backup even if old; taking a new backup of uncertain
|
|
mixed SQL/file state is not a substitute. No automatic fresh retry is allowed.
|
|
"""
|
|
if connection.vendor != "postgresql":
|
|
raise ValueError("PostgreSQL execution required")
|
|
binding = validate_backup(backup, image, recipient, now=now, require_fresh=not reconcile)
|
|
root, journal = Path(root), Path(journal)
|
|
name = f"spaces/{plan['selection']['space_slug']}/pages/index.md"
|
|
with _content_lock(root):
|
|
if reconcile:
|
|
record = json.loads(journal.read_bytes())
|
|
if (record["state"] != "pending" or record["plan"] != plan
|
|
or record["backup_receipt_sha256"] != binding):
|
|
raise ValueError("Original pending operation binding required")
|
|
original = base64.b64decode(record["original"], validate=True)
|
|
else:
|
|
original = (root / name).read_bytes()
|
|
patched = prepare_index_date_patch(plan, original)
|
|
before = _snapshot(root)
|
|
normalized = copy.deepcopy(before)
|
|
_set_date(normalized, plan, plan["observed_sql_created_at"])
|
|
normalized["files"][name] = sha(original)
|
|
if fingerprint(normalized) != baseline:
|
|
raise ValueError("Whole-store reviewed baseline changed")
|
|
expected = copy.deepcopy(normalized)
|
|
_set_date(expected, plan, plan["source_created_at"])
|
|
expected["files"][name] = sha(patched)
|
|
expected_fingerprint = fingerprint(expected)
|
|
|
|
def verify(phase):
|
|
if phase in {"intent", "locked"}:
|
|
_lock_tables()
|
|
current = _snapshot(root)
|
|
if reconcile:
|
|
# Only the selected date/index may be in either known state.
|
|
allowed_dates = {source_timestamp(plan["observed_sql_created_at"]),
|
|
source_timestamp(plan["source_created_at"])}
|
|
target = next(row for row in current["rows"]["spaces_space"]
|
|
if str(row["id"]) == plan["selection"]["space_id"])
|
|
if (target["created_at"] not in allowed_dates
|
|
or current["files"][name] not in {sha(original), sha(patched)}):
|
|
raise ValueError("Unknown date operation state")
|
|
corrected = target["created_at"] == source_timestamp(plan["source_created_at"])
|
|
verified = copy.deepcopy(current)
|
|
_set_date(verified, plan, plan["observed_sql_created_at"])
|
|
verified["files"][name] = sha(original)
|
|
if fingerprint(verified) != baseline:
|
|
raise ValueError("Unrelated rows or files changed")
|
|
outcome["fingerprint"] = expected_fingerprint if corrected else baseline
|
|
elif fingerprint(current) != baseline:
|
|
raise ValueError("Whole-store baseline changed after SQL locks")
|
|
elif phase == "sql":
|
|
sql_only = copy.deepcopy(expected)
|
|
sql_only["files"][name] = sha(original)
|
|
if fingerprint(_snapshot(root)) != fingerprint(sql_only):
|
|
raise ValueError("Unexpected SQL correction result")
|
|
elif phase in {"file", "commit"}:
|
|
if fingerprint(_snapshot(root)) != outcome["fingerprint"]:
|
|
raise ValueError("Whole-store correction postflight differs")
|
|
checkpoint(phase)
|
|
|
|
outcome = {"fingerprint": expected_fingerprint}
|
|
if reconcile:
|
|
result = reconcile_date_patch(root, journal, checkpoint=verify)
|
|
else:
|
|
if fingerprint(before) != baseline:
|
|
raise ValueError("Observed date operation state changed")
|
|
apply_date_patch(root, plan, journal, backup_sha256=binding, checkpoint=verify)
|
|
result = "completed"
|
|
if fingerprint(_snapshot(root)) != outcome["fingerprint"]:
|
|
raise ValueError("Final complete-store comparison failed")
|
|
return {"status": result, "whole_store_postflight": True,
|
|
"backup_metadata_validated": True, "backup_provider_readback_performed_here": False,
|
|
"external_writer_fence_verified_here": False,
|
|
"all_source_tables_compared": baseline["tables"],
|
|
"all_source_files_compared": baseline["files"]}
|