396 lines
14 KiB
Python
396 lines
14 KiB
Python
import hashlib
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gewu.application.uploads import CHUNK_BYTES, LocalArtifactStore, UploadError
|
|
from tests.domain.proposal_fixtures import EXTRACTION
|
|
|
|
|
|
def _begin(store: LocalArtifactStore, content: bytes, *, now: int = 1_000) -> str:
|
|
session = store.begin(
|
|
filename="selected.md",
|
|
media_type="text/markdown",
|
|
byte_length=len(content),
|
|
expected_sha256=hashlib.sha256(content).hexdigest(),
|
|
now=now,
|
|
)
|
|
return session.upload_id
|
|
|
|
|
|
def test_matching_chunk_retry_and_restart_are_idempotent(tmp_path: Path) -> None:
|
|
content = b"# Evidence\nA source block.\n"
|
|
first_store = LocalArtifactStore(tmp_path)
|
|
upload_id = _begin(first_store, content)
|
|
digest = hashlib.sha256(content).hexdigest()
|
|
assert first_store.put_chunk(
|
|
upload_id, offset=0, content=content, chunk_sha256=digest, now=1_001
|
|
) == (0,)
|
|
|
|
restarted = LocalArtifactStore(tmp_path)
|
|
assert restarted.verified_offsets(upload_id, now=1_002) == (0,)
|
|
assert restarted.put_chunk(
|
|
upload_id, offset=0, content=content, chunk_sha256=digest, now=1_002
|
|
) == (0,)
|
|
receipt = restarted.finalize(upload_id, now=1_003)
|
|
assert receipt.sha256 == digest
|
|
assert receipt.status == "staged"
|
|
with restarted.open_artifact(receipt.sha256) as artifact:
|
|
assert artifact.read() == content
|
|
|
|
|
|
def test_pasted_capture_receipt_survives_restart_and_conflicts_on_changed_input(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
first = LocalArtifactStore(tmp_path)
|
|
result = {"sha256": "a" * 64, "source_candidate": {"revision_id": "paste:r1"}}
|
|
with first.capture_receipt_guard():
|
|
first.save_capture_receipt(
|
|
"capture-1",
|
|
owner_id="agent-a",
|
|
project_id="project-a",
|
|
request_sha256="b" * 64,
|
|
result=result,
|
|
)
|
|
|
|
restarted = LocalArtifactStore(tmp_path)
|
|
with restarted.capture_receipt_guard():
|
|
assert (
|
|
restarted.get_capture_receipt(
|
|
"capture-1",
|
|
owner_id="agent-a",
|
|
project_id="project-a",
|
|
request_sha256="b" * 64,
|
|
)
|
|
== result
|
|
)
|
|
with pytest.raises(UploadError) as error:
|
|
restarted.get_capture_receipt(
|
|
"capture-1",
|
|
owner_id="agent-a",
|
|
project_id="project-a",
|
|
request_sha256="c" * 64,
|
|
)
|
|
assert error.value.code == "CAPTURE_CONFLICT"
|
|
with pytest.raises(UploadError) as error:
|
|
restarted.get_capture_receipt(
|
|
"capture-1",
|
|
owner_id="agent-a",
|
|
project_id="project-b",
|
|
request_sha256="b" * 64,
|
|
)
|
|
assert error.value.code == "CAPTURE_CONFLICT"
|
|
assert (
|
|
restarted.get_capture_receipt(
|
|
"capture-1",
|
|
owner_id="agent-b",
|
|
project_id="project-a",
|
|
request_sha256="b" * 64,
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
def test_upload_session_is_private_to_its_principal(tmp_path: Path) -> None:
|
|
store = LocalArtifactStore(tmp_path)
|
|
content = b"# Scoped upload\n"
|
|
upload_id = store.begin(
|
|
filename="scoped.md",
|
|
media_type="text/markdown",
|
|
byte_length=len(content),
|
|
expected_sha256=hashlib.sha256(content).hexdigest(),
|
|
owner_id="agent-a",
|
|
).upload_id
|
|
with pytest.raises(UploadError) as error:
|
|
store.verified_offsets(upload_id, owner_id="agent-b")
|
|
assert error.value.code == "UPLOAD_NOT_FOUND"
|
|
with pytest.raises(UploadError) as error:
|
|
store.put_chunk(
|
|
upload_id,
|
|
offset=0,
|
|
content=content,
|
|
chunk_sha256=hashlib.sha256(content).hexdigest(),
|
|
owner_id="agent-b",
|
|
)
|
|
assert error.value.code == "UPLOAD_NOT_FOUND"
|
|
|
|
|
|
def test_staged_extraction_is_immutable_and_principal_scoped(tmp_path: Path) -> None:
|
|
store = LocalArtifactStore(tmp_path)
|
|
store.save_extraction(EXTRACTION, owner_id="agent-a")
|
|
assert store.get_extraction(EXTRACTION.extraction_id, owner_id="agent-a") == EXTRACTION
|
|
assert store.get_extraction(EXTRACTION.extraction_id, owner_id="agent-b") is None
|
|
with pytest.raises(UploadError) as error:
|
|
store.save_extraction(EXTRACTION, owner_id="agent-b")
|
|
assert error.value.code == "EXTRACTION_STORE_CONFLICT"
|
|
|
|
|
|
def test_artifact_access_is_scoped_to_principal_and_project(tmp_path: Path) -> None:
|
|
store = LocalArtifactStore(tmp_path)
|
|
content = b"private project source"
|
|
digest = hashlib.sha256(content).hexdigest()
|
|
upload_id = store.begin(
|
|
filename="source.md",
|
|
media_type="text/markdown",
|
|
byte_length=len(content),
|
|
expected_sha256=digest,
|
|
owner_id="agent-a",
|
|
project_id="project-a",
|
|
).upload_id
|
|
store.put_chunk(
|
|
upload_id,
|
|
offset=0,
|
|
content=content,
|
|
chunk_sha256=digest,
|
|
owner_id="agent-a",
|
|
)
|
|
store.finalize(upload_id, owner_id="agent-a")
|
|
assert (
|
|
store.artifact_media_type(digest, owner_id="agent-a", project_id="project-a")
|
|
== "text/markdown"
|
|
)
|
|
with store.open_artifact(digest, owner_id="agent-a", project_id="project-a") as stream:
|
|
assert stream.read() == content
|
|
with pytest.raises(UploadError):
|
|
store.open_artifact(digest, owner_id="agent-b", project_id="project-a")
|
|
with pytest.raises(UploadError):
|
|
store.open_artifact(digest, owner_id="agent-a", project_id="project-b")
|
|
|
|
|
|
def test_changed_retry_at_same_offset_is_rejected(tmp_path: Path) -> None:
|
|
store = LocalArtifactStore(tmp_path)
|
|
upload_id = _begin(store, b"abc")
|
|
store.put_chunk(
|
|
upload_id,
|
|
offset=0,
|
|
content=b"abc",
|
|
chunk_sha256=hashlib.sha256(b"abc").hexdigest(),
|
|
now=1_001,
|
|
)
|
|
with pytest.raises(UploadError, match="different bytes") as error:
|
|
store.put_chunk(
|
|
upload_id,
|
|
offset=0,
|
|
content=b"xyz",
|
|
chunk_sha256=hashlib.sha256(b"xyz").hexdigest(),
|
|
now=1_002,
|
|
)
|
|
assert error.value.code == "IDEMPOTENCY_CONFLICT"
|
|
|
|
|
|
def test_chunk_and_complete_artifact_digests_are_verified(tmp_path: Path) -> None:
|
|
store = LocalArtifactStore(tmp_path)
|
|
content = b"verified bytes"
|
|
digest = hashlib.sha256(content).hexdigest()
|
|
upload_id = store.begin(
|
|
filename="verified.md",
|
|
media_type="text/markdown",
|
|
byte_length=len(content),
|
|
expected_sha256="f" * 64,
|
|
).upload_id
|
|
|
|
with pytest.raises(UploadError) as error:
|
|
store.put_chunk(
|
|
upload_id,
|
|
offset=0,
|
|
content=content,
|
|
chunk_sha256="0" * 64,
|
|
)
|
|
assert error.value.code == "CHUNK_DIGEST_MISMATCH"
|
|
|
|
store.put_chunk(upload_id, offset=0, content=content, chunk_sha256=digest)
|
|
with pytest.raises(UploadError) as error:
|
|
store.finalize(upload_id)
|
|
assert error.value.code == "ARTIFACT_DIGEST_MISMATCH"
|
|
with pytest.raises(UploadError) as error:
|
|
store.open_artifact("f" * 64)
|
|
assert error.value.code == "ARTIFACT_NOT_FOUND"
|
|
|
|
|
|
def test_interrupted_upload_does_not_publish_an_artifact(tmp_path: Path) -> None:
|
|
store = LocalArtifactStore(tmp_path)
|
|
content = b"a" * (CHUNK_BYTES + 1)
|
|
digest = hashlib.sha256(content).hexdigest()
|
|
upload_id = store.begin(
|
|
filename="partial.md",
|
|
media_type="text/markdown",
|
|
byte_length=len(content),
|
|
expected_sha256=digest,
|
|
owner_id="agent-a",
|
|
project_id="project-a",
|
|
).upload_id
|
|
first_chunk = content[:CHUNK_BYTES]
|
|
store.put_chunk(
|
|
upload_id,
|
|
offset=0,
|
|
content=first_chunk,
|
|
chunk_sha256=hashlib.sha256(first_chunk).hexdigest(),
|
|
owner_id="agent-a",
|
|
)
|
|
|
|
with pytest.raises(UploadError) as error:
|
|
store.finalize(upload_id, owner_id="agent-a")
|
|
assert error.value.code == "UPLOAD_INCOMPLETE"
|
|
with pytest.raises(UploadError) as error:
|
|
store.open_artifact(digest, owner_id="agent-a", project_id="project-a")
|
|
assert error.value.code == "ARTIFACT_NOT_FOUND"
|
|
|
|
|
|
def test_symlinked_upload_paths_are_not_followed(tmp_path: Path) -> None:
|
|
store = LocalArtifactStore(tmp_path)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
sensitive = outside / "chunk"
|
|
sensitive.write_bytes(b"data")
|
|
|
|
forged_id = "A" * 24
|
|
try:
|
|
(store.upload_root / forged_id).symlink_to(outside, target_is_directory=True)
|
|
except OSError as error:
|
|
pytest.skip(f"directory symlinks are unavailable: {error}")
|
|
with pytest.raises(UploadError) as error:
|
|
store.verified_offsets(forged_id)
|
|
assert error.value.code == "UPLOAD_NOT_FOUND"
|
|
|
|
upload_id = _begin(store, b"data")
|
|
chunk_link = store.upload_root / upload_id / f"{0:020d}.chunk"
|
|
chunk_link.symlink_to(sensitive)
|
|
assert store.verified_offsets(upload_id, now=1_001) == ()
|
|
with pytest.raises(UploadError) as error:
|
|
store.finalize(upload_id, now=1_001)
|
|
assert error.value.code == "UPLOAD_INCOMPLETE"
|
|
assert sensitive.read_bytes() == b"data"
|
|
|
|
|
|
def test_upload_requires_order_complete_digest_and_utf8(tmp_path: Path) -> None:
|
|
store = LocalArtifactStore(tmp_path)
|
|
content = b"a" * (CHUNK_BYTES + 1)
|
|
upload_id = _begin(store, content)
|
|
final_chunk = content[CHUNK_BYTES:]
|
|
with pytest.raises(UploadError, match="in order"):
|
|
store.put_chunk(
|
|
upload_id,
|
|
offset=CHUNK_BYTES,
|
|
content=final_chunk,
|
|
chunk_sha256=hashlib.sha256(final_chunk).hexdigest(),
|
|
now=1_001,
|
|
)
|
|
with pytest.raises(UploadError) as error:
|
|
store.finalize(upload_id, now=1_002)
|
|
assert error.value.code == "UPLOAD_INCOMPLETE"
|
|
|
|
invalid_bytes = b"\xff"
|
|
invalid_id = _begin(store, invalid_bytes)
|
|
store.put_chunk(
|
|
invalid_id,
|
|
offset=0,
|
|
content=invalid_bytes,
|
|
chunk_sha256=hashlib.sha256(invalid_bytes).hexdigest(),
|
|
now=1_001,
|
|
)
|
|
with pytest.raises(UploadError, match="UTF-8") as error:
|
|
store.finalize(invalid_id, now=1_002)
|
|
assert error.value.code == "TEXT_ENCODING_INVALID"
|
|
|
|
|
|
def test_rejects_unsupported_types_expiry_and_malicious_ids(tmp_path: Path) -> None:
|
|
store = LocalArtifactStore(tmp_path)
|
|
with pytest.raises(UploadError) as error:
|
|
store.begin(
|
|
filename="payload.exe",
|
|
media_type="application/octet-stream",
|
|
byte_length=4,
|
|
expected_sha256=hashlib.sha256(b"data").hexdigest(),
|
|
now=1_000,
|
|
)
|
|
assert error.value.code == "UNSUPPORTED_MEDIA_TYPE"
|
|
|
|
upload_id = _begin(store, b"data")
|
|
with pytest.raises(UploadError) as error:
|
|
store.verified_offsets(upload_id, now=1_601)
|
|
assert error.value.code == "UPLOAD_EXPIRED"
|
|
with pytest.raises(UploadError) as error:
|
|
store.verified_offsets("../../outside", now=1_001)
|
|
assert error.value.code == "UPLOAD_NOT_FOUND"
|
|
|
|
|
|
def test_pdf_upload_requires_matching_suffix_and_pdf_signature(tmp_path: Path) -> None:
|
|
store = LocalArtifactStore(tmp_path)
|
|
payload = b"%PDF-fixture"
|
|
digest = hashlib.sha256(payload).hexdigest()
|
|
with pytest.raises(UploadError) as error:
|
|
store.begin(
|
|
filename="source.txt",
|
|
media_type="application/pdf",
|
|
byte_length=len(payload),
|
|
expected_sha256=digest,
|
|
)
|
|
assert error.value.code == "UNSUPPORTED_MEDIA_TYPE"
|
|
upload_id = store.begin(
|
|
filename="source.pdf",
|
|
media_type="application/pdf",
|
|
byte_length=len(payload),
|
|
expected_sha256=digest,
|
|
).upload_id
|
|
store.put_chunk(upload_id, offset=0, content=payload, chunk_sha256=digest)
|
|
assert store.finalize(upload_id).media_type == "application/pdf"
|
|
|
|
invalid = b"not pdf"
|
|
invalid_digest = hashlib.sha256(invalid).hexdigest()
|
|
invalid_id = store.begin(
|
|
filename="invalid.pdf",
|
|
media_type="application/pdf",
|
|
byte_length=len(invalid),
|
|
expected_sha256=invalid_digest,
|
|
).upload_id
|
|
store.put_chunk(invalid_id, offset=0, content=invalid, chunk_sha256=invalid_digest)
|
|
with pytest.raises(UploadError) as error:
|
|
store.finalize(invalid_id)
|
|
assert error.value.code == "PDF_INVALID"
|
|
|
|
|
|
def test_identical_artifact_bytes_dedupe_without_collapsing_upload_receipts(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = LocalArtifactStore(tmp_path)
|
|
content = b"Shared source content."
|
|
digest = hashlib.sha256(content).hexdigest()
|
|
first_id = store.begin(
|
|
filename="shared.md",
|
|
media_type="text/markdown",
|
|
byte_length=len(content),
|
|
expected_sha256=digest,
|
|
owner_id="agent-shared",
|
|
project_id="project-a",
|
|
).upload_id
|
|
second_session = store.begin(
|
|
filename="second-copy.txt",
|
|
media_type="text/plain",
|
|
byte_length=len(content),
|
|
expected_sha256=digest,
|
|
owner_id="agent-shared",
|
|
project_id="project-b",
|
|
now=1_000,
|
|
)
|
|
for upload_id in (first_id, second_session.upload_id):
|
|
store.put_chunk(
|
|
upload_id,
|
|
offset=0,
|
|
content=content,
|
|
chunk_sha256=digest,
|
|
owner_id="agent-shared",
|
|
now=1_001,
|
|
)
|
|
first = store.finalize(first_id, owner_id="agent-shared", now=1_002)
|
|
second = store.finalize(second_session.upload_id, owner_id="agent-shared", now=1_002)
|
|
assert first.sha256 == second.sha256
|
|
assert first.byte_length == second.byte_length == len(content)
|
|
for project_id in ("project-a", "project-b"):
|
|
with store.open_artifact(
|
|
digest, owner_id="agent-shared", project_id=project_id
|
|
) as artifact:
|
|
assert artifact.read() == content
|
|
with pytest.raises(UploadError) as error:
|
|
store.open_artifact(digest, owner_id="agent-shared", project_id="project-c")
|
|
assert error.value.code == "ARTIFACT_NOT_FOUND"
|