From 59e3066f544eb644072b43b0841d566ba26a1136 Mon Sep 17 00:00:00 2001 From: Dickson Date: Fri, 11 Sep 2026 19:16:08 -0400 Subject: [PATCH] fix[tool]: reject ambiguous archive members --- tests/unit/compiler/test_zip_input_bundle.py | 108 +++++++++++++++++++ vyper/cli/compile_archive.py | 17 +-- vyper/compiler/input_bundle.py | 30 +++++- 3 files changed, 144 insertions(+), 11 deletions(-) create mode 100644 tests/unit/compiler/test_zip_input_bundle.py diff --git a/tests/unit/compiler/test_zip_input_bundle.py b/tests/unit/compiler/test_zip_input_bundle.py new file mode 100644 index 0000000000..605b2fc5c1 --- /dev/null +++ b/tests/unit/compiler/test_zip_input_bundle.py @@ -0,0 +1,108 @@ +import base64 +import io +import warnings +import zipfile + +import pytest + +from vyper import compile_code +from vyper.cli.compile_archive import compiler_data_from_zip +from vyper.compiler.input_bundle import ZipInputBundle +from vyper.exceptions import BadArchive +from vyper.utils import sha256sum + + +def make_archive(members): + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as archive: + with warnings.catch_warnings(): + warnings.filterwarnings("ignore", message="Duplicate name:", category=UserWarning) + for name, contents in members: + archive.writestr(name, contents) + return buf.getvalue() + + +@pytest.mark.parametrize("encoded", [False, True]) +@pytest.mark.parametrize("prefix", ["", "./"]) +def test_valid_archive(tmp_path, encoded, prefix): + source = "@external\ndef foo() -> uint256:\n return 42\n" + members = [ + ("MANIFEST/", ""), + ("MANIFEST/searchpaths", "."), + ("MANIFEST/compilation_targets", "src/main.vy"), + ("MANIFEST/settings.json", "{}"), + ("MANIFEST/integrity", sha256sum(sha256sum(source))), + ("src/", ""), + ("src/main.vy", source), + ("src/Main.vy", "# case-sensitive namespace"), + ("src/café.vy", "# Unicode names are valid"), + ("/absolute.vy", "# absolute POSIX names are valid"), + ("C:/drive.vy", "# drive-prefixed POSIX names are valid"), + ] + contents = make_archive([(prefix + name, value) for name, value in members]) + path = tmp_path / "input.vyz" + path.write_bytes(base64.b64encode(contents) if encoded else contents) + data = compiler_data_from_zip(path, None, False) + assert data.source_code == source + assert data.bytecode.hex() == compile_code(source, output_formats=["bytecode"])["bytecode"][2:] + bundle = data.input_bundle + assert bundle.load_file("src/./main.vy").source_id == bundle.load_file("src/main.vy").source_id + + +@pytest.mark.parametrize("entrypoint", ["bundle", "cli", "base64"]) +@pytest.mark.parametrize( + "names", + [ + ["main.vy", "main.vy"], + ["MANIFEST/compilation_targets", "MANIFEST/compilation_targets"], + ["MANIFEST/searchpaths", "MANIFEST/searchpaths"], + ["main.vy", "./main.vy"], + ["./main.vy", "main.vy"], + ["src/main.vy", "src/./main.vy"], + ["src/main.vy", "src//main.vy"], + ["src", "src/"], + ["MANIFEST/compilation_targets", "./MANIFEST/compilation_targets"], + ], +) +def test_duplicate_archive_members(tmp_path, monkeypatch, entrypoint, names): + contents = make_archive([(name, str(i)) for i, name in enumerate(names)]) + assert_rejected_before_read(tmp_path, monkeypatch, entrypoint, contents, "Duplicate archive") + + +@pytest.mark.parametrize("entrypoint", ["bundle", "cli", "base64"]) +@pytest.mark.parametrize( + "name", + [ + "", + ".", + "./", + "../main.vy", + "src/../main.vy", + "src/../../main.vy", + "src/..", + "src\\main.vy", + "C:\\main.vy", + "main.vy\x00suffix", + ], +) +def test_invalid_archive_member(tmp_path, monkeypatch, entrypoint, name): + # Substitute a same-length byte in both ZIP headers: writestr truncates NULs. + contents = make_archive([("unused.vy", ""), (name.replace("\x00", "!"), "")]) + if "\x00" in name: + contents = contents.replace(name.replace("\x00", "!").encode(), name.encode()) + assert_rejected_before_read(tmp_path, monkeypatch, entrypoint, contents, "Invalid archive") + + +def assert_rejected_before_read(tmp_path, monkeypatch, entrypoint, contents, message): + def unexpected_read(*args, **kwargs): + pytest.fail("Archive member contents read before namespace validation") + + monkeypatch.setattr(zipfile.ZipFile, "open", unexpected_read) + with pytest.raises(BadArchive, match=message): + if entrypoint == "bundle": + with zipfile.ZipFile(io.BytesIO(contents)) as archive: + ZipInputBundle(archive) + else: + path = tmp_path / "input.vyz" + path.write_bytes(base64.b64encode(contents) if entrypoint == "base64" else contents) + compiler_data_from_zip(path, None, False) diff --git a/vyper/cli/compile_archive.py b/vyper/cli/compile_archive.py index f1072de249..50e5265563 100644 --- a/vyper/cli/compile_archive.py +++ b/vyper/cli/compile_archive.py @@ -42,18 +42,21 @@ def compiler_data_from_zip(file_name, settings, no_bytecode_metadata): except (zipfile.BadZipFile, binascii.Error): raise NotZipInput() from e1 - fcontents = archive.read("MANIFEST/compilation_targets").decode("utf-8") + input_bundle = ZipInputBundle(archive) + fcontents = input_bundle.read_file("MANIFEST/compilation_targets").decode("utf-8") compilation_targets = fcontents.splitlines() if len(compilation_targets) != 1: raise BadArchive("Multiple compilation targets not supported!") - input_bundle = ZipInputBundle(archive) - storage_layout_path = "MANIFEST/storage_layout.json" storage_layout = None - if storage_layout_path in archive.namelist(): - storage_layout_map = json.loads(archive.read(storage_layout_path).decode("utf-8")) + try: + storage_layout_contents = input_bundle.read_file(storage_layout_path) + except KeyError: + pass + else: + storage_layout_map = json.loads(storage_layout_contents.decode("utf-8")) storage_layout = input_bundle.load_json_file(storage_layout_map[compilation_targets[0]]) mainpath = PurePath(compilation_targets[0]) @@ -62,10 +65,10 @@ def compiler_data_from_zip(file_name, settings, no_bytecode_metadata): settings = settings or Settings() - archive_settings_txt = archive.read("MANIFEST/settings.json").decode("utf-8") + archive_settings_txt = input_bundle.read_file("MANIFEST/settings.json").decode("utf-8") archive_settings = Settings.from_dict(json.loads(archive_settings_txt)) - integrity = archive.read("MANIFEST/integrity").decode("utf-8").strip() + integrity = input_bundle.read_file("MANIFEST/integrity").decode("utf-8").strip() settings = merge_settings( settings, archive_settings, lhs_source="command line", rhs_source="archive settings" diff --git a/vyper/compiler/input_bundle.py b/vyper/compiler/input_bundle.py index 76d1e4518a..6789e5fcc6 100644 --- a/vyper/compiler/input_bundle.py +++ b/vyper/compiler/input_bundle.py @@ -6,14 +6,14 @@ from pathlib import Path, PurePath from typing import TYPE_CHECKING, Any, Iterator, Optional -from vyper.exceptions import JSONError +from vyper.exceptions import BadArchive, JSONError from vyper.utils import sha256sum # a type to make mypy happy PathLike = Path | PurePath if TYPE_CHECKING: - from zipfile import ZipFile + from zipfile import ZipFile, ZipInfo # hacky sentinel to indicate that a file came from InputBundle for builtins BUILTIN = -2 @@ -257,14 +257,36 @@ def _load_from_path(self, resolved_path: PurePath, original_path: PurePath) -> C # a zipfile as input. class ZipInputBundle(InputBundle): def __init__(self, archive: "ZipFile"): + # Validate the entire namespace before testzip() or any manifest read. + # orig_filename preserves NULs which ZipInfo.filename truncates. + self._members: dict[str, "ZipInfo"] = {} + for member in archive.infolist(): + name = member.orig_filename + canonical = posixpath.normpath(name) + if ( + not name + or ".." in name.split("/") + or "\\" in name + or "\x00" in name + or canonical == "." + ): + raise BadArchive(f"Invalid archive member name: {name!r}") + if canonical in self._members: + previous = self._members[canonical].orig_filename + raise BadArchive(f"Duplicate archive member path: {previous!r} and {name!r}") + self._members[canonical] = member + assert archive.testzip() is None self.archive = archive - sp_str = archive.read("MANIFEST/searchpaths").decode("utf-8") + sp_str = self.read_file("MANIFEST/searchpaths").decode("utf-8") search_paths = [PurePath(p) for p in sp_str.splitlines()] super().__init__(search_paths) + def read_file(self, path: str) -> bytes: + return self.archive.read(self._members[posixpath.normpath(path)]) + def _normalize_path(self, path: PurePath) -> PurePath: return _normpath(path) @@ -272,7 +294,7 @@ def _load_from_path(self, resolved_path: PurePath, original_path: PurePath) -> C # zipfile.BadZipFile: File is not a zip file try: - value = self.archive.read(resolved_path.as_posix()).decode("utf-8") + value = self.read_file(resolved_path.as_posix()).decode("utf-8") except KeyError: # zipfile literally raises KeyError if the file is not there raise _NotFound(resolved_path)