120 lines
4.7 KiB
Python
120 lines
4.7 KiB
Python
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)
|