261 lines
9.3 KiB
Python
261 lines
9.3 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Node ledger backup or attended recovery; default performs preview only."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import fcntl
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import re
|
||
|
|
import shutil
|
||
|
|
import signal
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
from datetime import UTC, datetime
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from ledger_archive import (
|
||
|
|
PREFIX,
|
||
|
|
Refused,
|
||
|
|
checkpoint,
|
||
|
|
command,
|
||
|
|
digest,
|
||
|
|
private_directory,
|
||
|
|
private_file,
|
||
|
|
recover,
|
||
|
|
seal,
|
||
|
|
)
|
||
|
|
|
||
|
|
NAME = re.compile(r"\d{8}T\d{6}Z-[a-f0-9]{24}\.tar\.age")
|
||
|
|
PRIMARY_TOOL = Path("/opt/railiance-platform/scripts/primary_application_archive.py")
|
||
|
|
PRIMARY_PYTHON = Path("/opt/railiance-platform/backup-venv/bin/python")
|
||
|
|
RECIPIENT = "age1zvryunvjhvpkmasskauga2heeg0ztnte9ymgppvjge36ekumk50syr3tsz"
|
||
|
|
|
||
|
|
|
||
|
|
def publish(encrypted, fetched, receipt_path, captured, tool, python):
|
||
|
|
command(
|
||
|
|
[
|
||
|
|
str(python),
|
||
|
|
str(tool),
|
||
|
|
"transfer",
|
||
|
|
"--application",
|
||
|
|
"email-connect",
|
||
|
|
"--source",
|
||
|
|
str(encrypted),
|
||
|
|
"--output",
|
||
|
|
str(fetched),
|
||
|
|
"--receipt",
|
||
|
|
str(receipt_path),
|
||
|
|
"--captured-at",
|
||
|
|
captured["captured_at"],
|
||
|
|
"--execute",
|
||
|
|
]
|
||
|
|
)
|
||
|
|
private_file(fetched)
|
||
|
|
receipt = json.loads(receipt_path.read_text())
|
||
|
|
if (
|
||
|
|
receipt.get("schema") != "platform.application-primary-archive.v1"
|
||
|
|
or receipt.get("status") != "offsite_fetched_verified"
|
||
|
|
or receipt.get("application") != "email-connect"
|
||
|
|
or receipt.get("captured_at") != captured["captured_at"]
|
||
|
|
or receipt.get("remote_retrieval_verified") is not True
|
||
|
|
or receipt.get("immutable_create") is not True
|
||
|
|
or not receipt.get("version_id")
|
||
|
|
or receipt["version_id"] == "null"
|
||
|
|
or receipt.get("ciphertext_sha256") != digest(encrypted)
|
||
|
|
or receipt.get("ciphertext_sha256") != digest(fetched)
|
||
|
|
or receipt.get("ciphertext_bytes") != encrypted.stat().st_size
|
||
|
|
or not receipt.get("destination", "").startswith(PREFIX)
|
||
|
|
):
|
||
|
|
raise Refused("primary_receipt_not_verified")
|
||
|
|
return receipt
|
||
|
|
|
||
|
|
|
||
|
|
def backup(state, primary_tool, primary_python, recipient, resume=False):
|
||
|
|
state.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||
|
|
private_directory(state)
|
||
|
|
fd = os.open(state / "backup.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
|
||
|
|
with os.fdopen(fd, "w") as lock:
|
||
|
|
try:
|
||
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
|
|
except BlockingIOError:
|
||
|
|
raise Refused("backup_already_running") from None
|
||
|
|
pending = list(state.glob("pending-*"))
|
||
|
|
if resume:
|
||
|
|
if len(pending) != 1:
|
||
|
|
raise Refused("one_pending_capture_required")
|
||
|
|
stage = pending[0]
|
||
|
|
private_directory(stage)
|
||
|
|
else:
|
||
|
|
if pending:
|
||
|
|
raise Refused("pending_backup_requires_review")
|
||
|
|
stage = Path(tempfile.mkdtemp(prefix="pending-", dir=state))
|
||
|
|
command(
|
||
|
|
[
|
||
|
|
sys.executable,
|
||
|
|
str(Path(__file__).with_name("capture-ledger.py")),
|
||
|
|
"--execute",
|
||
|
|
"--destination",
|
||
|
|
str(stage / "capture"),
|
||
|
|
],
|
||
|
|
timeout=1800,
|
||
|
|
)
|
||
|
|
encrypted, captured = seal(stage / "capture", stage, recipient)
|
||
|
|
fd, name = tempfile.mkstemp(prefix="readback-", dir=stage)
|
||
|
|
os.close(fd)
|
||
|
|
fetched = Path(name)
|
||
|
|
fetched.unlink()
|
||
|
|
transfer = publish(
|
||
|
|
encrypted,
|
||
|
|
fetched,
|
||
|
|
stage / "primary-transfer.json",
|
||
|
|
captured,
|
||
|
|
primary_tool,
|
||
|
|
primary_python,
|
||
|
|
)
|
||
|
|
filename = transfer["object_key"].rsplit("/", 1)[-1]
|
||
|
|
if not NAME.fullmatch(filename):
|
||
|
|
raise Refused("invalid_archive_name")
|
||
|
|
receipt = dict(
|
||
|
|
transfer,
|
||
|
|
image=captured["image"],
|
||
|
|
database_sha256=captured["database_sha256"],
|
||
|
|
database_bytes=captured["database_bytes"],
|
||
|
|
ledger_fingerprint=captured["ledger_fingerprint"],
|
||
|
|
pause_and_capture_seconds=captured["elapsed_seconds"],
|
||
|
|
local_archive=filename,
|
||
|
|
application_restore_verified=False,
|
||
|
|
completed_at=datetime.now(UTC).isoformat(),
|
||
|
|
)
|
||
|
|
output = state / filename
|
||
|
|
if output.exists() or output.is_symlink():
|
||
|
|
private_file(output)
|
||
|
|
if digest(output) != receipt["ciphertext_sha256"]:
|
||
|
|
raise Refused("local_archive_collision")
|
||
|
|
else:
|
||
|
|
# Create private file from the start, including manual invocations.
|
||
|
|
fd = os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
|
||
|
|
with os.fdopen(fd, "wb") as dst, fetched.open("rb") as src:
|
||
|
|
shutil.copyfileobj(src, dst)
|
||
|
|
dst.flush()
|
||
|
|
os.fsync(dst.fileno())
|
||
|
|
inventory_path = state / "inventory.json"
|
||
|
|
inventory = json.loads(inventory_path.read_text()) if inventory_path.exists() else []
|
||
|
|
if not any(item["object_key"] == receipt["object_key"] for item in inventory):
|
||
|
|
inventory.append(receipt)
|
||
|
|
checkpoint(inventory_path, inventory)
|
||
|
|
checkpoint(state / "last-success.json", receipt)
|
||
|
|
shutil.rmtree(stage)
|
||
|
|
return receipt
|
||
|
|
|
||
|
|
|
||
|
|
def freshness(state, now=None, hours=26):
|
||
|
|
now = now or datetime.now(UTC)
|
||
|
|
private_directory(state)
|
||
|
|
private_file(state / "last-success.json")
|
||
|
|
receipt = json.loads((state / "last-success.json").read_text())
|
||
|
|
stamp = datetime.fromisoformat(receipt["captured_at"].replace("Z", "+00:00"))
|
||
|
|
age = (now - stamp).total_seconds()
|
||
|
|
if not 0 <= age <= hours * 3600:
|
||
|
|
raise Refused("backup_stale")
|
||
|
|
name = receipt.get("local_archive", "")
|
||
|
|
if (
|
||
|
|
receipt.get("status") != "offsite_fetched_verified"
|
||
|
|
or receipt.get("application") != "email-connect"
|
||
|
|
or receipt.get("remote_retrieval_verified") is not True
|
||
|
|
or not receipt.get("version_id")
|
||
|
|
or receipt["version_id"] == "null"
|
||
|
|
or not NAME.fullmatch(name)
|
||
|
|
):
|
||
|
|
raise Refused("verified_backup_required")
|
||
|
|
private_file(state / name)
|
||
|
|
if digest(state / name) != receipt["ciphertext_sha256"]:
|
||
|
|
raise Refused("local_ciphertext_mismatch")
|
||
|
|
return {"status": "fresh", "age_hours": round(age / 3600, 2), "remote_retrieval_verified": True}
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
||
|
|
parser.add_argument("action", choices=["backup", "recover", "freshness"])
|
||
|
|
parser.add_argument("--execute", action="store_true")
|
||
|
|
parser.add_argument("--resume-pending", action="store_true")
|
||
|
|
parser.add_argument("--state-dir", type=Path, default=Path("/var/lib/email-connect-backup"))
|
||
|
|
parser.add_argument("--primary-tool", type=Path, default=PRIMARY_TOOL)
|
||
|
|
parser.add_argument("--primary-python", type=Path, default=PRIMARY_PYTHON)
|
||
|
|
parser.add_argument("--age-recipient", default=RECIPIENT)
|
||
|
|
parser.add_argument("--archive", type=Path)
|
||
|
|
parser.add_argument("--transfer-receipt", type=Path)
|
||
|
|
parser.add_argument("--identity", type=Path)
|
||
|
|
parser.add_argument("--destination", type=Path)
|
||
|
|
args = parser.parse_args()
|
||
|
|
if args.action == "freshness":
|
||
|
|
print(json.dumps(freshness(args.state_dir)))
|
||
|
|
elif args.action == "recover":
|
||
|
|
if not args.execute:
|
||
|
|
print(json.dumps({"status": "reviewed", "decryption_performed": False}))
|
||
|
|
return 0
|
||
|
|
if not all([args.archive, args.transfer_receipt, args.identity, args.destination]):
|
||
|
|
raise Refused("recovery_arguments_required")
|
||
|
|
print(
|
||
|
|
json.dumps(
|
||
|
|
recover(args.archive, args.transfer_receipt, args.identity, args.destination)
|
||
|
|
)
|
||
|
|
)
|
||
|
|
elif args.execute:
|
||
|
|
if (
|
||
|
|
not args.state_dir.is_absolute()
|
||
|
|
or not args.primary_tool.is_file()
|
||
|
|
or not args.primary_python.is_file()
|
||
|
|
):
|
||
|
|
raise Refused("invalid_backup_configuration")
|
||
|
|
command(["age", "--version"])
|
||
|
|
command(
|
||
|
|
[
|
||
|
|
str(args.primary_python),
|
||
|
|
str(args.primary_tool),
|
||
|
|
"check",
|
||
|
|
"--application",
|
||
|
|
"email-connect",
|
||
|
|
]
|
||
|
|
)
|
||
|
|
print(
|
||
|
|
json.dumps(
|
||
|
|
backup(
|
||
|
|
args.state_dir,
|
||
|
|
args.primary_tool,
|
||
|
|
args.primary_python,
|
||
|
|
args.age_recipient,
|
||
|
|
args.resume_pending,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
if args.resume_pending:
|
||
|
|
raise Refused("resume_requires_execute")
|
||
|
|
command(
|
||
|
|
[
|
||
|
|
sys.executable,
|
||
|
|
str(Path(__file__).with_name("capture-ledger.py")),
|
||
|
|
"--destination",
|
||
|
|
"/tmp/email-ledger-capture-review",
|
||
|
|
]
|
||
|
|
)
|
||
|
|
print(
|
||
|
|
json.dumps({"status": "reviewed", "live_changes": False, "provider_contacted": False})
|
||
|
|
)
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
|
||
|
|
def terminate(signum, frame):
|
||
|
|
raise Refused("backup_interrupted")
|
||
|
|
|
||
|
|
signal.signal(signal.SIGTERM, terminate)
|
||
|
|
try:
|
||
|
|
raise SystemExit(main())
|
||
|
|
except Exception as error:
|
||
|
|
print(json.dumps({"status": "failed", "failure_type": type(error).__name__}))
|
||
|
|
raise SystemExit(1)
|