Add external metadata evidence ingestion

src/remove_ai_watermarks/__init__.py

@@ -25,7 +25,7 @@ _os.environ.setdefault("TRANSFORMERS_VERBOSITY", "error")
 _warnings.filterwarnings("ignore", message=r".*ImageProcessorFast.*")
 
 
-__version__ = "0.21.0"
+__version__ = "0.21.1"
 
 __all__ = ["__version__", "remove_visible", "visible_provenance"]
 

src/remove_ai_watermarks/identify.py

@@ -19,10 +19,12 @@ never as "clean". See CLAUDE.md "SynthID detection is metadata-only".
 
 from __future__ import annotations
 
+import base64
+import contextlib
 import itertools
 import logging
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, cast
 
 from remove_ai_watermarks.metadata import (
     AI_METADATA_KEYS,
@@ -30,17 +32,27 @@ from remove_ai_watermarks.metadata import (
     IPTC_AI_FIELD_MARKERS,
     IPTC_AI_MARKERS,
     aigc_label,
+    aigc_label_from_metadata,
     c2pa_cloud_manifest_in,
     c2pa_marker_in,
     exif_generator,
+    generator_from_metadata,
     get_ai_metadata,
     huggingface_job,
     iptc_ai_system,
+    iptc_ai_system_in,
     samsung_genai,
+    samsung_genai_in,
     scan_head,
     xai_signature,
+    xai_signature_pair,
+)
+from remove_ai_watermarks.noai.c2pa import (
+    c2pa_info_from_manifest_store,
+    cbor_text_after,
+    extract_c2pa_info,
+    soft_binding_vendors_in,
 )
-from remove_ai_watermarks.noai.c2pa import cbor_text_after, extract_c2pa_info, soft_binding_vendors_in
 from remove_ai_watermarks.noai.constants import (
     C2PA_AI_TOOLS,
     C2PA_AI_VENDORS,
@@ -51,7 +63,6 @@ from remove_ai_watermarks.watermark_registry import GEMINI_SPARKLE_TRUST_CONF
 
 if TYPE_CHECKING:
     from pathlib import Path
-    from typing import Any
 
     from numpy.typing import NDArray
 
@@ -154,6 +165,182 @@ class ProvenanceEvidence:
     samsung_genai: int | None
 
 
+def _external_metadata(value: Any) -> tuple[list[tuple[str, Any]], bytes]:
+    """Index nested metadata and recover common encoded binary values in one pass."""
+    pairs: list[tuple[str, Any]] = []
+    parts: list[bytes] = []
+
+    def visit(item: Any) -> None:
+        if isinstance(item, dict):
+            mapping = cast("dict[object, Any]", item)
+            for key, nested in mapping.items():
+                key_text = str(key)
+                pairs.append((key_text, nested))
+                parts.append(key_text.encode("utf-8", "replace"))
+                if isinstance(nested, str) and (key_text == "base64" or key_text.endswith("_base64")):
+                    encoded = nested.split("...TRUNCATED", 1)[0]
+                    with contextlib.suppress(ValueError, TypeError):
+                        parts.append(base64.b64decode(encoded, validate=True))
+                visit(nested)
+        elif isinstance(item, (list, tuple)):
+            sequence = cast("list[Any] | tuple[Any, ...]", item)
+            for nested in sequence:
+                visit(nested)
+        elif isinstance(item, bytes):
+            parts.append(item)
+        elif isinstance(item, str):
+            parts.append(item.encode("utf-8", "replace"))
+            if item.startswith("hex:"):
+                with contextlib.suppress(ValueError):
+                    parts.append(bytes.fromhex(item[4:]))
+        elif item is not None:
+            parts.append(str(item).encode("utf-8", "replace"))
+
+    visit(value)
+    return pairs, b"\n".join(parts)
+
+
+def _external_text(value: Any) -> str:
+    if isinstance(value, bytes):
+        return value.decode("latin-1", "replace").strip()
+    if not isinstance(value, str):
+        return str(value).strip()
+    if value.startswith("hex:"):
+        try:
+            return bytes.fromhex(value[4:]).decode("latin-1", "replace").strip()
+        except ValueError:
+            pass
+    return value.strip()
+
+
+def _external_exif_generator(pairs: list[tuple[str, Any]], scan: bytes) -> str | None:
+    candidate_keys = {
+        "software",
+        "make",
+        "artist",
+        "imagedescription",
+        "source",
+        "title",
+        "description",
+        "creatortool",
+    }
+    candidates = [
+        str(value)
+        for key, value in pairs
+        if key.lower().removeprefix("info:") in candidate_keys and isinstance(value, (str, bytes))
+    ]
+    return generator_from_metadata(candidates, scan)
+
+
+def evidence_from_metadata_record(
+    record: dict[str, Any], *, path: Path, c2pa_manifest_store: str | dict[str, Any] | None = None
+) -> ProvenanceEvidence:
+    """Normalize an externally collected metadata record into provenance evidence.
+
+    The record may contain arbitrary nested dictionaries and lists. Text, bytes,
+    hexadecimal values prefixed with ``hex:``, and fields named ``base64`` or
+    ending in ``_base64`` are included in the shared byte scan. No source file is
+    opened.
+    """
+    pairs, scan = _external_metadata(record)
+    store = c2pa_manifest_store
+    if store is None:
+        candidate = record.get("c2pa_store")
+        store = (
+            cast("dict[str, Any]", candidate)
+            if isinstance(candidate, dict)
+            else candidate
+            if isinstance(candidate, str)
+            else None
+        )
+    c2pa_info = c2pa_info_from_manifest_store(store) if store is not None else {}
+
+    ai_metadata: dict[str, str] = {}
+    pil_info = record.get("pil")
+    pil_pairs = cast("dict[str, Any]", pil_info).items() if isinstance(pil_info, dict) else ()
+    for key, value in pil_pairs:
+        normalized_key = key.lower().removeprefix("info:")
+        if normalized_key not in AI_METADATA_KEYS or isinstance(value, (dict, list, tuple)):
+            continue
+        text = value.decode("utf-8", "replace") if isinstance(value, bytes) else str(value)
+        ai_metadata.setdefault(normalized_key, text[:200] + ("…" if len(text) > 200 else ""))
+    for key, value in pairs:
+        if key != "text" or not isinstance(value, str) or "\x00" not in value:
+            continue
+        metadata_key, metadata_value = value.split("\x00", 1)
+        normalized_key = metadata_key.lower()
+        if normalized_key in AI_METADATA_KEYS:
+            ai_metadata.setdefault(
+                normalized_key,
+                metadata_value[:200] + ("…" if len(metadata_value) > 200 else ""),
+            )
+    for key in (
+        "c2pa_manifest",
+        "claim_generator",
+        "c2pa_spec",
+        "issuer",
+        "source_type",
+        "actions",
+        "synthid_watermark",
+        "soft_binding",
+    ):
+        if key in c2pa_info:
+            ai_metadata.setdefault(key, str(c2pa_info[key]))
+
+    iptc_system = iptc_ai_system_in(scan)
+
+    values_by_key: dict[str, str] = {}
+    for key, value in pairs:
+        if isinstance(value, (bytes, str)):
+            values_by_key.setdefault(key.lower(), _external_text(value))
+    description = values_by_key.get("imagedescription", "")
+    artist = values_by_key.get("artist", "")
+    xai = xai_signature_pair(description, artist)
+
+    hf_job = next(
+        (
+            str(value).strip()
+            for key, value in pairs
+            if key.lower().removeprefix("info:") == "hf-job-id" and str(value).strip()
+        ),
+        None,
+    )
+    samsung = samsung_genai_in(scan)
+
+    aigc_candidates = tuple(
+        value for key, value in pairs if key.lower().removeprefix("info:") == "aigc" and isinstance(value, str)
+    )
+    aigc = aigc_label_from_metadata(scan, aigc_candidates)
+    exif_gen = _external_exif_generator(pairs, scan)
+    if aigc is not None:
+        producer = aigc.get("ContentProducer", "")
+        ai_metadata.setdefault(
+            "aigc_label",
+            f"China AIGC label (TC260){f'; producer {producer}' if producer else ''}",
+        )
+    if xai:
+        ai_metadata.setdefault("xai_signature", "xAI/Grok EXIF signature (Artist UUID + Signature blob)")
+    if iptc_system:
+        ai_metadata.setdefault("ai_system", f"IPTC 2025.1 AI disclosure ({iptc_system})")
+    if hf_job:
+        ai_metadata.setdefault("huggingface_job", f"HuggingFace-hosted job ({hf_job})")
+    if samsung is not None:
+        ai_metadata.setdefault("samsung_genai", f"Samsung Galaxy AI editing marker (genAIType={samsung})")
+
+    return ProvenanceEvidence(
+        path=path,
+        c2pa_info=c2pa_info,
+        ai_metadata=ai_metadata,
+        scan=scan,
+        iptc_ai_system=iptc_system,
+        aigc_label=aigc,
+        exif_generator=exif_gen,
+        xai_signature=xai,
+        huggingface_job=hf_job,
+        samsung_genai=samsung,
+    )
+
+
 @dataclass
 class ProvenanceReport:
     """Aggregated provenance verdict for one image."""

src/remove_ai_watermarks/metadata.py

@@ -356,6 +356,54 @@ def has_ai_metadata(image_path: Path) -> bool:
     return xai_signature(image_path)
 
 
+def aigc_label_from_metadata(data: bytes, candidates: tuple[str, ...] = ()) -> dict[str, str] | None:
+    """Parse a China TC260 AI-labeling block from already collected metadata."""
+    import html
+    import json
+    from typing import cast
+
+    def _parse(text: str, *, require_tc260_field: bool) -> dict[str, str] | None:
+        try:
+            parsed = json.loads(text)
+        except ValueError:
+            return None
+        if not isinstance(parsed, dict):
+            return None
+        fields = {str(k): str(v) for k, v in cast("dict[object, object]", parsed).items()}
+        if require_tc260_field and not (_TC260_FIELDS & fields.keys()):
+            return None
+        return fields
+
+    for candidate in candidates:
+        if result := _parse(candidate, require_tc260_field=True):
+            return result
+
+    match = re.search(
+        rb'<TC260:AIGC>(.*?)</TC260:AIGC>|TC260:AIGC\s*=\s*"(.*?)"',
+        data,
+        re.DOTALL,
+    )
+    if match:
+        body = match.group(1) if match.group(1) is not None else match.group(2)
+        return _parse(html.unescape(body.decode("utf-8", "replace")), require_tc260_field=False)
+
+    text = data.decode("latin-1")
+    for needle in ('"AIGC"', "AIGC{"):
+        start = text.find(needle)
+        if start == -1:
+            continue
+        brace = text.find("{", start)
+        if brace == -1:
+            continue
+        try:
+            _, end = json.JSONDecoder().raw_decode(text, brace)
+        except ValueError:
+            continue
+        if result := _parse(text[brace:end], require_tc260_field=True):
+            return result
+    return None
+
+
 def aigc_label(image_path: Path) -> dict[str, str] | None:
     """Parse a China TC260 AI-labeling block, if present.
 
@@ -378,24 +426,6 @@ def aigc_label(image_path: Path) -> dict[str, str] | None:
     if they carry at least one known TC260 field (``_TC260_FIELDS``); the
     namespaced XMP element is unambiguous, so any JSON object is accepted.
     """
-    import html
-    import json
-    from typing import cast
-
-    def _parse(text: str, *, require_tc260_field: bool) -> dict[str, str] | None:
-        try:
-            parsed = json.loads(text)
-        except ValueError:
-            return None
-        if not isinstance(parsed, dict):
-            return None
-        fields = {str(k): str(v) for k, v in cast("dict[object, object]", parsed).items()}
-        if require_tc260_field and not (_TC260_FIELDS & fields.keys()):
-            return None
-        return fields
-
-    # PNG tEXt chunk keyed "AIGC" with raw JSON (Doubao and other China gens).
-    # The key is generic, so require a TC260 field to avoid a false positive.
     try:
         from PIL import Image
 
@@ -404,50 +434,9 @@ def aigc_label(image_path: Path) -> dict[str, str] | None:
     except Exception as exc:
         logger.debug("PIL could not open %s for AIGC chunk scan: %s", image_path, exc)
         value = None
-    if isinstance(value, str) and (result := _parse(value, require_tc260_field=True)):
-        return result
-
-    # XMP TC260:AIGC, namespaced (unambiguous) in either serialization RDF allows:
-    # an element  <TC260:AIGC>{...}</TC260:AIGC>  or an attribute  TC260:AIGC="{...}"
-    # (the attribute form is what PicWish writes). Both are HTML-entity encoded.
     data = scan_head(image_path)
-    match = re.search(
-        rb'<TC260:AIGC>(.*?)</TC260:AIGC>|TC260:AIGC\s*=\s*"(.*?)"',
-        data,
-        re.DOTALL,
-    )
-    if match:
-        body = match.group(1) if match.group(1) is not None else match.group(2)
-        return _parse(html.unescape(body.decode("utf-8", "replace")), require_tc260_field=False)
-
-    # Generic raw-JSON forms the PNG-chunk and XMP paths above both miss, each
-    # gated on a TC260 field: the ``"AIGC":{...}`` key wrapper (as written into
-    # JPEG EXIF UserComment) and the bare ``AIGC{...}`` blob (the label glued
-    # straight to its JSON, no key wrapper, in a JPEG APP segment near the JFIF
-    # header). `raw_decode` brace-matches the inner object (respecting nested
-    # braces / quoted strings); `_parse` applies the same dict coercion + TC260
-    # gate as the PNG-chunk path. A non-matching hit (no TC260 field, or an
-    # undecodable brace) must FALL THROUGH to the next form, never short-circuit:
-    # a quoted ``"AIGC"`` can appear later in an XMP packet while the real label
-    # is a bare ``AIGC{...}`` blob earlier in the file, so an unconditional return
-    # on the quoted form would shadow the bare form.
-    text = data.decode("latin-1")
-    for needle in ('"AIGC"', "AIGC{"):
-        start = text.find(needle)
-        if start == -1:
-            continue
-        # First brace at/after the needle: the object brace for ``"AIGC":{`` and
-        # the glued brace (at start+4) for the bare ``AIGC{`` -- one search covers both.
-        brace = text.find("{", start)
-        if brace == -1:
-            continue
-        try:
-            _, end = json.JSONDecoder().raw_decode(text, brace)
-        except ValueError:
-            continue
-        if result := _parse(text[brace:end], require_tc260_field=True):
-            return result
-    return None
+    candidates = (value,) if isinstance(value, str) else ()
+    return aigc_label_from_metadata(data, candidates)
 
 
 # C2PA "Durable Content Credentials" manifest repositories (C2PA 2.4). When the
@@ -541,6 +530,16 @@ def _read_file_tail(image_path: Path, size: int) -> bytes:
         return b""
 
 
+def samsung_genai_in(data: bytes) -> int | None:
+    """Return Samsung's non-zero ``genAIType`` from collected metadata bytes."""
+    if _SAMSUNG_EDITOR_MARKER not in data:
+        return None
+    match = _SAMSUNG_GENAI_RE.search(data)
+    if match is None:
+        return None
+    return int(match.group(1)) or None
+
+
 def samsung_genai(image_path: Path) -> int | None:
     """Return Samsung's non-zero ``genAIType`` value if the image carries the
     Galaxy AI editing marker, else None.
@@ -565,12 +564,17 @@ def samsung_genai(image_path: Path) -> int | None:
             oversize = False
         if oversize:
             data = _read_file_tail(image_path, _QUICK_SCAN_BYTES)
-    if _SAMSUNG_EDITOR_MARKER not in data:
+    return samsung_genai_in(data)
+
+
+def iptc_ai_system_in(data: bytes) -> str | None:
+    """Return an IPTC 2025.1 AI-disclosure note from collected metadata bytes."""
+    if not any(marker in data for marker in IPTC_AI_FIELD_MARKERS):
         return None
-    m = _SAMSUNG_GENAI_RE.search(data)
-    if m is None:
-        return None
-    return int(m.group(1)) or None
+    match = re.search(rb"AISystemUsed[=:\s]*[\"'>]\s*([^<\"']{1,120})", data)
+    if match and (value := match.group(1).decode("utf-8", "replace").strip()):
+        return value
+    return "fields present"
 
 
 def iptc_ai_system(image_path: Path) -> str | None:
@@ -583,13 +587,7 @@ def iptc_ai_system(image_path: Path) -> str | None:
     extractable, otherwise the literal ``"fields present"``. Container-agnostic
     raw-byte scan; handles both XMP element and attribute serializations.
     """
-    data = scan_head(image_path)
-    if not any(marker in data for marker in IPTC_AI_FIELD_MARKERS):
-        return None
-    match = re.search(rb"AISystemUsed[=:\s]*[\"'>]\s*([^<\"']{1,120})", data)
-    if match and (value := match.group(1).decode("utf-8", "replace").strip()):
-        return value
-    return "fields present"
+    return iptc_ai_system_in(scan_head(image_path))
 
 
 def synthid_source(image_path: Path) -> str | None:
@@ -632,6 +630,20 @@ def synthid_source(image_path: Path) -> str | None:
     return ", ".join(matched) if matched else None
 
 
+def generator_from_metadata(candidates: list[str], scan: bytes = b"") -> str | None:
+    """Return a known AI generator from collected EXIF, PNG, or XMP values."""
+    from remove_ai_watermarks.noai.constants import AI_GENERATOR_TOKENS
+
+    candidates.extend(
+        match.group(1).decode("latin1", "replace")
+        for match in re.finditer(rb"CreatorTool[>\"'=\s]{1,4}([^<\"']{1,80})", scan)
+    )
+    for value in candidates:
+        if any(token in value.lower() for token in AI_GENERATOR_TOKENS):
+            return value.strip()
+    return None
+
+
 def exif_generator(image_path: Path) -> str | None:
     """Return an AI-generator name from the EXIF ``Software`` / XMP ``CreatorTool``
     field (or a PNG text chunk), if it matches a known generator (see
@@ -644,10 +656,6 @@ def exif_generator(image_path: Path) -> str | None:
     chunks rather than EXIF. Only AI tokens match, so ordinary editors (plain
     "Adobe Photoshop", "GIMP") are not flagged.
     """
-    import re
-
-    from remove_ai_watermarks.noai.constants import AI_GENERATOR_TOKENS
-
     candidates: list[str] = []
 
     # EXIF Software / Artist / ImageDescription (0th IFD) via PIL exif bytes,
@@ -682,18 +690,12 @@ def exif_generator(image_path: Path) -> str | None:
     except Exception as exc:  # unopenable format / malformed EXIF
         logger.debug("EXIF generator read failed for %s: %s", image_path, exc)
 
-    # XMP CreatorTool: text, container-agnostic (covers HEIF/JXL via raw scan).
     try:
         head = scan_head(image_path)
-        for match in re.finditer(rb"CreatorTool[>\"'=\s]{1,4}([^<\"']{1,80})", head):
-            candidates.append(match.group(1).decode("latin1", "replace"))
     except Exception as exc:
         logger.debug("XMP CreatorTool scan failed for %s: %s", image_path, exc)
-
-    for value in candidates:
-        if any(token in value.lower() for token in AI_GENERATOR_TOKENS):
-            return value.strip()
-    return None
+        head = b""
+    return generator_from_metadata(candidates, head)
 
 
 # xAI / Grok EXIF signature scheme. A 64+ char base64 blob after "Signature:"
@@ -703,7 +705,7 @@ _XAI_SIGNATURE_RE = re.compile(r"Signature:\s*[A-Za-z0-9+/=]{64,}")
 _UUID_RE = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", re.IGNORECASE)
 
 
-def _is_xai_signature_pair(description: str, artist: str) -> bool:
+def xai_signature_pair(description: str, artist: str) -> bool:
     """True if an EXIF (ImageDescription, Artist) pair is xAI/Grok's scheme."""
     return _XAI_SIGNATURE_RE.match(description) is not None and _UUID_RE.fullmatch(artist) is not None
 
@@ -739,7 +741,7 @@ def xai_signature(image_path: Path) -> bool:
         logger.debug("xAI-signature EXIF read failed for %s: %s", image_path, exc)
         return False
 
-    return _is_xai_signature_pair(
+    return xai_signature_pair(
         _exif_text(tags, piexif.ImageIFD.ImageDescription), _exif_text(tags, piexif.ImageIFD.Artist)
     )
 
@@ -794,9 +796,7 @@ def _ai_exif_targets(loaded: dict[str, Any]) -> list[tuple[str, int, bytes, str]
             targets.append((ifd_key, tag, value, name))
 
     # (a) xAI / Grok: the Signature blob and the UUID Artist go together.
-    if _is_xai_signature_pair(
-        _exif_text(ifd0, piexif.ImageIFD.ImageDescription), _exif_text(ifd0, piexif.ImageIFD.Artist)
-    ):
+    if xai_signature_pair(_exif_text(ifd0, piexif.ImageIFD.ImageDescription), _exif_text(ifd0, piexif.ImageIFD.Artist)):
         add("0th", ifd0, piexif.ImageIFD.ImageDescription, "ImageDescription")
         add("0th", ifd0, piexif.ImageIFD.Artist, "Artist")
     # (b) known AI generator token in a 0th text tag.

src/remove_ai_watermarks/noai/c2pa.py

@@ -189,9 +189,8 @@ def _active_manifest(store: dict[str, Any]) -> dict[str, Any]:
     return cast("dict[str, Any]", active) if isinstance(active, dict) else {}
 
 
-def _info_from_store_json(store_json: str) -> dict[str, Any]:
-    """Build the C2PA info dict from a c2pa-python manifest-store JSON string."""
-    store_bytes = store_json.encode("utf-8")
+def _info_from_store(store: dict[str, Any], store_bytes: bytes) -> dict[str, Any]:
+    """Build normalized C2PA info from one parsed manifest store."""
     c2pa_info: dict[str, Any] = {
         "has_c2pa": True,
         "type": "C2PA (Coalition for Content Provenance and Authenticity)",
@@ -202,13 +201,6 @@ def _info_from_store_json(store_json: str) -> dict[str, Any]:
     # registry scan that runs on the raw caBX chunk applies unchanged here.
     _populate_registry_fields(store_bytes, c2pa_info)
 
-    try:
-        parsed: Any = json.loads(store_json)
-    except (ValueError, TypeError):
-        return c2pa_info
-    if not isinstance(parsed, dict):
-        return c2pa_info
-    store = cast("dict[str, Any]", parsed)
     if generator := _claim_generator_from_store(store):
         c2pa_info["claim_generator"] = generator
     sig: Any = _active_manifest(store).get("signature_info")
@@ -217,6 +209,44 @@ def _info_from_store_json(store_json: str) -> dict[str, Any]:
     return c2pa_info
 
 
+def _info_from_store_json(store_json: str) -> dict[str, Any]:
+    """Build the C2PA info dict from a c2pa-python manifest-store JSON string."""
+    store_bytes = store_json.encode("utf-8")
+    try:
+        parsed: Any = json.loads(store_json)
+    except (ValueError, TypeError):
+        parsed = {}
+    store = cast("dict[str, Any]", parsed) if isinstance(parsed, dict) else {}
+    return _info_from_store(store, store_bytes)
+
+
+def c2pa_info_from_manifest_store(store: str | dict[str, Any]) -> dict[str, Any]:
+    """Build normalized C2PA evidence from an externally collected manifest store.
+
+    ``store`` may be the JSON string returned by ``c2pa.Reader.json()`` or its
+    decoded dictionary form. This is the non-file-backed counterpart to
+    :func:`extract_c2pa_info`.
+    """
+    if isinstance(store, dict):
+        parsed = store
+        try:
+            store_json = json.dumps(store, ensure_ascii=False)
+        except (TypeError, ValueError):
+            return {}
+    else:
+        store_json = store
+        try:
+            decoded: Any = json.loads(store_json)
+        except (TypeError, ValueError):
+            return {}
+        if not isinstance(decoded, dict):
+            return {}
+        parsed = cast("dict[str, Any]", decoded)
+    if not store_json or not parsed or parsed.get("error"):
+        return {}
+    return _info_from_store(parsed, store_json.encode("utf-8"))
+
+
 def extract_c2pa_info(image_path: Path) -> dict[str, Any]:
     """
     Extract C2PA metadata information from an image.