fix(ci): stream OCI/archive scans; keep source nest fail-open

Avoid full-buffer gzip/layer decompression and hard-failing large members; chunk-scan with overlap and budgets. Opportunistic source nesting skips malformed magic without losing raw findings; explicit archive/image modes stay fail-closed.
2026-07-30 16:43:41 -07:00
parent cd48276e8c
commit 5022836354
5 changed files with 719 additions and 246 deletions
@@ -1,20 +1,35 @@
 #!/usr/bin/env python3
-"""Bounded nested content inspection for gzip/zip/tar payloads (stdlib only)."""
+"""Bounded nested content inspection for gzip/zip/tar payloads (stdlib only).
+
+Image/OCI layers stream gzip→tar without buffering full decompressed archives.
+Ordinary source scans treat nested detection as opportunistic: malformed magic
+keeps raw-byte findings and does not fail the whole scan. Explicit archive/image
+modes remain fail-closed.
+"""

 from __future__ import annotations

 import gzip
 import io
+import struct
 import tarfile
+import tempfile
 import zipfile
-from typing import List, Optional, Sequence, Tuple
+from typing import BinaryIO, List, Optional, Sequence

 import scan_text

-MAX_MEMBER_BYTES = 32 * 1024 * 1024
-MAX_TOTAL_BYTES = 512 * 1024 * 1024
+# Small members may be fully buffered for nested recursion.
+MAX_BUFFERED_MEMBER = 32 * 1024 * 1024
+# Realistic total uncompressed scan budget (GiB-level) per top-level scan root.
+MAX_TOTAL_BYTES = 4 * 1024 * 1024 * 1024
 MAX_MEMBERS = 200_000
 MAX_NEST_DEPTH = 5
+CHUNK_SIZE = 1024 * 1024
+CHUNK_OVERLAP = 64 * 1024
+
+# Back-compat aliases used by other modules/tests.
+MAX_MEMBER_BYTES = MAX_BUFFERED_MEMBER


 class ArchiveSafetyError(Exception):
@@ -22,17 +37,25 @@ class ArchiveSafetyError(Exception):


 class _Budget:
-    def __init__(self) -> None:
+    def __init__(self, limit: int = MAX_TOTAL_BYTES) -> None:
        self.total = 0
        self.members = 0
+        self.limit = limit

-    def add(self, n: int, label: str) -> None:
+    def add_member(self, label: str) -> None:
        self.members += 1
        if self.members > MAX_MEMBERS:
            raise ArchiveSafetyError(f"too many nested members near {label}")
+
+    def add_bytes(self, n: int, label: str) -> None:
+        if n < 0:
+            raise ArchiveSafetyError(f"negative budget add near {label}")
        self.total += n
-        if self.total > MAX_TOTAL_BYTES:
-            raise ArchiveSafetyError(f"nested uncompressed budget exceeded near {label}")
+        if self.total > self.limit:
+            raise ArchiveSafetyError(
+                f"uncompressed scan budget exceeded near {label} "
+                f"({self.total} > {self.limit})"
+            )


 def _validate_member_name(name: str) -> str:
@@ -54,114 +77,298 @@ def _is_zip(data: bytes) -> bool:
    return len(data) >= 4 and data[:2] == b"PK" and data[2] in (0x03, 0x05, 0x07)


-def _is_tar(data: bytes) -> bool:
-    if len(data) < 512:
+def _looks_tar_header(head: bytes) -> bool:
+    if len(head) < 262:
        return False
-    if data[257:262] == b"ustar":
+    if head[257:262] == b"ustar":
        return True
-    # POSIX tar with empty magic still has a plausible header checksum field.
-    try:
-        with tarfile.open(fileobj=io.BytesIO(data), mode="r:") as tf:
-            # Presence of at least one member confirms tar.
-            for _ in tf:
-                return True
-    except tarfile.TarError:
-        return False
    return False


-def _gunzip_limited(data: bytes, label: str) -> bytes:
+def _reject_link(member: tarfile.TarInfo) -> None:
+    link = member.linkname or ""
+    name = member.name or ""
+    if (
+        ".." in name.replace("\\", "/")
+        or ".." in link.replace("\\", "/")
+        or link.startswith("/")
+        or name.startswith("/")
+    ):
+        raise ArchiveSafetyError(f"dangerous link in archive: {name} -> {link}")
+    raise ArchiveSafetyError(f"link member forbidden in nested archive: {name}")
+
+
+def _prefer_tar_path(rel_path: str) -> bool:
+    # Only the current leaf segment (after the last nest marker) decides tar-ishness.
+    # Otherwise parent names like "layer.tar#tar/etc/issue" falsely force tar parsing.
+    leaf = rel_path.split("#")[-1].lower()
+    base = leaf.rsplit("/", 1)[-1]
+    return (
+        base.endswith(".tar")
+        or base.endswith(".tar.gz")
+        or base.endswith(".tgz")
+        or base == "layer.tar"
+        or leaf.endswith("/layer.tar")
+    )
+
+
+def _opportunistic_exc(exc: BaseException) -> bool:
+    return isinstance(
+        exc,
+        (
+            ArchiveSafetyError,
+            OSError,
+            EOFError,
+            tarfile.TarError,
+            zipfile.BadZipFile,
+            zipfile.LargeZipFile,
+            ValueError,
+            struct.error,
+        ),
+    )
+
+
+class _ChainReader(io.RawIOBase):
+    """Readable stream that yields prefix bytes then delegates to an underlying reader."""
+
+    def __init__(self, prefix: bytes, rest: BinaryIO) -> None:
+        super().__init__()
+        self._prefix = prefix
+        self._rest = rest
+        self._pos = 0
+
+    def readable(self) -> bool:
+        return True
+
+    def read(self, size: int = -1) -> bytes:  # type: ignore[override]
+        if size == 0:
+            return b""
+        if self._pos < len(self._prefix):
+            if size < 0:
+                out = self._prefix[self._pos :]
+                self._pos = len(self._prefix)
+                more = self._rest.read(-1) or b""
+                return out + more
+            take = self._prefix[self._pos : self._pos + size]
+            self._pos += len(take)
+            if len(take) == size:
+                return take
+            more = self._rest.read(size - len(take)) or b""
+            return take + more
+        if size < 0:
+            return self._rest.read(-1) or b""
+        return self._rest.read(size) or b""
+
+
+def scan_stream_chunks(
+    fh: BinaryIO,
+    rel_path: str,
+    rules: Sequence[scan_text.Rule],
+    allowlist: Sequence[str],
+    budget: _Budget,
+    *,
+    chunk_size: int = CHUNK_SIZE,
+    overlap: int = CHUNK_OVERLAP,
+) -> List[str]:
+    """Stream-scan a readable binary stream with overlap for boundary-spanning markers."""
+    findings: List[str] = []
+    prev = b""
+    while True:
+        chunk = fh.read(chunk_size)
+        if not chunk:
+            break
+        budget.add_bytes(len(chunk), rel_path)
+        window = prev + chunk
+        findings.extend(scan_text.match_rules(window, rel_path, rules, allowlist))
+        if len(window) > overlap:
+            prev = window[-overlap:]
+        else:
+            prev = window
+    return findings
+
+
+def _scan_tar_stream(
+    fh: BinaryIO,
+    rel_path: str,
+    rules: Sequence[scan_text.Rule],
+    allowlist: Sequence[str],
+    budget: _Budget,
+    *,
+    nest_depth: int,
+    strict: bool,
+    stream_mode: bool,
+) -> List[str]:
+    findings: List[str] = []
+    mode = "r|" if stream_mode else "r:*"
    try:
-        with gzip.GzipFile(fileobj=io.BytesIO(data), mode="rb") as gz:
-            out = gz.read(MAX_MEMBER_BYTES + 1)
+        with tarfile.open(fileobj=fh, mode=mode) as tf:
+            for member in tf:
+                if member.issym() or member.islnk():
+                    _reject_link(member)
+                if not member.isfile():
+                    continue
+                _validate_member_name(member.name)
+                nested_rel = f"{rel_path}#tar/{scan_text.norm_rel(member.name)}"
+                budget.add_member(nested_rel)
+
+                extracted = tf.extractfile(member)
+                if extracted is None:
+                    continue
+
+                size = int(member.size)
+                if size < 0:
+                    raise ArchiveSafetyError(f"negative tar size for {member.name}")
+
+                if size > MAX_BUFFERED_MEMBER:
+                    findings.extend(
+                        scan_stream_chunks(
+                            extracted, nested_rel, rules, allowlist, budget
+                        )
+                    )
+                    continue
+
+                blob = extracted.read(MAX_BUFFERED_MEMBER + 1)
+                if len(blob) > MAX_BUFFERED_MEMBER:
+                    raise ArchiveSafetyError(f"expanded member too large: {member.name}")
+                if size != len(blob):
+                    raise ArchiveSafetyError(
+                        f"declared/actual size mismatch for {member.name}"
+                    )
+                budget.add_bytes(len(blob), nested_rel)
+                findings.extend(
+                    scan_nested_bytes(
+                        blob,
+                        nested_rel,
+                        rules,
+                        allowlist,
+                        nest_depth=nest_depth + 1,
+                        budget=budget,
+                        strict=strict,
+                    )
+                )
+    except tarfile.TarError as exc:
+        raise ArchiveSafetyError(f"tar parse failed at {rel_path}: {exc}") from exc
+    return findings
+
+
+def _scan_gzip_payload(
+    fh: BinaryIO,
+    rel_path: str,
+    rules: Sequence[scan_text.Rule],
+    allowlist: Sequence[str],
+    budget: _Budget,
+    *,
+    nest_depth: int,
+    strict: bool,
+    prefer_tar: bool,
+) -> List[str]:
+    try:
+        gz = gzip.GzipFile(fileobj=fh, mode="rb")
    except OSError as exc:
-        raise ArchiveSafetyError(f"gzip parse failed at {label}: {exc}") from exc
-    if len(out) > MAX_MEMBER_BYTES:
-        raise ArchiveSafetyError(f"gzip member too large at {label}")
-    return out
+        raise ArchiveSafetyError(f"gzip parse failed at {rel_path}: {exc}") from exc
+
+    try:
+        head = gz.read(512)
+    except OSError as exc:
+        raise ArchiveSafetyError(f"gzip parse failed at {rel_path}: {exc}") from exc
+
+    rest = _ChainReader(head, gz)
+    gzip_rel = f"{rel_path}#gzip"
+
+    if prefer_tar or _looks_tar_header(head):
+        return _scan_tar_stream(
+            rest,
+            gzip_rel,
+            rules,
+            allowlist,
+            budget,
+            nest_depth=nest_depth,
+            strict=strict,
+            stream_mode=True,
+        )
+
+    return scan_stream_chunks(rest, gzip_rel, rules, allowlist, budget)


-def _read_zip_member(zf: zipfile.ZipFile, info: zipfile.ZipInfo) -> bytes:
+def _scan_zip_bytes(
+    data: bytes,
+    rel_path: str,
+    rules: Sequence[scan_text.Rule],
+    allowlist: Sequence[str],
+    budget: _Budget,
+    *,
+    nest_depth: int,
+    strict: bool,
+) -> List[str]:
+    findings: List[str] = []
+    try:
+        with zipfile.ZipFile(io.BytesIO(data)) as zf:
+            for info in zf.infolist():
+                if info.is_dir():
+                    continue
+                findings.extend(
+                    _scan_zip_member(
+                        zf,
+                        info,
+                        rel_path,
+                        rules,
+                        allowlist,
+                        budget,
+                        nest_depth=nest_depth,
+                        strict=strict,
+                    )
+                )
+    except zipfile.BadZipFile as exc:
+        raise ArchiveSafetyError(f"zip parse failed at {rel_path}: {exc}") from exc
+    return findings
+
+
+def _scan_zip_member(
+    zf: zipfile.ZipFile,
+    info: zipfile.ZipInfo,
+    rel_base: str,
+    rules: Sequence[scan_text.Rule],
+    allowlist: Sequence[str],
+    budget: _Budget,
+    *,
+    nest_depth: int,
+    strict: bool,
+) -> List[str]:
    name = info.filename
    _validate_member_name(name)
    mode = (info.external_attr >> 16) & 0o170000
    if mode == 0o120000:
        raise ArchiveSafetyError(f"symlink member forbidden: {name}")
-    if info.file_size > MAX_MEMBER_BYTES:
-        raise ArchiveSafetyError(f"member too large: {name} ({info.file_size})")
-    # Stream decompression; reject excess before allocating beyond the cap.
+    nested_rel = f"{rel_base}#zip/{scan_text.norm_rel(name)}"
+    budget.add_member(nested_rel)
+
    try:
-        with zf.open(info, "r") as fh:
-            data = fh.read(MAX_MEMBER_BYTES + 1)
+        fh = zf.open(info, "r")
    except (zipfile.BadZipFile, RuntimeError, OSError) as exc:
        raise ArchiveSafetyError(f"zip member read failed for {name}: {exc}") from exc
-    if len(data) > MAX_MEMBER_BYTES:
+
+    with fh:
+        if info.file_size > MAX_BUFFERED_MEMBER:
+            return scan_stream_chunks(fh, nested_rel, rules, allowlist, budget)
+        data = fh.read(MAX_BUFFERED_MEMBER + 1)
+
+    if len(data) > MAX_BUFFERED_MEMBER:
        raise ArchiveSafetyError(f"expanded member too large: {name}")
    if info.file_size != len(data):
        raise ArchiveSafetyError(
            f"declared/actual size mismatch for {name}: "
            f"declared={info.file_size} actual={len(data)}"
        )
-    return data
-
-
-def iter_zip_bytes(data: bytes, label: str, budget: _Budget) -> List[Tuple[str, bytes]]:
-    out: List[Tuple[str, bytes]] = []
-    try:
-        with zipfile.ZipFile(io.BytesIO(data)) as zf:
-            for info in zf.infolist():
-                if info.is_dir():
-                    continue
-                member = _read_zip_member(zf, info)
-                budget.add(len(member), f"{label}/{info.filename}")
-                out.append((info.filename, member))
-    except zipfile.BadZipFile as exc:
-        raise ArchiveSafetyError(f"zip parse failed at {label}: {exc}") from exc
-    return out
-
-
-def iter_tar_bytes(data: bytes, label: str, budget: _Budget) -> List[Tuple[str, bytes]]:
-    out: List[Tuple[str, bytes]] = []
-    try:
-        with tarfile.open(fileobj=io.BytesIO(data), mode="r:*") as tf:
-            for member in tf:
-                if member.issym() or member.islnk():
-                    link = member.linkname or ""
-                    if (
-                        ".." in member.name.replace("\\", "/")
-                        or ".." in link.replace("\\", "/")
-                        or link.startswith("/")
-                        or member.name.startswith("/")
-                    ):
-                        raise ArchiveSafetyError(
-                            f"dangerous link in archive: {member.name} -> {link}"
-                        )
-                    # Non-dangerous links still rejected fail-closed for nested payloads.
-                    raise ArchiveSafetyError(
-                        f"link member forbidden in nested archive: {member.name}"
-                    )
-                if not member.isfile():
-                    continue
-                _validate_member_name(member.name)
-                if member.size > MAX_MEMBER_BYTES:
-                    raise ArchiveSafetyError(
-                        f"member too large: {member.name} ({member.size})"
-                    )
-                extracted = tf.extractfile(member)
-                if extracted is None:
-                    continue
-                blob = extracted.read(MAX_MEMBER_BYTES + 1)
-                if len(blob) > MAX_MEMBER_BYTES:
-                    raise ArchiveSafetyError(f"expanded member too large: {member.name}")
-                if member.size != len(blob):
-                    raise ArchiveSafetyError(
-                        f"declared/actual size mismatch for {member.name}"
-                    )
-                budget.add(len(blob), f"{label}/{member.name}")
-                out.append((member.name, blob))
-    except tarfile.TarError as exc:
-        raise ArchiveSafetyError(f"tar parse failed at {label}: {exc}") from exc
-    return out
+    budget.add_bytes(len(data), nested_rel)
+    return scan_nested_bytes(
+        data,
+        nested_rel,
+        rules,
+        allowlist,
+        nest_depth=nest_depth + 1,
+        budget=budget,
+        strict=strict,
+    )


 def scan_nested_bytes(
@@ -172,67 +379,195 @@ def scan_nested_bytes(
    *,
    nest_depth: int = 0,
    budget: Optional[_Budget] = None,
+    strict: bool = True,
 ) -> List[str]:
    if budget is None:
        budget = _Budget()
-        budget.add(len(data), rel_path)

-    findings = scan_text.match_rules(data, rel_path, rules, allowlist)
+    # Raw-byte findings first (including NUL binaries / large blobs).
+    if len(data) > MAX_BUFFERED_MEMBER:
+        findings = scan_stream_chunks(
+            io.BytesIO(data), rel_path, rules, allowlist, budget
+        )
+    else:
+        budget.add_bytes(len(data), rel_path)
+        findings = scan_text.match_rules(data, rel_path, rules, allowlist)
+
    if nest_depth >= MAX_NEST_DEPTH:
        return findings

-    # Recognized compressed / archive payloads: fail closed on parse errors.
-    if _is_gzip(data):
-        inner = _gunzip_limited(data, rel_path)
-        budget.add(len(inner), rel_path + "#gzip")
-        findings.extend(
-            scan_nested_bytes(
-                inner,
-                f"{rel_path}#gzip",
-                rules,
-                allowlist,
-                nest_depth=nest_depth + 1,
-                budget=budget,
-            )
-        )
-        return findings
+    prefer_tar = _prefer_tar_path(rel_path)

-    if _is_zip(data):
-        for name, member in iter_zip_bytes(data, rel_path, budget):
-            nested_rel = f"{rel_path}#zip/{scan_text.norm_rel(name)}"
+    try:
+        if _is_gzip(data):
            findings.extend(
-                scan_nested_bytes(
-                    member,
-                    nested_rel,
+                _scan_gzip_payload(
+                    io.BytesIO(data),
+                    rel_path,
                    rules,
                    allowlist,
-                    nest_depth=nest_depth + 1,
-                    budget=budget,
+                    budget,
+                    nest_depth=nest_depth,
+                    strict=strict,
+                    prefer_tar=prefer_tar,
                )
            )
-        return findings
+            return findings

-    # Only attempt tar when magic/path strongly indicates tar (avoid false positives).
-    path_hint = rel_path.lower()
-    tarish = (
-        path_hint.endswith(".tar")
-        or path_hint.endswith(".tar.gz")
-        or path_hint.endswith(".tgz")
-        or "/layer.tar" in path_hint
-        or path_hint.endswith("#gzip")
-        or (len(data) >= 262 and data[257:262] == b"ustar")
-    )
-    if tarish and _is_tar(data):
-        for name, member in iter_tar_bytes(data, rel_path, budget):
-            nested_rel = f"{rel_path}#tar/{scan_text.norm_rel(name)}"
+        if _is_zip(data):
            findings.extend(
-                scan_nested_bytes(
-                    member,
-                    nested_rel,
+                _scan_zip_bytes(
+                    data,
+                    rel_path,
                    rules,
                    allowlist,
-                    nest_depth=nest_depth + 1,
-                    budget=budget,
+                    budget,
+                    nest_depth=nest_depth,
+                    strict=strict,
                )
            )
+            return findings
+
+        tarish = prefer_tar or _looks_tar_header(data[:512] if len(data) >= 512 else data)
+        if tarish:
+            findings.extend(
+                _scan_tar_stream(
+                    io.BytesIO(data),
+                    rel_path,
+                    rules,
+                    allowlist,
+                    budget,
+                    nest_depth=nest_depth,
+                    strict=strict,
+                    stream_mode=False,
+                )
+            )
+    except Exception as exc:  # noqa: BLE001 - opportunistic fallback is intentional
+        if strict or not _opportunistic_exc(exc):
+            if isinstance(exc, ArchiveSafetyError):
+                raise
+            raise ArchiveSafetyError(f"nested parse failed at {rel_path}: {exc}") from exc
+        return findings
+
    return findings
+
+
+def scan_nested_fileobj(
+    fh: BinaryIO,
+    rel_path: str,
+    rules: Sequence[scan_text.Rule],
+    allowlist: Sequence[str],
+    *,
+    nest_depth: int = 0,
+    budget: Optional[_Budget] = None,
+    strict: bool = True,
+) -> List[str]:
+    """Scan a file object using spooling/streaming (no multi-hundred-MiB RAM buffer)."""
+    if budget is None:
+        budget = _Budget()
+
+    # Ensure seekable backing store without holding the whole blob in RAM.
+    if not (hasattr(fh, "seek") and hasattr(fh, "tell")):
+        spool: BinaryIO = tempfile.SpooledTemporaryFile(max_size=8 * 1024 * 1024)
+        while True:
+            chunk = fh.read(CHUNK_SIZE)
+            if not chunk:
+                break
+            spool.write(chunk)
+        spool.seek(0)
+        fh = spool
+
+    head = fh.read(512)
+    fh.seek(0)
+
+    # Raw findings via streaming (counts compressed/raw blob bytes once).
+    findings = scan_stream_chunks(fh, rel_path, rules, allowlist, budget)
+    fh.seek(0)
+
+    if nest_depth >= MAX_NEST_DEPTH:
+        return findings
+
+    prefer_tar = _prefer_tar_path(rel_path)
+
+    try:
+        if _is_gzip(head):
+            findings.extend(
+                _scan_gzip_payload(
+                    fh,
+                    rel_path,
+                    rules,
+                    allowlist,
+                    budget,
+                    nest_depth=nest_depth,
+                    strict=strict,
+                    prefer_tar=prefer_tar,
+                )
+            )
+            return findings
+
+        if _is_zip(head):
+            with zipfile.ZipFile(fh) as zf:
+                for info in zf.infolist():
+                    if info.is_dir():
+                        continue
+                    findings.extend(
+                        _scan_zip_member(
+                            zf,
+                            info,
+                            rel_path,
+                            rules,
+                            allowlist,
+                            budget,
+                            nest_depth=nest_depth,
+                            strict=strict,
+                        )
+                    )
+            return findings
+
+        if prefer_tar or _looks_tar_header(head):
+            findings.extend(
+                _scan_tar_stream(
+                    fh,
+                    rel_path,
+                    rules,
+                    allowlist,
+                    budget,
+                    nest_depth=nest_depth,
+                    strict=strict,
+                    stream_mode=False,
+                )
+            )
+    except Exception as exc:  # noqa: BLE001
+        if strict or not _opportunistic_exc(exc):
+            if isinstance(exc, ArchiveSafetyError):
+                raise
+            raise ArchiveSafetyError(f"nested parse failed at {rel_path}: {exc}") from exc
+        return findings
+
+    return findings
+
+
+def _read_zip_member(zf: zipfile.ZipFile, info: zipfile.ZipInfo) -> bytes:
+    """Buffered zip read for small members (explicit archive mode helpers)."""
+    name = info.filename
+    _validate_member_name(name)
+    mode = (info.external_attr >> 16) & 0o170000
+    if mode == 0o120000:
+        raise ArchiveSafetyError(f"symlink member forbidden: {name}")
+    if info.file_size > MAX_BUFFERED_MEMBER:
+        raise ArchiveSafetyError(
+            f"member requires streaming extract: {name} ({info.file_size})"
+        )
+    try:
+        with zf.open(info, "r") as fh:
+            data = fh.read(MAX_BUFFERED_MEMBER + 1)
+    except (zipfile.BadZipFile, RuntimeError, OSError) as exc:
+        raise ArchiveSafetyError(f"zip member read failed for {name}: {exc}") from exc
+    if len(data) > MAX_BUFFERED_MEMBER:
+        raise ArchiveSafetyError(f"expanded member too large: {name}")
+    if info.file_size != len(data):
+        raise ArchiveSafetyError(
+            f"declared/actual size mismatch for {name}: "
+            f"declared={info.file_size} actual={len(data)}"
+        )
+    return data
@@ -2,6 +2,7 @@
 """Safely inspect source-release archive members and scan for private leakage.

 Rejects absolute paths, traversal, symlink escapes, and unreasonable sizes.
+Large members are stream-scanned rather than hard-failed solely for size.
 Uses only the Python standard library.
 """

@@ -12,7 +13,7 @@ import sys
 import tarfile
 import zipfile
 from pathlib import Path
-from typing import Iterable, List, Optional, Sequence, Tuple
+from typing import List, Optional, Sequence

 _SCAN_DIR = Path(__file__).resolve().parent
 if str(_SCAN_DIR) not in sys.path:
@@ -20,48 +21,100 @@ if str(_SCAN_DIR) not in sys.path:

 import scan_text  # noqa: E402
 from nested_content import (  # noqa: E402
-    MAX_MEMBER_BYTES,
+    MAX_BUFFERED_MEMBER,
    MAX_MEMBERS,
-    MAX_TOTAL_BYTES,
    ArchiveSafetyError,
-    _read_zip_member,
+    _Budget,
    _validate_member_name,
+    scan_nested_bytes,
+    scan_stream_chunks,
 )


-def _open_tar(path: Path) -> tarfile.TarFile:
-    return tarfile.open(path, mode="r:*")
-
-
 def _strip_single_top_dir(name: str) -> str:
-    """GitHub source archives nest files under Repo-tag/; normalize to repo paths."""
    parts = name.replace("\\", "/").split("/")
    if len(parts) >= 2 and parts[0]:
        return "/".join(parts[1:])
    return name.replace("\\", "/")


-def iter_zip_members(path: Path) -> Iterable[Tuple[str, bytes]]:
-    total = 0
-    count = 0
+def scan_archive(path: Path, allowlist_path: Path) -> List[str]:
+    allowlist = scan_text.load_allowlist(allowlist_path)
+    rules = scan_text.compile_rules()
+    budget = _Budget()
+
+    suffix = path.name.lower()
+    if suffix.endswith(".zip"):
+        return _scan_zip_archive(path, rules, allowlist, budget)
+    if (
+        suffix.endswith(".tar")
+        or suffix.endswith(".tar.gz")
+        or suffix.endswith(".tgz")
+        or suffix.endswith(".tar.bz2")
+        or suffix.endswith(".tar.xz")
+    ):
+        return _scan_tar_archive(path, rules, allowlist, budget)
+    try:
+        return _scan_tar_archive(path, rules, allowlist, budget)
+    except (tarfile.TarError, ArchiveSafetyError):
+        return _scan_zip_archive(path, rules, allowlist, budget)
+
+
+def _scan_zip_archive(
+    path: Path,
+    rules: Sequence[scan_text.Rule],
+    allowlist: Sequence[str],
+    budget: _Budget,
+) -> List[str]:
+    findings: List[str] = []
    with zipfile.ZipFile(path) as zf:
+        count = 0
        for info in zf.infolist():
            if info.is_dir():
                continue
            count += 1
            if count > MAX_MEMBERS:
                raise ArchiveSafetyError("too many archive members")
-            data = _read_zip_member(zf, info)
-            total += len(data)
-            if total > MAX_TOTAL_BYTES:
-                raise ArchiveSafetyError("archive uncompressed budget exceeded")
-            yield info.filename, data
+            rel = scan_text.norm_rel(_strip_single_top_dir(info.filename))
+            if not rel:
+                continue
+            _validate_member_name(info.filename)
+            mode = (info.external_attr >> 16) & 0o170000
+            if mode == 0o120000:
+                raise ArchiveSafetyError(f"symlink member forbidden: {info.filename}")
+
+            with zf.open(info, "r") as fh:
+                if info.file_size > MAX_BUFFERED_MEMBER:
+                    budget.add_member(rel)
+                    findings.extend(
+                        scan_stream_chunks(fh, rel, rules, allowlist, budget)
+                    )
+                    continue
+                data = fh.read(MAX_BUFFERED_MEMBER + 1)
+            if len(data) > MAX_BUFFERED_MEMBER:
+                raise ArchiveSafetyError(f"expanded member too large: {info.filename}")
+            if info.file_size != len(data):
+                raise ArchiveSafetyError(
+                    f"declared/actual size mismatch for {info.filename}"
+                )
+            budget.add_member(rel)
+            findings.extend(
+                scan_nested_bytes(
+                    data, rel, rules, allowlist, budget=budget, strict=True
+                )
+            )
+    return findings


-def iter_tar_members(path: Path) -> Iterable[Tuple[str, bytes]]:
-    total = 0
-    count = 0
-    with _open_tar(path) as tf:
+def _scan_tar_archive(
+    path: Path,
+    rules: Sequence[scan_text.Rule],
+    allowlist: Sequence[str],
+    budget: _Budget,
+) -> List[str]:
+    findings: List[str] = []
+    with tarfile.open(path, mode="r:*") as tf:
+        count = 0
        for member in tf:
            if not member.isfile():
                if member.issym() or member.islnk():
@@ -70,51 +123,32 @@ def iter_tar_members(path: Path) -> Iterable[Tuple[str, bytes]]:
            count += 1
            if count > MAX_MEMBERS:
                raise ArchiveSafetyError("too many archive members")
-            name = member.name
-            _validate_member_name(name)
-            if member.size > MAX_MEMBER_BYTES:
-                raise ArchiveSafetyError(f"member too large: {name} ({member.size})")
+            _validate_member_name(member.name)
+            rel = scan_text.norm_rel(_strip_single_top_dir(member.name))
+            if not rel:
+                continue
            extracted = tf.extractfile(member)
            if extracted is None:
                continue
-            data = extracted.read(MAX_MEMBER_BYTES + 1)
-            if len(data) > MAX_MEMBER_BYTES:
-                raise ArchiveSafetyError(f"expanded member too large: {name}")
-            if member.size != len(data):
-                raise ArchiveSafetyError(f"declared/actual size mismatch for {name}")
-            total += len(data)
-            if total > MAX_TOTAL_BYTES:
-                raise ArchiveSafetyError("archive uncompressed budget exceeded")
-            yield name, data
-
-
-def scan_archive(path: Path, allowlist_path: Path) -> List[str]:
-    allowlist = scan_text.load_allowlist(allowlist_path)
-    rules = scan_text.compile_rules()
-    findings: List[str] = []
-
-    suffix = path.name.lower()
-    if suffix.endswith(".zip"):
-        members = iter_zip_members(path)
-    elif (
-        suffix.endswith(".tar")
-        or suffix.endswith(".tar.gz")
-        or suffix.endswith(".tgz")
-        or suffix.endswith(".tar.bz2")
-        or suffix.endswith(".tar.xz")
-    ):
-        members = iter_tar_members(path)
-    else:
-        try:
-            members = list(iter_tar_members(path))
-        except (tarfile.TarError, ArchiveSafetyError):
-            members = iter_zip_members(path)
-
-    for name, data in members:
-        rel = scan_text.norm_rel(_strip_single_top_dir(name))
-        if not rel:
-            continue
-        findings.extend(scan_text.scan_bytes(data, rel, rules, allowlist))
+            size = int(member.size)
+            budget.add_member(rel)
+            if size > MAX_BUFFERED_MEMBER:
+                findings.extend(
+                    scan_stream_chunks(extracted, rel, rules, allowlist, budget)
+                )
+                continue
+            data = extracted.read(MAX_BUFFERED_MEMBER + 1)
+            if len(data) > MAX_BUFFERED_MEMBER:
+                raise ArchiveSafetyError(f"expanded member too large: {member.name}")
+            if size != len(data):
+                raise ArchiveSafetyError(
+                    f"declared/actual size mismatch for {member.name}"
+                )
+            findings.extend(
+                scan_nested_bytes(
+                    data, rel, rules, allowlist, budget=budget, strict=True
+                )
+            )
    return findings


@@ -1,13 +1,18 @@
 #!/usr/bin/env python3
-"""Inspect a docker-save tarball (configs + layer archives) without running containers."""
+"""Inspect a docker-save tarball (configs + layer archives) without running containers.
+
+Layer blobs are spooled to disk-backed temporary files and scanned with streaming
+gzip→tar iteration — never fully buffered at a 256MiB RAM cap.
+"""

 from __future__ import annotations

 import argparse
 import sys
 import tarfile
+import tempfile
 from pathlib import Path
-from typing import BinaryIO, List, Optional, Sequence
+from typing import List, Optional, Sequence

 _SCAN_DIR = Path(__file__).resolve().parent
 if str(_SCAN_DIR) not in sys.path:
@@ -15,24 +20,19 @@ if str(_SCAN_DIR) not in sys.path:

 import scan_text  # noqa: E402
 from nested_content import (  # noqa: E402
-    MAX_MEMBER_BYTES,
+    CHUNK_SIZE,
    ArchiveSafetyError,
+    _Budget,
    _validate_member_name,
-    scan_nested_bytes,
+    scan_nested_fileobj,
 )


-def _read_limited(fh: BinaryIO, limit: int) -> bytes:
-    data = fh.read(limit + 1)
-    if len(data) > limit:
-        raise ArchiveSafetyError("expanded member exceeds size limit")
-    return data
-
-
 def scan_docker_save(save_tar: Path, allowlist: Path, image_label: str) -> List[str]:
    allow = scan_text.load_allowlist(allowlist)
    rules = scan_text.compile_rules()
    findings: List[str] = []
+    budget = _Budget()

    with tarfile.open(save_tar, mode="r:*") as outer:
        for member in outer:
@@ -45,24 +45,30 @@ def scan_docker_save(save_tar: Path, allowlist: Path, image_label: str) -> List[
            if not member.isfile():
                continue
            _validate_member_name(name)
-            lower_probe = name.lower()
-            is_layerish = (
-                lower_probe.endswith(".tar")
-                or lower_probe.endswith(".tar.gz")
-                or lower_probe.endswith("/layer.tar")
-                or "/blobs/" in lower_probe
-            )
-            outer_limit = 256 * 1024 * 1024 if is_layerish else MAX_MEMBER_BYTES
-            if member.size > outer_limit:
-                raise ArchiveSafetyError(f"docker-save member too large: {name}")
            fh = outer.extractfile(member)
            if fh is None:
                continue
-            data = _read_limited(fh, outer_limit)
+
            rel = f"image/{image_label}/{scan_text.norm_rel(name)}"

-            # Config/manifest JSON and layer blobs: nested scan (gzip/tar/zip + raw).
-            findings.extend(scan_nested_bytes(data, rel, rules, allow))
+            # Stream outer member into a spooled temp file (small RAM, disk-backed).
+            with tempfile.SpooledTemporaryFile(max_size=8 * 1024 * 1024) as spool:
+                while True:
+                    chunk = fh.read(CHUNK_SIZE)
+                    if not chunk:
+                        break
+                    spool.write(chunk)
+                spool.seek(0)
+                findings.extend(
+                    scan_nested_fileobj(
+                        spool,
+                        rel,
+                        rules,
+                        allow,
+                        budget=budget,
+                        strict=True,
+                    )
+                )

    return findings

@@ -200,8 +200,14 @@ def scan_bytes(
    allowlist: Sequence[str],
    *,
    nest_depth: int = 0,
+    strict: bool = False,
 ) -> List[str]:
-    """Always raw-byte scan (including NUL binaries) and recurse into archives."""
+    """Always raw-byte scan (including NUL binaries) and recurse into archives.
+
+    Default strict=False: opportunistic nested detection for ordinary source/stdin
+    scans — malformed gzip/zip/tar magic keeps raw findings and does not fail the
+    whole scan. Explicit archive/image callers pass strict=True.
+    """
    # Import here to avoid circular import at module load for archive helpers.
    from nested_content import scan_nested_bytes  # noqa: WPS433

@@ -211,6 +217,7 @@ def scan_bytes(
        rules,
        allowlist,
        nest_depth=nest_depth,
+        strict=strict,
    )


@@ -203,24 +203,7 @@ printf '%s\n' "${out}" | grep -Eq 'Sirius-1\.1\.0/sirius-engine' \
  && fail "finding must not keep GitHub top directory prefix"
 pass "release-archive top-dir canary"

-echo "==> canary: zip bomb / oversize streaming rejection"
-zipbomb="${TMP_DIR}/zipbomb.zip"
-python3 - <<PY
-import zipfile, zlib
-from pathlib import Path
-# Craft a zip that declares a small size but would expand larger than MAX if trusted.
-# Our reader streams with MAX_MEMBER_BYTES+1 and also checks declared vs actual.
-payload = b"A" * (33 * 1024 * 1024)
-path = Path("${zipbomb}")
-with zipfile.ZipFile(path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
-    # Normal oversized member: declared size is honest and > MAX
-    zf.writestr("big.bin", payload)
-PY
-if python3 "${CI_DIR}/scan_archive.py" --archive "${zipbomb}" --allowlist "${ALLOWLIST}" 2>/dev/null; then
-  fail "expected oversize zip member to fail"
-fi
-# Declared/actual mismatch canary via manual zip header mutation is fragile;
-# exercise nested_content reader with a crafted ZipInfo mismatch using a temp helper.
+echo "==> canary: zip declared/actual size mismatch still fail-closed"
 python3 - <<PY
 import io, sys, zipfile
 sys.path.insert(0, "scripts/community-independence")
@@ -240,7 +223,7 @@ with zipfile.ZipFile(io.BytesIO(raw)) as zf:
    else:
        raise SystemExit("expected mismatch rejection")
 PY
-pass "zip oversize/mismatch canaries"
+pass "zip mismatch canary"

 echo "==> canary: NUL binary + nested gzip/tar/zip markers"
 python3 - <<PY
@@ -276,6 +259,114 @@ for rel in nul.bin nested.gz nested.tar nested.zip; do
 done
 pass "NUL/nested archive canaries"

+echo "==> canary: streaming large members / gzip tar / malformed magic / budgets"
+python3 - <<PY
+import gzip, io, sys, tarfile, tempfile
+from pathlib import Path
+
+sys.path.insert(0, "scripts/community-independence")
+import scan_text
+import scan_archive
+import scan_image_layers
+from nested_content import (
+    CHUNK_SIZE,
+    ArchiveSafetyError,
+    _Budget,
+    scan_nested_bytes,
+    scan_stream_chunks,
+)
+
+allow = Path("scripts/community-independence/policy/governance-allowlist.txt")
+allowlist = scan_text.load_allowlist(allow)
+rules = scan_text.compile_rules()
+marker = ("ghcr.io/" + "opensecurity" + "-infosec/").encode() + b"stream"
+tmp = Path("${TMP_DIR}") / "stream"
+tmp.mkdir(parents=True, exist_ok=True)
+
+# 1) Clean >32MiB tar member passes via production scan_archive path.
+big = b"C" * (33 * 1024 * 1024)
+tpath = tmp / "clean-large.tar"
+with tarfile.open(tpath, "w") as tf:
+    info = tarfile.TarInfo("blob.bin")
+    info.size = len(big)
+    tf.addfile(info, io.BytesIO(big))
+findings = scan_archive.scan_archive(tpath, allow)
+assert findings == [], findings
+print("OK clean >32MiB tar member")
+
+# 2) Marker spanning chunk boundary fails (production scan_stream_chunks).
+# Place marker so it crosses the first CHUNK_SIZE boundary.
+left = CHUNK_SIZE - (len(marker) // 2)
+payload = (b"A" * left) + marker + (b"B" * 4096)
+budget = _Budget()
+findings = scan_stream_chunks(io.BytesIO(payload), "span.bin", rules, allowlist, budget)
+assert any("private_registry" in f for f in findings), findings
+print("OK chunk-boundary marker")
+
+# 3) Gzip-compressed tar with >32MiB clean member passes via scan_image_layers.
+inner_tar = io.BytesIO()
+with tarfile.open(fileobj=inner_tar, mode="w") as tf:
+    info = tarfile.TarInfo("opt/big.bin")
+    info.size = len(big)
+    tf.addfile(info, io.BytesIO(big))
+gz_layer = gzip.compress(inner_tar.getvalue())
+save_path = tmp / "gzip-large.docker-save.tar"
+with tarfile.open(save_path, "w") as tf:
+    for name, data in [
+        ("manifest.json", b"[]"),
+        ("config.json", b"{}"),
+        ("layer.tar.gz", gz_layer),
+    ]:
+        info = tarfile.TarInfo(name)
+        info.size = len(data)
+        tf.addfile(info, io.BytesIO(data))
+findings = scan_image_layers.scan_docker_save(save_path, allow, "sirius-ui/linux-amd64")
+assert findings == [], findings
+print("OK gzip tar >32MiB via scan_image_layers")
+
+# 4) Malformed gzip magic in source (strict=False): raw-only, no traceback.
+bad = b"\x1f\x8b" + b"this-is-not-valid-gzip-payload"
+findings = scan_nested_bytes(bad, "weird.bin", rules, allowlist, strict=False)
+assert findings == [], findings
+# With a marker in the malformed payload, still report via raw scan.
+bad_mark = b"\x1f\x8b" + b"xx" + marker + b"yy"
+findings = scan_nested_bytes(bad_mark, "weird2.bin", rules, allowlist, strict=False)
+assert findings, "expected raw marker findings without nest parse"
+print("OK malformed magic source fallback")
+
+# 5) Malformed explicit image/archive fails closed (strict).
+try:
+    scan_nested_bytes(bad, "layer.tar.gz", rules, allowlist, strict=True)
+except ArchiveSafetyError:
+    print("OK malformed explicit nest fail-closed")
+else:
+    raise SystemExit("expected strict malformed gzip to fail")
+
+corrupt_save = tmp / "corrupt.docker-save.tar"
+with tarfile.open(corrupt_save, "w") as tf:
+    data = b"\x1f\x8b" + b"broken"
+    info = tarfile.TarInfo("layer.tar.gz")
+    info.size = len(data)
+    tf.addfile(info, io.BytesIO(data))
+try:
+    scan_image_layers.scan_docker_save(corrupt_save, allow, "sirius-api/linux-amd64")
+except ArchiveSafetyError:
+    print("OK malformed image layer fail-closed")
+else:
+    raise SystemExit("expected corrupt image layer to fail")
+
+# 6) Budget exceed fails cleanly.
+tiny = _Budget(limit=1024)
+try:
+    scan_nested_bytes(b"Z" * 4096, "huge.bin", rules, allowlist, budget=tiny, strict=True)
+except ArchiveSafetyError as exc:
+    assert "budget" in str(exc).lower(), exc
+    print("OK budget exceed")
+else:
+    raise SystemExit("expected budget exceed")
+PY
+pass "streaming/malformed/budget canaries"
+
 echo "==> canary: SBOM wrong-component and duplicate digest"
 sbom_dir="${TMP_DIR}/sboms"
 mkdir -p "${sbom_dir}"
@@ -325,14 +416,14 @@ fi
 pass "SBOM wrong-component + duplicate digest canaries"

 echo "==> canary: docker-save fixtures via scan_image_layers.py"
-python3 - <<'PY'
+python3 - <<PY
 import gzip, io, json, tarfile, sys
 from pathlib import Path
 sys.path.insert(0, "scripts/community-independence")
 import scan_image_layers
 from nested_content import ArchiveSafetyError

-tmp = Path("""${TMP_DIR}""") / "docker-save"
+tmp = Path("${TMP_DIR}") / "docker-save"
 tmp.mkdir(parents=True, exist_ok=True)
 allow = Path("scripts/community-independence/policy/governance-allowlist.txt")
 marker = ("ghcr.io/" + "opensecurity" + "-infosec/").encode() + b"img"