coulomb-social/scripts/rehearse-restored-storage.py

"""Rehearse a packaged storage candidate against an isolated recovered real pair.

Never opens production or prints rows, pages, identities or raw child failures.
The original encrypted point and private recovered files remain unchanged.
"""

from __future__ import annotations

import argparse
import ast
import hashlib
import json
import os
import re
import subprocess
from pathlib import Path


def run(arguments):
    return subprocess.run(arguments, capture_output=True, timeout=180, check=True).stdout


parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--restore-receipt", type=Path, required=True)
parser.add_argument("--candidate-image", required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
if args.output.exists() or args.output.is_symlink():
    raise SystemExit("fresh output required")
state = json.loads(args.restore_receipt.read_text())
assert re.fullmatch(r"csoc-recovery-\d{14}", state["database_container"])
assert state["network"] == state["database_container"] + "-net"
network = json.loads(run(["docker", "network", "inspect", state["network"]]))[0]
database = json.loads(run(["docker", "inspect", state["database_container"]]))[0]
assert network["Internal"] is True and database["State"]["Running"]
assert set(database["NetworkSettings"]["Networks"]) == {state["network"]}
assert not any(database["NetworkSettings"]["Ports"].values())
assert database["Image"] == state["postgres_image_id"]
image = run(["docker", "image", "inspect", args.candidate_image, "--format", "{{.Id}}"])
image = image.decode().strip()
assert re.fullmatch(r"sha256:[a-f0-9]{64}", image)
root = Path(state["restored_directory"]) / "content"


def content_hashes():
    return {
        str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest()
        for p in root.rglob("*")
        if p.is_file()
    }


before_files = content_hashes()
source = Path(__file__).with_name("verify-restored-application.py")
tree = ast.parse(source.read_text())
view_checks = next(
    node.value.value
    for node in tree.body
    if isinstance(node, ast.Assign)
    and isinstance(node.targets[0], ast.Name)
    and node.targets[0].id == "code"
)
migration_checks = r"""
import io, json, hashlib, django

django.setup()
from django.core.management import call_command
from django.db import connection, transaction, IntegrityError
from coulomb_social.apps.spaces.models import Space

assert connection.vendor == "postgresql"


def fingerprint():
    with connection.cursor() as cursor:
        cursor.execute(
            "SELECT tablename FROM pg_tables WHERE schemaname='public' "
            "AND tablename!='django_migrations' ORDER BY tablename"
        )
        tables = [row[0] for row in cursor.fetchall()]
        result = {}
        for table in tables:
            cursor.execute("SELECT * FROM " + connection.ops.quote_name(table))
            rows = sorted(
                json.dumps(row, default=str, separators=(",", ":")) for row in cursor.fetchall()
            )
            result[table] = {
                "rows": len(rows),
                "sha256": hashlib.sha256("\n".join(rows).encode()).hexdigest(),
            }
        return result


def constraints():
    with connection.cursor() as cursor:
        cursor.execute("SELECT conname FROM pg_constraint WHERE conrelid='spaces_space'::regclass")
        return {row[0] for row in cursor.fetchall()}


def migrate(target):
    call_command(
        "migrate",
        "spaces",
        target,
        verbosity=0,
        interactive=False,
        stdout=io.StringIO(),
        stderr=io.StringIO(),
    )


global_name = "spaces_space_slug_global_uniq"
tenant_name = "spaces_space_tenant_slug_uniq"
assert global_name not in constraints() and tenant_name in constraints()
baseline = fingerprint()
assert Space.objects.count() > 0
slug = Space.objects.first().slug
migrate("0003")
assert global_name in constraints() and tenant_name in constraints() and fingerprint() == baseline
try:
    with transaction.atomic():
        Space.objects.create(tenant_id="recovery:collision", slug=slug, title="Synthetic collision")
except IntegrityError:
    pass
else:
    raise RuntimeError("global_constraint_failed_to_refuse_collision")
assert fingerprint() == baseline
migrate("0002")
assert (
    global_name not in constraints() and tenant_name in constraints() and fingerprint() == baseline
)
with transaction.atomic():
    Space.objects.create(tenant_id="recovery:collision", slug=slug, title="Synthetic collision")
    held = fingerprint()
    try:
        migrate("0003")
    except RuntimeError as error:
        assert str(error).startswith("Refusing spaces_space_slug_global_uniq:")
    else:
        raise RuntimeError("legacy_collision_was_not_refused")
    assert global_name not in constraints() and fingerprint() == held
    transaction.set_rollback(True)
assert fingerprint() == baseline
migrate("0003")
assert global_name in constraints() and tenant_name in constraints() and fingerprint() == baseline
print(
    json.dumps(
        {
            "status": "recovered_data_migration_lifecycle_verified",
            "business_tables": len(baseline),
            "business_rows": sum(t["rows"] for t in baseline.values()),
            "forward_preserves_all_business_rows": True,
            "sql_collision_denied": True,
            "rollback_preserves_all_business_rows_and_tenant_constraint": True,
            "legacy_refusal_preserves_rows_and_grants": True,
            "reapply_preserves_all_business_rows": True,
            "global_constraint_present_after_reapply": True,
            "production_access": False,
        }
    )
)
"""
command = [
    "docker",
    "run",
    "--rm",
    "--network",
    state["network"],
    "--read-only",
    "--tmpfs",
    "/tmp",
    "--user",
    f"{os.getuid()}:{os.getgid()}",
    "--entrypoint",
    "/app/.venv/bin/python",
    "-e",
    "DJANGO_SETTINGS_MODULE=coulomb_social.settings.base",
    "-e",
    "DATABASE_URL=postgres://postgres@recovery-db/coulomb_social_restore",
    "-e",
    "CONTENT_ROOT=/recovered-content",
    "-e",
    "FORGEJO_BASE_URL=",
    "-e",
    "ALLOWED_HOSTS=testserver",
    "-e",
    "PYTHONDONTWRITEBYTECODE=1",
    "-v",
    str(root) + ":/recovered-content:ro",
    image,
    "-c",
    migration_checks + view_checks,
]
result = subprocess.run(command, capture_output=True, timeout=180, check=False)
if result.returncode:
    diagnostic = Path(state["restored_directory"]) / "private-candidate-diagnostics.txt"
    fd = os.open(diagnostic, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "wb") as output:
        output.write(result.stdout + result.stderr)
    print(
        json.dumps(
            {"status": "candidate_rehearsal_refused", "private_diagnostics": str(diagnostic)}
        )
    )
    raise SystemExit(1)
rows = [json.loads(line) for line in result.stdout.splitlines() if line.startswith(b"{")]
assert len(rows) == 2 and content_hashes() == before_files
proof = {
    "status": "recovered_data_candidate_verified",
    "candidate_image_id": image,
    "migration": rows[0],
    "application": rows[1],
    "content_files_unchanged": len(before_files),
    "network_internal": True,
    "published_ports": 0,
    "production_access": False,
    "shared_view_check_sha256": hashlib.sha256(source.read_bytes()).hexdigest(),
}
fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(fd, "w") as output:
    json.dump(proof, output, indent=2)
print(json.dumps(proof))