diff --git a/CHANGELOG.md b/CHANGELOG.md index 6deb8e4..76e35f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -67,6 +67,11 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). without changing the centrally managed review-agent credential contract. ### Security +- Preflight every direct release-SBOM wheel or source archive as one regular + file with a finite 256 MiB compressed-byte ceiling before ZIP or gzip/tar + parsing, metadata reads, or artifact hashing. Symlinks, directories, devices, + missing paths, and inspection failures now fail through stable non-leaking + errors while existing archive-member and decompression defenses remain intact. - Canonicalize the public manifest writer's optional `forbidden_root` before any output-parent creation or output-path access. Missing, non-directory, symlinked, unresolvable, or otherwise noncanonical roots now fail with one diff --git a/docs/sbom-release-evidence.md b/docs/sbom-release-evidence.md index a017716..8c2233f 100644 --- a/docs/sbom-release-evidence.md +++ b/docs/sbom-release-evidence.md @@ -20,19 +20,41 @@ No SLSA Build level is claimed merely because an SBOM or attestation exists. file as untrusted input and never imports EgressWeave. It must: 1. accept only a wheel or gzip source distribution; -2. reject unsafe or duplicate paths, links, devices, excessive member counts, +2. inspect the caller-supplied final release-artifact path without resolving that + final component, require a regular file, enforce a 256 MiB compressed-byte + ceiling, open it without following a final symbolic link where the platform + supports that flag, and require the opened descriptor to retain the accepted + device and inode before any parser; +3. parse and hash only that bound descriptor, keep every parser-visible read and + seek live-bounded by the same 256 MiB ceiling, bracket metadata parsing with + finite SHA-256 reads, and fail if the archive bytes change during verification; +4. reject unsafe or duplicate paths, links, devices, excessive member counts, ambiguous metadata, oversized metadata, and malformed archives; -3. check the declared wheel metadata size before decompression; -4. read exactly one wheel `METADATA` or root source `PKG-INFO` member; -5. verify package identity, license expression, and complete direct runtime +5. check the declared wheel metadata size before decompression; +6. read exactly one wheel `METADATA` or root source `PKG-INFO` member; +7. verify package identity, license expression, and complete direct runtime requirement declarations against the reviewed manifest; -6. verify every dependency version, SHA-256, and environment marker against the +8. verify every dependency version, SHA-256, and environment marker against the executable hash-locked subset in `requirements-ci.txt`, while rejecting dependency extras that could activate packages outside the reviewed graph; -7. validate identities, SPDX license identifiers, purls, graph references, +9. validate identities, SPDX license identifiers, purls, graph references, relationships, reachability, and acyclicity; -8. compute the artifact SHA-256 without trusting its filename; and -9. emit sorted UTF-8 CycloneDX 1.7 JSON without timestamps or random identifiers. +10. compute the artifact SHA-256 without trusting its filename; and +11. emit sorted UTF-8 CycloneDX 1.7 JSON without timestamps or random identifiers. + +The direct generator normalizes missing, uninspectable, symbolic-link, directory, +device, FIFO, socket, replaced, and other non-regular artifact inputs to +`release artifact is missing or unsafe`. The final artifact component remains +unresolved until no-follow validation binds the accepted path to its descriptor. +Inputs above the compressed-byte ceiling fail with +`release artifact exceeds the compressed-byte safety bound`; the parser-facing +wrapper rechecks the live regular descriptor before and after reads and seeks, so +an initially accepted archive that grows past the ceiling fails before the parser +can consume the expanded input. Bytes that change across the descriptor-bound +metadata pass fail with `release artifact changed during verification`. +Accepted-size archives remain subject to all member-count, path, link/device, +metadata-size, decompression, identity, dependency, and digest controls; the +compressed-input check does not replace those independent defenses. The root component uses a digest-derived `bom-ref`, preventing different artifacts from sharing evidence identity. The dependency graph is the union @@ -116,14 +138,25 @@ model-modified source under a write credential. ## Threats, failure, and recovery These controls address omitted inventory, evidence bound to the wrong artifact, -filename substitution, manifest-versus-lock drift, undeclared dependency extras, -mutable dependency resolution, nondeterministic evidence, unsafe archives, -metadata decompression, stale or wrong-workflow attestations, and publication -before exact verification. - -They do not detect every compromised upstream source, malicious but correctly -hashed package, license obligation, build-host compromise, or undisclosed -vulnerability. Those risks require provenance, reproducible builds, +filename substitution, path replacement between inspection and parsing, +manifest-versus-lock drift, undeclared dependency extras, mutable dependency +resolution, nondeterministic evidence, unsafe archives, compressed-input resource +exhaustion, metadata decompression, stale or wrong-workflow attestations, and +publication before exact verification. + +Descriptor identity, live parser bounds, and digest bracketing close ordinary +final-symlink, pathname-replacement, unbounded-growth, and in-place mutation races +during parsing. They do not convert a writable build host into an immutable-storage +system: a privileged writer able to alter and restore the same inode entirely +between verification observations remains a residual mutable-storage risk. Run +evidence generation from an isolated, read-only exact-artifact directory, and +rely on the later sealed-evidence descriptor, digest, and post-publication checks +before any credential-bearing use. No provenance or SLSA claim follows from these +direct-generator controls. + +These controls do not detect every compromised upstream source, malicious but +correctly hashed package, license obligation, build-host compromise, or +undisclosed vulnerability. Those risks require provenance, reproducible builds, vulnerability management, legal review, and hardened runners. On any generator, digest, semantic, manifest, lock, or attestation failure, @@ -150,6 +183,13 @@ https://docs.github.com/en/actions/how-tos/secure-your-work/use-artifact-attesta GitHub. (2026). *actions/attest* [Computer software]. GitHub. https://github.com/actions/attest +MITRE. (2026). *CWE-400: Uncontrolled resource consumption.* Common Weakness +Enumeration. https://cwe.mitre.org/data/definitions/400.html + +Python Software Foundation. (n.d.). *zipfile—Work with ZIP archives: +Decompression pitfalls.* Python 3 documentation. Retrieved August 6, 2026, from +https://docs.python.org/3/library/zipfile.html#decompression-pitfalls + Python Packaging Authority. (n.d.). *Core metadata specifications.* Python Packaging User Guide. Retrieved August 5, 2026, from https://packaging.python.org/en/latest/specifications/core-metadata/ diff --git a/scripts/ci/generate_release_sbom.py b/scripts/ci/generate_release_sbom.py index e72a2aa..90a19fa 100644 --- a/scripts/ci/generate_release_sbom.py +++ b/scripts/ci/generate_release_sbom.py @@ -9,15 +9,18 @@ import argparse import hashlib +import io import json +import os import re +import stat import tarfile import zipfile from email.message import Message from email.parser import BytesParser from email.policy import default from pathlib import Path, PurePosixPath -from typing import Any +from typing import Any, BinaryIO from packaging.markers import InvalidMarker, Marker from packaging.requirements import InvalidRequirement, Requirement @@ -28,6 +31,7 @@ CYCLONEDX_SPEC_VERSION = "1.7" MAX_METADATA_BYTES = MAX_MANIFEST_BYTES = 1_048_576 MAX_ARCHIVE_MEMBERS = 10_000 +MAX_RELEASE_ARTIFACT_BYTES = 256 * 1024 * 1024 NAME_SEPARATORS = re.compile(r"[-_.]+") SHA256 = re.compile(r"[0-9a-f]{64}") REVIEWED_SPDX_LICENSE_IDS = frozenset( @@ -49,6 +53,172 @@ def _parse_arguments() -> argparse.Namespace: return parser.parse_args() +def _unsafe_artifact_error(error: BaseException | None = None) -> SystemExit: + """Return one stable non-leaking failure for unsafe parser-visible state.""" + failure = SystemExit("release artifact is missing or unsafe") + if error is not None: + failure.__cause__ = error + return failure + + +def _require_live_artifact_descriptor(stream: BinaryIO) -> int: + """Return the live regular-file size or fail through a stable public error.""" + try: + descriptor = stream.fileno() + metadata = os.fstat(descriptor) + except (OSError, TypeError, ValueError) as error: + raise _unsafe_artifact_error(error) + if ( + isinstance(descriptor, bool) + or not isinstance(descriptor, int) + or descriptor < 0 + or not stat.S_ISREG(metadata.st_mode) + ): + raise _unsafe_artifact_error() + if metadata.st_size > MAX_RELEASE_ARTIFACT_BYTES: + raise SystemExit("release artifact exceeds the compressed-byte safety bound") + return metadata.st_size + + +def _require_artifact_position(stream: BinaryIO) -> int: + """Return one finite nonnegative parser position inside the byte ceiling.""" + try: + position = stream.tell() + except (OSError, TypeError, ValueError) as error: + raise _unsafe_artifact_error(error) + if ( + isinstance(position, bool) + or not isinstance(position, int) + or position < 0 + or position > MAX_RELEASE_ARTIFACT_BYTES + ): + raise _unsafe_artifact_error() + return position + + +class _LiveBoundedArtifactReader(io.BufferedIOBase): + """Expose parser reads and seeks while rechecking one finite live descriptor.""" + + def __init__(self, stream: BinaryIO) -> None: + super().__init__() + self._stream = stream + + def readable(self) -> bool: + """Report that the accepted artifact descriptor supports reads.""" + return True + + def seekable(self) -> bool: + """Report that archive parsers may seek within the accepted descriptor.""" + return True + + def fileno(self) -> int: + """Return the validated underlying descriptor without leaking failures.""" + try: + descriptor = self._stream.fileno() + except (OSError, TypeError, ValueError) as error: + raise _unsafe_artifact_error(error) + if isinstance(descriptor, bool) or not isinstance(descriptor, int) or descriptor < 0: + raise _unsafe_artifact_error() + return descriptor + + def tell(self) -> int: + """Return one validated finite parser position.""" + return _require_artifact_position(self._stream) + + def read(self, size: int = -1) -> bytes: + """Read no more than the finite ceiling and reject concurrent growth.""" + _require_live_artifact_descriptor(self._stream) + position_before = _require_artifact_position(self._stream) + if isinstance(size, bool) or not isinstance(size, int): + raise _unsafe_artifact_error() + remaining_with_tripwire = MAX_RELEASE_ARTIFACT_BYTES - position_before + 1 + bounded_size = ( + remaining_with_tripwire + if size < 0 or size > remaining_with_tripwire + else size + ) + try: + payload = self._stream.read(bounded_size) + except (OSError, TypeError, ValueError) as error: + raise _unsafe_artifact_error(error) + if not isinstance(payload, (bytes, bytearray, memoryview)): + raise _unsafe_artifact_error() + payload_bytes = bytes(payload) + if len(payload_bytes) > bounded_size: + raise _unsafe_artifact_error() + _require_live_artifact_descriptor(self._stream) + position_after = _require_artifact_position(self._stream) + if position_after != position_before + len(payload_bytes): + raise _unsafe_artifact_error() + if len(payload_bytes) > MAX_RELEASE_ARTIFACT_BYTES: + raise SystemExit( + "release artifact exceeds the compressed-byte safety bound" + ) + return payload_bytes + + def readinto(self, buffer: bytearray | memoryview) -> int: + """Fill a parser buffer through the same bounded read contract.""" + try: + payload = self.read(len(buffer)) + buffer[: len(payload)] = payload + except (OSError, TypeError, ValueError) as error: + raise _unsafe_artifact_error(error) + return len(payload) + + def seek(self, offset: int, whence: int = os.SEEK_SET) -> int: + """Seek only while descriptor state and parser position remain safe.""" + _require_live_artifact_descriptor(self._stream) + try: + position = self._stream.seek(offset, whence) + except (OSError, TypeError, ValueError) as error: + raise _unsafe_artifact_error(error) + if ( + isinstance(position, bool) + or not isinstance(position, int) + or position < 0 + or position > MAX_RELEASE_ARTIFACT_BYTES + ): + raise _unsafe_artifact_error() + _require_live_artifact_descriptor(self._stream) + if _require_artifact_position(self._stream) != position: + raise _unsafe_artifact_error() + return position + + +def _open_release_artifact(path: Path) -> BinaryIO: + """Open one preflighted regular archive and bind it to the accepted identity.""" + try: + path_metadata = path.lstat() + except OSError as error: + raise SystemExit("release artifact is missing or unsafe") from error + if not stat.S_ISREG(path_metadata.st_mode): + raise SystemExit("release artifact is missing or unsafe") + if path_metadata.st_size > MAX_RELEASE_ARTIFACT_BYTES: + raise SystemExit("release artifact exceeds the compressed-byte safety bound") + + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(path, flags) + except OSError as error: + raise SystemExit("release artifact is missing or unsafe") from error + try: + opened_metadata = os.fstat(descriptor) + if ( + not stat.S_ISREG(opened_metadata.st_mode) + or opened_metadata.st_dev != path_metadata.st_dev + or opened_metadata.st_ino != path_metadata.st_ino + ): + raise SystemExit("release artifact is missing or unsafe") + if opened_metadata.st_size > MAX_RELEASE_ARTIFACT_BYTES: + raise SystemExit( + "release artifact exceeds the compressed-byte safety bound" + ) + return os.fdopen(descriptor, "rb") + except BaseException: + os.close(descriptor) + raise + + def _name(value: str) -> str: """Return a canonical Python distribution name.""" return NAME_SEPARATORS.sub("-", value).lower() @@ -101,10 +271,11 @@ def _parse_metadata(payload: bytes, source: str) -> Message: return BytesParser(policy=default).parsebytes(payload) -def _wheel_metadata(path: Path) -> Message: - """Read the sole bounded wheel METADATA member.""" +def _wheel_metadata(stream: BinaryIO) -> Message: + """Read the sole bounded wheel METADATA member from the bound archive.""" + stream.seek(0) try: - with zipfile.ZipFile(path) as archive: + with zipfile.ZipFile(stream) as archive: members = archive.infolist() _check_archive_names([item.filename for item in members], "wheel") selected = [ @@ -119,10 +290,11 @@ def _wheel_metadata(path: Path) -> Message: raise SystemExit("release wheel is not a valid ZIP archive") from error -def _sdist_metadata(path: Path) -> Message: - """Read the sole bounded root PKG-INFO member.""" +def _sdist_metadata(stream: BinaryIO) -> Message: + """Read the sole bounded root PKG-INFO member from the bound archive.""" + stream.seek(0) try: - with tarfile.open(path, mode="r:gz") as archive: + with tarfile.open(fileobj=stream, mode="r:gz") as archive: members = archive.getmembers() _check_archive_names([item.name for item in members], "source distribution") if any(item.issym() or item.islnk() or item.isdev() for item in members): @@ -138,20 +310,21 @@ def _sdist_metadata(path: Path) -> Message: raise SystemExit("source distribution must contain one root PKG-INFO") if selected[0].size > MAX_METADATA_BYTES: raise SystemExit("source distribution metadata exceeds the safety bound") - stream = archive.extractfile(selected[0]) - if stream is None: + extracted = archive.extractfile(selected[0]) + if extracted is None: raise SystemExit("source distribution metadata could not be read") - return _parse_metadata(stream.read(), "source distribution") + return _parse_metadata(extracted.read(), "source distribution") except tarfile.TarError as error: raise SystemExit("release source distribution is not a valid gzip tar") from error -def _artifact_metadata(path: Path) -> Message: - """Read metadata from a wheel or gzip source distribution.""" - if path.name.endswith(".whl"): - return _wheel_metadata(path) - if path.name.endswith(".tar.gz"): - return _sdist_metadata(path) +def _artifact_metadata(stream: BinaryIO, filename: str) -> Message: + """Read metadata through a live-bounded wheel or source-archive descriptor.""" + bounded_stream = _LiveBoundedArtifactReader(stream) + if filename.endswith(".whl"): + return _wheel_metadata(bounded_stream) + if filename.endswith(".tar.gz"): + return _sdist_metadata(bounded_stream) raise SystemExit("release artifact must be a .whl or .tar.gz distribution") @@ -378,12 +551,19 @@ def _identity(metadata: Message) -> tuple[str, str, str, list[str]]: return _name(name), version, license_id, requirements -def _sha256_file(path: Path) -> str: - """Return an artifact SHA-256 without loading it into memory.""" +def _sha256_file(stream: BinaryIO) -> str: + """Hash the bound artifact while enforcing its live finite byte ceiling.""" digest = hashlib.sha256() - with path.open("rb") as stream: - for block in iter(lambda: stream.read(1_048_576), b""): - digest.update(block) + stream.seek(0) + consumed = 0 + while block := stream.read(1_048_576): + consumed += len(block) + if consumed > MAX_RELEASE_ARTIFACT_BYTES: + raise SystemExit( + "release artifact exceeds the compressed-byte safety bound" + ) + digest.update(block) + stream.seek(0) return digest.hexdigest() @@ -413,11 +593,15 @@ def _component_json(item: dict[str, Any]) -> dict[str, Any]: def build_sbom(artifact_path: Path, manifest_path: Path) -> dict[str, Any]: """Build deterministic CycloneDX evidence for one exact distribution.""" - if not artifact_path.is_file(): - raise SystemExit("release artifact does not exist or is not a regular file") - package, version, license_id, requirements = _identity( - _artifact_metadata(artifact_path) - ) + with _open_release_artifact(artifact_path) as artifact_stream: + digest_before = _sha256_file(artifact_stream) + package, version, license_id, requirements = _identity( + _artifact_metadata(artifact_stream, artifact_path.name) + ) + digest = _sha256_file(artifact_stream) + if digest != digest_before: + raise SystemExit("release artifact changed during verification") + root, components = _load_manifest(manifest_path) if package != root["name"] or license_id != root["license"]: raise SystemExit("artifact identity or license does not match the manifest") @@ -426,7 +610,6 @@ def build_sbom(artifact_path: Path, manifest_path: Path) -> dict[str, Any]: raise SystemExit("artifact direct runtime dependencies do not match the manifest") if requirements != root["requires_dist"]: raise SystemExit("artifact runtime requirement declarations do not match the manifest") - digest = _sha256_file(artifact_path) root_ref = f"urn:egressweave:artifact:sha256:{digest}" ordered = sorted(components.values(), key=lambda item: item["purl"]) dependencies = [ @@ -493,7 +676,7 @@ def main() -> int: arguments = _parse_arguments() validate_runtime_lock(arguments.manifest.resolve(), arguments.lock.resolve()) write_sbom( - build_sbom(arguments.artifact.resolve(), arguments.manifest.resolve()), + build_sbom(arguments.artifact, arguments.manifest.resolve()), arguments.output.resolve(), ) print(f"wrote CycloneDX {CYCLONEDX_SPEC_VERSION} SBOM: {arguments.output}") @@ -501,4 +684,4 @@ def main() -> int: if __name__ == "__main__": - raise SystemExit(main()) + raise SystemExit(main()) \ No newline at end of file diff --git a/tests/test_release_sbom_archive_bound.py b/tests/test_release_sbom_archive_bound.py new file mode 100644 index 0000000..db83fca --- /dev/null +++ b/tests/test_release_sbom_archive_bound.py @@ -0,0 +1,440 @@ +"""Regression tests for the direct release-SBOM archive-size boundary.""" + +from __future__ import annotations + +import importlib.util +import io +import os +import tarfile +import zipfile +from pathlib import Path +from types import ModuleType, SimpleNamespace + +import pytest + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +GENERATOR_PATH = REPOSITORY_ROOT / "scripts" / "ci" / "generate_release_sbom.py" +MANIFEST_PATH = REPOSITORY_ROOT / "scripts" / "ci" / "release_runtime_dependencies.json" +EXPECTED_MAX_ARTIFACT_BYTES = 256 * 1024 * 1024 + + +def _load_generator() -> ModuleType: + """Load the standalone generator without importing the package under test.""" + specification = importlib.util.spec_from_file_location( + "egressweave_generate_release_sbom_archive_bound", + GENERATOR_PATH, + ) + assert specification is not None and specification.loader is not None + module = importlib.util.module_from_spec(specification) + specification.loader.exec_module(module) + return module + + +def _write_sparse_oversized_file(path: Path) -> None: + """Create one cheap sparse fixture just above the accepted compressed bound.""" + with path.open("wb") as stream: + stream.truncate(EXPECTED_MAX_ARTIFACT_BYTES + 1) + + +def _write_minimal_archive(path: Path) -> None: + """Write one parseable wheel or source archive for live-growth regressions.""" + metadata = ( + b"Metadata-Version: 2.4\n" + b"Name: egressweave\n" + b"Version: 0.3.0\n" + b"License-Expression: Apache-2.0\n\n" + ) + if path.name.endswith(".whl"): + with zipfile.ZipFile(path, mode="w") as archive: + archive.writestr("egressweave-0.3.0.dist-info/METADATA", metadata) + return + with tarfile.open(path, mode="w:gz") as archive: + member = tarfile.TarInfo("egressweave-0.3.0/PKG-INFO") + member.size = len(metadata) + archive.addfile(member, io.BytesIO(metadata)) + + +def test_generator_uses_the_reviewed_compressed_archive_limit() -> None: + """Keep the standalone parser preflight aligned with release verification.""" + generator = _load_generator() + + assert generator.MAX_RELEASE_ARTIFACT_BYTES == EXPECTED_MAX_ARTIFACT_BYTES + + +def test_oversized_wheel_fails_before_zip_parser( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reject an oversized wheel before ZIP parser CPU or memory can be spent.""" + generator = _load_generator() + wheel_path = tmp_path / "egressweave-0.3.0-py3-none-any.whl" + _write_sparse_oversized_file(wheel_path) + + def unexpected_parser(*args, **kwargs): + pytest.fail("ZIP parser ran before the compressed-byte bound") + + monkeypatch.setattr(generator.zipfile, "ZipFile", unexpected_parser) + + with pytest.raises(SystemExit, match="compressed-byte safety bound"): + generator.build_sbom(wheel_path, MANIFEST_PATH) + + +def test_oversized_sdist_fails_before_tar_parser( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reject an oversized source archive before gzip/tar parsing begins.""" + generator = _load_generator() + sdist_path = tmp_path / "egressweave-0.3.0.tar.gz" + _write_sparse_oversized_file(sdist_path) + + def unexpected_parser(*args, **kwargs): + pytest.fail("tar parser ran before the compressed-byte bound") + + monkeypatch.setattr(generator.tarfile, "open", unexpected_parser) + + with pytest.raises(SystemExit, match="compressed-byte safety bound"): + generator.build_sbom(sdist_path, MANIFEST_PATH) + + +def test_symlinked_archive_fails_before_parser( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reject a direct archive symlink instead of following it into a parser.""" + generator = _load_generator() + target = tmp_path / "target.whl" + with zipfile.ZipFile(target, mode="w") as archive: + archive.writestr("egressweave-0.3.0.dist-info/METADATA", b"invalid") + alias = tmp_path / "egressweave-0.3.0-py3-none-any.whl" + try: + alias.symlink_to(target) + except OSError: + pytest.skip("symbolic links are unavailable on this platform") + + def unexpected_parser(*args, **kwargs): + pytest.fail("ZIP parser followed an unsafe archive link") + + monkeypatch.setattr(generator.zipfile, "ZipFile", unexpected_parser) + + with pytest.raises(SystemExit, match="missing or unsafe"): + generator.build_sbom(alias, MANIFEST_PATH) + + +def test_cli_rejects_symlinked_archive_before_parser( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Keep the CLI from resolving away the caller-supplied final symlink.""" + generator = _load_generator() + target = tmp_path / "target.whl" + _write_minimal_archive(target) + alias = tmp_path / "egressweave-0.3.0-py3-none-any.whl" + try: + alias.symlink_to(target) + except OSError: + pytest.skip("symbolic links are unavailable on this platform") + + monkeypatch.setattr( + generator, + "_parse_arguments", + lambda: SimpleNamespace( + artifact=alias, + manifest=MANIFEST_PATH, + lock=tmp_path / "runtime.lock", + output=tmp_path / "sbom.json", + ), + ) + monkeypatch.setattr(generator, "validate_runtime_lock", lambda *args: None) + + def unexpected_parser(*args, **kwargs): + pytest.fail("CLI resolved an unsafe archive link before validation") + + monkeypatch.setattr(generator.zipfile, "ZipFile", unexpected_parser) + + with pytest.raises(SystemExit, match="missing or unsafe"): + generator.main() + + +def test_archive_lstat_failure_is_normalized_before_parser( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Normalize filesystem inspection failure without exposing local details.""" + generator = _load_generator() + wheel_path = tmp_path / "egressweave-0.3.0-py3-none-any.whl" + with zipfile.ZipFile(wheel_path, mode="w") as archive: + archive.writestr("egressweave-0.3.0.dist-info/METADATA", b"invalid") + original_lstat = Path.lstat + + def fail_artifact_lstat(path: Path, *args, **kwargs): + if path == wheel_path: + raise OSError("sensitive local filesystem detail") + return original_lstat(path, *args, **kwargs) + + def unexpected_parser(*args, **kwargs): + pytest.fail("ZIP parser ran after archive inspection failed") + + monkeypatch.setattr(Path, "lstat", fail_artifact_lstat) + monkeypatch.setattr(generator.zipfile, "ZipFile", unexpected_parser) + + with pytest.raises(SystemExit, match="missing or unsafe"): + generator.build_sbom(wheel_path, MANIFEST_PATH) + + +def test_directory_archive_is_rejected_as_unsafe(tmp_path: Path) -> None: + """Reject a non-regular artifact with the same stable public failure.""" + generator = _load_generator() + directory = tmp_path / "egressweave-0.3.0-py3-none-any.whl" + directory.mkdir() + + with pytest.raises(SystemExit, match="missing or unsafe"): + generator.build_sbom(directory, MANIFEST_PATH) + + +def test_path_replacement_after_lstat_fails_before_parser( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Bind parser input to the file identity accepted by the path preflight.""" + generator = _load_generator() + wheel_path = tmp_path / "egressweave-0.3.0-py3-none-any.whl" + replacement = tmp_path / "replacement.whl" + with zipfile.ZipFile(wheel_path, mode="w") as archive: + archive.writestr("egressweave-0.3.0.dist-info/METADATA", b"original") + with zipfile.ZipFile(replacement, mode="w") as archive: + archive.writestr("egressweave-0.3.0.dist-info/METADATA", b"replacement") + original_lstat = Path.lstat + replaced = False + + def replace_after_artifact_lstat(path: Path, *args, **kwargs): + nonlocal replaced + metadata = original_lstat(path, *args, **kwargs) + if path == wheel_path and not replaced: + wheel_path.unlink() + replacement.replace(wheel_path) + replaced = True + return metadata + + def unexpected_parser(*args, **kwargs): + pytest.fail("parser opened a pathname replacement after preflight") + + monkeypatch.setattr(Path, "lstat", replace_after_artifact_lstat) + monkeypatch.setattr(generator.zipfile, "ZipFile", unexpected_parser) + + with pytest.raises(SystemExit, match="missing or unsafe"): + generator.build_sbom(wheel_path, MANIFEST_PATH) + + +@pytest.mark.parametrize( + "artifact_name", + ("egressweave-0.3.0-py3-none-any.whl", "egressweave-0.3.0.tar.gz"), +) +def test_archive_growth_after_initial_hash_is_bounded_inside_parser( + artifact_name: str, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reject a live archive that grows past the ceiling before parser reads.""" + generator = _load_generator() + artifact_path = tmp_path / artifact_name + _write_minimal_archive(artifact_path) + original_hash = generator._sha256_file + hash_calls = 0 + + def grow_after_initial_hash(stream): + nonlocal hash_calls + digest = original_hash(stream) + hash_calls += 1 + if hash_calls == 1: + with artifact_path.open("r+b") as mutable_artifact: + mutable_artifact.truncate(EXPECTED_MAX_ARTIFACT_BYTES + 1) + return digest + + monkeypatch.setattr(generator, "_sha256_file", grow_after_initial_hash) + + with pytest.raises(SystemExit, match="compressed-byte safety bound"): + generator.build_sbom(artifact_path, MANIFEST_PATH) + + assert hash_calls == 1 + + +def test_archive_mutation_during_metadata_parse_fails_closed( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Reject metadata evidence when the bound archive bytes change mid-read.""" + generator = _load_generator() + wheel_path = tmp_path / "egressweave-0.3.0-py3-none-any.whl" + metadata = ( + "Metadata-Version: 2.4\n" + "Name: egressweave\n" + "Version: 0.3.0\n" + "License-Expression: Apache-2.0\n" + "Requires-Dist: httpcore==1.0.9\n" + "Requires-Dist: httpx==0.28.1\n" + "Requires-Dist: idna==3.10\n" + "Requires-Dist: sniffio==1.3.1\n\n" + ) + with zipfile.ZipFile(wheel_path, mode="w") as archive: + archive.writestr("egressweave-0.3.0.dist-info/METADATA", metadata) + original_metadata = generator._artifact_metadata + + def mutate_after_metadata(*args, **kwargs): + result = original_metadata(*args, **kwargs) + with wheel_path.open("ab") as stream: + stream.write(b"mutated") + return result + + monkeypatch.setattr(generator, "_artifact_metadata", mutate_after_metadata) + + with pytest.raises(SystemExit, match="changed during verification"): + generator.build_sbom(wheel_path, MANIFEST_PATH) + + +def test_parser_read_all_is_capped_to_remaining_bytes_plus_tripwire( + tmp_path: Path, +) -> None: + """Never pass an unbounded parser read through to the artifact descriptor.""" + generator = _load_generator() + artifact_path = tmp_path / "artifact.whl" + artifact_path.write_bytes(b"abcdef") + requested_sizes: list[int] = [] + + with artifact_path.open("rb") as artifact_stream: + artifact_stream.seek(2) + + class RecordingStream: + """Record descriptor read sizes while delegating file operations.""" + + def fileno(self) -> int: + return artifact_stream.fileno() + + def tell(self) -> int: + return artifact_stream.tell() + + def seek(self, offset: int, whence: int = os.SEEK_SET) -> int: + return artifact_stream.seek(offset, whence) + + def read(self, size: int = -1) -> bytes: + requested_sizes.append(size) + return artifact_stream.read(size) + + reader = generator._LiveBoundedArtifactReader(RecordingStream()) + assert reader.read() == b"cdef" + + assert requested_sizes == [EXPECTED_MAX_ARTIFACT_BYTES - 2 + 1] + + +def test_growth_after_parser_seek_fails_before_the_next_read(tmp_path: Path) -> None: + """Recheck the live descriptor after a parser seek and before later reads.""" + generator = _load_generator() + artifact_path = tmp_path / "artifact.whl" + artifact_path.write_bytes(b"abcdef") + + with artifact_path.open("rb") as artifact_stream: + reader = generator._LiveBoundedArtifactReader(artifact_stream) + assert reader.seek(1) == 1 + with artifact_path.open("r+b") as mutable_artifact: + mutable_artifact.truncate(EXPECTED_MAX_ARTIFACT_BYTES + 1) + with pytest.raises(SystemExit, match="compressed-byte safety bound"): + reader.read(1) + + +def test_parser_read_rejects_out_of_range_position_before_underlying_read( + tmp_path: Path, +) -> None: + """Refuse an impossible live offset before it can turn read(-1) unbounded.""" + generator = _load_generator() + artifact_path = tmp_path / "artifact.whl" + artifact_path.write_bytes(b"abcdef") + + with artifact_path.open("rb") as artifact_stream: + + class OutOfRangeStream: + """Expose a valid descriptor but an impossible parser position.""" + + def fileno(self) -> int: + return artifact_stream.fileno() + + def tell(self) -> int: + return EXPECTED_MAX_ARTIFACT_BYTES + 1 + + def read(self, size: int = -1) -> bytes: + pytest.fail(f"underlying read received unsafe size {size}") + + reader = generator._LiveBoundedArtifactReader(OutOfRangeStream()) + with pytest.raises(SystemExit, match="missing or unsafe"): + reader.read() + + +def test_parser_seek_rejects_negative_result(tmp_path: Path) -> None: + """Reject a negative parser position even when a hostile stream returns it.""" + generator = _load_generator() + artifact_path = tmp_path / "artifact.whl" + artifact_path.write_bytes(b"abcdef") + + with artifact_path.open("rb") as artifact_stream: + + class NegativeSeekStream: + """Return a negative seek position without raising an OS error.""" + + def fileno(self) -> int: + return artifact_stream.fileno() + + def tell(self) -> int: + return artifact_stream.tell() + + def seek(self, offset: int, whence: int = os.SEEK_SET) -> int: + return -1 + + reader = generator._LiveBoundedArtifactReader(NegativeSeekStream()) + with pytest.raises(SystemExit, match="missing or unsafe"): + reader.seek(0) + + +@pytest.mark.parametrize("operation", ("fileno", "tell", "read", "seek")) +def test_parser_descriptor_failures_use_one_stable_error( + operation: str, + tmp_path: Path, +) -> None: + """Keep parser-visible descriptor failures behind the public safe boundary.""" + generator = _load_generator() + artifact_path = tmp_path / "artifact.whl" + artifact_path.write_bytes(b"abcdef") + + with artifact_path.open("rb") as artifact_stream: + + class FailingStream: + """Fail one selected descriptor operation with sensitive detail.""" + + def fileno(self) -> int: + if operation == "fileno": + raise OSError("sensitive descriptor detail") + return artifact_stream.fileno() + + def tell(self) -> int: + if operation == "tell": + raise OSError("sensitive descriptor detail") + return artifact_stream.tell() + + def read(self, size: int = -1) -> bytes: + if operation == "read": + raise OSError("sensitive descriptor detail") + return artifact_stream.read(size) + + def seek(self, offset: int, whence: int = os.SEEK_SET) -> int: + if operation == "seek": + raise OSError("sensitive descriptor detail") + return artifact_stream.seek(offset, whence) + + reader = generator._LiveBoundedArtifactReader(FailingStream()) + with pytest.raises(SystemExit, match="release artifact is missing or unsafe"): + if operation == "fileno": + reader.fileno() + elif operation == "tell": + reader.tell() + elif operation == "read": + reader.read(1) + else: + reader.seek(0)