fix[tool]: reject ambiguous archive members

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)

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"

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,22 +257,44 @@ 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)
 
     def _load_from_path(self, resolved_path: PurePath, original_path: PurePath) -> CompilerInput:
         # 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)