tegwick
80e8550f9b
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
241 lines
8.1 KiB
Python
241 lines
8.1 KiB
Python
"""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))
|