tegwick
11cab7298b
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 6m27s
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
292 lines
10 KiB
Python
292 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Mirror the latest verified pair and retain its public recovery receipt.
|
|
|
|
Preview has no effects. Execute uses only this worker's existing AWS environment,
|
|
the hash-pinned platform engine, and the dedicated Coulomb archives prefix.
|
|
It never captures, decrypts, deletes retained objects or reads recovery credentials.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import fcntl
|
|
import hashlib
|
|
import importlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import signal
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import backup_capacity as capacity
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"csoc_backup_health", Path(__file__).with_name("check-paired-backup-health.py")
|
|
)
|
|
health = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(health)
|
|
Refused = health.Refused
|
|
FILES = {
|
|
"primary_application_archive.py",
|
|
"application_recovery_mirror.py",
|
|
"application_recovery_resources.py",
|
|
"application_recovery_identities.py",
|
|
}
|
|
PUBLIC_FIELDS = (
|
|
"schema",
|
|
"status",
|
|
"application",
|
|
"captured_at",
|
|
"local_archive",
|
|
"destination",
|
|
"object_key",
|
|
"version_id",
|
|
"ciphertext_sha256",
|
|
"ciphertext_bytes",
|
|
"encrypted_sha256",
|
|
"encrypted_bytes",
|
|
"image",
|
|
"age_recipient",
|
|
"database_sha256",
|
|
"content_sha256",
|
|
"immutable_create",
|
|
"remote_retrieval_verified",
|
|
)
|
|
MIRROR_FIELDS = (
|
|
"project_id",
|
|
"bucket",
|
|
"object_key",
|
|
"version_id",
|
|
"ciphertext_sha256",
|
|
"ciphertext_bytes",
|
|
"hash_readback_verified",
|
|
)
|
|
|
|
|
|
def canonical(value):
|
|
return (
|
|
json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False) + "\n"
|
|
).encode("utf-8")
|
|
|
|
|
|
def private_directory(path):
|
|
info = path.lstat()
|
|
if (
|
|
not stat.S_ISDIR(info.st_mode)
|
|
or info.st_uid != os.geteuid()
|
|
or info.st_mode & 0o077
|
|
or path.resolve() != path.absolute()
|
|
):
|
|
raise Refused("private_owner_directory_required")
|
|
|
|
|
|
def load_runtime(directory, manifest_path):
|
|
manifest = json.loads(manifest_path.read_bytes())
|
|
if (
|
|
manifest.get("schema") != "coulomb.mirror-platform-runtime.v1"
|
|
or set(manifest["files"]) != FILES
|
|
or directory.resolve() != directory.absolute()
|
|
):
|
|
raise Refused("reviewed_platform_runtime_required")
|
|
for name, expected in manifest["files"].items():
|
|
path = directory / name
|
|
info = path.lstat()
|
|
if (
|
|
not stat.S_ISREG(info.st_mode)
|
|
or info.st_uid != os.geteuid()
|
|
or info.st_mode & 0o022
|
|
or hashlib.sha256(path.read_bytes()).hexdigest() != expected
|
|
):
|
|
raise Refused("platform_runtime_hash_differs")
|
|
loaded = sys.modules.get(path.stem)
|
|
if loaded is not None and Path(loaded.__file__).resolve() != path:
|
|
raise Refused("platform_module_already_loaded_elsewhere")
|
|
sys.path.insert(0, str(directory))
|
|
return importlib.import_module("application_recovery_mirror")
|
|
|
|
|
|
def public_receipt(primary, mirrored):
|
|
health.validate_primary(primary)
|
|
health.validate_mirror(primary, mirrored)
|
|
# Never upload arbitrary private fields present in local input/journal files.
|
|
return dict(
|
|
{key: primary[key] for key in PUBLIC_FIELDS},
|
|
application_restore_verified=False,
|
|
recovery_mirror={key: mirrored["recovery_mirror"][key] for key in MIRROR_FIELDS},
|
|
)
|
|
|
|
|
|
def retain_receipt(client, receipt, journal, save):
|
|
"""Conditionally publish bounded metadata, then GET the exact returned version."""
|
|
payload = canonical(receipt)
|
|
if len(payload) > 65536:
|
|
raise Refused("bounded_public_receipt_required")
|
|
digest = hashlib.sha256(payload).hexdigest()
|
|
key = "archives/" + receipt["object_key"] + "." + digest + ".receipt.json"
|
|
binding = {"object_key": key, "sha256": digest, "bytes": len(payload)}
|
|
previous = journal.get("public_receipt")
|
|
if previous is not None and any(previous.get(k) != v for k, v in binding.items()):
|
|
raise Refused("pending_receipt_binding_differs")
|
|
journal.setdefault("public_receipt", binding)
|
|
save()
|
|
try:
|
|
head = client.head_object(Bucket=health.MIRROR_BUCKET, Key=key)
|
|
except Exception as error:
|
|
if getattr(error, "response", {}).get("ResponseMetadata", {}).get("HTTPStatusCode") != 404:
|
|
raise
|
|
head = None
|
|
if head is None:
|
|
if previous and previous.get("version_id"):
|
|
raise Refused("retained_receipt_disappeared")
|
|
# On uncertain completion, leave this binding and adopt only an exact
|
|
# matching HEAD on retry. Never blindly PUT another retained version.
|
|
created = client.put_object(
|
|
Bucket=health.MIRROR_BUCKET,
|
|
Key=key,
|
|
Body=payload,
|
|
ContentType="application/json",
|
|
Metadata={"sha256": digest, "application": "coulomb-social"},
|
|
IfNoneMatch="*",
|
|
)
|
|
version = created.get("VersionId")
|
|
else:
|
|
if (
|
|
head.get("ContentLength") != len(payload)
|
|
or head.get("Metadata", {}).get("sha256") != digest
|
|
):
|
|
raise Refused("retained_receipt_binding_differs")
|
|
version = head.get("VersionId")
|
|
if (
|
|
not isinstance(version, str)
|
|
or not version
|
|
or version == "null"
|
|
or "\n" in version
|
|
or "\r" in version
|
|
):
|
|
raise Refused("immutable_receipt_version_required")
|
|
if previous and previous.get("version_id", version) != version:
|
|
raise Refused("retained_receipt_version_changed")
|
|
journal["public_receipt"]["version_id"] = version
|
|
save()
|
|
response = client.get_object(Bucket=health.MIRROR_BUCKET, Key=key, VersionId=version)
|
|
with response["Body"] as stream:
|
|
readback = stream.read(len(payload) + 1)
|
|
if (
|
|
response.get("VersionId") != version
|
|
or response.get("ContentLength") != len(payload)
|
|
or readback != payload
|
|
):
|
|
raise Refused("public_receipt_readback_differs")
|
|
journal["public_receipt"]["readback_verified"] = True
|
|
save()
|
|
return dict(binding, version_id=version, readback_verified=True)
|
|
|
|
|
|
def mirror_latest(state, engine, client):
|
|
private_directory(state)
|
|
lock_path = state / "backup.lock"
|
|
if lock_path.exists() or lock_path.is_symlink():
|
|
health.regular(lock_path)
|
|
fd = os.open(lock_path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
|
|
with os.fdopen(fd, "w") as lock:
|
|
health.regular(lock_path)
|
|
try:
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
raise Refused("backup_or_mirror_already_running") from None
|
|
capacity.require_capacity(state, phase="mirror")
|
|
primary = health.read_json(state / "last-success.json")
|
|
original = canonical(primary)
|
|
health.validate_primary(primary)
|
|
source = state / primary["local_archive"]
|
|
info = health.regular(source)
|
|
with source.open("rb") as stream:
|
|
digest = hashlib.file_digest(stream, "sha256").hexdigest()
|
|
if info.st_size != primary["ciphertext_bytes"] or digest != primary["ciphertext_sha256"]:
|
|
raise Refused("primary_ciphertext_differs")
|
|
directory = state / "mirror-jobs"
|
|
directory.mkdir(mode=0o700, exist_ok=True)
|
|
private_directory(directory)
|
|
identity = {key: primary[key] for key in health.BINDING}
|
|
job = directory / (hashlib.sha256(canonical(identity)).hexdigest() + ".json")
|
|
journal = health.read_json(job) if job.exists() or job.is_symlink() else {}
|
|
|
|
def save():
|
|
engine.checkpoint(job, journal)
|
|
|
|
with tempfile.TemporaryDirectory(prefix="mirror-readback-", dir=state) as temporary:
|
|
mirrored = engine.mirror(
|
|
client, source, Path(temporary) / "ciphertext.age", primary, journal, save
|
|
)
|
|
receipt = public_receipt(primary, mirrored)
|
|
reference = retain_receipt(client, receipt, journal, save)
|
|
if canonical(health.read_json(state / "last-success.json")) != original:
|
|
raise Refused("primary_changed_during_mirror")
|
|
receipt["recovery_mirror"].update(
|
|
receipt_object_key=reference["object_key"],
|
|
receipt_version_id=reference["version_id"],
|
|
receipt_sha256=reference["sha256"],
|
|
receipt_bytes=reference["bytes"],
|
|
receipt_readback_verified=True,
|
|
)
|
|
engine.checkpoint(state / "last-mirror.json", receipt)
|
|
return {
|
|
"status": "mirrored_with_retained_receipt",
|
|
"primary_recaptured": False,
|
|
"ciphertext_readback_verified": True,
|
|
"receipt_readback_verified": True,
|
|
"recovery_key_read": False,
|
|
"independent_reader_used": False,
|
|
"retained_objects_deleted": False,
|
|
"application_restore_verified": False,
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--execute", action="store_true")
|
|
parser.add_argument("--state-dir", type=Path, default=Path("/var/lib/coulomb-social-backup"))
|
|
parser.add_argument(
|
|
"--platform-scripts", type=Path, default=Path("/opt/railiance-platform/scripts")
|
|
)
|
|
parser.add_argument(
|
|
"--runtime-manifest",
|
|
type=Path,
|
|
default=Path(__file__).with_name("mirror-platform-runtime.json"),
|
|
)
|
|
args = parser.parse_args()
|
|
if not args.execute:
|
|
print(
|
|
json.dumps({"status": "preview", "provider_contacted": False, "state_modified": False})
|
|
)
|
|
return
|
|
engine = load_runtime(args.platform_scripts, args.runtime_manifest)
|
|
pair = {
|
|
name: os.environ.get(name, "") for name in ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY")
|
|
}
|
|
if any(not value or any(c in value for c in "\r\n") for value in pair.values()):
|
|
raise Refused("own_publisher_environment_required")
|
|
capacity.enforce_file_limit()
|
|
client = engine.s3(pair, health.PROJECT)
|
|
try:
|
|
print(json.dumps(mirror_latest(args.state_dir, engine, client)))
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
|
|
def terminate(signum, frame):
|
|
raise Refused("mirror_interrupted")
|
|
|
|
signal.signal(signal.SIGTERM, terminate)
|
|
main()
|
|
except Exception as error:
|
|
print(
|
|
json.dumps({"status": "mirror_requires_review", "failure_type": type(error).__name__})
|
|
)
|
|
raise SystemExit(1)
|