tegwick
f77e2333af
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11fd9-caf6-7a63-b578-9daaaa134ff2
45 lines
1.9 KiB
Python
45 lines
1.9 KiB
Python
"""Synthetic immutable mirror/readback and recovery-receipt checks."""
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.mark.parametrize("failure", [None, "ciphertext", "receipt", "unversioned"])
|
|
def test_mirror_refuses_corruption_before_success(monkeypatch, tmp_path, failure):
|
|
scripts = Path(__file__).resolve().parents[1] / "scripts"
|
|
monkeypatch.syspath_prepend(str(scripts))
|
|
spec = importlib.util.spec_from_file_location("mirror", scripts / "mirror-ledger.py")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
archive = tmp_path / "ledger.age"
|
|
archive.write_bytes(b"synthetic encrypted data")
|
|
archive.chmod(0o600)
|
|
objects = {}
|
|
|
|
class Client:
|
|
def put_object(self, **kwargs):
|
|
assert kwargs["IfNoneMatch"] == "*"
|
|
body = kwargs["Body"]
|
|
objects[kwargs["Key"]] = body.read() if hasattr(body, "read") else body
|
|
return {"VersionId": "null" if failure == "unversioned" else "synthetic-version"}
|
|
|
|
def get_object(self, **kwargs):
|
|
body = objects[kwargs["Key"]]
|
|
if ((failure == "ciphertext" and not kwargs["Key"].endswith("json"))
|
|
or (failure == "receipt" and kwargs["Key"].endswith("json"))):
|
|
body = b"corrupted"
|
|
return {"VersionId": kwargs["VersionId"], "Body": io.BytesIO(body)}
|
|
|
|
receipt = {"object_key": "platform-pg/application-archives/email-connect/synthetic.age",
|
|
"ciphertext_sha256": hashlib.sha256(archive.read_bytes()).hexdigest(),
|
|
"ciphertext_bytes": archive.stat().st_size}
|
|
if failure:
|
|
with pytest.raises(module.Refused):
|
|
module.mirror(Client(), archive, receipt)
|
|
else:
|
|
result = module.mirror(Client(), archive, receipt)
|
|
assert result["hash_readback_verified"]
|
|
assert result["receipt_key"] in objects
|