2c92b97f37
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02994-7685-7940-bf34-3555b8256018
290 lines
12 KiB
Python
Executable file
290 lines
12 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Inspect or extract an S1 backup into an isolated staging directory."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Any
|
|
|
|
from s1_backup import BackupError, _age_binary, _sha256
|
|
from s1_backup_contract import BackupContractError, declaration_sha256, load_spec
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
MANIFEST_KEYS = {
|
|
"artifacts",
|
|
"bundle_id",
|
|
"created_at",
|
|
"declaration_sha256",
|
|
"host",
|
|
"os_config_members",
|
|
"package_source",
|
|
"recipient_sha256",
|
|
"schema_version",
|
|
"source_revision",
|
|
}
|
|
BACKUP_RECEIPT_KEYS = {
|
|
"artifacts",
|
|
"bundle_id",
|
|
"created_at",
|
|
"declaration_sha256",
|
|
"event_type",
|
|
"host",
|
|
"manifest_sha256",
|
|
"receipt_id",
|
|
"retention",
|
|
"schema_version",
|
|
"source_revision",
|
|
"status",
|
|
}
|
|
|
|
|
|
class RestoreError(RuntimeError):
|
|
"""An artifact or restore destination is unsafe or inconsistent."""
|
|
|
|
|
|
def _json(path: Path, label: str) -> dict[str, Any]:
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise RestoreError(f"cannot read {label}") from exc
|
|
if not isinstance(payload, dict):
|
|
raise RestoreError(f"{label} must be a JSON object")
|
|
return payload
|
|
|
|
|
|
def validate_bundle(bundle: Path, spec_path: Path) -> dict[str, Any]:
|
|
bundle = bundle.resolve()
|
|
if not bundle.is_dir() or not bundle.name.startswith("s1-backup-"):
|
|
raise RestoreError("bundle must be one complete s1-backup-* directory")
|
|
declaration = load_spec(spec_path)
|
|
manifest_path = bundle / "manifest.json"
|
|
receipt_path = bundle / "receipt.json"
|
|
manifest = _json(manifest_path, "manifest")
|
|
receipt = _json(receipt_path, "receipt")
|
|
if set(manifest) != MANIFEST_KEYS:
|
|
raise RestoreError("manifest fields do not match the metadata-only contract")
|
|
if set(receipt) != BACKUP_RECEIPT_KEYS:
|
|
raise RestoreError("receipt fields do not match the metadata-only contract")
|
|
if manifest.get("schema_version") != "1.0" or receipt.get("schema_version") != "1.0":
|
|
raise RestoreError("unsupported backup evidence schema version")
|
|
if receipt.get("event_type") != "backup" or receipt.get("status") != "pass":
|
|
raise RestoreError("bundle has no passing backup receipt")
|
|
if manifest.get("bundle_id") != bundle.name or receipt.get("bundle_id") != bundle.name:
|
|
raise RestoreError("bundle identity does not match its directory")
|
|
for shared in (
|
|
"artifacts",
|
|
"created_at",
|
|
"declaration_sha256",
|
|
"host",
|
|
"source_revision",
|
|
):
|
|
if receipt.get(shared) != manifest.get(shared):
|
|
raise RestoreError(f"receipt and manifest {shared} differ")
|
|
if receipt.get("manifest_sha256") != _sha256(manifest_path):
|
|
raise RestoreError("manifest digest does not match the receipt")
|
|
expected_declaration = declaration_sha256(spec_path)
|
|
if manifest.get("declaration_sha256") != expected_declaration:
|
|
raise RestoreError("bundle was not produced from the supplied declaration")
|
|
if receipt.get("declaration_sha256") != expected_declaration:
|
|
raise RestoreError("receipt declaration digest is inconsistent")
|
|
artifacts = manifest.get("artifacts")
|
|
if not isinstance(artifacts, list) or {
|
|
item.get("artifact_class") for item in artifacts if isinstance(item, dict)
|
|
} != {"os-config", "packages"}:
|
|
raise RestoreError("manifest must describe exactly two artifact classes")
|
|
for item in artifacts:
|
|
filename = item.get("filename")
|
|
if not isinstance(filename, str) or Path(filename).name != filename:
|
|
raise RestoreError("artifact filename must be a basename")
|
|
artifact = bundle / filename
|
|
if not artifact.is_file():
|
|
raise RestoreError(f"artifact is missing: {filename}")
|
|
if item.get("encrypted_sha256") != _sha256(artifact):
|
|
raise RestoreError(f"artifact digest mismatch: {filename}")
|
|
if item.get("size_bytes") != artifact.stat().st_size:
|
|
raise RestoreError(f"artifact size mismatch: {filename}")
|
|
members = manifest.get("os_config_members")
|
|
if not isinstance(members, list) or not members:
|
|
raise RestoreError("manifest contains no OS configuration membership")
|
|
member_by_path = {
|
|
item.get("path"): item for item in members if isinstance(item, dict)
|
|
}
|
|
for entry in declaration["artifacts"]["os-config"]["paths"]:
|
|
if not entry["required"]:
|
|
continue
|
|
relative = str(PurePosixPath(entry["path"]).relative_to("/"))
|
|
member = member_by_path.get(relative)
|
|
if not member or member.get("mode") != entry["expected_mode"]:
|
|
raise RestoreError(f"required member or declared critical mode differs: {entry['path']}")
|
|
return {"bundle": bundle, "declaration": declaration, "manifest": manifest, "receipt": receipt}
|
|
|
|
|
|
def _safe_member(member: tarfile.TarInfo) -> dict[str, Any]:
|
|
path = PurePosixPath(member.name)
|
|
if path.is_absolute() or ".." in path.parts or str(path) != member.name.rstrip("/"):
|
|
raise RestoreError(f"unsafe archive member path: {member.name}")
|
|
if member.isdev() or member.isfifo():
|
|
raise RestoreError(f"unsupported archive member type: {member.name}")
|
|
if not (member.isfile() or member.isdir() or member.issym() or member.islnk()):
|
|
raise RestoreError(f"unsupported archive member type: {member.name}")
|
|
if member.issym() or member.islnk():
|
|
target = PurePosixPath(member.linkname)
|
|
combined = target if target.is_absolute() else path.parent / target
|
|
if target.is_absolute() or ".." in combined.parts:
|
|
raise RestoreError(f"archive link escapes staging root: {member.name}")
|
|
kind = "file" if member.isfile() else "directory" if member.isdir() else "symlink"
|
|
return {
|
|
"kind": kind,
|
|
"mode": f"{member.mode:04o}",
|
|
"path": str(path),
|
|
"size_bytes": member.size if member.isfile() else 0,
|
|
}
|
|
|
|
|
|
def _decrypt(age_binary: str, identity: Path, source: Path, destination: Path) -> None:
|
|
try:
|
|
subprocess.run(
|
|
[age_binary, "-d", "-i", str(identity), "-o", str(destination), str(source)],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
check=True,
|
|
)
|
|
except (OSError, subprocess.CalledProcessError) as exc:
|
|
raise RestoreError("artifact decryption failed with output suppressed") from exc
|
|
|
|
|
|
def _safe_destination(destination: Path) -> Path:
|
|
resolved = destination.resolve()
|
|
if not resolved.is_absolute() or resolved in {
|
|
Path("/"),
|
|
Path("/etc"),
|
|
Path("/home"),
|
|
Path("/root"),
|
|
Path("/var"),
|
|
}:
|
|
raise RestoreError("restore destination must be a bounded isolated directory")
|
|
try:
|
|
resolved.relative_to(ROOT)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise RestoreError("restore destination must not be inside the repository")
|
|
if not resolved.parent.is_dir():
|
|
raise RestoreError("restore destination parent must already exist")
|
|
if resolved.exists() and (not resolved.is_dir() or any(resolved.iterdir())):
|
|
raise RestoreError("restore destination must be absent or an empty directory")
|
|
return resolved
|
|
|
|
|
|
def extract_bundle(
|
|
*, bundle: Path, spec_path: Path, destination: Path, identity: Path, age_binary: str | None
|
|
) -> Path:
|
|
validated = validate_bundle(bundle, spec_path)
|
|
if not identity.is_file():
|
|
raise RestoreError("age identity must be supplied as a readable file")
|
|
final = _safe_destination(destination)
|
|
stage = Path(tempfile.mkdtemp(prefix=f".{final.name}.", dir=final.parent))
|
|
os.chmod(stage, 0o700)
|
|
try:
|
|
artifacts = {
|
|
item["artifact_class"]: validated["bundle"] / item["filename"]
|
|
for item in validated["manifest"]["artifacts"]
|
|
}
|
|
executable = _age_binary(age_binary)
|
|
plain_archive = stage / ".os-config.tar.gz"
|
|
_decrypt(executable, identity, artifacts["os-config"], plain_archive)
|
|
with tarfile.open(plain_archive, "r:gz") as archive:
|
|
observed = [_safe_member(member) for member in archive.getmembers()]
|
|
expected = validated["manifest"]["os_config_members"]
|
|
if observed != expected:
|
|
raise RestoreError("decrypted archive membership or modes differ from manifest")
|
|
# Python 3.12 added the extraction filter argument. On older
|
|
# supported runtimes the complete member/link/type validation
|
|
# above is the equivalent fail-closed gate before extraction.
|
|
if sys.version_info >= (3, 12):
|
|
archive.extractall(path=stage, filter="data")
|
|
else:
|
|
archive.extractall(path=stage)
|
|
plain_archive.unlink()
|
|
packages = stage / "packages.txt"
|
|
_decrypt(executable, identity, artifacts["packages"], packages)
|
|
os.chmod(packages, 0o600)
|
|
if packages.stat().st_size == 0 or b"\x00" in packages.read_bytes():
|
|
raise RestoreError("package-selection artifact is empty or not text")
|
|
|
|
receipt = {
|
|
"schema_version": "1.0",
|
|
"receipt_id": str(uuid.uuid4()),
|
|
"event_type": "restore",
|
|
"status": "pass",
|
|
"created_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"bundle_id": validated["manifest"]["bundle_id"],
|
|
"source_revision": validated["manifest"]["source_revision"],
|
|
"declaration_sha256": validated["manifest"]["declaration_sha256"],
|
|
"manifest_sha256": _sha256(validated["bundle"] / "manifest.json"),
|
|
"member_count": len(validated["manifest"]["os_config_members"]),
|
|
"isolated": True,
|
|
}
|
|
(stage / "restore-receipt.json").write_text(
|
|
json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|
|
if final.exists():
|
|
final.rmdir()
|
|
os.replace(stage, final)
|
|
return final
|
|
except Exception:
|
|
shutil.rmtree(stage, ignore_errors=True)
|
|
raise
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("bundle", type=Path)
|
|
parser.add_argument("--spec", type=Path, default=ROOT / "spec" / "s1-backup.yaml")
|
|
parser.add_argument("--extract-to", type=Path)
|
|
parser.add_argument("--identity", type=Path)
|
|
parser.add_argument("--age-bin")
|
|
args = parser.parse_args()
|
|
try:
|
|
if args.extract_to is None:
|
|
if args.identity is not None:
|
|
parser.error("--identity is used only with --extract-to")
|
|
validated = validate_bundle(args.bundle, args.spec)
|
|
result = {
|
|
"artifacts": len(validated["manifest"]["artifacts"]),
|
|
"bundle": validated["manifest"]["bundle_id"],
|
|
"member_count": len(validated["manifest"]["os_config_members"]),
|
|
"status": "pass",
|
|
}
|
|
else:
|
|
if args.identity is None:
|
|
parser.error("--extract-to requires --identity")
|
|
restored = extract_bundle(
|
|
bundle=args.bundle,
|
|
spec_path=args.spec,
|
|
destination=args.extract_to,
|
|
identity=args.identity,
|
|
age_binary=args.age_bin,
|
|
)
|
|
result = {"restored_to": str(restored), "status": "pass"}
|
|
print(json.dumps(result, sort_keys=True))
|
|
return 0
|
|
except (BackupContractError, BackupError, RestoreError, OSError, tarfile.TarError) as exc:
|
|
print(f"S1 restore failed closed: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|