import gzip import tarfile from dataclasses import replace import pytest from viano_bug_analysis.profiles.vcore import AdapterError, ArchiveLimits, ArchiveReader @pytest.mark.parametrize("name", ["../escape", "/absolute", "C:/escape", "a\\b", "a//b", "a/./b", "a:stream"]) def test_unsafe_paths(make_tar, name): info = tarfile.TarInfo(name) info.size = 1 path = make_tar({"manifest.json": {}}, extra=[(info, b"x")]) with pytest.raises(AdapterError) as error: with ArchiveReader(path): pass assert error.value.code == "UNSAFE_PATH" @pytest.mark.parametrize("type_", [tarfile.SYMTYPE, tarfile.LNKTYPE, tarfile.FIFOTYPE, tarfile.CHRTYPE, tarfile.BLKTYPE, tarfile.GNUTYPE_SPARSE]) def test_special_members_rejected(make_tar, type_): info = tarfile.TarInfo("wrapper/link") info.type = type_ info.linkname = "../escape" path = make_tar({"manifest.json": {}}, extra=[(info, None)]) with pytest.raises(AdapterError): with ArchiveReader(path): pass @pytest.mark.parametrize("entries", [{"a": b"x", "A": b"y"}, {"a": b"x", "a/b": b"y"}]) def test_member_collisions(make_tar, entries): path = make_tar({"manifest.json": {}} | entries) with pytest.raises(AdapterError) as error: with ArchiveReader(path): pass assert error.value.code == "MEMBER_COLLISION" @pytest.mark.parametrize("damage", ["gzip", "truncated", "crc", "tar"]) def test_corrupt_source_rejected(make_tar, damage): path = make_tar({"manifest.json": {}}) raw = bytearray(path.read_bytes()) if damage == "gzip": raw = b"not gzip" elif damage == "truncated": raw = raw[:-4] elif damage == "crc": raw[-8] ^= 1 else: unpacked = bytearray(gzip.decompress(raw)) unpacked[0] ^= 1 raw = gzip.compress(unpacked) path.write_bytes(raw) with pytest.raises(AdapterError): with ArchiveReader(path): pass @pytest.mark.parametrize("field,value", [("max_archive_bytes", 1), ("max_tar_bytes", 512), ("max_file_bytes", 1), ("max_entries", 1), ("max_json_bytes", 1), ("max_ratio", 1)]) def test_configurable_limits(make_tar, field, value): path = make_tar({"manifest.json": {}, "data": b"abc"}) with pytest.raises(AdapterError): with ArchiveReader(path, replace(ArchiveLimits(), **{field: value})): pass def test_pax_metadata_bound(make_tar): info = tarfile.TarInfo("extra") info.pax_headers = {"comment": "x" * 2000} info.size = 1 path = make_tar({"manifest.json": {}}, extra=[(info, b"x")]) with pytest.raises(AdapterError) as error: with ArchiveReader(path, ArchiveLimits(max_header_bytes=100)): pass assert error.value.code == "HEADER_LIMIT" def test_ambiguous_manifest(make_tar): path = make_tar({"one/manifest.json": {}, "two/manifest.json": {}}, root="") with pytest.raises(AdapterError) as error: with ArchiveReader(path): pass assert error.value.code == "MANIFEST_AMBIGUOUS" def test_snapshot_is_immutable_and_nothing_is_extracted(make_tar, tmp_path): path = make_tar({"manifest.json": {}, "unknown": b"data"}) raw = path.read_bytes() with ArchiveReader(path) as reader: path.write_bytes(b"changed external file") assert reader.original_bytes() == raw assert reader.read_member(reader.member_name("unknown")) == b"data" with pytest.raises(AdapterError): reader.member_name("../escape") assert not (tmp_path / "arbitrary-wrapper").exists() def test_sparse_rejected_before_sparse_map_processing(make_tar, monkeypatch): info = tarfile.TarInfo("sparse") info.type = tarfile.GNUTYPE_SPARSE path = make_tar({"manifest.json": {}}, extra=[(info, None)]) def never_parse(*args): pytest.fail("Sparse map processing should not execute") monkeypatch.setattr(tarfile.TarInfo, "_proc_sparse", never_parse) with pytest.raises(AdapterError) as error: with ArchiveReader(path): pass assert error.value.code == "SPECIAL_MEMBER"