railiance-platform/tests/test_mas_evidence_backup.py

"""MAS backup worker: capture, seal, publish with readback, freshness. Synthetic only."""

import hashlib
import importlib.util
import io
import json
import sys
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
spec = importlib.util.spec_from_file_location("mas_backup", ROOT / "scripts/mas_evidence_backup.py")
worker = importlib.util.module_from_spec(spec)
spec.loader.exec_module(worker)

NOW = datetime(2026, 11, 3, 4, 15, tzinfo=UTC)
CAPTURED = "2026-11-03T04:15:00Z"


class NotFound(Exception):
    response = {"ResponseMetadata": {"HTTPStatusCode": 404}}


class Provider:
    def __init__(self, fail_complete=False):
        self.objects, self.calls, self.fail_complete = {}, [], fail_complete
        self.parts = []

    def head_object(self, Bucket, Key):
        self.calls.append("head")
        raise NotFound()

    def create_multipart_upload(self, **kw):
        self.calls.append("start")
        self.meta = kw["Metadata"]
        return {"UploadId": "u1"}

    def upload_part(self, **kw):
        self.parts.append(kw["Body"])
        return {"ETag": "e"}

    def complete_multipart_upload(self, **kw):
        self.calls.append("complete")
        assert kw["IfNoneMatch"] == "*"
        if self.fail_complete:
            raise TimeoutError("synthetic")
        self.objects[kw["Key"]] = b"".join(self.parts)
        return {"VersionId": "v1"}

    def abort_multipart_upload(self, **kw):
        self.calls.append("abort")

    def get_object(self, Bucket, Key, VersionId):
        data = self.objects[Key]
        return {"ContentLength": len(data), "VersionId": VersionId, "Body": io.BytesIO(data)}

    def delete_object(self, **kw):  # pragma: no cover - must never be called
        raise AssertionError("worker must never delete")


class FakeCapture:
    def quiesced_copy(self, destination):
        destination.mkdir(mode=0o700)
        for name in worker.archive.FILES:
            (destination / name).write_bytes(b"plaintext " + name.encode())
        (destination / "receipt.json").write_text("{}")
        return {"captured_at": CAPTURED, "image": "mas-live:test", "volume": "pvc-x",
                "files": {n: {"present": True, "bytes": 10} for n in worker.archive.FILES}}


@pytest.fixture
def env(tmp_path, monkeypatch):
    state = tmp_path / "state"
    recipient = tmp_path / "age-recipient"
    recipient.write_text("age1examplerecipient\n")
    monkeypatch.setattr(worker, "RECIPIENT_FILE", recipient)

    def seal(directory, stage, age_recipient):
        assert age_recipient == "age1examplerecipient"
        out = stage / "packet.tar.age"
        out.write_bytes(b"age-encryption.org/v1\nsealed")
        out.chmod(0o600)
        return {"ciphertext_sha256": hashlib.sha256(out.read_bytes()).hexdigest(),
                "ciphertext_bytes": out.stat().st_size}

    monkeypatch.setattr(worker.archive, "seal", seal)
    return state


def test_backup_publishes_verified_and_leaves_no_plaintext(env):
    provider = Provider()
    result = worker.backup(env, FakeCapture(), lambda: provider, now=NOW)
    assert result["status"] == "offsite_fetched_verified" and result["remote_retrieval_verified"]
    assert result["object_key"].startswith("platform-pg/application-archives/mas/2026/11/03/")
    assert result["version_id"] == "v1" and result["retained_objects_deleted"] is False
    run = env / result["run"]
    assert sorted(p.name for p in run.iterdir()) == ["capture-public.json", "offsite-receipt.json", "packet.tar.age"]
    latest = json.loads((env / worker.LATEST).read_text())
    assert latest["version_id"] == "v1" and latest["capture"]["mail_sent"] is False
    assert worker.freshness(env, now=NOW + timedelta(hours=1))["status"] == "fresh"


def test_failed_upload_keeps_no_plaintext_and_no_latest(env):
    with pytest.raises(TimeoutError):
        worker.backup(env, FakeCapture(), lambda: Provider(fail_complete=True), now=NOW)
    run = env / "20261103T041500Z"
    assert not any((run / name).exists() for name in worker.archive.FILES)
    assert json.loads((run / "offsite-receipt.json").read_text())["status"] == "failed"
    assert not (env / worker.LATEST).exists()


def test_missing_custody_refuses_before_quiescing(env):
    class Boom(FakeCapture):
        def quiesced_copy(self, destination):  # pragma: no cover
            raise AssertionError("must not quiesce")

    def no_custody():
        raise ValueError("owner_delivered_primary_custody_required")

    with pytest.raises(ValueError):
        worker.backup(env, Boom(), no_custody, now=NOW)


def test_freshness_fails_when_stale_or_tampered(env):
    result = worker.backup(env, FakeCapture(), lambda: Provider(), now=NOW)
    assert worker.freshness(env, now=NOW + timedelta(hours=27))["checks"]["fresh"] is False
    (env / result["run"] / "packet.tar.age").write_bytes(b"age-encryption.org/v1\nchanged")
    assert worker.freshness(env, now=NOW)["checks"]["local_ciphertext_matches"] is False


def test_recipient_must_be_a_public_age_key(tmp_path):
    bad = tmp_path / "r"
    bad.write_text("AGE-SECRET-KEY-1XYZ")
    with pytest.raises(worker.archive.Refused):
        worker.recipient(bad)