898918c8a3
Owner-scope continuation of TASK-DR-RUNBOOK-BOUNDED-FIX-20261023 after the
independent review of 86683500. Same DR session and branch; backend tree
unchanged; no extraction, no DB/PG, no n2/full/lab/backup/restore/push.
Reproduced blocker: tarfile strips trailing slashes from directory names, so raw
"/", "///", "//" (and "") arrive as an empty member name. The inventory skipped
an empty name as a root marker before _unsafe_name ran, so a full check() over a
hostile archive with consistent manifest digest/size and listing returned [].
Fix: the archive inventory now refuses a raw empty or "/"-prefixed member name
with BACKUP_UNSAFE_ENTRY before any "." / "./" root-marker normalization; only
an explicitly relative "." (from "." or "./") remains a valid root directory
marker. Recorded listing entries are checked for raw absolute paths the same
way before normalization. No extraction, no escape claim: this is read-only
preflight acceptance of forbidden absolute paths.
New full-preflight fixtures (existing cases retained unchanged): "/", "///",
"//" and "" directory members refused; absolute regular member refused;
absolute listing entries "/", "///", "/drafts", "/etc/synthetic" refused; and
"." / "./" directory markers still accepted. The review's authoritative
artifacts/.orchestrator/dr-review-86683500/test_absolute_root_gap.py now passes
3/3 against this source (previously 2 failed).
Offline evidence: focused run RC 0, 67 passed (36 preflight + 13 flow + 18
unchanged accepted helper fixtures), JUnit 0 failures/errors/skips;
artifacts/.orchestrator/dr-bounded-fix3-20261023/.
389 lines
18 KiB
Python
389 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
# #region Tooling.DrBackupPreflight [C:4] [TYPE Module] [SEMANTICS dr,backup,preflight,integrity,restore]
|
|
# @ingroup Tooling
|
|
# @BRIEF Read-only gate between a backup set and any restore/extraction: manifest contract, digests,
|
|
# listings from the same artifacts, and a safe type-aware archive inventory.
|
|
# @PRE RUNBOOK §2 invokes this before pg_restore and tar extraction; BACKUP_DIR names the exact
|
|
# backup directory written by the backup step.
|
|
# @POST Nothing is written or extracted; a typed code list is printed and exit is non-zero on refusal.
|
|
# @INVARIANT Every refusal happens before restore/extraction; no fallback, default or first/latest choice.
|
|
# @INVARIANT Manifest shape and types are exact; unknown fields and wrong nested types are refused.
|
|
# @INVARIANT Only directory/regular tar members are accepted; links, devices, FIFOs and unsafe names
|
|
# are refused even when archive/list hashes agree.
|
|
# @RATIONALE Restoring an unverified or hostile archive destroys the stand it is meant to prove.
|
|
# @REJECTED Trusting a listing file without the archive was rejected: it cannot prove the archive bytes.
|
|
# @REJECTED Printing raw manifest/archive values in errors was rejected: they can carry private paths.
|
|
"""Offline, read-only DR backup preflight.
|
|
|
|
Exit codes: 0 = set is usable; 1 = typed refusal (codes printed, no raw values).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import tarfile
|
|
|
|
SCHEMA_VERSION = 1
|
|
MODES = {"quiesced", "hot"}
|
|
TARGET_KEYS = ("host", "port", "database", "user", "schema")
|
|
IMMUTABLE_TABLES = (
|
|
"baseline_catalog_heads", "baseline_catalog_revisions", "baseline_catalog_memberships",
|
|
"baseline_catalog_entry_authorities", "baseline_catalog_dashboard_uuid_bindings",
|
|
"dashboard_reference_draft_approval_receipts", "publication_operations",
|
|
)
|
|
MANIFEST_KEYS = ("schema_version", "mode", "t_snap", "t_dump_start", "t_dump_end", "t_storage_end",
|
|
"target", "alembic_version", "counts", "reference_ids", "dump", "storage",
|
|
"catalog_ref")
|
|
REFERENCE_KEYS = ("catalog_revisions", "approval_receipts", "publication_operations")
|
|
CATALOG_REVISION_KEYS = ("catalog_id", "catalog_revision_id", "catalog_digest", "integrity_sha256")
|
|
APPROVAL_RECEIPT_KEYS = ("receipt_id", "receipt_sha256")
|
|
PUBLICATION_KEYS = ("id", "catalog_revision_id", "commit_sha", "render_sha256")
|
|
ALLOWED_TOP_LEVEL = {"drafts", ".native-result-holds", "repositories", "backups",
|
|
"chat_uploads", "migrations_failed"}
|
|
UUID = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
|
|
HEX64 = re.compile(r"^[0-9a-f]{64}$")
|
|
HEX40 = re.compile(r"^[0-9a-f]{40}$")
|
|
MAX_LISTING_LINES = 200_000
|
|
MAX_LISTING_BYTES = 8_000_000
|
|
LISTING_TIMEOUT_SECONDS = 60
|
|
|
|
|
|
# #region Tooling.DrBackupPreflight.Hash [C:1] [TYPE Function]
|
|
def _sha256(path: Path) -> str:
|
|
hasher = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1 << 20), b""):
|
|
hasher.update(chunk)
|
|
return hasher.hexdigest()
|
|
# #endregion Tooling.DrBackupPreflight.Hash
|
|
|
|
|
|
# #region Tooling.DrBackupPreflight.Manifest [C:4] [TYPE Function]
|
|
def _is_str(value) -> bool:
|
|
return isinstance(value, str) and bool(value) and len(value) <= 4096
|
|
|
|
|
|
def _target_ok(value) -> bool:
|
|
return (isinstance(value, dict) and tuple(sorted(value)) == tuple(sorted(TARGET_KEYS))
|
|
and all(_is_str(value[key]) for key in TARGET_KEYS))
|
|
|
|
|
|
def _stat_ok(value) -> bool:
|
|
return (isinstance(value, dict) and tuple(sorted(value)) == ("bytes", "sha256")
|
|
and isinstance(value.get("bytes"), int) and not isinstance(value.get("bytes"), bool)
|
|
and value["bytes"] > 0 and isinstance(value.get("sha256"), str)
|
|
and HEX64.fullmatch(value["sha256"]) is not None)
|
|
|
|
|
|
def _list_of_dicts(value, keys, digits) -> bool:
|
|
if not isinstance(value, list):
|
|
return False
|
|
for item in value:
|
|
if not isinstance(item, dict) or tuple(sorted(item)) != tuple(sorted(keys)):
|
|
return False
|
|
for key in keys:
|
|
text = item[key]
|
|
if not isinstance(text, str):
|
|
return False
|
|
if key in digits and digits[key].fullmatch(text) is None:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _manifest_errors(manifest) -> list[str]:
|
|
codes: list[str] = []
|
|
if not isinstance(manifest, dict) or tuple(sorted(manifest)) != tuple(sorted(MANIFEST_KEYS)):
|
|
return ["BACKUP_MANIFEST_SCHEMA"]
|
|
if manifest["schema_version"] != SCHEMA_VERSION or isinstance(manifest["schema_version"], bool):
|
|
codes.append("BACKUP_MANIFEST_SCHEMA")
|
|
if manifest["mode"] not in MODES:
|
|
codes.append("BACKUP_MANIFEST_MODE")
|
|
for key in ("t_snap", "t_dump_start", "t_dump_end", "t_storage_end", "alembic_version"):
|
|
if not _is_str(manifest[key]):
|
|
codes.append("BACKUP_MANIFEST_FIELD")
|
|
if not _target_ok(manifest["target"]):
|
|
codes.append("BACKUP_MANIFEST_TARGET")
|
|
counts = manifest["counts"]
|
|
if (not isinstance(counts, dict)
|
|
or tuple(sorted(counts)) != tuple(sorted(IMMUTABLE_TABLES))
|
|
or any(not isinstance(value, int) or isinstance(value, bool) or value < 0
|
|
for value in counts.values())):
|
|
codes.append("BACKUP_MANIFEST_COUNTS")
|
|
references = manifest["reference_ids"]
|
|
if (not isinstance(references, dict)
|
|
or tuple(sorted(references)) != tuple(sorted(REFERENCE_KEYS))):
|
|
codes.append("BACKUP_MANIFEST_REFERENCES")
|
|
else:
|
|
if not _list_of_dicts(references["catalog_revisions"], CATALOG_REVISION_KEYS,
|
|
{"catalog_id": UUID, "catalog_revision_id": UUID,
|
|
"catalog_digest": HEX64, "integrity_sha256": HEX64}):
|
|
codes.append("BACKUP_MANIFEST_REFERENCES")
|
|
if not _list_of_dicts(references["approval_receipts"], APPROVAL_RECEIPT_KEYS,
|
|
{"receipt_id": UUID, "receipt_sha256": HEX64}):
|
|
codes.append("BACKUP_MANIFEST_REFERENCES")
|
|
if not _list_of_dicts(references["publication_operations"], PUBLICATION_KEYS,
|
|
{"id": UUID, "catalog_revision_id": UUID,
|
|
"commit_sha": re.compile(r"^(?:[0-9a-f]{40}|[0-9a-f]{64})$"),
|
|
"render_sha256": HEX64}):
|
|
codes.append("BACKUP_MANIFEST_REFERENCES")
|
|
for key in ("dump", "storage"):
|
|
if not _stat_ok(manifest[key]):
|
|
codes.append("BACKUP_MANIFEST_DIGEST_FIELD")
|
|
catalog_ref = manifest["catalog_ref"]
|
|
if not isinstance(catalog_ref, dict):
|
|
codes.append("BACKUP_MANIFEST_CATALOG_REF")
|
|
elif catalog_ref.get("state") == "captured":
|
|
remote = catalog_ref.get("ls_remote")
|
|
if (not isinstance(remote, list) or not remote
|
|
or any(not isinstance(item, list) or len(item) != 2
|
|
or not isinstance(item[0], str) or not isinstance(item[1], str)
|
|
or not (HEX40.fullmatch(item[0]) or HEX64.fullmatch(item[0]))
|
|
or not _is_str(item[1]) for item in remote)
|
|
or tuple(sorted(catalog_ref)) != ("ls_remote", "state")):
|
|
codes.append("BACKUP_MANIFEST_CATALOG_REF")
|
|
elif catalog_ref.get("state") == "unavailable":
|
|
if (tuple(sorted(catalog_ref)) != ("reason", "state")
|
|
or not _is_str(catalog_ref.get("reason"))):
|
|
codes.append("BACKUP_MANIFEST_CATALOG_REF")
|
|
else:
|
|
codes.append("BACKUP_MANIFEST_CATALOG_REF")
|
|
return sorted(set(codes))
|
|
# #endregion Tooling.DrBackupPreflight.Manifest
|
|
|
|
|
|
# #region Tooling.DrBackupPreflight.Mapping [C:3] [TYPE Function]
|
|
def _mapping_errors(mapping, manifest) -> list[str]:
|
|
codes: list[str] = []
|
|
expected_keys = ("approved_at", "destination", "reviewed_by", "schema_version", "source")
|
|
if not isinstance(mapping, dict) or tuple(sorted(mapping)) != expected_keys:
|
|
return ["BACKUP_DESTINATION_MAPPING_INVALID"]
|
|
if mapping["schema_version"] != 1 or isinstance(mapping["schema_version"], bool):
|
|
codes.append("BACKUP_DESTINATION_MAPPING_INVALID")
|
|
if not _is_str(mapping["reviewed_by"]) or not _is_str(mapping["approved_at"]):
|
|
codes.append("BACKUP_DESTINATION_MAPPING_INVALID")
|
|
if not _target_ok(mapping["source"]) or not _target_ok(mapping["destination"]):
|
|
codes.append("BACKUP_DESTINATION_MAPPING_INVALID")
|
|
elif mapping["source"] != manifest["target"]:
|
|
codes.append("BACKUP_DESTINATION_MAPPING_SOURCE_MISMATCH")
|
|
if not codes and mapping["source"] == mapping["destination"]:
|
|
codes.append("BACKUP_DESTINATION_MAPPING_NOT_DISTINCT")
|
|
return sorted(set(codes))
|
|
# #endregion Tooling.DrBackupPreflight.Mapping
|
|
|
|
|
|
# #region Tooling.DrBackupPreflight.Archive [C:4] [TYPE Function]
|
|
def _listing_entries(lines) -> list[str]:
|
|
return [line.strip() for line in lines
|
|
if line.strip() and not line.strip().startswith(";")]
|
|
|
|
|
|
def _normalize_listing(lines) -> list[str]:
|
|
normalized = []
|
|
for line in lines:
|
|
value = line.strip()
|
|
if not value or value.startswith(";"):
|
|
continue
|
|
while value.startswith("./"):
|
|
value = value[2:]
|
|
value = value.rstrip("/")
|
|
if value:
|
|
normalized.append(value)
|
|
return normalized
|
|
|
|
|
|
def _unsafe_name(name: str) -> bool:
|
|
parts = name.split("/")
|
|
return (not name or name.startswith("/") or chr(92) in name or ".." in parts
|
|
or any(part in {"", "."} for part in parts) or len(name) > 512
|
|
or parts[0] not in ALLOWED_TOP_LEVEL)
|
|
|
|
|
|
def _forbidden_member_code(member) -> str:
|
|
if member.issym() or member.islnk():
|
|
return "BACKUP_MEMBER_LINK_FORBIDDEN"
|
|
if member.ischr() or member.isblk() or member.isfifo() or member.isdev():
|
|
return "BACKUP_MEMBER_DEVICE_FORBIDDEN"
|
|
return "BACKUP_MEMBER_TYPE_FORBIDDEN"
|
|
|
|
|
|
def _archive_inventory(archive: Path):
|
|
"""Return (normalized names, safe?) using only member metadata; nothing is extracted."""
|
|
codes: list[str] = []
|
|
names: list[str] = []
|
|
try:
|
|
with tarfile.open(archive, "r:gz") as handle:
|
|
seen: set[str] = set()
|
|
for member in handle:
|
|
# @INVARIANT tarfile strips trailing slashes from directory names, so raw "/", "///"
|
|
# and "" all arrive as an empty name. An empty or absolute raw name is
|
|
# ambiguous/absolute, never a legitimate relative root marker, and must be
|
|
# refused before the "."/""./" root-marker normalization below; only "."
|
|
# (from a relative "." or "./" entry) is a valid root marker.
|
|
if member.name == "" or member.name.startswith("/"):
|
|
codes.append("BACKUP_UNSAFE_ENTRY")
|
|
continue
|
|
name = member.name
|
|
while name.startswith("./"):
|
|
name = name[2:]
|
|
name = name.rstrip("/")
|
|
if not name or name == ".":
|
|
# @INVARIANT A root marker is accepted only as a directory; its type is checked
|
|
# before the name is skipped, otherwise a link named "./" bypasses
|
|
# the forbidden-member contract.
|
|
if member.type != tarfile.DIRTYPE:
|
|
codes.append(_forbidden_member_code(member))
|
|
continue
|
|
if _unsafe_name(name):
|
|
codes.append("BACKUP_UNSAFE_ENTRY")
|
|
continue
|
|
if name in seen:
|
|
codes.append("BACKUP_DUPLICATE_ENTRY")
|
|
continue
|
|
seen.add(name)
|
|
if member.type in (tarfile.REGTYPE, tarfile.AREGTYPE, tarfile.DIRTYPE):
|
|
names.append(name)
|
|
else:
|
|
codes.append(_forbidden_member_code(member))
|
|
except (tarfile.TarError, OSError, EOFError):
|
|
codes.append("BACKUP_ARCHIVE_UNREADABLE")
|
|
return [], codes
|
|
return sorted(names), sorted(set(codes))
|
|
|
|
|
|
def _bound_listing(command: list[str]) -> tuple[int, str]:
|
|
try:
|
|
process = subprocess.run(command, capture_output=True, text=True,
|
|
timeout=LISTING_TIMEOUT_SECONDS)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return 124, ""
|
|
output = process.stdout or ""
|
|
if len(output) > MAX_LISTING_BYTES or output.count("\n") > MAX_LISTING_LINES:
|
|
return 125, ""
|
|
return process.returncode, output
|
|
# #endregion Tooling.DrBackupPreflight.Archive
|
|
|
|
|
|
# #region Tooling.DrBackupPreflight.Check [C:5] [TYPE Function]
|
|
def _destination_errors(destination, manifest, mapping, mapping_present) -> list[str]:
|
|
"""Bind the reviewed manifest source and optional mapping to the actually selected destination."""
|
|
expected_keys = ("host", "port", "database", "user", "schema")
|
|
if (not isinstance(destination, dict)
|
|
or tuple(sorted(destination)) != tuple(sorted(expected_keys))
|
|
or not all(_is_str(destination[key]) for key in expected_keys)):
|
|
return ["BACKUP_DESTINATION_INVALID"]
|
|
codes: list[str] = []
|
|
if destination == manifest["target"]:
|
|
if mapping_present:
|
|
codes.append("BACKUP_DESTINATION_MAPPING_UNEXPECTED")
|
|
elif not mapping_present:
|
|
codes.append("BACKUP_DESTINATION_MAPPING_REQUIRED")
|
|
elif isinstance(mapping, dict) and mapping.get("destination") != destination:
|
|
codes.append("BACKUP_DESTINATION_MAPPING_MISMATCH")
|
|
return codes
|
|
|
|
|
|
def check(backup: Path, mapping_path: Path | None = None,
|
|
destination: dict | None = None) -> list[str]:
|
|
codes: list[str] = []
|
|
if not backup.is_absolute() or not backup.is_dir():
|
|
return ["BACKUP_DIR_INVALID"]
|
|
manifest_path = backup / "manifest.json"
|
|
if not manifest_path.is_file():
|
|
return ["BACKUP_MANIFEST_MISSING"]
|
|
try:
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return ["BACKUP_MANIFEST_INVALID"]
|
|
codes.extend(_manifest_errors(manifest))
|
|
mapping = None
|
|
if mapping_path is not None:
|
|
try:
|
|
mapping = json.loads(mapping_path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
codes.append("BACKUP_DESTINATION_MAPPING_INVALID")
|
|
if isinstance(manifest, dict) and tuple(sorted(manifest)) == tuple(sorted(MANIFEST_KEYS)):
|
|
if mapping_path is not None:
|
|
if mapping is not None:
|
|
codes.extend(_mapping_errors(mapping, manifest))
|
|
else:
|
|
codes.append("BACKUP_DESTINATION_MAPPING_INVALID")
|
|
if destination is not None:
|
|
codes.extend(_destination_errors(destination, manifest, mapping,
|
|
mapping_path is not None))
|
|
if codes:
|
|
return sorted(set(codes))
|
|
dump = backup / "ss_tools.dump"
|
|
archive = backup / "storage.tgz"
|
|
archive_list = backup / "storage.tgz.list"
|
|
dump_list = backup / "ss_tools.dump.list"
|
|
for path, prefix in ((dump, "BACKUP_DUMP"), (archive, "BACKUP_ARCHIVE"),
|
|
(archive_list, "BACKUP_ARCHIVE_LISTING"),
|
|
(dump_list, "BACKUP_DUMP_LISTING")):
|
|
if not path.is_file():
|
|
codes.append(prefix + "_MISSING")
|
|
elif path.stat().st_size <= 0:
|
|
codes.append(prefix + "_EMPTY")
|
|
if codes:
|
|
return sorted(set(codes))
|
|
if manifest["dump"]["bytes"] != dump.stat().st_size:
|
|
codes.append("BACKUP_DUMP_SIZE_MISMATCH")
|
|
if manifest["storage"]["bytes"] != archive.stat().st_size:
|
|
codes.append("BACKUP_ARCHIVE_SIZE_MISMATCH")
|
|
if manifest["dump"]["sha256"] != _sha256(dump):
|
|
codes.append("BACKUP_DUMP_DIGEST_MISMATCH")
|
|
if manifest["storage"]["sha256"] != _sha256(archive):
|
|
codes.append("BACKUP_ARCHIVE_DIGEST_MISMATCH")
|
|
archive_lines = archive_list.read_text(encoding="utf-8").splitlines()
|
|
if any(entry.startswith("/") for entry in _listing_entries(archive_lines)):
|
|
codes.append("BACKUP_UNSAFE_ENTRY")
|
|
recorded_archive = _normalize_listing(archive_lines)
|
|
if not recorded_archive or len(recorded_archive) != len(set(recorded_archive)):
|
|
codes.append("BACKUP_ARCHIVE_LISTING_EMPTY" if not recorded_archive
|
|
else "BACKUP_DUPLICATE_ENTRY")
|
|
if any(_unsafe_name(name) for name in recorded_archive):
|
|
codes.append("BACKUP_UNSAFE_ENTRY")
|
|
names, inventory_codes = _archive_inventory(archive)
|
|
codes.extend(inventory_codes)
|
|
if names and sorted(names) != sorted(recorded_archive):
|
|
codes.append("BACKUP_ARCHIVE_LISTING_MISMATCH")
|
|
return_code, dump_listing = _bound_listing(["pg_restore", "--list", str(dump)])
|
|
if return_code != 0:
|
|
codes.append("BACKUP_DUMP_LISTING_UNREADABLE")
|
|
else:
|
|
recorded_dump = _normalize_listing(dump_list.read_text(encoding="utf-8").splitlines())
|
|
if not recorded_dump:
|
|
codes.append("BACKUP_DUMP_LISTING_EMPTY")
|
|
elif _normalize_listing(dump_listing.splitlines()) != recorded_dump:
|
|
codes.append("BACKUP_DUMP_LISTING_MISMATCH")
|
|
return sorted(set(codes))
|
|
# #endregion Tooling.DrBackupPreflight.Check
|
|
|
|
|
|
# #region Tooling.DrBackupPreflight.Main [C:2] [TYPE Function]
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Read-only DR backup preflight (no restore, no extraction)")
|
|
parser.add_argument("--backup-dir", required=True)
|
|
parser.add_argument("--destination-mapping", default=None)
|
|
for name in ("host", "port", "database", "user", "schema"):
|
|
parser.add_argument("--destination-" + name, default=None)
|
|
args = parser.parse_args(argv)
|
|
mapping_path = Path(args.destination_mapping) if args.destination_mapping else None
|
|
values = {name: getattr(args, "destination_" + name) for name in
|
|
("host", "port", "database", "user", "schema")}
|
|
destination = values if any(value is not None for value in values.values()) else None
|
|
try:
|
|
codes = check(Path(args.backup_dir), mapping_path, destination)
|
|
except Exception: # noqa: BLE001 — malformed input must never print a traceback or raw values
|
|
codes = ["BACKUP_UNREADABLE_INPUT"]
|
|
print(json.dumps({"status": "REFUSED" if codes else "OK", "codes": codes}))
|
|
return 1 if codes else 0
|
|
# #endregion Tooling.DrBackupPreflight.Main
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|