"""Private operative recovery drill; no body/identity values in output.""" import argparse import hashlib import json import os import re import subprocess from pathlib import Path parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--restore-receipt", type=Path, required=True) parser.add_argument("--snapshot", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) parser.add_argument( "--kit-manifest", type=Path, help="Bind offline image IDs to the independently verified kit manifest", ) 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()) capture = json.loads((args.snapshot / "receipt.json").read_text()) assert capture["status"] == "captured" and capture["app_resumed"] and capture["writers_quiesced"] image = capture["image"] if args.kit_manifest: kit = json.loads(args.kit_manifest.read_text()) assert kit["schema"] == "platform.coulomb-recovery-kit-manifest.v1" assert kit["application_image"] == capture["image"] assert kit["captured_at"] == capture["started_at"] assert kit["postgres_image_id"] == state["postgres_image_id"] image = kit["application_image_id"] assert re.fullmatch(r"sha256:[a-f0-9]{64}", image) if not re.fullmatch(r"csoc-recovery-\d{14}", state["database_container"]): raise SystemExit("isolated recovery container required") if state["network"] != state["database_container"] + "-net": raise SystemExit("exact recovery network required") network = json.loads( subprocess.run( ["docker", "network", "inspect", state["network"]], capture_output=True, text=True, check=True, ).stdout )[0] database = json.loads( subprocess.run( ["docker", "inspect", state["database_container"]], capture_output=True, text=True, check=True, ).stdout )[0] assert network["Internal"] is True assert database["Image"] == state["postgres_image_id"] and database["State"]["Running"] assert set(database["NetworkSettings"]["Networks"]) == {state["network"]} assert not any(database["NetworkSettings"]["Ports"].values()) assert ( state["status"] == "restored_verified" and state["network_internal"] and state["published_ports"] == 0 ) root = Path(state["restored_directory"]) / "content" before = { str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest() for p in root.rglob("*") if p.is_file() } code = r""" import hashlib, json, pathlib, uuid, django django.setup() from django.test import RequestFactory from django.urls import reverse from django.db import transaction from coulomb_social.apps.members.models import Member, User from coulomb_social.apps.spaces.models import Space, SpaceMembership from coulomb_social.apps.spaces import pageops, views from coulomb_social.apps.spaces.content import load_space_page from django.conf import settings assert ( settings.CONTENT_ROOT == "/recovered-content" and not settings.FLEX_AUTH_BASE_URL and not settings.USER_ENGINE_BASE_URL ) factory = RequestFactory() pages = 0 owners = 0 viewer_reads = 0 viewer_denials = 0 outsider_denials = 0 def request(user, slug, page, method="get"): route = reverse("spaces:detail", kwargs={"slug": slug}) r = ( factory.get(route, {"page": page}) if method == "get" else factory.post( reverse("spaces:page_edit", kwargs={"slug": slug, "page_slug": page}), {"title": "synthetic denied write", "body": "synthetic denied write"}, ) ) r.user = user r.session = {} return ( views.space_detail(r, slug=slug) if method == "get" else views.page_edit(r, slug=slug, page_slug=page) ) with transaction.atomic(): for space in Space.objects.filter(is_active=True).order_by("id"): paths = sorted( (pathlib.Path(settings.CONTENT_ROOT) / "spaces" / space.slug / "pages").glob("*.md") ) if not paths: continue owner = space.created_by if owner is None: owner = ( space.memberships.filter(role="owner").select_related("member__user").first().member ) owners += 1 def member(kind): token = "recovery-" + kind + "-" + uuid.uuid4().hex return Member.objects.create( tenant_id=space.tenant_id, user=User.objects.create_user(username=token), user_engine_user_id=token, issuer="https://recovery.invalid", subject=token, ) outsider = member("outsider") viewer = member("viewer") SpaceMembership.objects.create(space=space, member=viewer, role="viewer") for path in paths: page = path.stem loaded = load_space_page(space, page) assert loaded.error is None response = request(owner.user, space.slug, page) assert response.status_code == 200 and loaded.html.encode() in response.content pages += 1 denied = request(outsider.user, space.slug, page) assert denied.status_code == 404 if loaded.html: assert loaded.html.encode() not in denied.content outsider_denials += 1 page = paths[0].stem assert request(viewer.user, space.slug, page).status_code == 200 viewer_reads += 1 assert request(viewer.user, space.slug, page, "post").status_code == 403 viewer_denials += 1 transaction.set_rollback(True) assert pages > 0 and viewer_reads > 0 and outsider_denials == pages print( json.dumps( { "status": "isolated_application_recovery_verified", "recovered_spaces_with_content": owners, "actual_pages_rendered_through_view": pages, "restored_owner_access_verified": True, "synthetic_viewer_reads": viewer_reads, "synthetic_viewer_mutation_denials": viewer_denials, "same_tenant_ungranted_page_denials": outsider_denials, "fixture_transaction_rolled_back": True, "production_access": False, "external_identity_provider_used": False, "real_browser_journey_verified": False, } ) ) """ docker_args = [ "docker", "run", "--rm", "--network", state["network"], "--read-only", "--tmpfs", "/tmp", "--user", str(os.getuid()) + ":" + str(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", code, ] p = subprocess.run(docker_args, capture_output=True, check=False, timeout=120) if p.returncode: # Keep exceptions private: traceback may include page/principal content. diagnostics = Path(state["restored_directory"]) / "private-app-diagnostics.txt" fd = os.open(diagnostics, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "wb") as stream: stream.write(p.stdout + p.stderr) print( json.dumps( { "status": "isolated_application_drill_refused", "exit_code": p.returncode, "private_diagnostics": str(diagnostics), } ) ) raise SystemExit(1) result = json.loads(p.stdout.splitlines()[-1]) after = { str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest() for p in root.rglob("*") if p.is_file() } assert before == after result.update( content_files_unchanged=len(after), packaged_application_image=image, database_restore=state ) out = args.output fd = os.open(out, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w") as stream: json.dump(result, stream, indent=2) print(json.dumps(result))