fix: keep the truncated tail instead of dropping it in png/isobmff strips (#182)

* fix: keep the truncated tail instead of dropping it in png/isobmff strips

strip_png stopped walking at the first chunk it could not parse and
never copied the remainder; strip_isobmff rebuilt only the boxes that
parsed. A truncated IDAT/mdat — the actual coded image — was dropped
from the output, the run reported "already clean", the exit code was
0: a user with a recoverable (viewable) image ended with an
unopenable husk and the tool saying it was fine (#170).

Both strippers now keep the unparseable tail verbatim and append a
"kept N bytes of truncated ... tail" action — a real report, so the
run is never mistaken for a no-op clean and every original byte
survives (truncated-image-capable readers can still open the output).

* refactor: surface parser walk end; stop false truncation reports

- _parse_isobmff_boxes returns (boxes, scanned_end), so strip_isobmff no longer re-walks parsed boxes to recover the stop offset.

- Fewer than 8 trailing bytes is trailing junk, not truncation: kept verbatim without a 'file truncated' action.

- Tests: drop unused noqa (RUF100) and unpack (RUF059), simplify the PNG tail assertion to byte-equality, cover the trailing-junk case.

---------

Co-authored-by: yzxcj797 <yzxcj797@users.noreply.github.com>
Co-authored-by: Guillaume Meyer (The Opinionated Man) <1385518+guillaumemeyer@users.noreply.github.com>

service/scripts/av_meta.py

@@ -96,10 +96,10 @@ def _inspect_moov_udta(data: bytes) -> tuple[bool, bool, list[str]]:
     has_c2pa = False
     has_ai = False
     findings: list[str] = []
-    for fourcc, payload, _size, _hdr in _parse_isobmff_boxes(data):
+    for fourcc, payload, _size, _hdr in _parse_isobmff_boxes(data)[0]:
         if fourcc != b"moov":
             continue
-        for s_fourcc, s_payload, _s_size, _s_hdr in _parse_isobmff_boxes(payload):
+        for s_fourcc, s_payload, _s_size, _s_hdr in _parse_isobmff_boxes(payload)[0]:
             if s_fourcc != b"udta":
                 continue
             hits = _contains_any(s_payload, AI_META_HINTS)
@@ -114,12 +114,12 @@ def _inspect_moov_udta(data: bytes) -> tuple[bool, bool, list[str]]:
 def _strip_moov_udta(data: bytes, *, strip_all_metadata: bool) -> tuple[bytes, list[str]]:
     actions: list[str] = []
     out = bytearray()
-    for fourcc, payload, _size, hdr in _parse_isobmff_boxes(data):
+    for fourcc, payload, _size, hdr in _parse_isobmff_boxes(data)[0]:
         if fourcc != b"moov":
             out.extend(_build_isobmff_box(fourcc, payload, hdr))
             continue
         new_moov = bytearray()
-        for s_fourcc, s_payload, s_size, s_hdr in _parse_isobmff_boxes(payload):
+        for s_fourcc, s_payload, s_size, s_hdr in _parse_isobmff_boxes(payload)[0]:
             if s_fourcc == b"udta" and (
                 strip_all_metadata or _contains_any(s_payload, AI_META_HINTS)
             ):

service/scripts/image_meta.py

@@ -476,10 +476,14 @@ XMP_UUID = b"\xbe\x7a\xcf\xcb\x97\xa9\x42\xe8\x9c\x71\x99\x94\x91\xe3\xaf\xac"
 
 def _parse_isobmff_boxes(
     data: bytes, start: int = 0, end: int | None = None
-) -> list[tuple[bytes, bytes, int, int]]:
+) -> tuple[list[tuple[bytes, bytes, int, int]], int]:
     """Parse top-level or container ISOBMFF boxes.
 
-    Returns list of (fourcc, payload, total_box_size, header_size).
+    Returns (boxes, scanned_end): the list of (fourcc, payload,
+    total_box_size, header_size) tuples and the offset where the walk
+    stopped -- the start of the first box that could not be parsed (an
+    overrunning box, or a run of fewer than 8 trailing bytes), or `end`
+    when the whole buffer parsed.
     """
     if end is None:
         end = len(data)
@@ -502,7 +506,7 @@ def _parse_isobmff_boxes(
         payload = data[pos + header_size : pos + size]
         boxes.append((fourcc, payload, size, header_size))
         pos += size
-    return boxes
+    return boxes, pos
 
 
 def _build_isobmff_box(fourcc: bytes, payload: bytes, header_size: int = 8) -> bytes:
@@ -523,7 +527,7 @@ def inspect_isobmff(data: bytes, fmt: str = "avif") -> tuple[bool, bool, list[st
     has_c2pa = False
     has_ai = False
 
-    boxes = _parse_isobmff_boxes(data)
+    boxes, _ = _parse_isobmff_boxes(data)
     if not boxes:
         # Box parsing failed (e.g. the first box's size overruns a truncated
         # download) — that is exactly when the whole-file byte scan below is
@@ -563,7 +567,7 @@ def inspect_isobmff(data: bytes, fmt: str = "avif") -> tuple[bool, bool, list[st
                     if any(h.lower() in ("c2pa", "contentcredentials", "jumb") for h in hits):
                         has_c2pa = True
         elif fourcc == b"meta":
-            meta_sub = _parse_isobmff_boxes(payload, start=4)
+            meta_sub, _ = _parse_isobmff_boxes(payload, start=4)
             for s_fourcc, s_payload, _, _ in meta_sub:
                 s_name = s_fourcc.decode("latin-1", errors="replace")
                 if s_fourcc in (b"jumb", b"c2pa") or s_name.lower().startswith("c2"):
@@ -1752,6 +1756,15 @@ def strip_png(data: bytes, *, strip_all_text: bool = True) -> tuple[bytes, list[
         chunk_start = pos + 8
         chunk_end = chunk_start + length
         if chunk_end + 4 > len(data):
+            # A truncated chunk (interrupted download): copy the remainder
+            # verbatim instead of dropping it — dropping turned a recoverable
+            # image into an unopenable husk while reporting "already clean"
+            # (#170). Note it as an action so the run is never mistaken for
+            # an ordinary no-op clean.
+            out.extend(data[pos:])
+            actions.append(
+                f"kept {len(data) - pos} bytes of truncated chunk {ctype.decode('latin-1', errors='replace')} tail (file truncated)"
+            )
             break
         payload = data[chunk_start:chunk_end]
         crc_bytes = data[chunk_end : chunk_end + 4]
@@ -1914,10 +1927,15 @@ def strip_webp(data: bytes, *, strip_all_metadata: bool = True) -> tuple[bytes,
 def strip_isobmff(
     data: bytes, fmt: str = "avif", *, strip_all_metadata: bool = True
 ) -> tuple[bytes, list[str]]:
-    boxes = _parse_isobmff_boxes(data)
+    boxes, scanned_end = _parse_isobmff_boxes(data)
     if not boxes:
         raise ValueError(f"not a valid {fmt.upper()} (no ISOBMFF boxes)")
 
+    # scanned_end is where the box walk stopped: the parser halts at the first
+    # box whose declared size overruns the data (a truncated download). The
+    # rebuild below used to emit only the boxes that parsed — dropping a
+    # truncated mdat, the actual coded image, while reporting "already clean"
+    # (#170); the tail is appended verbatim afterwards instead.
     actions: list[str] = []
     out = bytearray()
 
@@ -1940,7 +1958,7 @@ def strip_isobmff(
 
         if fourcc == b"meta":
             meta_verflags = payload[:4] if len(payload) >= 4 else b"\x00\x00\x00\x00"
-            sub_boxes = _parse_isobmff_boxes(payload, start=4)
+            sub_boxes, _ = _parse_isobmff_boxes(payload, start=4)
             clean_sub = bytearray()
             for s_fourcc, s_payload, s_size, s_hdr in sub_boxes:
                 s_name = s_fourcc.decode("latin-1", errors="replace")
@@ -1971,6 +1989,15 @@ def strip_isobmff(
 
         out.extend(_build_isobmff_box(fourcc, payload, header_size))
 
+    if scanned_end < len(data):
+        tail = data[scanned_end:]
+        out.extend(tail)
+        # An 8-byte header whose box overruns the data is a truncated download
+        # worth reporting; fewer than 8 leftover bytes is merely trailing junk
+        # and is kept verbatim without claiming the file was truncated (#170).
+        if len(tail) >= 8:
+            actions.append(f"kept {len(tail)} bytes of truncated tail (file truncated)")
+
     if not actions:
         actions.append(f"no {fmt.upper()} metadata boxes removed (already clean or none matched)")
 

tests/test_truncated_tail_preserved.py

@@ -0,0 +1,87 @@
+"""Truncated-tail preservation in strip_png / strip_isobmff (#170).
+
+An unparseable chunk/box used to end the walk with the remainder dropped from
+the output — a recoverable image became an unopenable husk while the run
+reported "already clean". The tail must be kept and the run must say so.
+"""
+
+from __future__ import annotations
+
+import struct
+import sys
+import zlib
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "service" / "scripts"
+sys.path.insert(0, str(SCRIPTS))
+
+import image_meta
+
+
+def _chunk(ctype: bytes, payload: bytes) -> bytes:
+    return (
+        struct.pack(">I", len(payload))
+        + ctype
+        + payload
+        + struct.pack(">I", zlib.crc32(ctype + payload) & 0xFFFFFFFF)
+    )
+
+
+def _truncated_png() -> bytes:
+    ihdr = struct.pack(">IIBBBBB", 1, 1, 8, 0, 0, 0, 0)
+    idat_payload = b"\x78\x9c\x00" + b"\x00" * 40
+    head = b"\x89PNG\r\n\x1a\n" + _chunk(b"IHDR", ihdr)
+    # An IDAT whose declared length overruns the file (interrupted download):
+    # 8-byte header + declared payload + 4-byte CRC > remaining bytes.
+    truncated_idat = struct.pack(">I", len(idat_payload)) + b"IDAT" + idat_payload[:8]
+    return head + truncated_idat  # no CRC, no IEND
+
+
+def test_strip_png_keeps_truncated_tail():
+    data = _truncated_png()
+    out, actions = image_meta.strip_png(data)
+    # No chunk in this fixture is droppable, so every input byte must survive
+    # byte-for-byte: nothing is silently dropped.
+    assert out == data, (len(out), len(data))
+    # And the run says the tail was kept — never "already clean".
+    assert any("truncated" in a.lower() for a in actions), actions
+
+
+def test_strip_png_intact_unchanged():
+    ihdr = struct.pack(">IIBBBBB", 1, 1, 8, 0, 0, 0, 0)
+    data = b"\x89PNG\r\n\x1a\n" + _chunk(b"IHDR", ihdr) + _chunk(b"IEND", b"")
+    out, actions = image_meta.strip_png(data)
+    assert out == data
+    assert not any("truncated" in a.lower() for a in actions)
+
+
+def test_strip_isobmff_keeps_truncated_tail():
+    # ftyp + a truncated mdat (declared size overruns the file).
+    ftyp = struct.pack(">I", 16) + b"ftypavif" + b"\x00\x00\x00\x00"
+    mdat_declared = 200
+    mdat = struct.pack(">I", mdat_declared) + b"mdat" + b"\x00" * 40
+    data = ftyp + mdat
+
+    out, actions = image_meta.strip_isobmff(data, fmt="avif")
+    # The truncated mdat tail is preserved, not dropped.
+    assert len(out) >= len(data), (len(out), len(data))
+    assert any("truncated" in a.lower() for a in actions), actions
+
+
+def test_strip_isobmff_intact_unchanged():
+    ftyp = struct.pack(">I", 16) + b"ftypavif" + b"\x00\x00\x00\x00"
+    full = struct.pack(">I", 12) + b"mdat" + b"\x00" * 4
+    data = ftyp + full
+    _, actions = image_meta.strip_isobmff(data, fmt="avif")
+    assert not any("truncated" in a.lower() for a in actions)
+
+
+def test_strip_isobmff_trailing_junk_kept_without_truncation():
+    # A few trailing bytes after the last box are kept verbatim but are not
+    # reported as truncation -- the walk simply ran out of box headers.
+    ftyp = struct.pack(">I", 16) + b"ftypavif" + b"\x00\x00\x00\x00"
+    data = ftyp + b"\x01\x02\x03"
+    out, actions = image_meta.strip_isobmff(data, fmt="avif")
+    assert out == data
+    assert not any("truncated" in a.lower() for a in actions)