284 lines
13 KiB
Python
284 lines
13 KiB
Python
import hashlib
|
|
import os
|
|
import zipfile
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.package import (
|
|
PackageLimits, PackageReader, PackageValidator, PackageWriteError, PackageWriter,
|
|
)
|
|
|
|
|
|
def assert_rejected(writer, source, target, code=None, **kwargs):
|
|
before = source.read_bytes()
|
|
target_before = target.read_bytes() if target.exists() else None
|
|
with pytest.raises(PackageWriteError) as error:
|
|
writer.write(source, target, **kwargs)
|
|
if code:
|
|
assert error.value.result.issues[0].code == code
|
|
assert source.read_bytes() == before
|
|
assert (target.read_bytes() if target.exists() else None) == target_before
|
|
assert not list(target.parent.glob(".vbcase-*.tmp"))
|
|
|
|
|
|
def test_full_roundtrip(manifest, make_package, tmp_path):
|
|
manifest["extension"] = {"unknown": [1, 2]}
|
|
manifest["routing"]["custom"] = "preserve"
|
|
manifest["artifacts"]["vendor"] = "unknown/binary"
|
|
originals = {"original/report.json": b'{"report":"generic"}',
|
|
"unknown/binary": bytes(range(256)), "empty/": b"",
|
|
"analysis/revisions/0001/data.json": b"old revision",
|
|
"analysis/current.json": b'{"revision":1}'}
|
|
source = make_package(entries=originals)
|
|
before = source.read_bytes()
|
|
target = tmp_path / "renamed.anything"
|
|
result = PackageWriter().write(
|
|
source, target,
|
|
additions={"analysis/revisions/0002/data.json": b"new revision"},
|
|
index_updates={"analysis/current.json": b'{"revision":2}'},
|
|
manifest_updates={"state": "VALIDATED", "routing": {"destination": "cases"},
|
|
"analysis_revision": 2, "history": {"analysis_revisions": 2},
|
|
"artifacts": {"case": "analysis/revisions/0002/data.json"}},
|
|
)
|
|
assert result.valid and PackageValidator().validate(target).valid
|
|
assert source.read_bytes() == before
|
|
with PackageReader(target) as reader:
|
|
updated = reader.manifest
|
|
for field in ("package_id", "case_id", "created_at", "project", "extension"):
|
|
assert updated[field] == manifest[field]
|
|
assert updated["package_generation"] == 1
|
|
assert updated["analysis_revision"] == 2
|
|
assert updated["routing"]["custom"] == "preserve"
|
|
assert updated["artifacts"]["vendor"] == "unknown/binary"
|
|
assert updated["updated_at"] >= manifest["updated_at"]
|
|
for name, data in originals.items():
|
|
if name != "analysis/current.json" and not name.endswith("/"):
|
|
assert reader.read_bytes(name) == data
|
|
assert updated["integrity"][name]["sha256"] == hashlib.sha256(data).hexdigest()
|
|
with zipfile.ZipFile(target) as archive:
|
|
assert "empty/" in archive.namelist()
|
|
next_target = tmp_path / "generation2"
|
|
PackageWriter().write(target, next_target)
|
|
with PackageReader(next_target) as reader:
|
|
assert reader.manifest["package_generation"] == 2
|
|
assert reader.manifest["analysis_revision"] == 2
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["original/report", "unknown/file", "analysis/revisions/0001/data"])
|
|
def test_immutable_existing_files(name, make_package, tmp_path):
|
|
source = make_package(entries={name: b"old"})
|
|
assert_rejected(PackageWriter(), source, tmp_path / "out", "ARTIFACT_EXISTS", additions={name: b"new"})
|
|
|
|
|
|
def test_frozen_revision_namespace(make_package, tmp_path):
|
|
source = make_package(entries={"analysis/revisions/0001/data": b"old"})
|
|
assert_rejected(PackageWriter(), source, tmp_path / "out", "REVISION_IMMUTABLE", additions={"analysis/revisions/0001/new": b"new"})
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["original/report", "unknown/file", "analysis/revisions/0001/data", "manifest.json"])
|
|
def test_index_allowlist(name, make_package, tmp_path):
|
|
source = make_package(entries={name: b"old"} if name != "manifest.json" else {})
|
|
assert_rejected(PackageWriter(), source, tmp_path / "out", index_updates={name: b"new"})
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["../escape", "/abs", "x\\y", "manifest.json", "a:stream"])
|
|
def test_unsafe_additions(name, make_package, tmp_path):
|
|
assert_rejected(PackageWriter(), make_package(), tmp_path / "out", "UNSAFE_PATH", additions={name: b"new"})
|
|
|
|
|
|
@pytest.mark.parametrize("entries,addition", [({"a": b"x"}, "a/b"), ({"a/b": b"x"}, "a"), ({"A": b"x"}, "a"), ({"dir/": b""}, "dir")])
|
|
def test_colliding_paths(entries, addition, make_package, tmp_path):
|
|
assert_rejected(PackageWriter(), make_package(entries=entries), tmp_path / "out", "ENTRY_COLLISION", additions={addition: b"new"})
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [("package_id", "new"), ("case_id", "new"), ("package_generation", 99), ("created_at", "2026-01-01T00:00:00Z"), ("integrity", {}), ("project", {}), ("format_version", 2)])
|
|
def test_reserved_manifest_fields(field, value, make_package, tmp_path):
|
|
assert_rejected(PackageWriter(), make_package(), tmp_path / "out", "MANIFEST_UPDATE_FORBIDDEN", manifest_updates={field: value})
|
|
|
|
|
|
@pytest.mark.parametrize("updates", [{"state": "WRONG"}, {"artifacts": {"case": "absent"}}, {"routing": {"destination": "wrong"}}, {"analysis_revision": -1}])
|
|
def test_invalid_manifest_update(updates, make_package, tmp_path):
|
|
assert_rejected(PackageWriter(), make_package(), tmp_path / "out", manifest_updates=updates)
|
|
|
|
|
|
def test_revision_and_history_cannot_decrease(manifest, make_package, tmp_path):
|
|
manifest["analysis_revision"] = 2
|
|
manifest["history"]["evidence"] = 3
|
|
source = make_package()
|
|
assert_rejected(PackageWriter(), source, tmp_path / "out", "REVISION_REGRESSION", manifest_updates={"analysis_revision": 1})
|
|
assert_rejected(PackageWriter(), source, tmp_path / "out", "HISTORY_REGRESSION", manifest_updates={"history": {"evidence": 2}})
|
|
|
|
|
|
def test_invalid_input(make_package, tmp_path):
|
|
assert_rejected(PackageWriter(), make_package(raw=b"{}"), tmp_path / "out")
|
|
|
|
|
|
def test_input_alias_and_hardlink(make_package, tmp_path):
|
|
source = make_package()
|
|
assert_rejected(PackageWriter(), source, source, "INPUT_TARGET_ALIAS", overwrite=True)
|
|
alias = tmp_path / "alias"
|
|
os.link(source, alias)
|
|
assert_rejected(PackageWriter(), source, alias, "INPUT_TARGET_ALIAS", overwrite=True)
|
|
|
|
|
|
def test_explicit_target_replacement(make_package, tmp_path):
|
|
source, target = make_package(), tmp_path / "out"
|
|
target.write_bytes(b"old target")
|
|
assert_rejected(PackageWriter(), source, target, "TARGET_EXISTS")
|
|
assert PackageWriter().write(source, target, overwrite=True).valid
|
|
|
|
|
|
def test_failure_preserves_existing_target(make_package, tmp_path):
|
|
source, target = make_package(), tmp_path / "out"
|
|
target.write_bytes(b"old target")
|
|
assert_rejected(PackageWriter(), source, target, overwrite=True, manifest_updates={"artifacts": {"case": "absent"}})
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [("max_file_bytes", 1024), ("max_total_bytes", 2000), ("max_entries", 1), ("max_archive_bytes", 2000), ("max_manifest_bytes", 800)])
|
|
def test_output_limits(field, value, make_package, tmp_path):
|
|
limits = PackageLimits(**{field: value})
|
|
assert_rejected(PackageWriter(limits), make_package(), tmp_path / "out", additions={"new": b"x" * 3000})
|
|
|
|
|
|
def test_compressible_artifact_is_valid_with_default_policy(make_package, tmp_path):
|
|
source, target = make_package(), tmp_path / "out"
|
|
assert PackageWriter().write(source, target, additions={"new": b"x" * 100_000}).valid
|
|
|
|
|
|
def test_integrity_extensions_retained(manifest, make_package, tmp_path):
|
|
manifest["integrity"]["analysis/current.json"] = {"size": 3, "sha256": hashlib.sha256(b"old").hexdigest(), "vendor": "preserve"}
|
|
source = make_package(entries={"analysis/current.json": b"old"})
|
|
result = PackageWriter().write(source, tmp_path / "out", index_updates={"analysis/current.json": b"new"})
|
|
assert result.manifest["integrity"]["analysis/current.json"]["vendor"] == "preserve"
|
|
|
|
|
|
def test_atomic_replace_failure(make_package, tmp_path, monkeypatch):
|
|
source, target = make_package(), tmp_path / "out"
|
|
target.write_bytes(b"old")
|
|
def fail(*args):
|
|
raise OSError("simulated rename failure")
|
|
monkeypatch.setattr(os, "replace", fail)
|
|
assert_rejected(PackageWriter(), source, target, "WRITE_ERROR", overwrite=True)
|
|
|
|
|
|
def test_competing_target_not_clobbered(make_package, tmp_path, monkeypatch):
|
|
source, target = make_package(), tmp_path / "out"
|
|
original_link = os.link
|
|
def race(src, dst):
|
|
target.write_bytes(b"competitor")
|
|
original_link(src, dst)
|
|
monkeypatch.setattr(os, "link", race)
|
|
with pytest.raises(PackageWriteError):
|
|
PackageWriter().write(source, target)
|
|
assert target.read_bytes() == b"competitor"
|
|
assert not list(tmp_path.glob(".vbcase-*.tmp"))
|
|
|
|
|
|
def test_preservation_check_detects_valid_tampered_output(make_package, tmp_path, monkeypatch):
|
|
source = make_package(entries={"unknown": b"original"})
|
|
writer = PackageWriter()
|
|
original_validate = writer.validator.validate
|
|
def tamper(path):
|
|
with zipfile.ZipFile(path) as archive:
|
|
entries = {name: archive.read(name) for name in archive.namelist()}
|
|
import json
|
|
doc = json.loads(entries["manifest.json"])
|
|
entries["unknown"] = b"tampered"
|
|
doc["integrity"]["unknown"].update(size=8, sha256=hashlib.sha256(b"tampered").hexdigest())
|
|
entries["manifest.json"] = json.dumps(doc).encode()
|
|
with zipfile.ZipFile(path, "w") as archive:
|
|
for name, data in entries.items():
|
|
archive.writestr(name, data)
|
|
return original_validate(path)
|
|
monkeypatch.setattr(writer.validator, "validate", tamper)
|
|
assert_rejected(writer, source, tmp_path / "out", "MANIFEST_CHANGED")
|
|
|
|
|
|
def test_preservation_check_detects_unlisted_file_drop(make_package, tmp_path, monkeypatch):
|
|
source = make_package(entries={"unknown": b"original"})
|
|
writer = PackageWriter()
|
|
original_verify = writer._verify_preservation
|
|
def drop(reader, path, indexes):
|
|
with zipfile.ZipFile(path) as archive:
|
|
entries = {name: archive.read(name) for name in archive.namelist() if name != "unknown"}
|
|
with zipfile.ZipFile(path, "w") as archive:
|
|
for name, data in entries.items():
|
|
archive.writestr(name, data)
|
|
original_verify(reader, path, indexes)
|
|
monkeypatch.setattr(writer, "_verify_preservation", drop)
|
|
assert_rejected(writer, source, tmp_path / "out", "PRESERVATION_FAILED")
|
|
|
|
|
|
def test_frozen_empty_revision_directory(make_package, tmp_path):
|
|
source = make_package(entries={"analysis/revisions/0001/": b""})
|
|
assert_rejected(PackageWriter(), source, tmp_path / "out", "REVISION_IMMUTABLE", additions={"analysis/revisions/0001/new": b"new"})
|
|
|
|
|
|
def test_missing_target_parent(make_package, tmp_path):
|
|
source = make_package()
|
|
with pytest.raises(PackageWriteError):
|
|
PackageWriter().write(source, tmp_path / "absent" / "out")
|
|
|
|
|
|
def test_new_index_and_type_validation(make_package, tmp_path):
|
|
source = make_package()
|
|
assert PackageWriter().write(source, tmp_path / "out", index_updates={"analysis/current.json": b"{}"}).valid
|
|
assert_rejected(PackageWriter(), source, tmp_path / "other", "ARTIFACT_TYPE", additions={"new": "string"})
|
|
|
|
|
|
def test_future_timestamp_does_not_regress(manifest, make_package, tmp_path):
|
|
manifest["updated_at"] = "2099-01-01T00:00:00Z"
|
|
result = PackageWriter().write(make_package(), tmp_path / "out")
|
|
from datetime import datetime
|
|
assert datetime.fromisoformat(result.manifest["updated_at"]) == datetime.fromisoformat("2099-01-01T00:00:00+00:00")
|
|
|
|
|
|
def test_fsync_failure(make_package, tmp_path, monkeypatch):
|
|
source, target = make_package(), tmp_path / "out"
|
|
target.write_bytes(b"old")
|
|
def fail(*args):
|
|
raise OSError("simulated disk failure")
|
|
monkeypatch.setattr(os, "fsync", fail)
|
|
assert_rejected(PackageWriter(), source, target, "WRITE_ERROR", overwrite=True)
|
|
|
|
|
|
def test_byte_comparison_detects_changed_file(make_package, tmp_path, monkeypatch):
|
|
source = make_package(entries={"unknown": b"original"})
|
|
writer = PackageWriter()
|
|
original_verify = writer._verify_preservation
|
|
def change(reader, path, indexes):
|
|
with zipfile.ZipFile(path) as archive:
|
|
entries = {name: archive.read(name) for name in archive.namelist()}
|
|
entries["unknown"] = b"changed"
|
|
with zipfile.ZipFile(path, "w") as archive:
|
|
for name, data in entries.items():
|
|
archive.writestr(name, data)
|
|
original_verify(reader, path, indexes)
|
|
monkeypatch.setattr(writer, "_verify_preservation", change)
|
|
assert_rejected(writer, source, tmp_path / "out", "PRESERVATION_FAILED")
|
|
|
|
|
|
def test_zip_comments_preserved(make_package, tmp_path):
|
|
source = make_package(entries={"unknown": b"data"})
|
|
with zipfile.ZipFile(source, "a") as archive:
|
|
archive.comment = b"vendor archive comment"
|
|
PackageWriter().write(source, tmp_path / "out")
|
|
with zipfile.ZipFile(tmp_path / "out") as archive:
|
|
assert archive.comment == b"vendor archive comment"
|
|
|
|
|
|
def test_cleanup_failure_after_publication_reports_success(make_package, tmp_path, monkeypatch):
|
|
from pathlib import Path
|
|
source = make_package()
|
|
original_unlink = Path.unlink
|
|
def fail(path, *args, **kwargs):
|
|
if path.name.startswith(".vbcase-"):
|
|
raise OSError("simulated cleanup failure")
|
|
return original_unlink(path, *args, **kwargs)
|
|
monkeypatch.setattr(Path, "unlink", fail)
|
|
with pytest.warns(ResourceWarning, match="Package published"):
|
|
assert PackageWriter().write(source, tmp_path / "out").valid
|
|
assert PackageValidator().validate(tmp_path / "out").valid
|
|
for path in tmp_path.glob(".vbcase-*.tmp"):
|
|
original_unlink(path)
|