mirror of
https://github.com/wiltodelta/remove-ai-watermarks.git
synced 2026-06-05 02:28:00 +02:00
feat(identify): integrity-clash detection for contradictory provenance (v0.6.7)
Surface contradictions between independent provenance signals instead of collapsing to a single verdict -- a strong tell of spoofed, transplanted, or laundered metadata. Inspired by arXiv:2603.02378. Two rules in the new _integrity_clashes helper: - Conflicting AI-origin attributions: two or more distinct AI vendors named by independent generator stamps (e.g. a C2PA OpenAI manifest on an image whose EXIF says Make="Ideogram AI"). - Camera + AI: a camera-capture C2PA device (Pixel/Leica/Sony/Nikon/Truepic) coexisting with an AI-generation marker -- a genuine capture is not AI. High-precision by design: only hard generator stamps feed it (C2PA issuer when the source is AI, SynthID proxy, EXIF/XMP generator, IPTC AISystemUsed, xAI, AIGC). The fuzzy visible sparkle and the open invisible watermark are excluded -- the latter can be a by-product of our own SDXL removal pass. Vendor normalization (_vendor_of over _AI_VENDOR_TOKENS) keeps consistent signals from clashing (C2PA "Google (Gemini)" + SynthID-Google agree); the C2PA vendor is read from the issuer attribution, not the resolved platform, so a camera label like "Google Pixel" cannot mis-normalize to an AI vendor. Surfaced as ProvenanceReport.integrity_clashes (red in the table view, included in --json). 19 new tests; all real single-origin fixtures (chatgpt/firefly/ doubao/grok/mj) verified to produce zero clashes (false-positive guard). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -325,7 +325,7 @@ Tracked but not yet implemented:
|
||||
- **Real non-PNG C2PA fixtures**. SynthID-source detection for JPEG / WebP / AVIF is currently covered only by synthetic byte blobs; replace with real vendor-emitted files to ground the binary-scan path.
|
||||
- **Maintenance debt**. Clear strict-pyright debt in `remove_ai_metadata` / `cli.py` (untyped piexif / PIL / click / rich) so `maintain.sh` can finish green. (`uv-secure` is already clean since `idna` was bumped to 3.16.)
|
||||
- **AVIF / HEIF EXIF/XMP inside the `meta` box**. Removal already strips top-level C2PA `uuid` / JUMBF `jumb` boxes and any AI-labelled top-level XMP `uuid` box, and non-ISOBMFF audio/video (WebM, MP3, WAV, FLAC, OGG) is stripped losslessly via ffmpeg. Still open: EXIF/XMP stored as *items inside the `meta` box* (typical for AVIF/HEIF stills) — needs `meta`-box surgery (iinf/iloc + mdat splice) or `exiftool` (a non-bundled binary dependency).
|
||||
- **Multi-signal contradiction reporting ("Integrity Clash")**. When a C2PA manifest claims human authorship but a watermark / IPTC signal indicates AI (or signals otherwise disagree), `identify` should surface the *contradiction* rather than collapse to one verdict (per [arXiv:2603.02378](https://arxiv.org/abs/2603.02378)). Pure aggregation logic — no new dependency or sample needed.
|
||||
- **Multi-signal contradiction reporting ("Integrity Clash")** — *shipped (v0.6.7)*. `identify` now surfaces contradictions between independent provenance signals (two different AI vendors named by separate stamps, or camera-capture C2PA credentials next to AI-generation markers) as `integrity_clashes` (shown in red in the table view and in `--json`), rather than collapsing to a single verdict. Inspired by [arXiv:2603.02378](https://arxiv.org/abs/2603.02378).
|
||||
- **More C2PA device signers**. Leica, Nikon, Google Pixel, Sony, and Truepic are mapped (each verified against a real signed file). Canon and Samsung Galaxy (AI-edit) are deferred until a real signed sample surfaces — no public direct-download C2PA file exists for them today (upload-to-verify / news-agency-licensed only).
|
||||
- **C2PA detection window for streaming MP4**. Non-PNG detection scans the first 1 MB; a manifest placed after a large `mdat` in a streaming MP4 can be missed (front-placed manifests, the common case, are caught).
|
||||
- **Resemble PerTh audio detection** — evaluated, not feasible with the public API: `get_watermark()` returns a raw bit array with no presence/confidence flag, so watermarked vs. clean audio can't be reliably separated without Resemble's fixed payload or a confidence service. Same wall as the SynthID pixel detector.
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "remove-ai-watermarks"
|
||||
version = "0.6.6"
|
||||
version = "0.6.7"
|
||||
description = "Remove visible and invisible AI watermarks from images (Gemini / Nano Banana, ChatGPT, Stable Diffusion)"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.10"
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
"""Remove-AI-Watermarks: Unified tool for removing visible and invisible AI watermarks."""
|
||||
|
||||
__version__ = "0.6.6"
|
||||
__version__ = "0.6.7"
|
||||
|
||||
@@ -618,6 +618,11 @@ def cmd_identify(ctx: click.Context, source: Path, no_visible: bool, as_json: bo
|
||||
console.print(f"\n Verdict: {verdict} [dim](confidence: {report.confidence})[/]")
|
||||
console.print(f" Platform: {report.platform or '[dim]undetermined[/]'}")
|
||||
|
||||
if report.integrity_clashes:
|
||||
console.print("\n [bold red]⚠ Integrity clash[/] [dim](provenance signals contradict each other)[/]")
|
||||
for clash in report.integrity_clashes:
|
||||
console.print(f" [red]- {clash}[/]")
|
||||
|
||||
if report.watermarks:
|
||||
table = Table(show_header=True, header_style="bold", title="Watermarks / provenance markers")
|
||||
table.add_column("Marker", style="cyan")
|
||||
|
||||
@@ -110,6 +110,11 @@ class ProvenanceReport:
|
||||
watermarks: list[str] = field(default_factory=list[str])
|
||||
signals: list[Signal] = field(default_factory=list["Signal"])
|
||||
caveats: list[str] = field(default_factory=list[str])
|
||||
# Contradictions between independent provenance signals (e.g. two different
|
||||
# AI vendors both claiming the image, or camera-capture credentials next to
|
||||
# AI-generation markers). Non-empty means the provenance is internally
|
||||
# inconsistent -- a strong tell of spoofed, transplanted, or laundered metadata.
|
||||
integrity_clashes: list[str] = field(default_factory=list[str])
|
||||
|
||||
|
||||
def _issuers_in(data: bytes) -> list[str]:
|
||||
@@ -188,6 +193,88 @@ def _attribute_platform(issuers: list[str], *, is_ai: bool = True) -> str | None
|
||||
return None
|
||||
|
||||
|
||||
# Coarse origin-vendor normalization for integrity-clash detection. Two signals
|
||||
# that resolve to the SAME key are consistent (a C2PA "Google (Gemini)" issuer
|
||||
# and a SynthID-Google proxy, or Adobe Firefly + its Adobe TrustMark soft
|
||||
# binding); two DIFFERENT keys from independent generator stamps are a
|
||||
# contradiction (a C2PA OpenAI manifest on an image whose EXIF says "Ideogram
|
||||
# AI"). Substring match on the lowercased platform/detail string; first hit wins,
|
||||
# so order specific tokens before brand umbrellas where they overlap.
|
||||
_AI_VENDOR_TOKENS: tuple[tuple[str, str], ...] = (
|
||||
("gpt-image", "OpenAI"),
|
||||
("dall", "OpenAI"),
|
||||
("sora", "OpenAI"),
|
||||
("openai", "OpenAI"),
|
||||
("gemini", "Google"),
|
||||
("imagen", "Google"),
|
||||
("nano banana", "Google"),
|
||||
("google", "Google"),
|
||||
("firefly", "Adobe"),
|
||||
("adobe", "Adobe"),
|
||||
("bing", "Microsoft"),
|
||||
("designer", "Microsoft"),
|
||||
("microsoft", "Microsoft"),
|
||||
("stability", "Stability AI"),
|
||||
("stable diffusion", "Stability AI"),
|
||||
("sdxl", "Stability AI"),
|
||||
("ideogram", "Ideogram"),
|
||||
("grok", "xAI"),
|
||||
("aurora", "xAI"),
|
||||
("xai", "xAI"),
|
||||
)
|
||||
|
||||
|
||||
def _vendor_of(text: str | None) -> str | None:
|
||||
"""Normalize a platform/generator string to a coarse origin-vendor key, or None."""
|
||||
if not text:
|
||||
return None
|
||||
low = text.lower()
|
||||
for token, vendor in _AI_VENDOR_TOKENS:
|
||||
if token in low:
|
||||
return vendor
|
||||
return None
|
||||
|
||||
|
||||
def _integrity_clashes(
|
||||
ai_vendors: dict[str, str], camera_label: str | None, *, camera_has_ai_marker: bool
|
||||
) -> list[str]:
|
||||
"""Surface contradictions between independent provenance signals.
|
||||
|
||||
Args:
|
||||
ai_vendors: family name -> normalized AI-origin vendor, one entry per
|
||||
generator-stamped signal (C2PA issuer when the source is AI, SynthID
|
||||
proxy, EXIF/XMP generator tag, IPTC AISystemUsed, xAI, AIGC label).
|
||||
camera_label: a camera/verified-capture C2PA device platform, if one was
|
||||
identified (Pixel, Leica, Sony, Nikon, Truepic), else None.
|
||||
camera_has_ai_marker: True when an AI-generation stamp coexists with the
|
||||
camera credentials.
|
||||
|
||||
Returns:
|
||||
Human-readable clash descriptions; empty when the signals agree.
|
||||
"""
|
||||
clashes: list[str] = []
|
||||
|
||||
by_vendor: dict[str, list[str]] = {}
|
||||
for family, vendor in ai_vendors.items():
|
||||
by_vendor.setdefault(vendor, []).append(family)
|
||||
if len(by_vendor) >= 2:
|
||||
parts = [f"{vendor} (via {', '.join(sorted(fams))})" for vendor, fams in sorted(by_vendor.items())]
|
||||
clashes.append(
|
||||
"Conflicting AI-origin attributions from independent signals: "
|
||||
+ " vs ".join(parts)
|
||||
+ " -- one provenance set was likely spoofed, transplanted, or laundered."
|
||||
)
|
||||
|
||||
if camera_label and camera_has_ai_marker:
|
||||
vendors = ", ".join(sorted(set(ai_vendors.values()))) or "present"
|
||||
clashes.append(
|
||||
f"Camera-capture C2PA credentials ({camera_label}) coexist with AI-generation markers "
|
||||
f"({vendors}) -- a genuine camera capture is not AI-generated, so the provenance is inconsistent."
|
||||
)
|
||||
|
||||
return clashes
|
||||
|
||||
|
||||
def _visible_sparkle(image_path: Path) -> float | None:
|
||||
"""Visible Gemini-sparkle confidence in [0, 1], or None if unavailable.
|
||||
|
||||
@@ -251,6 +338,13 @@ def identify(image_path: Path, *, check_visible: bool = True, check_invisible: b
|
||||
signals: list[Signal] = []
|
||||
watermarks: list[str] = []
|
||||
caveats: list[str] = []
|
||||
# One normalized origin vendor per generator-stamped signal, for integrity-
|
||||
# clash detection (see _integrity_clashes). Visible sparkle and the open
|
||||
# invisible watermark are deliberately excluded: the former is a fuzzy visual
|
||||
# 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)
|
||||
|
||||
# ── C2PA Content Credentials ────────────────────────────────────
|
||||
has_c2pa = bool(info) or b"c2pa" in head.lower() or C2PA_UUID in head
|
||||
@@ -271,11 +365,17 @@ def identify(image_path: Path, *, check_visible: bool = True, check_invisible: b
|
||||
# signer/producer), with the issuer byte-scan only as fallback. The issuer
|
||||
# scan alone mis-attributed real samples (Leica->Truepic timestamp authority,
|
||||
# Nikon->Adobe namespace, Pixel->Google Gemini) -- the device scan fixes that.
|
||||
platform = (_device_platform(head) or _attribute_platform(issuers, is_ai=c2pa_is_ai)) if has_c2pa else None
|
||||
platform = (camera_label or _attribute_platform(issuers, is_ai=c2pa_is_ai)) if has_c2pa else None
|
||||
if has_c2pa:
|
||||
detail = ", ".join(filter(None, [", ".join(issuers), generator, info.get("source_type")]))
|
||||
signals.append(Signal("c2pa", detail or "C2PA manifest present", "high"))
|
||||
watermarks.append(f"C2PA Content Credentials ({', '.join(issuers) or 'unknown signer'})")
|
||||
# Record the AI-origin vendor for clash detection only when the source is
|
||||
# actually AI -- classify the issuer attribution / generator, NOT the
|
||||
# resolved `platform` (which may be a camera device token whose label,
|
||||
# e.g. "Google Pixel", would mis-normalize to an AI vendor).
|
||||
if c2pa_is_ai and (v := (_vendor_of(_attribute_platform(issuers, is_ai=True)) or _vendor_of(generator))):
|
||||
ai_vendor_claims["c2pa"] = v
|
||||
|
||||
# ── SynthID metadata proxy ──────────────────────────────────────
|
||||
# get_ai_metadata already sets synthid_watermark for both PNG (caBX parser)
|
||||
@@ -286,6 +386,8 @@ def identify(image_path: Path, *, check_visible: bool = True, check_invisible: b
|
||||
caveats.append(_SYNTHID_CAVEAT)
|
||||
if "OpenAI" in (" ".join(issuers) + synthid):
|
||||
caveats.append(_OPENAI_CAVEAT)
|
||||
if v := _vendor_of(synthid):
|
||||
ai_vendor_claims["synthid"] = v
|
||||
|
||||
# ── C2PA soft-binding: a named forensic/third-party watermark vendor ─
|
||||
# (Adobe TrustMark, Digimarc, Imatag, ...). Present in the manifest even when
|
||||
@@ -315,6 +417,8 @@ def identify(image_path: Path, *, check_visible: bool = True, check_invisible: b
|
||||
watermarks.append(f"IPTC 2025.1 AI disclosure ({system})" if named else "IPTC 2025.1 AI disclosure fields")
|
||||
if platform is None and named:
|
||||
platform = f"{system} (IPTC AISystemUsed)"
|
||||
if named and (v := _vendor_of(system)):
|
||||
ai_vendor_claims["iptc_ai_system"] = v
|
||||
|
||||
# ── China TC260 AIGC label (Doubao and other China-served gens) ──
|
||||
aigc = any(m in head for m in AIGC_MARKERS)
|
||||
@@ -324,6 +428,7 @@ def identify(image_path: Path, *, check_visible: bool = True, check_invisible: b
|
||||
watermarks.append("China AIGC label (TC260 standard)")
|
||||
if platform is None:
|
||||
platform = "China AIGC-labeled generator (TC260; e.g. Doubao)"
|
||||
ai_vendor_claims["aigc"] = "China AIGC (TC260)"
|
||||
|
||||
# ── Local diffusion parameters (Stable Diffusion / ComfyUI) ──────
|
||||
local_keys = sorted(k for k in meta if k.lower() in _LOCAL_GEN_KEYS)
|
||||
@@ -340,6 +445,8 @@ def identify(image_path: Path, *, check_visible: bool = True, check_invisible: b
|
||||
watermarks.append(f"Embedded generator tag: {generator_tag}")
|
||||
if platform is None:
|
||||
platform = f"{generator_tag} (EXIF/XMP generator tag)"
|
||||
if v := _vendor_of(generator_tag):
|
||||
ai_vendor_claims["exif_generator"] = v
|
||||
|
||||
# ── xAI / Grok EXIF signature scheme (no C2PA/SynthID/IPTC) ──────
|
||||
# Grok's only provenance signal: EXIF ImageDescription "Signature: <base64>"
|
||||
@@ -350,6 +457,7 @@ def identify(image_path: Path, *, check_visible: bool = True, check_invisible: b
|
||||
watermarks.append("xAI/Grok EXIF signature")
|
||||
if platform is None:
|
||||
platform = "xAI (Grok / Aurora)"
|
||||
ai_vendor_claims["xai"] = "xAI"
|
||||
|
||||
# ── Open invisible watermark (SD / SDXL / FLUX, dwtDct) ──────────
|
||||
# Public decoder, no key -- a definitive embedded signal on pristine files.
|
||||
@@ -404,6 +512,9 @@ def identify(image_path: Path, *, check_visible: bool = True, check_invisible: b
|
||||
is_ai = None
|
||||
confidence = "none"
|
||||
|
||||
# ── Integrity clashes: contradictions between independent signals ─
|
||||
clashes = _integrity_clashes(ai_vendor_claims, camera_label, camera_has_ai_marker=bool(ai_vendor_claims))
|
||||
|
||||
caveats.append(_STRIP_CAVEAT)
|
||||
# De-duplicate while preserving order.
|
||||
caveats = list(dict.fromkeys(caveats))
|
||||
@@ -416,4 +527,5 @@ def identify(image_path: Path, *, check_visible: bool = True, check_invisible: b
|
||||
watermarks=watermarks,
|
||||
signals=signals,
|
||||
caveats=caveats,
|
||||
integrity_clashes=clashes,
|
||||
)
|
||||
|
||||
@@ -17,7 +17,9 @@ from remove_ai_watermarks.identify import (
|
||||
ProvenanceReport,
|
||||
_ai_tools_in,
|
||||
_attribute_platform,
|
||||
_integrity_clashes,
|
||||
_issuers_in,
|
||||
_vendor_of,
|
||||
identify,
|
||||
)
|
||||
|
||||
@@ -447,3 +449,101 @@ class TestIdentifyAIGC:
|
||||
r = identify(self._aigc_png(tmp_path), check_visible=False)
|
||||
sig = next(s for s in r.signals if s.name == "aigc")
|
||||
assert "BYTEDANCE001" in sig.detail
|
||||
|
||||
|
||||
# ── Integrity clashes (contradictions between independent signals) ──────
|
||||
|
||||
|
||||
class TestVendorOf:
|
||||
def test_openai_variants(self):
|
||||
assert _vendor_of("OpenAI (ChatGPT / gpt-image / DALL-E / Sora)") == "OpenAI"
|
||||
assert _vendor_of("DALL-E 3") == "OpenAI"
|
||||
|
||||
def test_google_variants(self):
|
||||
assert _vendor_of("Google (Gemini / Imagen)") == "Google"
|
||||
assert _vendor_of("Imagen 3") == "Google"
|
||||
|
||||
def test_other_vendors(self):
|
||||
assert _vendor_of("Ideogram AI") == "Ideogram"
|
||||
assert _vendor_of("Adobe Firefly") == "Adobe"
|
||||
assert _vendor_of("Stability AI (Stable Image)") == "Stability AI"
|
||||
|
||||
def test_camera_label_is_not_an_ai_vendor(self):
|
||||
# Camera platform labels must NOT normalize to an AI vendor, or a camera
|
||||
# capture would be mistaken for AI-generation in clash detection.
|
||||
assert _vendor_of("Leica (camera, C2PA capture)") is None
|
||||
|
||||
def test_unknown_is_none(self):
|
||||
assert _vendor_of("a regular photo") is None
|
||||
assert _vendor_of(None) is None
|
||||
|
||||
|
||||
class TestIntegrityClashesHelper:
|
||||
def test_two_ai_vendors_clash(self):
|
||||
clashes = _integrity_clashes({"c2pa": "OpenAI", "exif_generator": "Ideogram"}, None, camera_has_ai_marker=True)
|
||||
assert len(clashes) == 1
|
||||
assert "OpenAI" in clashes[0]
|
||||
assert "Ideogram" in clashes[0]
|
||||
|
||||
def test_same_vendor_two_signals_no_clash(self):
|
||||
# C2PA Google + SynthID-Google proxy is consistent, not a contradiction.
|
||||
assert _integrity_clashes({"c2pa": "Google", "synthid": "Google"}, None, camera_has_ai_marker=True) == []
|
||||
|
||||
def test_single_vendor_no_clash(self):
|
||||
assert _integrity_clashes({"c2pa": "OpenAI"}, None, camera_has_ai_marker=True) == []
|
||||
|
||||
def test_empty_no_clash(self):
|
||||
assert _integrity_clashes({}, None, camera_has_ai_marker=False) == []
|
||||
|
||||
def test_camera_plus_ai_marker_clashes(self):
|
||||
clashes = _integrity_clashes(
|
||||
{"exif_generator": "Ideogram"},
|
||||
"Google Pixel (camera, C2PA capture)",
|
||||
camera_has_ai_marker=True,
|
||||
)
|
||||
assert any("Camera-capture" in c and "Pixel" in c for c in clashes)
|
||||
|
||||
def test_camera_without_ai_marker_no_clash(self):
|
||||
# A clean camera capture (the normal case for our Pixel/Leica/Sony files)
|
||||
# must NOT raise a clash.
|
||||
assert _integrity_clashes({}, "Leica (camera, C2PA capture)", camera_has_ai_marker=False) == []
|
||||
|
||||
|
||||
class TestIntegrityClashEndToEnd:
|
||||
def _c2pa_jpeg(self, tmp_path: Path, blob: bytes) -> Path:
|
||||
path = tmp_path / "img.jpg"
|
||||
path.write_bytes(b"\xff\xd8\xff\xe1jumbc2pa" + blob + b"\xff\xd9")
|
||||
return path
|
||||
|
||||
def test_two_generator_stamps_clash(self, tmp_path: Path):
|
||||
# An OpenAI C2PA manifest (AI source) on an image that ALSO carries a
|
||||
# China TC260 AIGC label = two independent generator stamps naming
|
||||
# different origins -> a laundering tell.
|
||||
path = self._c2pa_jpeg(tmp_path, b"OpenAI ... trainedAlgorithmicMedia ... TC260:AIGC label")
|
||||
r = identify(path, check_visible=False, check_invisible=False)
|
||||
assert r.integrity_clashes
|
||||
assert any("Conflicting AI-origin" in c for c in r.integrity_clashes)
|
||||
|
||||
def test_single_stamp_no_clash(self, tmp_path: Path):
|
||||
path = self._c2pa_jpeg(tmp_path, b"OpenAI ... trainedAlgorithmicMedia")
|
||||
r = identify(path, check_visible=False, check_invisible=False)
|
||||
assert r.integrity_clashes == []
|
||||
|
||||
def test_clash_serializes_to_json(self, tmp_path: Path):
|
||||
path = self._c2pa_jpeg(tmp_path, b"OpenAI ... trainedAlgorithmicMedia ... TC260:AIGC label")
|
||||
r = identify(path, check_visible=False, check_invisible=False)
|
||||
payload = json.loads(json.dumps(asdict(r), default=str))
|
||||
assert payload["integrity_clashes"] == r.integrity_clashes
|
||||
|
||||
|
||||
@pytest.mark.skipif(not SAMPLES_DIR.exists(), reason="data/samples not present")
|
||||
@pytest.mark.parametrize("fixture", ["chatgpt-1.png", "firefly-1.png", "doubao-1.png", "grok-1.jpg", "mj-1.png"])
|
||||
class TestRealSamplesHaveNoClash:
|
||||
"""Every real single-origin fixture must report zero clashes (false-positive guard)."""
|
||||
|
||||
def test_no_false_positive_clash(self, fixture: str):
|
||||
path = SAMPLES_DIR / fixture
|
||||
if not path.exists():
|
||||
pytest.skip(f"{fixture} not present")
|
||||
r = identify(path, check_visible=False, check_invisible=False)
|
||||
assert r.integrity_clashes == []
|
||||
|
||||
Reference in New Issue
Block a user