diff --git a/src/specify_cli/commands/bundle/__init__.py b/src/specify_cli/commands/bundle/__init__.py index d8fb22e511..868f0b4db5 100644 --- a/src/specify_cli/commands/bundle/__init__.py +++ b/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"): diff --git a/tests/integration/test_bundler_local_install.py b/tests/integration/test_bundler_local_install.py index 32972ac684..19fb94fcfe 100644 --- a/tests/integration/test_bundler_local_install.py +++ b/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")