diff --git a/docs/module-internals.md b/docs/module-internals.md index 3f8ecfe..108c869 100644 --- a/docs/module-internals.md +++ b/docs/module-internals.md @@ -391,6 +391,12 @@ metadata extraction from verdict logic: metadata record into the same evidence type without file access. Diagnostic values under `error` and `kind` are excluded from evidence while nested raw bytes remain available through encoded binary fields. +- The vendor registries are matched over `_metadata_region(head)`, not the whole scan + buffer: they see the container's metadata and not its coded pixels. The tokens are + raw substrings and the shortest are four and five bytes, so over a megabyte of + compressed data one turns up by chance, and the entry it hits may assert AI. The + trim happens only when the container parses -- a malformed or unknown one is left + whole, because dropping real evidence to avoid a chance match is the wrong trade. - `identify_from_evidence` evaluates that evidence without reopening the source. Rules that decide a verdict live here, not in extraction: extraction has two implementations, and a rule in only one of them is a rule the other lacks. The diff --git a/scripts/record_parity_audit.py b/scripts/record_parity_audit.py index 554401f..acb0226 100644 --- a/scripts/record_parity_audit.py +++ b/scripts/record_parity_audit.py @@ -162,18 +162,26 @@ def _summarize(rows: list[dict[str, Any]], baseline: Path | None) -> None: was = previous.get(Path(row["path"]).name) if was is None: continue - before = was.get("file") or {"confidence": was.get("confidence"), "signals": was.get("signals")} - if before.get("confidence") != row["file"]["confidence"] or before.get("signals") != row["file"]["signals"]: + # The WHOLE verdict, not a chosen subset. A first version compared confidence + # and signals only and reported "0 changed" for a run whose single intended + # correction was a watermark line -- the change it existed to show. + before = was.get("file") or {} + if before and before != row["file"]: changed.append((row["path"], before, row["file"])) print(f"\nverdicts changed against the baseline: {len(changed)}") - gained: collections.Counter[str] = collections.Counter() + moved: collections.Counter[str] = collections.Counter() for _, before, after in changed: for name in set(after["signals"]) - set(before.get("signals") or []): - gained[f"gained {name}"] += 1 + moved[f"gained signal {name}"] += 1 for name in set(before.get("signals") or []) - set(after["signals"]): - gained[f"LOST {name}"] += 1 - for label, count in gained.most_common(): + moved[f"LOST signal {name}"] += 1 + for field in (*COMPARED, "watermarks"): + if before.get(field) != after.get(field): + moved[f"{field} changed"] += 1 + for label, count in moved.most_common(): print(f" {label}: {count}") + for path, before, after in changed[:10]: + print(f" {Path(path).name}\n before: {before}\n after: {after}") def main() -> int: diff --git a/src/remove_ai_watermarks/identify.py b/src/remove_ai_watermarks/identify.py index b6054bc..3f84a38 100644 --- a/src/remove_ai_watermarks/identify.py +++ b/src/remove_ai_watermarks/identify.py @@ -22,6 +22,7 @@ from __future__ import annotations import base64 import itertools import logging +import struct from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, cast @@ -479,6 +480,72 @@ _DEVICE_C2PA_PLATFORM: tuple[tuple[bytes, str], ...] = ( ) +def _metadata_region(head: bytes) -> bytes: + """The part of the scan buffer that can hold metadata, with the coded pixels cut out. + + The vendor registries are matched as raw substrings, and the shortest tokens are + four and five bytes (``Bria``, ``Adobe``, ``Canva``). Over a megabyte of compressed + pixel data a four-byte sequence appears by chance about once in three thousand + images -- measured: ``Bria`` matched inside the entropy-coded scan of 4 of 14,707 + corpus JPEGs, in none of which the manifest names Bria. That is not a cosmetic + mislabel, because the Bria entry carries ``asserts_ai``: a chance match can declare + an image AI-generated. + + ``c2pa_marker_in`` already refuses a bare ``c2pa`` substring for the same reason. + This is the same defence for the registries: they see the container's metadata and + not its pixels. + + JPEG keeps the marker segments before the entropy-coded scan, PNG every chunk but + ``IDAT``, and both keep the trailer past the end marker. Anything ``scan_head`` + APPENDED past the window is metadata by construction (late chunks, boxes, decoder + text), so it is always kept and never walked -- walking it is what produced 11 MB + records and a phantom AIGC signal in the record collector. + + Trimming happens only when the container actually parses: a JPEG whose marker walk + reaches the coded scan, a PNG whose chunk walk reaches ``IDAT``. Anything else -- + a malformed container, a synthetic blob, a format with no walker here -- is + returned whole. Cutting a buffer this function did not understand would drop real + evidence to avoid a chance match, which is the wrong way round. + """ + raw, appended = head[:_SCAN_BYTES], head[_SCAN_BYTES:] + if raw[:2] == b"\xff\xd8": + index, size = 2, len(raw) + while index + 1 < size: + if raw[index] != 0xFF: + return head # not a marker boundary: the walk is lost, keep everything + marker = raw[index + 1] + if marker in (0xDA, 0xD9): # SOS / EOI: the coded scan follows + end = raw.rfind(b"\xff\xd9") + return raw[:index] + (raw[end + 2 :] if end >= index else b"") + appended + if 0xD0 <= marker <= 0xD7 or marker == 0x01: + index += 2 + continue + if index + 4 > size: + break + length = int.from_bytes(raw[index + 2 : index + 4], "big") + if length < 2 or index + 2 + length > size: + break + index += 2 + length + return head # ran out of buffer before the scan: nothing was skipped anyway + if raw[:8] == b"\x89PNG\r\n\x1a\n": + out = bytearray() + position, size, saw_idat = 8, len(raw), False + while position + 8 <= size: + (length,) = struct.unpack(">I", raw[position : position + 4]) + chunk_type = raw[position + 4 : position + 8] + start = position + 8 + if chunk_type == b"IDAT": + saw_idat = True + else: + out += chunk_type + raw[start : start + min(length, size - start)] + position = start + length + 4 + if chunk_type == b"IEND": + out += raw[position:] + break + return bytes(out) + appended if saw_idat else head + return head + + def _first_token_match(head: bytes, table: tuple[tuple[bytes, str], ...]) -> str | None: """First platform in ``table`` whose token appears in ``head``, else None. @@ -915,12 +982,16 @@ def _identify_from_evidence( # score, the latter can be a by-product of our own SDXL removal pass, so # neither is a trustworthy "the generator stamped its identity" claim. ai_vendor_claims: dict[str, str] = {} - camera_label = _device_platform(head) - signer_label = _signer_platform(head) + # The vendor registries match short raw substrings, so they read the container's + # metadata rather than its pixels -- see `_metadata_region`. Every other check + # below keeps the full buffer: their markers are long and distinctive. + region = _metadata_region(head) + camera_label = _device_platform(region) + signer_label = _signer_platform(region) # ── C2PA Content Credentials ──────────────────────────────────── has_c2pa = bool(info) or c2pa_marker_in(head) - issuers = [info["issuer"]] if info.get("issuer") else _issuers_in(head) + issuers = [info["issuer"]] if info.get("issuer") else _issuers_in(region) # Full AI generation (trainedAlgorithmicMedia) vs an AI-enhanced real photo # (compositeWithTrainedAlgorithmicMedia). The structured kind is parsed once in # _internal.c2pa._populate_registry_fields (covers PNG + any container the c2pa-python @@ -947,7 +1018,7 @@ def _identify_from_evidence( generator = ( info.get("claim_generator") or cbor_text_after(head, b"claim_generator") - or (", ".join(tools) if (tools := _ai_tools_in(head)) else None) + or (", ".join(tools) if (tools := _ai_tools_in(region)) else None) ) # Platform: a distinctive device/camera token in the manifest wins (it is the # signer/producer), then an editing-app/AI-device signer (Samsung Galaxy, @@ -998,7 +1069,7 @@ def _identify_from_evidence( # reusing the derived `has_c2pa` / `c2pa_source_kind` above, which are broader: # the file path's answer must not move. trained_source = b"trainedAlgorithmicMedia" in head or b"TrainedAlgorithmicMedia" in head - if not synthid and trained_source and c2pa_marker_in(head) and (vendors := synthid_vendors_in(head)): + if not synthid and trained_source and c2pa_marker_in(head) and (vendors := synthid_vendors_in(region)): synthid = synthid_verdict(", ".join(vendors)) if synthid: watermarks.append(f"SynthID watermark, inferred from C2PA metadata ({synthid})") @@ -1011,7 +1082,7 @@ def _identify_from_evidence( # ── C2PA soft-binding: a named forensic/third-party watermark vendor ─ # (Adobe TrustMark, Digimarc, Imatag, ...). Present in the manifest even when # the watermark itself can't be decoded; names whose watermark stamped the pixels. - soft_binding = meta.get("soft_binding") or (", ".join(v) if (v := soft_binding_vendors_in(head)) else None) + soft_binding = meta.get("soft_binding") or (", ".join(v) if (v := soft_binding_vendors_in(region)) else None) if soft_binding: signals.append(Signal("soft_binding", f"C2PA soft binding: {soft_binding}", "high")) watermarks.append(f"Forensic watermark soft binding ({soft_binding})") diff --git a/tests/test_identify.py b/tests/test_identify.py index 26a7933..a277b29 100644 --- a/tests/test_identify.py +++ b/tests/test_identify.py @@ -1406,3 +1406,53 @@ class TestSynthIdProxyIsDecidedInTheVerdict: report = identify(path, check_visible=False, check_invisible=False) assert not any("SynthID" in mark for mark in report.watermarks) + + +class TestRegistryScansSkipTheCodedPixels: + """The vendor registries match short raw substrings -- the shortest are four and + five bytes. Over a megabyte of compressed pixel data such a sequence turns up by + chance: `Bria` matched inside the entropy-coded scan of 4 of 14,707 corpus JPEGs, + and that entry asserts AI, so a chance match can declare an image AI-generated. + `c2pa_marker_in` already refuses a bare `c2pa` substring for the same reason.""" + + def _jpeg(self, path: Path, *, in_segment: bytes = b"", in_scan: bytes = b"") -> Path: + import numpy as np + from PIL import Image + + Image.fromarray(np.zeros((32, 32, 3), dtype=np.uint8)).save(path, "JPEG") + data = path.read_bytes() + if in_segment: + payload = b"jumb c2pa trainedAlgorithmicMedia " + in_segment + data = data[:2] + b"\xff\xeb" + (len(payload) + 2).to_bytes(2, "big") + payload + data[2:] + if in_scan: + # After SOS, i.e. inside the entropy-coded scan the walk skips. + sos = data.index(b"\xff\xda") + data = data[: sos + 16] + in_scan + data[sos + 16 :] + path.write_bytes(data) + return path + + def test_a_token_in_a_marker_segment_is_attributed(self, tmp_path: Path): + from remove_ai_watermarks.identify import _issuers_in, _metadata_region + from remove_ai_watermarks.metadata import scan_head + + path = self._jpeg(tmp_path / "signed.jpg", in_segment=b"Bria") + + assert _issuers_in(_metadata_region(scan_head(path))) == ["Bria Artificial Intelligence"] + + def test_a_token_in_the_coded_scan_is_not(self, tmp_path: Path): + from remove_ai_watermarks.identify import _issuers_in, _metadata_region + from remove_ai_watermarks.metadata import scan_head + + path = self._jpeg(tmp_path / "chance.jpg", in_segment=b"OpenAI", in_scan=b"Bria") + region = _metadata_region(scan_head(path)) + + assert _issuers_in(region) == ["OpenAI"] + + def test_a_container_that_does_not_parse_is_left_whole(self, tmp_path: Path): + """Cutting a buffer the walk did not understand would drop real evidence to + avoid a chance match, which is the wrong way round.""" + from remove_ai_watermarks.identify import _metadata_region + + blob = b"\xff\xd8" + b"not really a jpeg, no valid marker chain here" * 4 + + assert _metadata_region(blob) == blob