tegwick
bf77e378a8
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1abf1-7e95-7141-87c9-ee78332bf790
117 lines
5.1 KiB
Python
117 lines
5.1 KiB
Python
"""Validated point selection for attended backups of the admitted demo release.
|
|
|
|
The original acceptance modules remain byte-for-byte frozen. Only their point
|
|
parameters are selected here; image, schema, custody, crypto and limits remain.
|
|
"""
|
|
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import re
|
|
|
|
import vergabe_pilot_export as export
|
|
from vergabe_pilot_recovery_store import digest, private_directory, require
|
|
|
|
BASELINE = "20261105T233543Z-89bfbe04dece"
|
|
PATTERN = r"[0-9]{8}T[0-9]{6}Z-[0-9a-f]{12}"
|
|
FILES = {"database.dump", "stores.tar.gz", "manifest.private.json", "receipt.json"}
|
|
|
|
|
|
def run_time(run):
|
|
require(isinstance(run, str) and re.fullmatch(PATTERN, run), "invalid_backup_run")
|
|
try:
|
|
value = dt.datetime.strptime(run[:16], "%Y%m%dT%H%M%SZ").replace(tzinfo=dt.UTC)
|
|
except ValueError:
|
|
require(False, "invalid_backup_date")
|
|
require(value >= dt.datetime(2026, 11, 5, tzinfo=dt.UTC)
|
|
and value <= dt.datetime.now(dt.UTC) + dt.timedelta(minutes=5), "backup_date_outside_scope")
|
|
return value
|
|
|
|
|
|
def read_private(path, maximum=16384):
|
|
info = export.regular(path, os.geteuid())
|
|
require(0 < info.st_size <= maximum, "point_metadata_bound")
|
|
return json.loads(path.read_text())
|
|
|
|
|
|
def select(run, files):
|
|
run_time(run)
|
|
require(set(files) == FILES, "point_file_set_changed")
|
|
for value in files.values():
|
|
require(set(value) == {"bytes", "sha256"} and type(value["bytes"]) is int
|
|
and 0 < value["bytes"] <= 1024**2 and isinstance(value["sha256"], str)
|
|
and re.fullmatch(r"[0-9a-f]{64}", value["sha256"]), "point_file_metadata_invalid")
|
|
export.RUN_ID = run
|
|
export.NAME = "vergabe-pilot-" + run
|
|
export.EXPECTED = {n: (files[n]["bytes"], files[n]["sha256"]) for n in ("database.dump", "stores.tar.gz")}
|
|
|
|
|
|
def select_node(run):
|
|
run_time(run)
|
|
directory = export.BASE / run
|
|
private_directory(export.BASE)
|
|
private_directory(directory)
|
|
receipt = read_private(directory / "receipt.json")
|
|
require(receipt["run_id"] == run and receipt["status"] == "captured_and_resumed", "capture_not_complete")
|
|
files = {}
|
|
for name in FILES:
|
|
path = directory / name
|
|
info = export.regular(path, os.geteuid())
|
|
require(0 < info.st_size <= 1024**2, "point_file_size_bound")
|
|
files[name] = {"bytes": info.st_size, "sha256": digest(path)}
|
|
select(run, files)
|
|
# Reuses completed-writer/resume, revision, image and receipt-hash guards.
|
|
require(export.capture_files(directory) == files, "node_capture_changed")
|
|
return files
|
|
|
|
|
|
def select_transfer(directory, run):
|
|
run_time(run)
|
|
private_directory(directory)
|
|
metadata = read_private(directory / ("vergabe-pilot-" + run + ".transfer.json"))
|
|
packet = metadata["packet"]
|
|
require(metadata["schema"] == "railiance.vergabe-pilot-sealed-export.v1"
|
|
and metadata["run_id"] == packet["run_id"] == run
|
|
and metadata["recipient"] == export.RECIPIENT and packet["image"] == export.IMAGE
|
|
and packet["database"] == "vergabe_demo_company", "point_binding_changed")
|
|
require(type(metadata["ciphertext_bytes"]) is int and 0 < metadata["ciphertext_bytes"] <= 4 * 1024**2
|
|
and re.fullmatch(r"[0-9a-f]{64}", metadata["ciphertext_sha256"]), "ciphertext_metadata_invalid")
|
|
select(run, packet["files"])
|
|
return metadata
|
|
|
|
|
|
def successful_provider(body, run, sealed):
|
|
return (body.get("run_id") == run and body.get("status") == "published_read_back_and_decrypted"
|
|
and body.get("warden_exit_code") == 0
|
|
and all(body.get(k) is True for k in ("provider_publication_verified", "independent_readback_verified",
|
|
"decryption_verified", "session_cleanup_confirmed"))
|
|
and body.get("ciphertext_sha256") == sealed["ciphertext_sha256"]
|
|
and body.get("ciphertext_bytes") == sealed["ciphertext_bytes"]
|
|
and body.get("object_path") == "r01-platform-pg/vergabe-pilot-" + run + ".zip.age")
|
|
|
|
|
|
def retention_plan(points):
|
|
"""Seven daily plus two weekly representatives; always protect baseline/newest.
|
|
|
|
Input is the owner's locally verified provider inventory. Unknown remote
|
|
objects are never inferred eligible. Multiple points on a day don't consume
|
|
extra daily slots; weekly representatives may overlap the daily set.
|
|
"""
|
|
require(len(points) <= 1000 and len({p["run_id"] for p in points}) == len(points), "inventory_bound_or_duplicate")
|
|
ordered = sorted(points, key=lambda p: run_time(p["run_id"]), reverse=True)
|
|
keep = {BASELINE}
|
|
if ordered:
|
|
keep.add(ordered[0]["run_id"])
|
|
days, weeks = set(), set()
|
|
for point in ordered:
|
|
date = run_time(point["run_id"]).date()
|
|
day, week = date.isoformat(), tuple(date.isocalendar()[:2])
|
|
if day not in days and len(days) < 7:
|
|
keep.add(point["run_id"])
|
|
days.add(day)
|
|
if week not in weeks and len(weeks) < 2:
|
|
keep.add(point["run_id"])
|
|
weeks.add(week)
|
|
return {"keep": [p for p in ordered if p["run_id"] in keep],
|
|
"candidates": [p for p in ordered if p["run_id"] not in keep],
|
|
"baseline_protected": BASELINE, "unknown_objects_eligible": False}
|