From 1124c591be590f4ee2de68a7d6bd1ab14786eede Mon Sep 17 00:00:00 2001 From: Victor Kuznetsov Date: Wed, 5 Aug 2026 17:32:44 -0700 Subject: [PATCH] Match the vendor registries against metadata, not coded pixels The registries are raw substrings and the shortest tokens are four and five bytes (`Bria`, `Adobe`, `Canva`). 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, in none of which the manifest names Bria. The rate is what a four-byte pattern predicts on that corpus, and the Bria entry asserts AI, so a chance match can declare an image AI-generated rather than merely mislabel its signer. `_metadata_region` gives the registry scans the container's metadata: JPEG marker segments before the coded scan, PNG chunks other than IDAT, both trailers, and whatever `scan_head` appended past the window. Every other check keeps the full buffer -- their markers are long and distinctive. A container that does not parse is returned whole, since dropping real evidence to avoid a chance match is the wrong trade. `c2pa_marker_in` already refuses a bare `c2pa` substring for this reason; this is the same defence for the registries. Verified the way the rules require for a change that MOVES a verdict: over all 48,905 corpus images, exactly one file changed, the one named in advance, from "C2PA Content Credentials (Bria Artificial Intelligence)" to "(unknown signer)". Record-path parity is 0 disagreements, down from 75 when this work started. The audit's own baseline comparison is fixed here too. It compared confidence and signals only, and so reported "0 changed" for the run whose single intended correction was a watermark line -- the change it exists to show. Co-Authored-By: Claude Opus 5 --- docs/module-internals.md | 6 ++ scripts/record_parity_audit.py | 20 +++++-- src/remove_ai_watermarks/identify.py | 83 ++++++++++++++++++++++++++-- tests/test_identify.py | 50 +++++++++++++++++ 4 files changed, 147 insertions(+), 12 deletions(-) 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