fix(bundle): wrap local .zip errors in BundlerError instead of raw crash

src/specify_cli/commands/bundle/__init__.py

@@ -712,17 +712,72 @@ def _local_manifest_source(arg: str):
     if candidate.suffix == ".zip":
         import io
         import zipfile
+        import zlib
 
         import yaml as _yaml
-
-        with zipfile.ZipFile(candidate) as archive:
-            try:
+        from rich.markup import escape as _escape_markup
+
+        # A .zip-suffixed file that is not a valid archive would otherwise raise
+        # a raw BadZipFile that escapes bundle_install's `except BundlerError`
+        # and dumps a traceback. Report it cleanly, like the remote-download
+        # path already does for the same source.
+        #
+        # The boundary has to cover the ENTRY READ too, not just the open:
+        # ZipFile() validates only the central directory, so a corrupt member
+        # still fails later. None of these are OSError subclasses, so each is
+        # listed explicitly (verified against CPython's zipfile):
+        #   open  -- BadZipFile (bad magic/directory), NotImplementedError
+        #            (unsupported zip version), ValueError
+        #   read  -- BadZipFile ("Bad CRC-32 for file ..."), zlib.error (corrupt
+        #            deflate stream), EOFError (truncated), RuntimeError
+        #            ("... is encrypted, password required for extraction"),
+        #            NotImplementedError ("That compression method is not
+        #            supported")
+        _ZIP_OPEN_ERRORS = (
+            zipfile.BadZipFile,
+            OSError,
+            NotImplementedError,
+            ValueError,
+        )
+        _ZIP_READ_ERRORS = _ZIP_OPEN_ERRORS + (zlib.error, EOFError, RuntimeError)
+
+        # The messages below escape BOTH interpolated values, because they reach
+        # the user through _fail(), which renders Rich markup:
+        #   * the exception text -- zipfile embeds raw bytes from the archive in
+        #     some messages, e.g. BadZipFile("File name in directory
+        #     'bundle.yml' and header b'[/red]abcd' differ.")
+        #   * the path -- a file reachable as ``.../[/red].zip`` (a directory
+        #     named ``[``) puts an unmatched closing tag in the message
+        # Unescaped, either would raise MarkupError instead of reporting the
+        # actual corruption.
+        safe_path = _escape_markup(str(candidate))
+        try:
+            archive = zipfile.ZipFile(candidate)
+        except _ZIP_OPEN_ERRORS as exc:
+            raise BundlerError(
+                f"Artifact '{safe_path}' is not a valid .zip bundle: "
+                f"{_escape_markup(str(exc))}"
+            ) from exc
+        try:
+            with archive:
                 raw = archive.read("bundle.yml")
-            except KeyError as exc:
-                raise BundlerError(
-                    f"Artifact '{candidate}' does not contain a bundle.yml."
-                ) from exc
-        data = _yaml.safe_load(io.BytesIO(raw))
+        except KeyError as exc:
+            raise BundlerError(
+                f"Artifact '{safe_path}' does not contain a bundle.yml."
+            ) from exc
+        except _ZIP_READ_ERRORS as exc:
+            raise BundlerError(
+                f"Artifact '{safe_path}' is not a valid .zip bundle: "
+                f"{_escape_markup(str(exc))}"
+            ) from exc
+        try:
+            data = _yaml.safe_load(io.BytesIO(raw))
+        except _yaml.YAMLError as exc:
+            # Malformed embedded bundle.yml — same rationale as above.
+            raise BundlerError(
+                f"Artifact '{safe_path}' contains an invalid bundle.yml: "
+                f"{_escape_markup(str(exc))}"
+            ) from exc
         return BundleManifest.from_dict(data)
 
     if candidate.name == "bundle.yml" or candidate.suffix in (".yml", ".yaml"):

tests/integration/test_bundler_local_install.py

@@ -6,6 +6,7 @@
 """
 from __future__ import annotations
 
+import io
 import os
 from pathlib import Path
 
@@ -53,6 +54,141 @@ def test_local_source_from_zip_artifact(tmp_path: Path):
     assert manifest.bundle.id == "demo-bundle"
 
 
+def test_local_source_corrupt_zip_raises_clean_error(tmp_path: Path):
+    """A .zip-suffixed file that is not a valid archive raises a clean
+    BundlerError, not a raw zipfile.BadZipFile that escapes the install
+    handler's `except BundlerError` and dumps a traceback."""
+    bad = tmp_path / "artifact.zip"
+    bad.write_text("this is not a zip", encoding="utf-8")
+    with pytest.raises(BundlerError, match="not a valid .zip bundle"):
+        _local_manifest_source(str(bad))
+
+
+def test_local_source_zip_with_malformed_bundle_yml_raises_clean_error(tmp_path: Path):
+    """A valid .zip whose embedded bundle.yml is malformed YAML raises a clean
+    BundlerError, not a raw yaml.YAMLError."""
+    import zipfile
+
+    artifact = tmp_path / "artifact.zip"
+    with zipfile.ZipFile(artifact, "w") as zf:
+        zf.writestr("bundle.yml", "key: [unterminated\n")  # invalid YAML
+    with pytest.raises(BundlerError, match="invalid bundle.yml"):
+        _local_manifest_source(str(artifact))
+
+
+def test_local_source_zip_with_corrupt_entry_raises_clean_error(tmp_path: Path):
+    """A .zip whose CENTRAL DIRECTORY is intact but whose member data is corrupt
+    must also become a BundlerError. ZipFile() only validates the directory, so
+    these surface at read() time: a bad CRC raises BadZipFile and corrupt deflate
+    data raises zlib.error -- neither an OSError, so both escaped the open-only
+    boundary as a traceback.
+    """
+    import zipfile
+
+    # Stored (uncompressed) member with a flipped payload byte -> CRC mismatch.
+    crc_bad = tmp_path / "crc.zip"
+    with zipfile.ZipFile(crc_bad, "w", zipfile.ZIP_STORED) as zf:
+        zf.writestr("bundle.yml", "schema_version: '1.0'\n")
+    raw = bytearray(crc_bad.read_bytes())
+    idx = raw.find(b"schema_version")
+    raw[idx] ^= 0xFF
+    crc_bad.write_bytes(bytes(raw))
+    with pytest.raises(BundlerError, match="not a valid .zip bundle"):
+        _local_manifest_source(str(crc_bad))
+
+    # Deflated member with a mangled compressed stream -> zlib.error.
+    zlib_bad = tmp_path / "zlib.zip"
+    with zipfile.ZipFile(zlib_bad, "w", zipfile.ZIP_DEFLATED) as zf:
+        zf.writestr("bundle.yml", "schema_version: '1.0'\n")
+    raw2 = bytearray(zlib_bad.read_bytes())
+    start = raw2.find(b"bundle.yml") + len(b"bundle.yml")
+    for k in range(start, min(start + 12, len(raw2))):
+        raw2[k] ^= 0xFF
+    zlib_bad.write_bytes(bytes(raw2))
+    with pytest.raises(BundlerError, match="not a valid .zip bundle"):
+        _local_manifest_source(str(zlib_bad))
+
+
+def test_local_source_encrypted_zip_raises_clean_error(tmp_path: Path):
+    """A password-protected artifact (``zip -e``) must report cleanly. CPython
+    raises RuntimeError("... is encrypted, password required for extraction") at
+    read() time -- not a BadZipFile/OSError, so it escaped the first boundary."""
+    import zipfile
+
+    artifact = tmp_path / "enc.zip"
+    with zipfile.ZipFile(artifact, "w") as zf:
+        zf.writestr("bundle.yml", "schema_version: '1.0'\n")
+    raw = bytearray(artifact.read_bytes())
+    raw[raw.find(b"PK\x03\x04") + 6] |= 0x01  # local header: encrypted bit
+    raw[raw.find(b"PK\x01\x02") + 8] |= 0x01  # central dir: encrypted bit
+    artifact.write_bytes(bytes(raw))
+    with pytest.raises(BundlerError, match="not a valid .zip bundle"):
+        _local_manifest_source(str(artifact))
+
+
+def test_local_source_unsupported_compression_raises_clean_error(tmp_path: Path):
+    """An unsupported compression method raises NotImplementedError at read()
+    time -- also not a BadZipFile/OSError."""
+    import zipfile
+
+    artifact = tmp_path / "comp.zip"
+    with zipfile.ZipFile(artifact, "w") as zf:
+        zf.writestr("bundle.yml", "schema_version: '1.0'\n")
+    raw = bytearray(artifact.read_bytes())
+    raw[raw.find(b"PK\x01\x02") + 10] = 99  # central dir compress_type
+    raw[raw.find(b"PK\x03\x04") + 8] = 99   # local header compress_type
+    artifact.write_bytes(bytes(raw))
+    with pytest.raises(BundlerError, match="not a valid .zip bundle"):
+        _local_manifest_source(str(artifact))
+
+
+def test_local_source_zip_error_message_escapes_archive_markup(tmp_path: Path):
+    """zipfile embeds bytes taken FROM THE ARCHIVE in some messages, e.g.
+    BadZipFile("File name in directory 'bundle.yml' and header b'[/red]abcd'
+    differ."). That text reaches the user through _fail(), which renders Rich
+    markup, so it must be escaped or a crafted member name raises MarkupError
+    instead of reporting the corruption."""
+    import zipfile
+
+    from rich.console import Console
+
+    artifact = tmp_path / "markup.zip"
+    with zipfile.ZipFile(artifact, "w") as zf:
+        zf.writestr("bundle.yml", "schema_version: '1.0'\n")
+    raw = bytearray(artifact.read_bytes())
+    name_at = raw.find(b"PK\x03\x04") + 30
+    raw[name_at:name_at + 10] = b"[/red]abcd"  # same length keeps offsets valid
+    artifact.write_bytes(bytes(raw))
+
+    with pytest.raises(BundlerError, match="not a valid .zip bundle") as excinfo:
+        _local_manifest_source(str(artifact))
+
+    # The message must survive Rich rendering (this is what _fail does).
+    Console(file=io.StringIO()).print(f"[red]Error:[/red] {excinfo.value}")
+
+
+def test_local_source_zip_error_message_escapes_path_markup(tmp_path: Path):
+    """The ARTIFACT PATH is interpolated into the same message, so it needs the
+    same escaping: a file reachable as ``.../[/red].zip`` -- a directory named
+    ``[`` -- puts an unmatched closing tag in the text and would raise
+    MarkupError when _fail() renders it (POSIX keeps the forward slash; Windows
+    normalizes it to a backslash, which is inert)."""
+    from rich.console import Console
+
+    bracket_dir = tmp_path / "["
+    bracket_dir.mkdir()
+    artifact = bracket_dir / "red].zip"
+    artifact.write_text("this is not a zip", encoding="utf-8")
+
+    # Pass the path with forward slashes so the message carries "[/red].zip"
+    # exactly as a POSIX caller would produce it.
+    posix_style = f"{bracket_dir.as_posix()}/red].zip"
+    with pytest.raises(BundlerError, match="not a valid .zip bundle") as excinfo:
+        _local_manifest_source(posix_style)
+
+    Console(file=io.StringIO()).print(f"[red]Error:[/red] {excinfo.value}")
+
+
 def test_local_source_rejects_unknown_file(tmp_path: Path):
     weird = tmp_path / "thing.txt"
     weird.write_text("nope", encoding="utf-8")