feat: add case package validation foundation

tests/conftest.py

@@ -0,0 +1,31 @@
+import json
+import zipfile
+
+import pytest
+
+
+@pytest.fixture
+def manifest():
+    return {
+        "format": "viano-bug-case-package", "format_version": 1,
+        "package_id": "stable-package", "package_generation": 0, "case_id": "case-A",
+        "project": {"type": "generic", "profile": "test"},
+        "created_at": "2026-01-01T00:00:00Z", "updated_at": "2026-01-01T00:00:00Z",
+        "state": "NEW", "routing": {"destination": "inbox", "reason": None, "retry_count": 0, "retry_not_before": None},
+        "analysis_revision": 0,
+        "artifacts": {"case": None, "diagnosis": None, "change_request": None},
+        "history": {"analysis_revisions": 0, "reproductions": 0, "evidence": 0, "handoffs": 0, "feedback": 0, "events_ref": None},
+        "integrity": {}
+    }
+
+
+@pytest.fixture
+def make_package(tmp_path, manifest):
+    def make(*, entries=None, document=None, raw=None, name="arbitrary.bin", compression=zipfile.ZIP_STORED):
+        path = tmp_path / name
+        with zipfile.ZipFile(path, "w", compression=compression) as archive:
+            archive.writestr("manifest.json", raw if raw is not None else json.dumps(document if document is not None else manifest))
+            for key, value in (entries or {}).items():
+                archive.writestr(key, value)
+        return path
+    return make

tests/test_manifest.py

@@ -0,0 +1,51 @@
+
+import pytest
+from jsonschema import Draft202012Validator
+import json
+from importlib.resources import files
+
+from viano_bug_analysis.package.manifest import manifest_issues, parse_manifest
+
+
+def test_schema_is_valid():
+    Draft202012Validator.check_schema(json.loads(files("viano_bug_analysis.schemas").joinpath("manifest-v1.schema.json").read_text()))
+
+
+@pytest.mark.parametrize("field,value", [("format_version", 2), ("format_version", True), ("package_generation", -1), ("analysis_revision", True), ("state", "UNKNOWN"), ("created_at", "2026-01-01"), ("package_id", " ")])
+def test_invalid_fields(manifest, field, value):
+    manifest[field] = value
+    assert manifest_issues(manifest)
+
+
+def test_missing_state_and_routing(manifest):
+    del manifest["state"]
+    manifest["status"] = "NEW"
+    assert manifest_issues(manifest)
+
+
+def test_independent_generation_revision_and_routing(manifest):
+    manifest.update(package_generation=15, analysis_revision=2, state="BLOCKED")
+    manifest["routing"].update(destination="inbox", reason="Dependency unavailable", retry_count=3, retry_not_before="2026-01-02T00:00:00Z")
+    assert not manifest_issues(manifest)
+
+
+def test_time_order(manifest):
+    manifest["updated_at"] = "2025-01-01T00:00:00Z"
+    assert manifest_issues(manifest)[0].code == "MANIFEST_TIME"
+
+
+@pytest.mark.parametrize("raw", [b'{"a":1,"a":2}', b'{"a":NaN}', b'\xff'])
+def test_strict_json(raw):
+    with pytest.raises((ValueError, UnicodeError)):
+        parse_manifest(raw)
+
+
+@pytest.mark.parametrize("timestamp", ["2026-02-30T00:00:00Z", "2026-01-01T00:00:00", "2026-01-01T00:00:00+00:60", "2026-01-01T00:00:60Z"])
+def test_invalid_timestamps(timestamp, manifest):
+    manifest["routing"]["retry_not_before"] = timestamp
+    assert manifest_issues(manifest)
+
+
+def test_lowercase_timestamp(manifest):
+    manifest["created_at"] = "2026-01-01t00:00:00z"
+    assert not manifest_issues(manifest)

tests/test_package_reader.py

@@ -0,0 +1,36 @@
+import pytest
+
+from viano_bug_analysis.package import PackageReader, PackageReadError
+
+
+def test_read_only_and_filename_independent(make_package):
+    path = make_package(entries={"unknown/data.bin": b"\x00\xff"})
+    before = path.read_bytes()
+    renamed = path.with_name("different.name")
+    path.rename(renamed)
+    with PackageReader(renamed) as reader:
+        assert reader.manifest["package_id"] == "stable-package"
+        assert reader.read_bytes("unknown/data.bin") == b"\x00\xff"
+        modified = reader.manifest
+        modified["case_id"] = "changed"
+        assert reader.manifest["case_id"] == "case-A"
+        with pytest.raises(PackageReadError):
+            reader.read_bytes("../outside")
+    assert renamed.read_bytes() == before
+    with pytest.raises(RuntimeError):
+        reader.read_bytes("manifest.json")
+
+
+def test_reader_rejection_and_cleanup(make_package):
+    reader = PackageReader(make_package(raw=b"{}"))
+    with pytest.raises(PackageReadError) as error:
+        with reader:
+            pytest.fail("Invalid package opened")
+    assert not error.value.result.valid
+    assert reader._source is None
+
+
+def test_reader_missing_file(tmp_path):
+    with pytest.raises(PackageReadError):
+        with PackageReader(tmp_path / "absent"):
+            pass

tests/test_package_security.py

@@ -0,0 +1,119 @@
+import dataclasses
+import stat
+import struct
+import warnings
+import zipfile
+
+import pytest
+
+from viano_bug_analysis.package import PackageLimits, PackageReadError, PackageReader, PackageValidator
+
+
+@pytest.mark.parametrize("name", ["../escape", "a/../../escape", "/absolute", "C:/escape", "//server/file", "a\\b", "a/./b", "a//b", "a:stream", "a\x01b"])
+def test_unsafe_paths(name, make_package):
+    path = make_package(entries={name.replace("\\", "/"): b"x"})
+    if "\\" in name:
+        path.write_bytes(path.read_bytes().replace(name.replace("\\", "/").encode(), name.encode()))
+    assert PackageValidator().validate(path).issues[0].code == "UNSAFE_PATH"
+
+
+@pytest.mark.parametrize("entries", [{"A": b"x", "a": b"y"}, {"a": b"x", "a/b": b"y"}])
+def test_collisions(entries, make_package):
+    assert PackageValidator().validate(make_package(entries=entries)).issues[0].code == "ENTRY_COLLISION"
+
+
+def test_duplicate_manifest(make_package):
+    path = make_package()
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", UserWarning)
+        with zipfile.ZipFile(path, "a") as archive:
+            archive.writestr("manifest.json", "{}")
+    assert PackageValidator().validate(path).issues[0].code == "ENTRY_COLLISION"
+
+
+def test_symlink(make_package):
+    path = make_package()
+    with zipfile.ZipFile(path, "a") as archive:
+        info = zipfile.ZipInfo("link")
+        info.create_system = 3
+        info.external_attr = (stat.S_IFLNK | 0o777) << 16
+        archive.writestr(info, "../outside")
+    assert PackageValidator().validate(path).issues[0].code == "SPECIAL_ENTRY"
+
+
+@pytest.mark.parametrize("field,value,code", [("max_archive_bytes", 1, "ARCHIVE_LIMIT"), ("max_entries", 1, "ENTRY_LIMIT"), ("max_file_bytes", 1, "FILE_LIMIT"), ("max_manifest_bytes", 1, "FILE_LIMIT"), ("max_total_bytes", 1, "TOTAL_LIMIT")])
+def test_configurable_limits(field, value, code, make_package):
+    limits = dataclasses.replace(PackageLimits(), **{field: value})
+    path = make_package(entries={"data": b"abc"})
+    assert PackageValidator(limits).validate(path).issues[0].code == code
+    assert PackageValidator().validate(path).valid
+
+
+def test_compression_ratio(make_package):
+    path = make_package(entries={"data": b"a" * 100_000}, compression=zipfile.ZIP_DEFLATED)
+    assert PackageValidator().validate(path).issues[0].code == "RATIO_LIMIT"
+    assert PackageValidator(PackageLimits(max_compression_ratio=10000)).validate(path).valid
+
+
+@pytest.mark.parametrize("value", [0, -1, float("inf"), True, 1.5])
+def test_invalid_limit_configuration(value):
+    with pytest.raises(ValueError):
+        PackageLimits(max_entries=value)
+
+
+@pytest.mark.parametrize("damage", ["not_zip", "truncated", "crc"])
+def test_corrupt_archive(damage, make_package):
+    path = make_package(entries={"data": b"abcdef"})
+    raw = bytearray(path.read_bytes())
+    if damage == "not_zip":
+        raw = bytearray(b"not an archive")
+    elif damage == "truncated":
+        raw = raw[:-25]
+    else:
+        with zipfile.ZipFile(path) as archive:
+            offset = archive.getinfo("data").header_offset
+        name_len, extra_len = struct.unpack_from("<HH", raw, offset + 26)
+        raw[offset + 30 + name_len + extra_len] ^= 1
+    path.write_bytes(raw)
+    assert not PackageValidator().validate(path).valid
+    with pytest.raises(PackageReadError):
+        with PackageReader(path):
+            pass
+
+
+def test_unsupported_compression(make_package):
+    path = make_package(entries={"data": b"abc"}, compression=zipfile.ZIP_BZIP2)
+    assert PackageValidator().validate(path).issues[0].code == "UNSUPPORTED_ENTRY"
+
+
+def test_missing_root_manifest(tmp_path):
+    path = tmp_path / "package"
+    with zipfile.ZipFile(path, "w") as archive:
+        archive.writestr("nested/manifest.json", "{}")
+    assert PackageValidator().validate(path).issues[0].code == "MANIFEST_MISSING"
+
+
+def test_encrypted_flag(make_package):
+    path = make_package()
+    raw = bytearray(path.read_bytes())
+    for signature, field_offset in [(b"PK\x03\x04", 6), (b"PK\x01\x02", 8)]:
+        offset = raw.index(signature) + field_offset
+        struct.pack_into("<H", raw, offset, struct.unpack_from("<H", raw, offset)[0] | 1)
+    path.write_bytes(raw)
+    assert PackageValidator().validate(path).issues[0].code == "UNSUPPORTED_ENTRY"
+
+
+def test_nul_filename(make_package):
+    path = make_package(entries={"nulXname": b"x"})
+    path.write_bytes(path.read_bytes().replace(b"nulXname", b"nul\x00name"))
+    assert PackageValidator().validate(path).issues[0].code == "UNSAFE_PATH"
+
+
+def test_explicit_directories(make_package):
+    assert PackageValidator().validate(make_package(entries={"unknown/": b"", "unknown/file": b"x"})).valid
+
+
+def test_bounded_read():
+    import io
+    with pytest.raises(ValueError):
+        PackageValidator._read_bounded(io.BytesIO(b"abcd"), 3)

tests/test_package_validator.py

@@ -0,0 +1,50 @@
+import hashlib
+
+import pytest
+
+from viano_bug_analysis.package import PackageValidator
+
+
+def test_minimal_package(make_package):
+    result = PackageValidator().validate(make_package())
+    assert result.valid
+    assert result.manifest["case_id"] == "case-A"
+
+
+def test_returned_package_and_unknown_data(manifest, make_package):
+    payload = b"immutable case"
+    manifest.update(state="REOPENED", package_generation=8, analysis_revision=3)
+    manifest["artifacts"]["case"] = "case/case.json"
+    manifest["history"]["analysis_revisions"] = 3
+    manifest["integrity"]["case/case.json"] = {"size": len(payload), "sha256": hashlib.sha256(payload).hexdigest()}
+    manifest["extension"] = {"vendor": "anything"}
+    result = PackageValidator().validate(make_package(entries={"case/case.json": payload, "unknown/data.bin": b"opaque"}))
+    assert result.valid
+    assert result.manifest["extension"] == {"vendor": "anything"}
+    assert "unknown/data.bin" in result.files
+
+
+@pytest.mark.parametrize("kind", ["missing", "hash", "size"])
+def test_invalid_integrity(kind, manifest, make_package):
+    manifest["integrity"]["original/raw"] = {"size": 3, "sha256": hashlib.sha256(b"abc").hexdigest()}
+    entries = {} if kind == "missing" else {"original/raw": b"abc"}
+    if kind == "hash":
+        manifest["integrity"]["original/raw"]["sha256"] = "0" * 64
+    if kind == "size":
+        manifest["integrity"]["original/raw"]["size"] = 4
+    assert not PackageValidator().validate(make_package(entries=entries)).valid
+
+
+@pytest.mark.parametrize("ref", ["absent.json", "../escape", "/absolute", "manifest.json"])
+def test_invalid_pointer(ref, manifest, make_package):
+    manifest["artifacts"]["case"] = ref
+    assert not PackageValidator().validate(make_package()).valid
+
+
+@pytest.mark.parametrize("raw", [b"{}", b"[]", b"{", b'{"state":"NEW","state":"REOPENED"}', b"\xff", b"[" * 2000])
+def test_invalid_manifest_controlled(raw, make_package):
+    assert not PackageValidator().validate(make_package(raw=raw)).valid
+
+
+def test_missing_archive(tmp_path):
+    assert PackageValidator().validate(tmp_path / "missing").issues[0].code == "ARCHIVE_ERROR"