a8fc1a3438
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11538-05ea-7b63-b3c7-a2df5f38898d
204 lines
8.2 KiB
Python
204 lines
8.2 KiB
Python
"""Immutable archive transfer with a synthetic provider; no live credentials."""
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def archive():
|
|
spec = importlib.util.spec_from_file_location(
|
|
"primary_application_archive",
|
|
Path(__file__).resolve().parents[1] / "scripts/primary_application_archive.py",
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class NotFound(Exception):
|
|
response = {"ResponseMetadata": {"HTTPStatusCode": 404}}
|
|
|
|
|
|
class Provider:
|
|
def __init__(self, data, failure=None, exists=False):
|
|
self.data, self.failure, self.exists = data, failure, exists
|
|
self.calls = []
|
|
self.version = "synthetic-version-1"
|
|
|
|
def head_object(self, **kwargs):
|
|
self.calls.append(("head", kwargs))
|
|
if self.failure == "permission":
|
|
raise PermissionError("synthetic provider failure")
|
|
if not self.exists:
|
|
raise NotFound()
|
|
return {
|
|
"ContentLength": len(self.data),
|
|
"VersionId": self.version,
|
|
"Metadata": {"sha256": hashlib.sha256(self.data).hexdigest()},
|
|
}
|
|
|
|
def create_multipart_upload(self, **kwargs):
|
|
self.calls.append(("start", kwargs))
|
|
return {"UploadId": "own-unfinished-upload"}
|
|
|
|
def upload_part(self, **kwargs):
|
|
self.calls.append(("part", kwargs))
|
|
if self.failure == "part":
|
|
raise TimeoutError("synthetic timeout")
|
|
assert kwargs["Body"] == self.data
|
|
return {"ETag": "synthetic-part-etag"}
|
|
|
|
def complete_multipart_upload(self, **kwargs):
|
|
self.calls.append(("complete", kwargs))
|
|
assert kwargs["IfNoneMatch"] == "*"
|
|
self.exists = True
|
|
if self.failure == "lost-response":
|
|
raise TimeoutError("synthetic lost acknowledgement")
|
|
return {"VersionId": None if self.failure == "unversioned" else self.version}
|
|
|
|
def abort_multipart_upload(self, **kwargs):
|
|
self.calls.append(("abort", kwargs))
|
|
assert kwargs["UploadId"] == "own-unfinished-upload"
|
|
|
|
def get_object(self, **kwargs):
|
|
self.calls.append(("get", kwargs))
|
|
assert kwargs["VersionId"] == self.version
|
|
data = b"X" * len(self.data) if self.failure == "corrupt" else self.data
|
|
if self.failure == "oversize":
|
|
data += b"extra"
|
|
return {
|
|
"ContentLength": len(self.data),
|
|
"VersionId": "other-version" if self.failure == "wrong-version" else self.version,
|
|
"Body": io.BytesIO(data),
|
|
}
|
|
|
|
|
|
def fixture(archive, tmp_path, failure=None, exists=False):
|
|
tmp_path.chmod(0o700)
|
|
source = tmp_path / "packet.tar.age"
|
|
source.write_bytes(b"age-encryption.org/v1\nsynthetic-encrypted-pair")
|
|
source.chmod(0o600)
|
|
receipt = {"object_key": archive.key_for("coulomb-social", "2026-10-07T20:00:00Z")}
|
|
return source, tmp_path / "fetched.age", receipt, Provider(source.read_bytes(), failure, exists)
|
|
|
|
|
|
def test_conditional_create_and_version_pinned_readback(archive, tmp_path):
|
|
source, output, receipt, client = fixture(archive, tmp_path)
|
|
saved = []
|
|
result = archive.transfer(
|
|
client, "coulomb-social", source, output, receipt, lambda: saved.append(dict(receipt))
|
|
)
|
|
assert result["status"] == "offsite_fetched_verified"
|
|
assert result["remote_retrieval_verified"] and result["immutable_create"]
|
|
assert result["version_id"] == client.version
|
|
assert output.read_bytes() == source.read_bytes()
|
|
assert output.stat().st_mode & 0o777 == 0o600
|
|
assert [call[0] for call in client.calls] == ["head", "start", "part", "complete", "get"]
|
|
assert not saved[0]["remote_retrieval_verified"]
|
|
assert saved[-1]["remote_retrieval_verified"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"failure",
|
|
["part", "lost-response", "unversioned", "corrupt", "oversize", "wrong-version", "permission"],
|
|
)
|
|
def test_failure_never_declares_retrieval_or_deletes_retained_objects(archive, tmp_path, failure):
|
|
source, output, receipt, client = fixture(archive, tmp_path, failure)
|
|
with pytest.raises((ValueError, TimeoutError, PermissionError)):
|
|
archive.transfer(client, "coulomb-social", source, output, receipt)
|
|
assert not receipt["remote_retrieval_verified"]
|
|
assert not any(name == "delete" for name, _ in client.calls)
|
|
if failure in {"part", "lost-response"}:
|
|
assert len([call for call in client.calls if call[0] == "abort"]) == 1
|
|
else:
|
|
assert not any(name == "abort" for name, _ in client.calls)
|
|
|
|
|
|
def test_uncertain_completion_retries_readback_without_another_upload(archive, tmp_path):
|
|
source, output, receipt, client = fixture(archive, tmp_path, "lost-response")
|
|
with pytest.raises(TimeoutError):
|
|
archive.transfer(client, "coulomb-social", source, output, receipt)
|
|
key = receipt["object_key"]
|
|
client.failure = None
|
|
archive.transfer(client, "coulomb-social", source, output, receipt)
|
|
assert receipt["object_key"] == key and receipt["reused_verified_binding"]
|
|
assert len([call for call in client.calls if call[0] == "start"]) == 1
|
|
assert receipt["remote_retrieval_verified"]
|
|
|
|
|
|
def test_existing_key_must_bind_bytes_and_previous_version(archive, tmp_path):
|
|
source, output, receipt, client = fixture(archive, tmp_path, exists=True)
|
|
client.data = b"other ciphertext"
|
|
with pytest.raises(ValueError, match="existing_archive_binding_mismatch"):
|
|
archive.transfer(client, "coulomb-social", source, output, receipt)
|
|
assert [name for name, _ in client.calls] == ["head"]
|
|
client.data = source.read_bytes()
|
|
receipt["version_id"] = "previous-version"
|
|
with pytest.raises(ValueError, match="existing_archive_version_changed"):
|
|
archive.transfer(client, "coulomb-social", source, output, receipt)
|
|
assert not output.exists()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"invalid", ["plaintext", "scope", "receipt-hash", "linked-source", "linked-output"]
|
|
)
|
|
def test_invalid_packet_refuses_before_provider_contact(archive, tmp_path, invalid):
|
|
source, output, receipt, client = fixture(archive, tmp_path)
|
|
if invalid == "plaintext":
|
|
source.write_bytes(b"private unencrypted data")
|
|
elif invalid == "scope":
|
|
receipt["object_key"] = archive.key_for("email-connect", "2026-10-07T20:00:00Z")
|
|
elif invalid == "receipt-hash":
|
|
receipt["ciphertext_sha256"] = "bad"
|
|
elif invalid == "linked-source":
|
|
target = tmp_path / "real.age"
|
|
source.rename(target)
|
|
source.symlink_to(target)
|
|
else:
|
|
output.symlink_to(tmp_path / "absent")
|
|
with pytest.raises(ValueError):
|
|
archive.transfer(client, "coulomb-social", source, output, receipt)
|
|
assert client.calls == []
|
|
|
|
|
|
def test_retrieve_after_source_loss_uses_saved_version_without_provider_writes(archive, tmp_path):
|
|
source, output, receipt, client = fixture(archive, tmp_path)
|
|
receipt.update(
|
|
schema="platform.application-primary-archive.v1", captured_at="2026-10-07T20:00:00Z"
|
|
)
|
|
archive.transfer(client, "coulomb-social", source, output, receipt)
|
|
source.unlink()
|
|
output.unlink()
|
|
client.calls.clear()
|
|
result = archive.retrieve(client, "coulomb-social", output, receipt)
|
|
assert result["status"] == "retrieved_verified"
|
|
assert not result["provider_mutations"] and not result["application_restore_verified"]
|
|
assert output.read_bytes() == client.data
|
|
assert [name for name, _ in client.calls] == ["get"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"field,value",
|
|
[
|
|
("remote_retrieval_verified", False),
|
|
("destination", "s3://another-bucket/key"),
|
|
("ciphertext_sha256", "bad"),
|
|
("version_id", "null"),
|
|
("ciphertext_bytes", -1),
|
|
("application", "email-connect"),
|
|
],
|
|
)
|
|
def test_retrieval_rejects_unbound_receipt_before_contact(archive, tmp_path, field, value):
|
|
source, output, receipt, client = fixture(archive, tmp_path)
|
|
receipt.update(schema="platform.application-primary-archive.v1")
|
|
archive.transfer(client, "coulomb-social", source, output, receipt)
|
|
output.unlink()
|
|
client.calls.clear()
|
|
receipt[field] = value
|
|
with pytest.raises(ValueError):
|
|
archive.retrieve(client, "coulomb-social", output, receipt)
|
|
assert client.calls == []
|