feat(identify): detect open SD/SDXL/FLUX invisible watermark

Research found one locally-fillable detection gap: Stable Diffusion, SDXL,
and FLUX all embed an open DWT-DCT watermark via the invisible-watermark
(imwatermark) library -- a PUBLIC decoder, no secret key, unlike SynthID.
New invisible_watermark.py decodes the known fixed patterns (verified
against upstream source: diffusers SDXL WATERMARK_MESSAGE, FLUX.2
src/flux2/watermark.py, and the 'StableDiffusionV1' default string) and
identify() reports the scheme as a high-confidence signal.

Verified locally end-to-end: embedding SDXL's exact 48-bit message and
decoding it back recovers 48/48 bits; a clean image and our own fal-SDXL
outputs decode to ~21/48 (no match). Caveat baked into the report: the
watermark is fragile -- gone after JPEG q90 -- so it confirms origin only
on pristine files; absence is never proof.

imwatermark is an optional dep (extra 'detect'; pulls non-headless opencv),
so the import is guarded and the signal is skipped when absent. CLI
--no-visible now means metadata-only (skips both pixel-domain detectors).

Also records the broader watermarking landscape in CLAUDE.md: which
services are locally detectable (SD/SDXL/FLUX), C2PA-covered (Bing/Canva/
Getty/Shutterstock unsampled), or proprietary-only like SynthID (Amazon
Titan/Nova, Kakao). Midjourney embeds neither C2PA nor an invisible mark.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/remove_ai_watermarks/cli.py

@@ -335,7 +335,11 @@ def cmd_metadata(
 
 @main.command("identify")
 @click.argument("source", type=click.Path(exists=True, path_type=Path))
-@click.option("--no-visible", is_flag=True, help="Skip the visible-sparkle detector (metadata-only, no cv2).")
+@click.option(
+    "--no-visible",
+    is_flag=True,
+    help="Skip pixel-domain detectors (visible sparkle + invisible watermark); metadata-only.",
+)
 @click.option("--json", "as_json", is_flag=True, help="Emit the report as JSON instead of a table.")
 @click.pass_context
 def cmd_identify(ctx: click.Context, source: Path, no_visible: bool, as_json: bool) -> None:
@@ -351,7 +355,7 @@ def cmd_identify(ctx: click.Context, source: Path, no_visible: bool, as_json: bo
     from remove_ai_watermarks.identify import identify
 
     source = _validate_image(source)
-    report = identify(source, check_visible=not no_visible)
+    report = identify(source, check_visible=not no_visible, check_invisible=not no_visible)
 
     if as_json:
         click.echo(json.dumps(asdict(report), default=str, indent=2))

src/remove_ai_watermarks/identify.py

@@ -75,6 +75,10 @@ _OPENAI_CAVEAT = (
     "before the rollout carry C2PA without SynthID, so the SynthID verdict is 'likely'."
 )
 _IPTC_ONLY_CAVEAT = "The IPTC 'Made with AI' tag flags AI provenance but does not identify the specific platform."
+_INVISIBLE_WM_CAVEAT = (
+    "The open invisible watermark is fragile: it does not survive JPEG re-encoding "
+    "or resizing, so it confirms origin only on a pristine (un-re-encoded) file."
+)
 
 
 @dataclass
@@ -140,13 +144,26 @@ def _visible_sparkle(image_path: Path) -> float | None:
     return detect_sparkle_confidence(image_path)
 
 
-def identify(image_path: Path, *, check_visible: bool = True) -> ProvenanceReport:
+def _invisible_watermark(image_path: Path) -> str | None:
+    """Open invisible-watermark scheme name (SD/SDXL/FLUX) or None.
+
+    Optional: needs the imwatermark decoder (extra ``detect``). Returns None if
+    it is not installed or no known watermark decodes.
+    """
+    from remove_ai_watermarks.invisible_watermark import detect_invisible_watermark
+
+    return detect_invisible_watermark(image_path)
+
+
+def identify(image_path: Path, *, check_visible: bool = True, check_invisible: bool = True) -> ProvenanceReport:
     """Identify an image's origin platform and watermark inventory.
 
     Args:
         image_path: Path to the image (PNG, JPEG, WebP, or ISOBMFF container).
         check_visible: Also run the visible Gemini-sparkle detector (cv2). Set
             False for a pure-metadata, dependency-light scan.
+        check_invisible: Also decode open invisible watermarks (SD/SDXL/FLUX) via
+            the optional imwatermark library. No-op when it is not installed.
 
     Returns:
         A :class:`ProvenanceReport`. ``is_ai_generated`` is True when any AI
@@ -206,8 +223,18 @@ def identify(image_path: Path, *, check_visible: bool = True) -> ProvenanceRepor
         if platform is None:
             platform = "Stable Diffusion / local pipeline (Automatic1111, ComfyUI, InvokeAI)"
 
-    # ── Verdict so far (metadata) ───────────────────────────────────
-    ai_from_metadata = bool((has_c2pa and (c2pa_is_ai or synthid)) or iptc or local_keys)
+    # ── Open invisible watermark (SD / SDXL / FLUX, dwtDct) ──────────
+    # Public decoder, no key -- a definitive embedded signal on pristine files.
+    if check_invisible and (scheme := _invisible_watermark(image_path)) is not None:
+        signals.append(Signal("invisible_watermark", scheme, "high"))
+        watermarks.append(f"Open invisible watermark: {scheme}")
+        caveats.append(_INVISIBLE_WM_CAVEAT)
+        if platform is None:
+            platform = f"{scheme} (open DWT-DCT watermark)"
+
+    # ── Verdict so far (metadata + embedded watermark) ──────────────
+    invisible_wm = any(s.name == "invisible_watermark" for s in signals)
+    ai_from_metadata = bool((has_c2pa and (c2pa_is_ai or synthid)) or iptc or local_keys or invisible_wm)
 
     # ── Visible Gemini sparkle (fallback for stripped-metadata case) ─
     if check_visible and (conf := _visible_sparkle(image_path)) is not None and conf >= _SPARKLE_THRESHOLD:

src/remove_ai_watermarks/invisible_watermark.py

@@ -0,0 +1,108 @@
+"""Detect open invisible watermarks embedded by the ``invisible-watermark``
+(imwatermark) library -- used by Stable Diffusion, SDXL, and FLUX.
+
+Unlike SynthID (proprietary, no local decoder), these are DWT-DCT watermarks
+with a PUBLIC decoder and no secret key, so a fresh, un-re-encoded output can be
+identified locally. The known fixed patterns were verified against upstream
+source:
+
+- **Stable Diffusion XL** -- diffusers ``StableDiffusionXLWatermarker``
+  ``WATERMARK_MESSAGE`` (48-bit).
+- **FLUX.2** -- ``black-forest-labs/flux2`` ``src/flux2/watermark.py`` (48-bit).
+- **Stable Diffusion 1.x / 2.x** -- the library's default ``"StableDiffusionV1"``
+  string (136-bit).
+
+The watermark is fragile: it does NOT survive JPEG re-encoding or resizing
+(verified -- gone after JPEG q90), so detection works only on pristine PNG
+originals. Absence is never proof. Requires the optional ``invisible-watermark``
+package (extra: ``detect``); ``detect_invisible_watermark`` returns None when it
+is not installed.
+"""
+
+# imwatermark ships no type stubs (like cv2); its decoder returns are Unknown.
+# Relax the untyped-library diagnostics for this thin wrapper module only.
+# pyright: reportMissingTypeStubs=false, reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, cast
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+log = logging.getLogger(__name__)
+
+# Known 48-bit ``bits`` watermarks (dwtDct, no key), name -> message integer.
+_BITS_48: dict[str, int] = {
+    "Stable Diffusion XL": 0b101100111110110010010000011110111011000110011110,
+    "FLUX.2 (Black Forest Labs)": 0b001010101111111010000111100111001111010100101110,
+}
+# The invisible-watermark default string watermark (SD 1.x / 2.x).
+_SD1_STRING = b"StableDiffusionV1"
+
+# Decoded bits/bytes never match a 48-bit pattern by chance: random decode lands
+# near 24/48, an exact embed at 48/48 (measured). 44 (<=4 bit errors) is a safe
+# floor that tolerates light perturbation without risking a false positive.
+_MATCH_48 = 44
+_MATCH_SD1_FRAC = 0.92  # fraction of the 136 string bits that must match
+
+
+def is_available() -> bool:
+    """True if the optional imwatermark decoder is installed."""
+    import importlib.util
+
+    return importlib.util.find_spec("imwatermark") is not None
+
+
+def _bits_match(value: int, ref: int, width: int = 48) -> int:
+    """Number of matching bits between two ``width``-bit integers."""
+    return width - bin(value ^ ref).count("1")
+
+
+def _bytes_match_frac(a: bytes, b: bytes) -> float:
+    """Fraction of matching bits between two equal-length byte strings."""
+    if len(a) != len(b) or not a:
+        return 0.0
+    diff = sum(bin(x ^ y).count("1") for x, y in zip(a, b, strict=True))
+    return 1.0 - diff / (8 * len(b))
+
+
+def detect_invisible_watermark(image_path: Path) -> str | None:
+    """Return the embedding scheme name if a known open watermark is decoded.
+
+    Returns e.g. ``"Stable Diffusion XL"`` / ``"FLUX.2 (Black Forest Labs)"`` /
+    ``"Stable Diffusion 1.x / 2.x"``, or None if none matches, the decoder is
+    unavailable, or the image can't be read. Meaningful only on pristine
+    (un-re-encoded) images.
+    """
+    if not is_available():
+        return None
+    import cv2
+    from imwatermark import WatermarkDecoder
+
+    img = cv2.imread(str(image_path))
+    if img is None:
+        return None
+
+    # 48-bit fixed-message watermarks (SDXL, FLUX.2).
+    try:
+        bits = WatermarkDecoder("bits", 48).decode(img, "dwtDct")
+        value = 0
+        for bit in bits:
+            value = (value << 1) | (1 if bit else 0)
+        for name, ref in _BITS_48.items():
+            if _bits_match(value, ref) >= _MATCH_48:
+                return name
+    except Exception as exc:  # decode can fail on tiny images
+        log.debug("48-bit watermark decode failed for %s: %s", image_path, exc)
+
+    # 136-bit default string watermark (SD 1.x / 2.x).
+    try:
+        raw = cast("bytes", WatermarkDecoder("bytes", 8 * len(_SD1_STRING)).decode(img, "dwtDct"))
+        if _bytes_match_frac(raw, _SD1_STRING) >= _MATCH_SD1_FRAC:
+            return "Stable Diffusion 1.x / 2.x"
+    except Exception as exc:
+        log.debug("string watermark decode failed for %s: %s", image_path, exc)
+
+    return None