111 lines
4.0 KiB
Python
111 lines
4.0 KiB
Python
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"
|