fix: metadata-strip parity, input robustness, and detection/clash coverage

Bug fixes (each with a regression test):
- metadata strip parity across every marker placement: IPTC digitalSourceType
  in XMP, the Samsung post-EOI trailer, the China TC260 AIGC block in EXIF
  UserComment, a bare AIGC block in a non-standard APP segment, and the ISOBMFF
  EXIF path (AIGC + xAI) are all now stripped -- anything a scanner flags, the
  strip reaches
- Samsung genAIType detected when its trailer sits past the 512 KB scan window
  (file-tail read on large photos)
- crashes on edge inputs: Gemini detector on images with a short side < 16px,
  footprint_mask on a zero-size ndarray, the humanizer on chromatic_shift >=
  width, and the CLI on unreadable/corrupt/empty input (clean error, not a
  traceback)
- WebP written losslessly (cv2 quality 101), not lossy at 100
- the IPTC digitalSourceType algorithmicMedia (procedural, not trained on
  sampled data) is no longer flagged as AI-generated, so clean procedural
  content is not scrubbed
- c2pa source-type: compositeWithTrainedAlgorithmicMedia is checked before the
  bare algorithmicMedia token, so an AI-enhanced composite is not misclassified

Detection:
- integrity-clash coverage now normalizes ByteDance / Canva / ElevenLabs /
  Black Forest Labs, so a transplanted manifest next to an independent
  conflicting stamp is caught; the generic China TC260 AIGC label is attributed
  to a co-present TC260 vendor, so a legit Doubao image (its own C2PA + TC260
  label) does not clash (corpus-validated: 0 new clashes on 5000 carriers)

CLI:
- batch exits non-zero (with a warning) when any image errors or a GPU-missing
  SynthID scrub is skipped, and copies the input through so the output dir stays
  complete -- it used to always exit 0 and could silently drop files

Perf:
- GeminiEngine reused as a process-wide singleton with a precomputed template
  ladder: -24% on the identify sparkle path, detection byte-identical

Internal: one shared _ai_exif_targets rule set feeds both EXIF scrubbers so
their coverage cannot drift; docs synced; maintain.sh hardened so the uv-secure
internal teardown crash no longer aborts the gate (still fails on a real finding).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Victor Kuznetsov
2026-07-13 10:49:24 +03:00
co-authored by Claude Opus 4.8
parent 190dc89d23
commit a4c901ff39
25 changed files with 894 additions and 97 deletions
+8 -8
View File
File diff suppressed because one or more lines are too long
+1 -1
View File
@@ -74,7 +74,7 @@ Metadata detection for AVIF/HEIF/JPEG-XL relies on a binary scan for `C2PA_UUID`
**Meta-box XMP now handled (`isobmff.blank_ai_xmp_packets`, v0.6.9):** an AI-label XMP packet stored as a meta-box `mime` item (AVIF/HEIF) is blanked in place (overwritten with spaces of the same length, so `iloc` offsets and the coded image stay valid).
**`Exif` item inside the `meta` box (AVIF/HEIF), now handled in place (2026-06-19):** an AI-generator token in an EXIF item (its TIFF bytes live in `mdat`/`idat`) is blanked by `isobmff.blank_ai_exif_tokens` — it finds EXIF TIFF blocks by their II/MM byte-order header, validates each with **piexif** (a coincidental II/MM run in pixel data won't parse as a TIFF IFD, so it is ignored), and overwrites any `Software`/`Make`/`Artist`/`ImageDescription` value carrying an `AI_GENERATOR_TOKENS` token with spaces of the **same length**. Same-length means every box size and `iloc` offset stays valid and the coded image is untouched — so it avoids the full `iinf`/`iloc` surgery (offset rewrite) that exiftool would need (exiftool is a non-installed binary dep, deliberately not used). It scrubs only the AI-token value; camera/editor EXIF is preserved. Wired into `remove_ai_metadata`'s ISOBMFF path after `blank_ai_xmp_packets`. Limitation: covers the AI-generator-token case (the realistic one); a future xAI-signature-in-meta-box-EXIF (Grok is JPEG-only today) is not separately handled. **Still NOT built:** Resemble PerTh audio detection (no presence/confidence flag exists).
**`Exif` item inside the `meta` box (AVIF/HEIF), now handled in place (2026-06-19):** an AI-generator token in an EXIF item (its TIFF bytes live in `mdat`/`idat`) is blanked by `isobmff.blank_ai_exif_tokens` — it finds EXIF TIFF blocks by their II/MM byte-order header, validates each with **piexif** (a coincidental II/MM run in pixel data won't parse as a TIFF IFD, so it is ignored), and overwrites any `Software`/`Make`/`Artist`/`ImageDescription` value carrying an `AI_GENERATOR_TOKENS` token with spaces of the **same length**. Same-length means every box size and `iloc` offset stays valid and the coded image is untouched — so it avoids the full `iinf`/`iloc` surgery (offset rewrite) that exiftool would need (exiftool is a non-installed binary dep, deliberately not used). It scrubs only the AI value; camera/editor EXIF is preserved. Wired into `remove_ai_metadata`'s ISOBMFF path after `blank_ai_xmp_packets`. Because the ISOBMFF branch never runs the JPEG `_scrub_ai_exif`, this is the ONLY EXIF scrubber on that path and must stay in PARITY with it: it now also blanks the China TC260 `{"AIGC":{...}}` block in `ImageDescription`/`UserComment` (via `_is_aigc_exif_value` — Doubao producer + Tencent service-provider schemas) and the xAI/Grok `Signature:` + UUID-`Artist` pair, not just `AI_GENERATOR_TOKENS` in `Software`/`Make`/`Artist`/`ImageDescription` (regression `test_noai.py::TestISOBMFF::{test_blank_aigc_block_in_exif, test_blank_xai_signature_pair_in_exif}`). **Still NOT built:** Resemble PerTh audio detection (no presence/confidence flag exists).
**Audio watermark DETECTION (Resemble PerTh) was evaluated and NOT built (2026-05-26):** `resemble-perth`'s `PerthImplicitWatermarker.get_watermark()` returns a raw bit-array with **no presence/confidence flag** (clean audio decodes to arbitrary bits too), so reliably distinguishing watermarked-from-clean needs either Resemble's fixed payload or a confidence API -- neither is public, and there's no real Resemble sample to calibrate against. Same wall-class as the SynthID pixel detector: the decode exists, reliable presence-detection does not. (perth's top-level `PerthImplicitWatermarker` is also gated to None unless `librosa` is importable.)
+4 -4
View File
@@ -41,7 +41,7 @@ module.
**Samsung Galaxy + ASUS Gallery live in a separate `_SIGNER_C2PA_PLATFORM` (scanned after `_device_platform`, before the issuer fallback), NOT in `_DEVICE_C2PA_PLATFORM`** — verified on real signed files 2026-05-29. Reason: a Galaxy phone stamps BOTH its device cert AND a `trainedAlgorithmicMedia`/genAIType AI marker on a Generative-Edit image, so treating it as a "genuine camera capture" would false-fire integrity-clash rule 2 on every Galaxy AI edit. The signer tokens (`b"Samsung Galaxy"` cert org — distinct from the EXIF `SM-xxxx` model string on ordinary Samsung photos; `b"com.asus.gallery"` claim generator) only resolve the platform label; the AI verdict still comes from the source-type / genAIType. ASUS Gallery is a C2PA-signed edit with no AI marker, so it attributes the platform without asserting `is_ai`.
**Samsung's `genAIType` (in the proprietary `PhotoEditor_Re_Edit_Data` JSON) is an undocumented Galaxy-AI editing marker** (`metadata.samsung_genai`, gated on the `PhotoEditor_Re_Edit_Data` container; non-zero value = AI tool used, values {1,5} observed): medium-confidence because the field has no public spec (verified 2026-05-29: absent from C2PA spec + Samsung docs), but it co-occurred with `trainedAlgorithmicMedia` in 3/3 verified files that record a source-type and was the SOLE AI marker on a Galaxy S24 file that omits the source type. Camera C2PA marks capture authenticity, not AI (Pixel carries `computationalCapture`, not `trainedAlgorithmicMedia`), so these never set `is_ai` -- that stays driven by digital-source-type. `c2pa.cbor_text_after` (now public) is best-effort for the `generator` detail string only and can be None when the manifest keys it `claim_generator_info` (Pixel).
**Samsung's `genAIType` (in the proprietary `PhotoEditor_Re_Edit_Data` JSON) is an undocumented Galaxy-AI editing marker** (`metadata.samsung_genai`, gated on the `PhotoEditor_Re_Edit_Data` container; non-zero value = AI tool used, values {1,5} observed; Galaxy AI appends it as a trailer AFTER the JPEG EOI, so `samsung_genai` reads the file TAIL when the 512 KB quick-scan head misses it — else a multi-MB photo's trailer past the window went undetected while removal, which reads the whole file, would still strip it; removal truncates the post-EOI Samsung trailer via `metadata._strip_samsung_trailer`, pixels bit-identical): medium-confidence because the field has no public spec (verified 2026-05-29: absent from C2PA spec + Samsung docs), but it co-occurred with `trainedAlgorithmicMedia` in 3/3 verified files that record a source-type and was the SOLE AI marker on a Galaxy S24 file that omits the source type. Camera C2PA marks capture authenticity, not AI (Pixel carries `computationalCapture`, not `trainedAlgorithmicMedia`), so these never set `is_ai` -- that stays driven by digital-source-type. `c2pa.cbor_text_after` (now public) is best-effort for the `generator` detail string only and can be None when the manifest keys it `claim_generator_info` (Pixel).
**Issuer→generator mapping is `is_ai`-gated** (`_attribute_platform(issuers, is_ai=c2pa_is_ai)`): a specific AI-generator platform is named only when the digital-source-type is `trainedAlgorithmicMedia`; on a non-AI source an issuer substring is treated as incidental (an "Adobe XMP" toolkit string in an *unmapped* Canon/Sony capture would otherwise mislabel it "Adobe Firefly"), so it degrades to the neutral "C2PA signer: X" label. **The one exception is an identity-AI issuer** (`c2pa_is_ai = c2pa_source_kind is not None or c2pa_identity_ai`, where `c2pa_identity_ai` is any resolved issuer org in `C2PA_IDENTITY_AI_ORGS`): a vendor flagged `asserts_ai` (today only Dreamina) sets `c2pa_is_ai` True on its own, so its platform resolves even though the manifest carries no `trainedAlgorithmicMedia`. This is safe precisely because the flag is restricted to distinctive brand strings, not the incidental-mention-prone common words. Real Firefly/OpenAI/Google output carries the AI source-type, so it is unaffected (verified: chatgpt-1.png→OpenAI, firefly-1.png→Adobe Firefly still attribute). `_attribute_platform` defaults `is_ai=True` so the mapping stays unit-testable in isolation. Add capture-camera tokens to `_DEVICE_C2PA_PLATFORM`, editing-app/AI-device signer tokens to `_SIGNER_C2PA_PLATFORM`, generator/issuer platforms to the `C2PA_AI_VENDORS` registry in `constants.py` (which derives `_ISSUER_PLATFORM`), not inline. For non-PNG containers (JPEG/WebP/AVIF/HEIF/JXL) the caBX parser returns nothing, so issuer (`_issuers_in`) and generator (`_ai_tools_in`, reusing `C2PA_AI_TOOLS`) are recovered by binary-scanning the first MB. EXIF `Software` / `Make` / `Artist` / `ImageDescription`, XMP `CreatorTool`, and PNG `tEXt` chunks (`Software`/`Source`/`Title`/`Description` — NovelAI stamps its generator there, not EXIF) are read by `metadata.exif_generator` (PIL+piexif for any format PIL opens incl. AVIF, plus a container-agnostic XMP raw-byte scan that also covers HEIF/JXL), matched against `AI_GENERATOR_TOKENS` so ordinary editors (plain "Adobe Photoshop") and real-camera `Make` ("Apple"/"Canon") are not flagged. Tokens mined from the retained corpus 2026-06-22: `novelai`, `reve.com` (full token, not bare `reve`), `aphrodite ai` — all no-C2PA generator stamps that previously read as no-signal (and under the P0#5 no-signal skip would have skipped the scrub).
@@ -51,7 +51,7 @@ module.
**Rule 2's independence gate (added 2026-06-11):** a device that both captures and runs on-device generative AI (Google Pixel Magic Editor / Pixel Studio) records the capture AND the AI edit in ONE C2PA manifest — so the AI vendor is named only from that same manifest (`c2pa` issuer + `synthid` proxy, both `c2pa_manifest` source) — a legitimate edit chain, NOT a clash. Rule 2 therefore fires only when some `ai_vendor_claims` family has a source `!= "c2pa_manifest"` (EXIF/XMP generator, IPTC, TC260 AIGC, a second manifest naming AI on a camera capture — the real laundering tell). This killed a false-positive class on the corpus: 2 real Pixel generative-edit PNGs (`computationalCapture` + `trainedAlgorithmicMedia` + "Applied imperceptible SynthID watermark" in one Google manifest) read as camera-vs-AI clashes before the gate. Pure cameras (Leica/Sony/Nikon/Truepic) that do NOT generate AI still clash on any within-manifest AI marker only if it is independent — they never legitimately carry one, so the gate is behavior-neutral for them while fixing Pixel (regression-guarded by `test_identify.py::TestIntegrityClashesHelper::{test_pixel_generative_edit_same_manifest_no_clash,test_camera_plus_independent_ai_marker_still_clashes}` + `TestIntegrityClashEndToEnd::test_pixel_generative_edit_no_clash`).
**Independence is source-grouped (`_CLASH_SOURCE`, added 2026-06-02):** the C2PA issuer attribution (`c2pa`) and the SynthID proxy (`synthid`) are NOT independent — the proxy is inferred from the *same* manifest — so they share one source and two vendors named within a single manifest do not clash. This killed a false-positive class found on the spaces corpus: legitimate multi-actor manifests where a product wraps another vendor's engine (Microsoft Designer on OpenAI → `OpenAI, Microsoft`; Microsoft on Google → `Microsoft, Google LLC, Google C2PA Core Generator Library`) or an edit chain re-signs (Adobe over a Gemini original → Adobe c2pa + Google synthid) — 19 such files across the 2026-06-01/02 batches read as clashes before the fix. Rule 1 still fires when a manifest vendor disagrees with a genuinely independent stamp (EXIF/XMP generator, IPTC `AISystemUsed`, AIGC, xAI); each non-`c2pa`/`synthid` family is its own source (`test_identify.py::TestIntegrityClashes::{test_multi_actor_manifest_no_clash,test_manifest_vendor_vs_independent_signal_clashes}`). Vendor normalization is `_vendor_of` over `_AI_VENDOR_TOKENS` (so a C2PA "Google (Gemini)" issuer and a SynthID-Google proxy agree, while different vendors clash).
**Independence is source-grouped (`_CLASH_SOURCE`, added 2026-06-02):** the C2PA issuer attribution (`c2pa`) and the SynthID proxy (`synthid`) are NOT independent — the proxy is inferred from the *same* manifest — so they share one source and two vendors named within a single manifest do not clash. This killed a false-positive class found on the spaces corpus: legitimate multi-actor manifests where a product wraps another vendor's engine (Microsoft Designer on OpenAI → `OpenAI, Microsoft`; Microsoft on Google → `Microsoft, Google LLC, Google C2PA Core Generator Library`) or an edit chain re-signs (Adobe over a Gemini original → Adobe c2pa + Google synthid) — 19 such files across the 2026-06-01/02 batches read as clashes before the fix. Rule 1 still fires when a manifest vendor disagrees with a genuinely independent stamp (EXIF/XMP generator, IPTC `AISystemUsed`, AIGC, xAI); each non-`c2pa`/`synthid` family is its own source (`test_identify.py::TestIntegrityClashes::{test_multi_actor_manifest_no_clash,test_manifest_vendor_vs_independent_signal_clashes}`). Vendor normalization is `_vendor_of` over `_AI_VENDOR_TOKENS` (so a C2PA "Google (Gemini)" issuer and a SynthID-Google proxy agree, while different vendors clash). `_AI_VENDOR_TOKENS` covers ByteDance (all brands: bytedance/doubao/jimeng/dreamina/volcengine), Canva, ElevenLabs, and Black Forest Labs in addition to the OpenAI/Google/Adobe/... set — without them a transplanted ByteDance/Canva/BFL C2PA manifest next to an independent conflicting stamp was silently missed. **The generic `China AIGC (TC260)` label names no SPECIFIC vendor** (any Chinese generator applies it), so it cannot vendor-conflict in the spoofing sense: when a Chinese TC260-applying vendor (`_TC260_VENDORS`, today `{ByteDance}`) is co-attributed, Rule 1 attributes the label to that vendor (a legit Doubao image carries BOTH a ByteDance C2PA manifest and its own TC260 label and must not clash); against a NON-TC260 vendor (OpenAI etc.) the label stays generic and still clashes as a laundering tell (`test_bytedance_c2pa_plus_own_aigc_no_clash`, `test_foreign_vendor_plus_aigc_still_clashes`, `test_bytedance_c2pa_plus_foreign_generator_clashes`). Corpus-validated: 0 new clashes on 5000 ByteDance/AIGC/Canva/FLUX carriers.
**High-precision by design:** only hard generator stamps feed it (C2PA-issuer when source is AI, SynthID, 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). The c2pa vendor is classified from the issuer attribution / generator, NOT the resolved `platform` (a camera label like "Google Pixel" would mis-normalize to "Google"). All real single-origin fixtures (chatgpt/firefly/doubao/grok/mj) verified to produce **zero** clashes (false-positive guard in `test_identify.py::TestRealSamplesHaveNoClash`).
@@ -131,7 +131,7 @@ The cost (mislabel ~8-33% of non-Gemini content as Gemini) outweighs the benefit
**Removal is localize -> fill:** the glyph blob is localized to a solid, dilated footprint mask and the shared `watermark_registry.fill` inpaints it. Verified on a real 2958-wide @f-liva photo: re-detect 0.79→0.00, no readable text or outline on the recovered wooden table — checked **visually**, not just by the detector. The registry gates removal on `detect`.
**Detection is locale-specific** (the string differs per language); this build detects only the Italian "Contenuti generati dall'AI" variant, so non-Italian Samsung locales are not detected — and, because detection gates removal, not removed — even though the fill mask itself is locale-independent. Other locales need their own captured detection template. This is a pre-existing limit, unchanged by the localize -> fill refactor.
**Detection is locale-specific** (the string differs per language); this build detects only the Italian "Contenuti generati dall'AI" variant, so non-Italian Samsung locales are not detected — and, because detection gates removal, not removed — even though the fill mask itself is locale-independent. Other locales need their own detection silhouette — the locale string font-rendered and calibrated on real positives (the pill's `scripts/render_pill_silhouette.py` pattern), NOT an app capture (the solid/gray/white capture workflow retired with reverse-alpha). This is a pre-existing limit, unchanged by the localize -> fill refactor.
**No committed real sample** (only the flat calibration captures are committed) — `tests/test_samsung_engine.py` synthesizes a mark from the bundled template (bottom-left geometry), with `test_recovers_shifted_mark_on_texture` guarding the localize-on-shift path. Samsung Galaxy AI edits are independently caught by C2PA + the `genAIType` marker in `metadata`/`identify`, so this engine is the visible-mark *removal* path; it also feeds `identify` as the medium-confidence `visible_samsung` signal via the registry (the stripped-metadata fallback).
@@ -234,4 +234,4 @@ Known-visible-mark removal by **localize -> fill**: each detected mark is locali
### `batch`
Process every supported image in a directory (output defaults to `<directory>_clean/`, set with `-o`). `--mode visible|invisible|metadata|all` (default `visible`); the invisible/all path reuses the **full `invisible` knob set** (`--strength`/`--steps`/`--guidance-scale`/`--pipeline`/`--controlnet-scale`/`--model`/`--device`/`--max-resolution`/`--min-resolution`/`--upscaler`/`--seed`/`--hf-token`/`--humanize`/`--unsharp`/`--adaptive-polish`/`--tile`/`--tile-size`/`--tile-overlap`/`--force`), plus `--backend` for the visible localize -> fill pass. `--adaptive-polish` is ON by default; `--auto` is deprecated and a no-op that only warns. **No-signal skip (P0#5):** in invisible/all mode each image runs the same `has_invisible_target` gate — a signal-less image is skipped (no diffusion); in `invisible` mode the input is copied through to the output dir so it stays complete, in `all` mode the visible-removed result is kept and metadata is still stripped. `--force` scrubs every image regardless. One engine cached per pipeline; the polish is resolved once before the loop.
Process every supported image in a directory (output defaults to `<directory>_clean/`, set with `-o`). `--mode visible|invisible|metadata|all` (default `visible`); the invisible/all path reuses the **full `invisible` knob set** (`--strength`/`--steps`/`--guidance-scale`/`--pipeline`/`--controlnet-scale`/`--model`/`--device`/`--max-resolution`/`--min-resolution`/`--upscaler`/`--seed`/`--hf-token`/`--humanize`/`--unsharp`/`--adaptive-polish`/`--tile`/`--tile-size`/`--tile-overlap`/`--force`), plus `--backend` for the visible localize -> fill pass. `--adaptive-polish` is ON by default; `--auto` is deprecated and a no-op that only warns. **No-signal skip (P0#5):** in invisible/all mode each image runs the same `has_invisible_target` gate — a signal-less image is skipped (no diffusion); in `invisible` mode the input is copied through to the output dir so it stays complete, in `all` mode the visible-removed result is kept and metadata is still stripped. `--force` scrubs every image regardless. One engine cached per pipeline; the polish is resolved once before the loop. **Exit code (`batch` used to always exit 0, hiding failures):** `cmd_batch` raises `SystemExit(1)` when any image errored, OR when a `--mode invisible`/`all` image carried an invisible signal but the GPU extra was absent so its SynthID scrub was skipped — mirroring single `all`, it emits a loud "the invisible watermark was NOT removed on N image(s)" warning and (invisible mode) copies the input through so the output dir stays complete, rather than silently dropping the signal-bearing files that most needed processing. `_process_batch_image` returns that skipped-scrub flag; the loop tallies it. Regression-guarded by `tests/test_cli.py::TestBatchCommand::{test_batch_errors_exit_nonzero, test_batch_invisible_gpu_missing_writes_output_and_exits_nonzero}`.
+16 -4
View File
@@ -4,11 +4,23 @@ set -euo pipefail
uv sync --all-extras
# uv-outdated / uv-secure run via uvx (isolated env), NOT `uv run`: resolving them
# inside the project env crashes (uv-secure -> "annotated-doc raised exception") and,
# with set -e, aborts the whole gate before ruff/pyright/tests. uvx sidesteps the
# in-project dependency conflict (see CLAUDE.md "Test and lint").
# inside the project env crashes and, with set -e, aborts the whole gate before
# ruff/pyright/tests (see CLAUDE.md "Test and lint").
uvx uv-outdated
uvx uv-secure --ignore-unfixed
# uv-secure prints its verdict but can then crash in an internal teardown with a
# NON-ZERO exit -- observed as "annotated-doc raised exception" and later "anyio raised
# exception"; both are bugs in uv-secure's OWN uvx env, not a project vulnerability. With
# set -e that teardown crash aborts the whole gate before ruff/pyright/tests. So gate on
# the VERDICT, not the exit code: capture the output, accept the run when uv-secure
# reported all-safe (even if it then crashed), but still FAIL on a real finding (no
# all-safe line) so a genuine CVE is never masked, and fail loud if it never got a
# verdict at all (so a broken run is never silently skipped).
secure_out="$(uvx uv-secure --ignore-unfixed 2>&1)" || true
printf '%s\n' "$secure_out"
if ! grep -qE "No vulnerabilities or maintenance issues detected|All dependencies appear safe" <<<"$secure_out"; then
echo "maintain.sh: uv-secure reported a finding or failed before its verdict -- triage before committing." >&2
exit 1
fi
uv run ruff check --fix
uv run ruff format
# Scoped to src/: a full-project pyright run OOM-crashes node on this ML-heavy
@@ -315,6 +315,8 @@ class TextMarkEngine:
With ``force`` and no glyph found, falls back to the whole geometry box (the
``--no-detect`` path). The caller gates on detection.
"""
if image is None or image.size == 0:
return None # guard before to_bgr (cvtColor raises on an empty Mat); mirror detect()
image = image_io.to_bgr(image)
h, w = image.shape[:2]
if h < 32 or w < 64:
+55 -7
View File
@@ -595,6 +595,9 @@ def cmd_visible(
except RuntimeError as e: # e.g. a selected migan/lama backend whose extra is absent
console.print(f" Error: {e}")
raise SystemExit(1) from e
except (ValueError, OSError) as e: # unreadable / truncated / non-image input
console.print(f" Error: cannot read image {source.name}: {e}")
raise SystemExit(1) from e
elapsed = time.monotonic() - t0
h, w = result.shape[:2]
console.print(f" Input: {source.name} ({w}x{h})")
@@ -933,7 +936,11 @@ def cmd_metadata(
return
# Remove
out = remove_ai_metadata(source, output, keep_standard=keep_standard)
try:
out = remove_ai_metadata(source, output, keep_standard=keep_standard)
except (OSError, ValueError) as e: # unreadable / truncated / non-image (PIL raises OSError subclasses)
console.print(f" Error: cannot process {source.name}: {e}")
raise SystemExit(1) from e
console.print(f" AI metadata stripped -> {out}")
@@ -1246,6 +1253,14 @@ def cmd_all(
# ── Batch command ──
def _passthrough_copy(img_path: Path, out_path: Path) -> None:
"""Copy the input's pixels through to ``out_path`` unchanged (the invisible-mode skip
paths), so the output dir stays complete without touching the pixels."""
src_bgr, src_alpha = image_io.read_bgr_and_alpha(img_path)
if src_bgr is not None:
image_io.write_bgr_with_alpha(out_path, src_bgr, src_alpha)
def _process_batch_image(
ctx: click.Context,
img_path: Path,
@@ -1272,16 +1287,21 @@ def _process_batch_image(
tile_size: int = 1024,
tile_overlap: int = 128,
force: bool = False,
) -> None:
) -> bool:
"""Process a single image for batch mode.
Applies the requested watermark removal steps (visible, invisible,
metadata) to *img_path* and writes the result to *out_path*.
Returns True if the invisible (SynthID) scrub was skipped because the GPU deps
are missing while a signal was present -- so the batch caller can warn + exit
non-zero, mirroring the single ``all`` command.
Raises:
ValueError: If the image cannot be opened.
"""
saved_alpha: NDArray[Any] | None = None
synthid_skipped = False
if mode in ("visible", "all"):
# Always read the ORIGINAL source: the visible pass is the first step, so a
@@ -1341,13 +1361,22 @@ def _process_batch_image(
# visible-processed `out_path` whose C2PA is already gone.
vendor=vendor_for_strength(img_path),
)
elif not invisible_available() and not skip_no_signal:
# An invisible signal IS present but the GPU deps are missing, so the
# SynthID scrub cannot run. Mirror the single `all` command's loud skip:
# flag it for a batch-level warning + non-zero exit (a silently retained
# SynthID watermark is the #1 "it didn't work" report). For invisible mode
# nothing wrote out_path yet -> copy the input through so the output dir is
# complete with the pixels deliberately left intact (without this, a
# signal-bearing image in a GPU-less --mode invisible run got NO output).
synthid_skipped = True
if mode == "invisible" and not out_path.exists():
_passthrough_copy(img_path, out_path)
elif skip_no_signal and mode == "invisible" and not out_path.exists():
# No invisible target and the visible/all pass did not write out_path
# (invisible mode): copy the input through so the output dir is complete
# with the pixels deliberately left intact.
src_bgr, src_alpha = image_io.read_bgr_and_alpha(img_path)
if src_bgr is not None:
image_io.write_bgr_with_alpha(out_path, src_bgr, src_alpha)
_passthrough_copy(img_path, out_path)
if mode in ("metadata", "all"):
from remove_ai_watermarks.metadata import remove_ai_metadata
@@ -1361,6 +1390,8 @@ def _process_batch_image(
if final_bgr is not None:
image_io.write_bgr_with_alpha(out_path, final_bgr, saved_alpha)
return synthid_skipped
@main.command("batch")
@click.argument("directory", type=click.Path(exists=True, file_okay=False, path_type=Path))
@@ -1457,6 +1488,7 @@ def cmd_batch(
processed = 0
errors = 0
synthid_skipped_count = 0
with Progress(
SpinnerColumn(),
@@ -1473,7 +1505,7 @@ def cmd_batch(
progress.update(task, description=f"{img_path.name}")
try:
_process_batch_image(
if _process_batch_image(
ctx=ctx,
img_path=img_path,
out_path=out_path,
@@ -1499,7 +1531,8 @@ def cmd_batch(
tile_size=tile_size,
tile_overlap=tile_overlap,
force=force,
)
):
synthid_skipped_count += 1
processed += 1
except Exception as e:
@@ -1511,6 +1544,21 @@ def cmd_batch(
console.print(f"\n {processed} processed" + (f" {errors} errors" if errors else ""))
if synthid_skipped_count:
# Mirror the single `all` command: a silently retained SynthID watermark is the
# #1 "it didn't work" report, so make the skipped scrub impossible to miss.
console.print(
f"\n WARNING: the invisible (SynthID) watermark was NOT removed on "
f"{synthid_skipped_count} image(s) -- the GPU dependencies are not installed, "
f"so those outputs still carry the invisible watermark.\n"
f" Install the extra and rerun: pip install 'remove-ai-watermarks[gpu]'"
)
# Non-zero exit so a wrapping service detects an incomplete/failed run (batch used
# to always exit 0, hiding both per-image errors and skipped SynthID scrubs).
if errors or synthid_skipped_count:
raise SystemExit(1)
if __name__ == "__main__":
main()
+40 -4
View File
@@ -20,6 +20,7 @@ to DETECT and to shape the removal mask -- the old reverse-alpha pixel recovery
# pyright: reportUnknownMemberType=false, reportUnknownArgumentType=false, reportUnknownVariableType=false, reportUnknownParameterType=false, reportMissingTypeArgument=false, reportMissingTypeStubs=false, reportMissingImports=false, reportArgumentType=false, reportAssignmentType=false, reportReturnType=false, reportCallIssue=false, reportIndexIssue=false, reportOperatorIssue=false, reportOptionalMemberAccess=false, reportOptionalCall=false, reportOptionalSubscript=false, reportOptionalOperand=false, reportAttributeAccessIssue=false, reportPrivateImportUsage=false, reportPrivateUsage=false, reportInvalidTypeForm=false, reportConstantRedefinition=false, reportUnnecessaryComparison=false
from __future__ import annotations
import functools
import logging
from dataclasses import dataclass
from enum import Enum
@@ -126,6 +127,12 @@ def _load_embedded_asset(name: str) -> NDArray[Any]:
return img
# Single source of truth for the multi-scale template ladder (aggressively downscaled to
# slightly upscaled): the precomputed `_tmpl_cache` and the `_scan_scales` loop must use
# the SAME scales or a scan scale would miss the cache and KeyError.
_TEMPLATE_SCALES: tuple[int, ...] = tuple(range(16, 120, 2))
class GeminiEngine:
"""Detects and localizes the visible Gemini sparkle for the shared fill removal.
@@ -242,6 +249,16 @@ class GeminiEngine:
self._alpha_small = _calculate_alpha_map(bg_small)
self._alpha_large = _calculate_alpha_map(bg_large)
# Per-scale resized templates are constant (``_alpha_large`` never changes),
# so precompute the whole fixed 16..118 ladder once: ``_scan_scales`` runs it on
# every image (twice -- global + corner), and re-``resize``-ing the 96x96 source
# each time is pure repeated work. Prebuilt (not lazy) so the dict is read-only
# after construction and safe to share across threads via the module singleton.
self._tmpl_cache: dict[int, NDArray[Any]] = {
scale: cv2.resize(self._alpha_large, (scale, scale), interpolation=cv2.INTER_AREA)
for scale in _TEMPLATE_SCALES
}
logger.debug(
"Alpha maps loaded: small=%s, large=%s",
self._alpha_small.shape,
@@ -275,11 +292,10 @@ class GeminiEngine:
``_alpha_large`` is the high-quality source downscaled per scale; the range
covers aggressively downscaled to slightly upscaled logos.
"""
for scale in range(16, 120, 2):
for scale in _TEMPLATE_SCALES:
if scale > gray.shape[0] or scale > gray.shape[1]:
continue
tmpl = cv2.resize(self._alpha_large, (scale, scale), interpolation=cv2.INTER_AREA)
match_res = cv2.matchTemplate(gray, tmpl, cv2.TM_CCOEFF_NORMED)
match_res = cv2.matchTemplate(gray, self._tmpl_cache[scale], cv2.TM_CCOEFF_NORMED)
_, max_val, _, max_loc = cv2.minMaxLoc(match_res)
yield scale, float(max_val), max_loc
@@ -361,6 +377,12 @@ class GeminiEngine:
if promoted is not None:
candidates.append(promoted)
# No candidate at any scale: the search region is smaller than the 16px template
# floor (an image whose short side is < 16px), so nothing is detectable. Return
# the empty (detected=False) result rather than dereferencing candidates[0].
if not candidates:
return result
# Select the candidate with the highest full-fusion confidence (pre-FP-gate).
best_scale, pos_x, pos_y, best_raw_ncc = candidates[0]
grad_score, var_score, best_fused = 0.0, 0.0, -1.0
@@ -543,6 +565,8 @@ class GeminiEngine:
yield no mask (reported-removed-but-unchanged). Absent ``region``, direct callers
keep the detect-then-force behavior.
"""
if image is None or image.size == 0:
return None # guard before to_bgr (cvtColor raises on an empty Mat); mirror detect_watermark
image = image_io.to_bgr(image)
h, w = image.shape[:2]
if region is not None:
@@ -681,6 +705,18 @@ class GeminiEngine:
return float(np.median((hi - lo) / (hi + 1.0)))
@functools.lru_cache(maxsize=1)
def _shared_engine() -> GeminiEngine:
"""Process-wide default ``GeminiEngine`` singleton.
The engine holds only constant assets (embedded captures, alpha maps, the
precomputed template ladder) and takes the image as a method argument, so one
instance is reused across every ``detect_sparkle_confidence`` call instead of
reloading assets + recomputing alpha maps + rebuilding the template cache on
each of the ~34k images an ``identify`` batch scans. Output is identical."""
return GeminiEngine()
def detect_sparkle_confidence(image_path: Path, *, image: NDArray[Any] | None = None) -> float | None:
"""Visible-sparkle detection confidence for a file, for provenance use.
@@ -698,4 +734,4 @@ def detect_sparkle_confidence(image_path: Path, *, image: NDArray[Any] | None =
img = image if image is not None else image_io.imread(image_path)
if img is None:
return None
return float(GeminiEngine().detect_watermark(img).confidence)
return float(_shared_engine().detect_watermark(img).confidence)
+8 -5
View File
@@ -41,11 +41,14 @@ def apply_analog_humanizer(image: NDArray, grain_intensity: float = 4.0, chromat
# Shift R channel left, B channel right. np.roll is circular, so it wraps
# the opposite edge into a thin colored fringe at the L/R borders; replicate
# the original edge columns there to keep the intended offset interior-only.
if chromatic_shift > 0:
r = np.roll(r, -chromatic_shift, axis=1)
r[:, -chromatic_shift:] = r[:, -chromatic_shift - 1 : -chromatic_shift]
b = np.roll(b, chromatic_shift, axis=1)
b[:, :chromatic_shift] = b[:, chromatic_shift : chromatic_shift + 1]
# Clamp so the edge-replication slices below always have a source column: a shift
# >= width would leave them empty and crash the broadcast (r[:, -shift:] = (H, 0)).
shift = min(chromatic_shift, image.shape[1] - 1)
if shift > 0:
r = np.roll(r, -shift, axis=1)
r[:, -shift:] = r[:, -shift - 1 : -shift]
b = np.roll(b, shift, axis=1)
b[:, :shift] = b[:, shift : shift + 1]
merged = cv2.merge((b, g, r))
+37 -1
View File
@@ -299,6 +299,19 @@ _AI_VENDOR_TOKENS: tuple[tuple[str, str], ...] = (
("grok", "xAI"),
("aurora", "xAI"),
("xai", "xAI"),
# ByteDance family (all its brands normalize to one origin, mirroring constants.py):
# without these a transplanted ByteDance C2PA manifest next to an independent
# conflicting stamp went undetected by the clash check.
("bytedance", "ByteDance"),
("doubao", "ByteDance"),
("jimeng", "ByteDance"),
("dreamina", "ByteDance"),
("volcengine", "ByteDance"),
("volcano engine", "ByteDance"),
("canva", "Canva"),
("elevenlabs", "ElevenLabs"),
("eleven labs", "ElevenLabs"),
("black forest", "Black Forest Labs"),
)
@@ -328,6 +341,17 @@ def _vendor_of(text: str | None) -> str | None:
_C2PA_MANIFEST_SOURCE = "c2pa_manifest"
_CLASH_SOURCE: dict[str, str] = {"c2pa": _C2PA_MANIFEST_SOURCE, "synthid": _C2PA_MANIFEST_SOURCE}
# The generic China TC260 AIGC vendor label -- a COUNTRY-LEVEL regulatory "this is AI"
# stamp any Chinese generator applies to its own output, naming no specific vendor.
_GENERIC_AIGC_VENDOR = "China AIGC (TC260)"
# Vendors that apply the TC260 label to their OWN output. When one is co-attributed with
# the generic AIGC label, the label is that vendor's own stamp (not an independent
# competing origin), so the clash check attributes the AIGC label to it -- else a legit
# ByteDance/Doubao image (C2PA "ByteDance" + its own TC260 label) would false-clash once
# ByteDance normalizes via _vendor_of. Chinese generators only (Canva/BFL/ElevenLabs,
# also added to _vendor_of, are NOT TC260 appliers).
_TC260_VENDORS: frozenset[str] = frozenset({"ByteDance"})
def _integrity_clashes(
ai_vendors: dict[str, str], camera_label: str | None, *, camera_has_ai_marker: bool
@@ -352,6 +376,18 @@ def _integrity_clashes(
# families clash only when they belong to different provenance sources (see
# _CLASH_SOURCE) AND name different vendors -- so multiple vendors named within
# one C2PA manifest (c2pa issuer + synthid proxy) do not flag.
# The generic TC260 AIGC label is a Chinese regulatory "this is AI" stamp. When a
# Chinese TC260-applying vendor (ByteDance) is ALSO attributed, the label is that
# vendor's own stamp on its own output, so attribute it to that vendor -- a legit
# Doubao image carries BOTH a ByteDance C2PA manifest and its own TC260 label and
# must not clash. Against a NON-TC260 vendor (OpenAI, Google, ...) the label stays
# generic and still clashes as a laundering tell (a foreign-vendor image carrying a
# Chinese TC260 label names two different origins).
if ai_vendors.get("aigc") == _GENERIC_AIGC_VENDOR:
own = next((v for f, v in ai_vendors.items() if f != "aigc" and v in _TC260_VENDORS), None)
if own:
ai_vendors = {**ai_vendors, "aigc": own} # copy co-located with the relabel
source = {fam: _CLASH_SOURCE.get(fam, fam) for fam in ai_vendors}
independent_conflict = any(
source[a] != source[b] and ai_vendors[a] != ai_vendors[b] for a, b in itertools.combinations(ai_vendors, 2)
@@ -625,7 +661,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)"
ai_vendor_claims["aigc"] = _GENERIC_AIGC_VENDOR
# ── Local diffusion parameters (Stable Diffusion / ComfyUI) ──────
local_keys = sorted(k for k in meta if k.lower() in _LOCAL_GEN_KEYS)
+5 -1
View File
@@ -145,7 +145,11 @@ def _encode_params(ext: str) -> list[int]:
params += [sf, sf444]
return params
if ext == ".webp":
return [cv2.IMWRITE_WEBP_QUALITY, 100]
# cv2 WebP: quality 1-100 is LOSSY; a value > 100 selects LOSSLESS mode.
# "work with originals" requires lossless so a mark-removal re-encode does not
# degrade the untouched pixels the fill composites over (regression: q100
# round-tripped a random image at maxdiff ~230, q101 at 0).
return [cv2.IMWRITE_WEBP_QUALITY, 101]
return []
+139 -31
View File
@@ -93,9 +93,17 @@ def c2pa_marker_in(data: bytes) -> bool:
IPTC_AI_MARKERS: tuple[bytes, ...] = (
b"trainedAlgorithmicMedia",
b"compositeSynthetic",
b"algorithmicMedia",
b"compositeWithTrainedAlgorithmicMedia",
)
# NOTE: bare ``algorithmicMedia`` is deliberately NOT here. That IPTC digitalSourceType
# means "created purely by an algorithm, NOT from sampled training data" (procedural /
# generative-code art) -- it is NOT AI/ML generation. Real "Made with AI" labels
# (Meta / Instagram / MidJourney) use ``trainedAlgorithmicMedia``. Including the bare
# token flagged clean procedural images as AI (is_ai=high + has_invisible_target=True ->
# a diffusion scrub of clean content), contradicting the c2pa layer, which sets
# source_type without ai_source for it (tests/test_noai.py::test_plain_algorithmic_media_not_flagged_ai).
# It is not a substring of the trained/composite tokens, so its removal does not affect
# their detection.
# IPTC Photo Metadata 2025.1 (published 2025-11-27) added explicit AI-disclosure
# XMP properties in the Iptc4xmpExt namespace. Their mere presence is an AI
@@ -521,18 +529,45 @@ _SAMSUNG_GENAI_RE = re.compile(rb'genAIType"\s*:\s*(-?\d+)')
_SAMSUNG_EDITOR_MARKER = b"PhotoEditor_Re_Edit_Data"
def _read_file_tail(image_path: Path, size: int) -> bytes:
"""Return the last ``size`` bytes of the file (or the whole file if smaller)."""
try:
file_size = image_path.stat().st_size
with open(image_path, "rb") as f:
if file_size > size:
f.seek(file_size - size)
return f.read()
except OSError:
return b""
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.
See the module note above ``_SAMSUNG_GENAI_RE``: detection is empirical and
gated on the ``PhotoEditor_Re_Edit_Data`` container so an incidental
``genAIType`` token cannot false-positive.
``genAIType`` token cannot false-positive. Galaxy AI appends the marker as a
trailer AFTER the JPEG EOI, so on a multi-MB phone photo it sits past the quick-
scan window; when the head misses it, also read the file tail (else detection
and removal disagree -- the strip reads the whole file and would drop a marker
detection never reported).
"""
head = scan_head(image_path, _QUICK_SCAN_BYTES)
if _SAMSUNG_EDITOR_MARKER not in head:
data = scan_head(image_path, _QUICK_SCAN_BYTES)
if _SAMSUNG_EDITOR_MARKER not in data:
# The marker is a post-EOI trailer, so only a file LARGER than the quick-scan
# window can hide it past the head (`scan_head` already read a smaller file
# whole). Gate the extra tail read on that — `samsung_genai` is on the identify
# hot path, so a redundant 512 KB re-read per small image is not free.
try:
oversize = image_path.stat().st_size > _QUICK_SCAN_BYTES
except OSError:
oversize = False
if oversize:
data = _read_file_tail(image_path, _QUICK_SCAN_BYTES)
if _SAMSUNG_EDITOR_MARKER not in data:
return None
m = _SAMSUNG_GENAI_RE.search(head)
m = _SAMSUNG_GENAI_RE.search(data)
if m is None:
return None
return int(m.group(1)) or None
@@ -709,47 +744,87 @@ def xai_signature(image_path: Path) -> bool:
)
def _scrub_ai_exif(exif_dict: dict[str, Any]) -> list[str]:
"""Delete AI-provenance tags from a piexif dict's ``0th`` IFD, in place.
def _is_aigc_exif_value(raw: object) -> bool:
"""Whether an EXIF tag value carries a China TC260 AIGC producer/service block.
Removes (a) the xAI/Grok signature pair (``ImageDescription`` "Signature: ..."
+ UUID ``Artist``) and (b) any ``Software`` / ``Make`` / ``Artist`` /
``ImageDescription`` tag whose value carries an ``AI_GENERATOR_TOKENS`` token
(Ideogram's ``Make``, Firefly's ``Software``, etc.). Mirrors the detection in
``xai_signature`` / ``exif_generator`` so removal scrubs exactly what
``identify`` flags, while leaving genuine camera/editor EXIF intact. Returns
the names of the removed tags (for logging).
Mirrors ``aigc_label``'s EXIF path: the ``{"AIGC":{...}}`` wrapper embedded in
``UserComment`` / ``ImageDescription`` by China-served generators (Doubao's
producer schema AND Tencent Cloud's service-provider schema, both keyed under
``_TC260_FIELDS``). Gated on both the ``AIGC`` marker and a TC260 field so a
coincidental token cannot false-drop a genuine caption/comment.
"""
if not isinstance(raw, (bytes, bytearray)):
return False
if b"AIGC" not in raw:
return False
text = bytes(raw).decode("latin-1", "ignore")
return any(field in text for field in _TC260_FIELDS)
def _ai_exif_targets(loaded: dict[str, Any]) -> list[tuple[str, int, bytes, str]]:
"""The SINGLE AI-EXIF rule set, as ``(ifd_key, tag, value_bytes, name)`` entries.
Shared by both EXIF scrubbers so their coverage cannot drift: the JPEG-path
:func:`_scrub_ai_exif` pops each tag, and the ISOBMFF-path
``isobmff.blank_ai_exif_tokens`` blanks each value's bytes in place. Covers
(a) the xAI/Grok ``Signature:`` + UUID-``Artist`` pair, (b) any ``Software`` /
``Make`` / ``Artist`` / ``ImageDescription`` tag carrying an ``AI_GENERATOR_TOKENS``
token, and (c) the China TC260 ``{"AIGC":{...}}`` block in ``ImageDescription``
(0th) or ``UserComment`` (Exif). De-duplicated by ``(ifd_key, tag)`` so a value
flagged by two rules is removed and named once. Mirrors the detection in
``xai_signature`` / ``exif_generator`` / ``aigc_label``; adding a new AI EXIF
placement here reaches BOTH containers.
"""
import piexif
from remove_ai_watermarks.noai.constants import AI_GENERATOR_TOKENS
ifd = exif_dict.get("0th")
if not ifd:
return []
ifd0: dict[int, Any] = loaded.get("0th") or {}
ifde: dict[int, Any] = loaded.get("Exif") or {}
seen: set[tuple[str, int]] = set()
targets: list[tuple[str, int, bytes, str]] = []
drop: dict[int, str] = {}
def add(ifd_key: str, ifd: dict[int, Any], tag: int, name: str) -> None:
value = ifd.get(tag)
if isinstance(value, bytes) and (ifd_key, tag) not in seen:
seen.add((ifd_key, tag))
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(ifd, piexif.ImageIFD.ImageDescription), _exif_text(ifd, piexif.ImageIFD.Artist)
_exif_text(ifd0, piexif.ImageIFD.ImageDescription), _exif_text(ifd0, piexif.ImageIFD.Artist)
):
drop[piexif.ImageIFD.ImageDescription] = "ImageDescription"
drop[piexif.ImageIFD.Artist] = "Artist"
# (b) Known AI generator token in any of the text tags.
add("0th", ifd0, piexif.ImageIFD.ImageDescription, "ImageDescription")
add("0th", ifd0, piexif.ImageIFD.Artist, "Artist")
# (b) known AI generator token in a 0th text tag.
for tag, name in (
(piexif.ImageIFD.Software, "Software"),
(piexif.ImageIFD.Make, "Make"),
(piexif.ImageIFD.Artist, "Artist"),
(piexif.ImageIFD.ImageDescription, "ImageDescription"),
):
if any(token in _exif_text(ifd, tag).lower() for token in AI_GENERATOR_TOKENS):
drop[tag] = name
if any(token in _exif_text(ifd0, tag).lower() for token in AI_GENERATOR_TOKENS):
add("0th", ifd0, tag, name)
# (c) TC260 AIGC block in ImageDescription (0th) or UserComment (Exif sub-IFD).
if _is_aigc_exif_value(ifd0.get(piexif.ImageIFD.ImageDescription)):
add("0th", ifd0, piexif.ImageIFD.ImageDescription, "ImageDescription")
if _is_aigc_exif_value(ifde.get(piexif.ExifIFD.UserComment)):
add("Exif", ifde, piexif.ExifIFD.UserComment, "UserComment")
for tag in drop:
ifd.pop(tag, None)
return list(drop.values())
return targets
def _scrub_ai_exif(exif_dict: dict[str, Any]) -> list[str]:
"""Delete the AI-provenance EXIF tags (`_ai_exif_targets`) from a piexif dict's
``0th`` / ``Exif`` IFDs in place; return the removed tag names (for logging).
Genuine camera/editor EXIF is left intact."""
removed: list[str] = []
for ifd_key, tag, _value, name in _ai_exif_targets(exif_dict):
ifd = exif_dict.get(ifd_key)
if ifd is not None:
ifd.pop(tag, None)
removed.append(name)
return removed
def get_ai_metadata(image_path: Path) -> dict[str, str]:
@@ -879,17 +954,50 @@ def _jpeg_app_carries_ai(marker: int, payload: bytes) -> bool:
if marker == 0xEB: # APP11: C2PA / JUMBF manifest
return c2pa_marker_in(payload) or b"jumb" in payload[:256].lower()
if marker == 0xE1 and payload.startswith(b"http://ns.adobe.com/xap/"): # APP1 XMP
return c2pa_marker_in(payload) or any(m in payload for m in AIGC_MARKERS)
return (
c2pa_marker_in(payload)
or any(m in payload for m in AIGC_MARKERS)
or any(m in payload for m in IPTC_AI_MARKERS) # digitalSourceType in XMP, not only APP13
or any(m in payload for m in IPTC_AI_FIELD_MARKERS) # IPTC 2025.1 AI-disclosure fields
)
if marker == 0xED: # APP13: Photoshop / IPTC
return any(m in payload for m in IPTC_AI_MARKERS) or any(m in payload for m in IPTC_AI_FIELD_MARKERS)
# A bare / wrapped China TC260 AIGC block (``AIGC{...}`` or ``{"AIGC":{...}}``) that
# some China-served generators glue into a non-standard APP segment near the JFIF
# header. ``aigc_label`` detects it anywhere in the scan head, so removal must drop
# the carrying segment too (detection<->removal parity). Skip APP1-EXIF (0xE1
# ``Exif``): its camera tags are scrubbed tag-by-tag via piexif, and the AIGC-in-
# UserComment/ImageDescription placement is handled there, so it must not be dropped
# wholesale here.
if 0xE0 <= marker <= 0xEF and not (marker == 0xE1 and payload.startswith(b"Exif")):
return _is_aigc_exif_value(payload)
return False
def _strip_samsung_trailer(scan_and_tail: bytes) -> bytes:
"""Drop a Samsung Galaxy AI editing trailer appended AFTER the JPEG EOI.
Galaxy AI records its ``PhotoEditor_Re_Edit_Data`` (``genAIType``) blob as a
proprietary trailer past the final ``FFD9`` end-of-image, so the verbatim
scan copy in :func:`_strip_jpeg_metadata_lossless` would carry it through. If
the marker is present in the post-EOI trailer, truncate at EOI (the coded scan
is untouched, pixels stay bit-identical). A JPEG with no such trailer -- or a
non-Samsung trailer (e.g. an MPF multi-picture block) -- is returned unchanged.
"""
if _SAMSUNG_EDITOR_MARKER not in scan_and_tail:
return scan_and_tail
eoi = scan_and_tail.rfind(b"\xff\xd9")
if eoi == -1 or _SAMSUNG_EDITOR_MARKER not in scan_and_tail[eoi:]:
return scan_and_tail # marker not in the post-EOI trailer; leave the scan alone
return scan_and_tail[: eoi + 2]
def _strip_jpeg_metadata_lossless(source_path: Path, output_path: Path) -> bool:
"""Remove AI metadata from a JPEG WITHOUT re-encoding the DCT scan, so the pixels
stay bit-identical (the point of "work with originals" -- a metadata strip must not
degrade the image). Walks the marker segments up to SOS, drops the AI-bearing APP
segments (:func:`_jpeg_app_carries_ai`), copies the entropy-coded scan verbatim,
segments (:func:`_jpeg_app_carries_ai`), copies the entropy-coded scan verbatim
(minus a Samsung Galaxy AI trailer past EOI, via :func:`_strip_samsung_trailer`),
then scrubs AI EXIF tags in place via piexif (which rewrites only the APP1 EXIF,
leaving genuine camera EXIF and the scan untouched). Returns False if the bytes are
not a parseable JPEG, so the caller falls back to the near-lossless PIL re-save."""
@@ -905,7 +1013,7 @@ def _strip_jpeg_metadata_lossless(source_path: Path, output_path: Path) -> bool:
return False # malformed marker boundary: defer to the PIL re-encode fallback
marker = data[i + 1]
if marker in (0xDA, 0xD9): # SOS / EOI -> the coded scan follows; copy verbatim
out += data[i:]
out += _strip_samsung_trailer(data[i:])
break
if 0xD0 <= marker <= 0xD7 or marker == 0x01: # standalone markers carry no length
out += data[i : i + 2]
+5 -2
View File
@@ -374,12 +374,15 @@ def _populate_registry_fields(buf: bytes, c2pa_info: dict[str, Any]) -> bool:
c2pa_info["source_type"] = "trainedAlgorithmicMedia (AI-generated)"
c2pa_info["ai_source_kind"] = "generated"
ai_source = True
elif b"algorithmicMedia" in buf:
c2pa_info["source_type"] = "algorithmicMedia"
elif b"compositeWithTrainedAlgorithmicMedia" in buf:
# Checked BEFORE bare ``algorithmicMedia``: a manifest can carry both tokens
# (an AI-enhanced composite with a procedural ingredient), and the bare-token
# branch would otherwise fire first and misclassify the AI composite as non-AI.
c2pa_info["source_type"] = "compositeWithTrainedAlgorithmicMedia (AI-enhanced)"
c2pa_info["ai_source_kind"] = "enhanced"
ai_source = True
elif b"algorithmicMedia" in buf:
c2pa_info["source_type"] = "algorithmicMedia"
# SynthID pixel-watermark proxy: a C2PA manifest from a SynthID-using
# vendor (Google/OpenAI) on AI-generated content implies an invisible
+22 -25
View File
@@ -246,42 +246,39 @@ def blank_ai_exif_tokens(data: bytes) -> tuple[bytes, int]:
``remove_ai_metadata`` (a documented gap). This locates EXIF TIFF blocks by
their byte-order header, **validates each with piexif** (so a coincidental
II/MM run in pixel data is ignored -- it will not parse as a TIFF IFD), and
overwrites any value carrying an ``AI_GENERATOR_TOKENS`` token with spaces of
the SAME length. Because the replacement is same-length, every box size and
``iloc`` offset stays valid and the coded image is untouched -- only the AI tag
content is destroyed; camera/editor EXIF without an AI token is left intact
(mirrors ``metadata._scrub_ai_exif`` and ``blank_ai_xmp_packets``).
overwrites any AI value with spaces of the SAME length. Because the replacement
is same-length, every box size and ``iloc`` offset stays valid and the coded image
is untouched -- only the AI tag content is destroyed; camera/editor EXIF without an
AI token is left intact. This mirrors ``metadata._scrub_ai_exif`` in what it removes
-- generator tokens (``Software``/``Make``/``Artist``/``ImageDescription``), the
China TC260 ``{"AIGC":{...}}`` block (``ImageDescription``/``UserComment``), and the
xAI/Grok ``Signature:`` + UUID-``Artist`` pair -- since on the ISOBMFF path this is
the ONLY EXIF scrubber (``_scrub_ai_exif`` never runs there), so without parity a
HEIC/AVIF AIGC/xAI tag is detected but not removed.
"""
import piexif
from remove_ai_watermarks.noai.constants import AI_GENERATOR_TOKENS
# The AI-EXIF rule set is defined ONCE in metadata._ai_exif_targets and shared by both
# EXIF scrubbers (the JPEG _scrub_ai_exif pops the tag; here we blank the value bytes),
# so their coverage cannot drift. Imported lazily to avoid import-order coupling with
# metadata (which imports this module); a deliberate cross-module use, not an API leak.
from remove_ai_watermarks.metadata import _ai_exif_targets # pyright: ignore[reportPrivateUsage]
ai_tags = (
piexif.ImageIFD.Software,
piexif.ImageIFD.Make,
piexif.ImageIFD.Artist,
piexif.ImageIFD.ImageDescription,
)
out = bytearray(data)
blanked = 0
for header in _TIFF_HEADERS:
pos = data.find(header)
while pos != -1:
window = bytes(out[pos : pos + _EXIF_WINDOW])
ifd: dict[int, Any] = {}
try:
ifd = piexif.load(window).get("0th", {})
loaded: dict[str, Any] = piexif.load(window)
except Exception:
ifd = {}
for tag in ai_tags:
value = ifd.get(tag)
if not isinstance(value, bytes):
continue
if any(token in value.decode("latin1", "replace").lower() for token in AI_GENERATOR_TOKENS):
# Blank the value bytes in place, within this EXIF block only.
vpos = out.find(value, pos, pos + _EXIF_WINDOW)
if vpos != -1:
out[vpos : vpos + len(value)] = b" " * len(value)
blanked += 1
loaded = {}
for _ifd_key, _tag, value, _name in _ai_exif_targets(loaded):
# Blank the value bytes in place, within this EXIF block only.
vpos = out.find(value, pos, pos + _EXIF_WINDOW)
if vpos != -1:
out[vpos : vpos + len(value)] = b" " * len(value)
blanked += 1
pos = data.find(header, pos + len(header))
return bytes(out), blanked
@@ -964,8 +964,9 @@ def remove_watermark(
"""Convenience function to remove watermark from an image.
``strength=None`` lets the profile pick its vendor-adaptive default
(0.20 OpenAI / 0.30 Google / 0.30 unknown, from the C2PA SynthID proxy on the
input; same ladder for the controlnet and sdxl pipelines). Pass a value to override.
(0.10 OpenAI / 0.15 Google / 0.15 unknown, from the C2PA SynthID proxy on the
input; same ladder for the controlnet and sdxl pipelines -- the single source of
truth is ``watermark_profiles.py``). Pass a value to override.
``region=(x, y, w, h)`` restricts the regeneration to that box and preserves the
real photo elsewhere -- for AI-enhanced composites (see
+73
View File
@@ -573,6 +573,16 @@ class TestMetadataCommand:
assert result.exit_code == 0
assert "stripped" in result.output
def test_metadata_remove_in_place(self, runner, tmp_png_with_ai_metadata):
"""With ``-o`` omitted, the strip overwrites the source in place (default
output_path=None). Previously every test passed an explicit ``-o``."""
from remove_ai_watermarks.metadata import has_ai_metadata
assert has_ai_metadata(tmp_png_with_ai_metadata) # precondition
result = runner.invoke(main, ["metadata", str(tmp_png_with_ai_metadata), "--remove"])
assert result.exit_code == 0, result.output
assert not has_ai_metadata(tmp_png_with_ai_metadata) # source overwritten, AI metadata gone
class TestIdentifyCommand:
"""Tests for the 'identify' subcommand."""
@@ -600,6 +610,18 @@ class TestIdentifyCommand:
assert "AI-generated" in result.output
assert "Stable Diffusion" in result.output
def test_identify_reports_generated_source_kind(self, runner):
"""The C2PA trainedAlgorithmicMedia source type sharpens the verdict to
'AI-generated (fully synthetic)' at the CLI (the ai_source_kind branch)."""
from pathlib import Path
sample = Path(__file__).resolve().parent.parent / "data" / "samples" / "chatgpt-1.png"
if not sample.exists():
pytest.skip("chatgpt sample not present")
result = runner.invoke(main, ["identify", str(sample), "--no-visible"])
assert result.exit_code == 0
assert "AI-generated (fully synthetic)" in result.output
def test_identify_json_is_valid(self, runner, tmp_png_with_ai_metadata):
result = runner.invoke(main, ["identify", str(tmp_png_with_ai_metadata), "--no-visible", "--json"])
assert result.exit_code == 0
@@ -774,6 +796,34 @@ class TestBatchCommand:
expected_dir = tmp_path / "input_clean"
assert expected_dir.exists()
def test_batch_errors_exit_nonzero(self, runner, tmp_path):
"""Regression: batch used to always exit 0 even when every image errored,
hiding failure from a wrapping service. A corrupt image must yield a non-zero
exit and an error count."""
input_dir = tmp_path / "input"
input_dir.mkdir()
(input_dir / "corrupt.png").write_bytes(b"this is not a PNG at all" * 50)
result = runner.invoke(main, ["batch", str(input_dir), "--mode", "visible"])
assert result.exit_code != 0, result.output
assert "error" in result.output.lower()
def test_batch_invisible_gpu_missing_writes_output_and_exits_nonzero(self, runner, tmp_path):
"""Regression: batch --mode invisible with a signal-bearing image but no GPU
deps used to write NO output for that image and still exit 0, silently dropping
the files that most needed processing. It must now copy the input through (so the
output dir is complete), warn about the retained SynthID watermark, and exit
non-zero -- mirroring the single ``all`` command."""
input_dir = _make_batch_dir_with_metadata(tmp_path, count=3) # SD params = invisible signal
output_dir = tmp_path / "output"
with patch("remove_ai_watermarks.invisible_engine.is_available", return_value=False):
result = runner.invoke(
main,
["batch", str(input_dir), "-o", str(output_dir), "--mode", "invisible"],
)
assert result.exit_code != 0, result.output
assert "NOT removed" in result.output
assert len(list(output_dir.glob("*.png"))) == 3 # every input copied through, none dropped
class TestGpuHintMarkup:
"""The GPU-extra install hint must reach the user with the ``[gpu]`` token
@@ -887,3 +937,26 @@ def test_visible_backend_runtime_error_exits_cleanly(runner, tmp_path, monkeypat
result = runner.invoke(main, ["visible", str(doubao), "-o", str(out), "--backend", "migan"])
assert result.exit_code == 1
assert not isinstance(result.exception, RuntimeError), "RuntimeError leaked as a traceback"
@pytest.mark.parametrize(
("name", "content"),
[
("empty.png", b""),
("notimage.jpg", b"plain text, not an image at all " * 20),
("truncated.png", b"\x89PNG\r\n\x1a\n" + b"\x00" * 40),
],
)
@pytest.mark.parametrize("cmd", [["metadata", "--remove"], ["visible", "--backend", "cv2"]])
def test_unreadable_input_exits_cleanly(runner, tmp_path, name, content, cmd):
"""Regression: a corrupt / empty / non-image file (real prod uploads include
truncated files) must produce a clean 'Error: cannot read/process' + exit 1, NOT a
raw PIL.UnidentifiedImageError / OSError / ValueError traceback. Found by the runtime
mode fuzz across metadata --remove and visible."""
bad = tmp_path / name
bad.write_bytes(content)
out = tmp_path / "out.png"
result = runner.invoke(main, [cmd[0], str(bad), "-o", str(out), *cmd[1:]])
assert result.exit_code == 1, result.output
assert isinstance(result.exception, SystemExit), f"leaked a raw traceback: {result.exception!r}"
assert "Error" in result.output
+26
View File
@@ -195,3 +195,29 @@ class TestDegenerateAndChannelInputs:
bgra = np.zeros((2048, 2048, 4), np.uint8)
mask = eng.footprint_mask(bgra, force=True)
assert mask is None or mask.shape == (2048, 2048)
def test_template_match_score_guards_return_zero(self):
# Guards return 0.0 (never a false positive) for a mask that cannot hold a
# glyph: empty, narrower than min_gw, or shorter than the 4-px floor.
assert _template_match_score(np.zeros((0, 5), np.uint8), 1000) == 0.0
assert _template_match_score(np.zeros((10, 3), np.uint8), 1000) == 0.0 # width-1 < min_gw
assert _template_match_score(np.zeros((3, 200), np.uint8), 1000) == 0.0 # height-1 < 4
@pytest.mark.parametrize("shape", [(20, 20, 3), (10, 400, 3), (400, 10, 3), (1, 1, 3), (2000, 2000, 3)])
def test_locate_box_stays_in_bounds(self, shape):
"""locate() must clamp its geometry box inside the image for ANY size/aspect --
wide-short, tall-narrow, 1x1, huge -- for both bottom corners (br + bl)."""
from remove_ai_watermarks._text_mark_engine import TextMarkEngine
from remove_ai_watermarks.doubao_engine import _CONFIG as BR_CONFIG
from remove_ai_watermarks.samsung_engine import _CONFIG as BL_CONFIG
h, w = shape[:2]
img = np.zeros(shape, np.uint8)
for cfg in (BR_CONFIG, BL_CONFIG):
loc = TextMarkEngine(cfg).locate(img)
assert loc.x >= 0
assert loc.y >= 0
assert loc.x + loc.w <= w
assert loc.y + loc.h <= h
assert loc.w > 0
assert loc.h > 0
+26
View File
@@ -1,4 +1,5 @@
import numpy as np
import pytest
from remove_ai_watermarks.humanizer import apply_analog_humanizer, unsharp_mask
@@ -72,6 +73,15 @@ def test_chromatic_shift_does_not_wrap_opposite_edge():
assert result[:, -shift:, 2].min() > 195
@pytest.mark.parametrize(("width", "shift"), [(1, 1), (3, 3), (3, 5), (5, 10)])
def test_chromatic_shift_wider_than_image_no_crash(width: int, shift: int):
"""Regression: a chromatic_shift >= image width left the edge-replication slices
empty and crashed the broadcast (ValueError). The shift must clamp to width-1."""
img = np.full((4, width, 3), 120, np.uint8)
result = apply_analog_humanizer(img, grain_intensity=0.0, chromatic_shift=shift)
assert result.shape == img.shape
def test_unsharp_disabled_returns_unchanged_copy():
img = np.full((20, 20, 3), 128, dtype=np.uint8)
img[10, 10] = [100, 150, 200]
@@ -147,3 +157,19 @@ class TestAdaptivePolish:
a = adaptive_polish(soft, reference, seed=7)
b = adaptive_polish(soft, reference, seed=7)
assert np.array_equal(a, b)
def test_all_edges_reference_grain_mask_near_zero(self):
# An all-high-frequency target: _smooth_grain_mask suppresses edges, so the grain
# mask is ~all-zero (grain adds nothing) -- adaptive_polish must still return a
# valid same-shape image, not crash on the empty-mask branch.
import cv2
from remove_ai_watermarks.humanizer import _smooth_grain_mask, adaptive_polish
rng = np.random.default_rng(5)
edges = rng.integers(0, 256, (120, 120, 3), dtype=np.uint8) # all high-frequency
assert _smooth_grain_mask(edges).mean() < _smooth_grain_mask(np.full((120, 120, 3), 128, np.uint8)).mean()
soft = cv2.GaussianBlur(edges, (0, 0), sigmaX=3.0)
out = adaptive_polish(soft, edges, seed=0)
assert out.shape == soft.shape
assert out.dtype == np.uint8
+34
View File
@@ -926,6 +926,15 @@ class TestVendorOf:
assert _vendor_of("a regular photo") is None
assert _vendor_of(None) is None
def test_registered_vendors_normalize(self):
# Regression: these registered C2PA vendors returned None, so their claims never
# entered clash detection (a coverage hole). They now normalize to one origin.
assert _vendor_of("ByteDance (Doubao / Jimeng / Volcano Engine)") == "ByteDance"
assert _vendor_of("Dreamina/1.2") == "ByteDance"
assert _vendor_of("Canva (Magic Media)") == "Canva"
assert _vendor_of("Black Forest Labs (FLUX)") == "Black Forest Labs"
assert _vendor_of("Eleven Labs Inc.") == "ElevenLabs"
class TestIntegrityClashesHelper:
def test_two_ai_vendors_clash(self):
@@ -949,6 +958,31 @@ class TestIntegrityClashesHelper:
== []
)
def test_bytedance_c2pa_plus_own_aigc_no_clash(self):
# A legit ByteDance/Doubao image carries BOTH a ByteDance C2PA manifest and its
# own China TC260 AIGC label. The label is ByteDance's own regulatory stamp, so
# it must be attributed to ByteDance and NOT read as a competing origin.
assert (
_integrity_clashes({"c2pa": "ByteDance", "aigc": "China AIGC (TC260)"}, None, camera_has_ai_marker=True)
== []
)
def test_foreign_vendor_plus_aigc_still_clashes(self):
# But a NON-Chinese vendor's C2PA next to a China TC260 label names two different
# origins -- a laundering tell that must still fire (the generic label stays generic).
clashes = _integrity_clashes({"c2pa": "OpenAI", "aigc": "China AIGC (TC260)"}, None, camera_has_ai_marker=True)
assert len(clashes) == 1
assert "Conflicting AI-origin" in clashes[0]
def test_bytedance_c2pa_plus_foreign_generator_clashes(self):
# Coverage win: a transplanted ByteDance C2PA manifest next to an independent
# foreign generator stamp is a laundering tell that went undetected before
# ByteDance was added to _vendor_of.
clashes = _integrity_clashes({"c2pa": "ByteDance", "exif_generator": "OpenAI"}, None, camera_has_ai_marker=True)
assert len(clashes) == 1
assert "ByteDance" in clashes[0]
assert "OpenAI" in clashes[0]
def test_manifest_vendor_vs_independent_signal_clashes(self):
# A vendor named only inside the manifest still clashes with a genuinely
# independent stamp (here an EXIF/XMP generator tag) naming a third vendor.
+11
View File
@@ -183,6 +183,17 @@ class TestQualityPreservingWrite:
assert back is not None
assert float(np.abs(img.astype(int) - back.astype(int)).mean()) < 1.0
def test_webp_written_lossless(self, tmp_path: Path) -> None:
# Regression: cv2 WebP quality 1-100 is LOSSY; lossless needs > 100. A
# mark-removal .webp re-encode must NOT degrade the untouched pixels, so
# a full-frame round-trip of random data must be bit-identical.
img = np.random.default_rng(0).integers(0, 256, (80, 80, 3), dtype=np.uint8)
p = tmp_path / "x.webp"
assert image_io.imwrite(p, img) is True
back = image_io.imread(p)
assert back is not None
assert np.array_equal(back, img), "WebP re-encode was lossy"
@pytest.mark.skipif(not _heif_writable("HEIF"), reason="no HEIC encoder in this env")
def test_heic_write_roundtrips(self, tmp_path: Path) -> None:
# cv2 cannot encode HEIC (used to raise); imwrite must route through Pillow.
+22
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
from types import SimpleNamespace
import pytest
from PIL import Image
from remove_ai_watermarks.invisible_engine import InvisibleEngine, _target_size, is_available
@@ -195,3 +196,24 @@ class TestEsrganUpscale:
out = InvisibleEngine._esrgan_upscale(self._fake_engine(), img, (512, 341))
assert out.size == (512, 341)
assert np.array_equal(np.asarray(out), np.asarray(img.resize((512, 341), Image.Resampling.LANCZOS)))
class TestCannyControlImage:
"""The ControlNet canny conditioning image builder (pure cv2/numpy; behind the gpu
extra since it lives on WatermarkRemover). Skips when torch/diffusers are absent."""
def test_edge_map_is_3channel_rgb(self):
if not is_available():
pytest.skip("gpu extra (torch/diffusers) not installed")
import numpy as np
from remove_ai_watermarks.noai.watermark_remover import WatermarkRemover
rng = np.random.default_rng(0)
img = Image.fromarray(rng.integers(0, 256, (64, 80, 3), dtype=np.uint8))
# The method uses no instance state, so call it unbound with a dummy self.
out = WatermarkRemover._build_canny_control_image(None, img) # type: ignore[arg-type]
arr = np.array(out)
assert out.mode == "RGB"
assert arr.shape == (64, 80, 3)
assert arr.max() <= 255
+160 -1
View File
@@ -179,6 +179,53 @@ class TestHasAiMetadata:
assert np.array_equal(before, after), f"{name}: pixels changed (DCT was re-encoded)"
assert not has_ai_metadata(out), f"{name}: AI metadata survived the strip"
@staticmethod
def _xmp_iptc_jpeg(tmp_path: Path, name: str, marker: bytes) -> Path:
"""A real (decodable) JPEG carrying the IPTC AI marker in a well-formed APP1
XMP segment -- the layout Instagram/Facebook/X and MidJourney/Meta use, where
``digitalSourceType`` lives in XMP rather than the APP13 IPTC-IIM record.
Synthetic (no corpus): a solid cv2 JPEG with the APP1 spliced in after SOI."""
import cv2
import numpy as np
real = tmp_path / f"real-{name}"
cv2.imwrite(str(real), np.full((32, 32, 3), 200, np.uint8), [cv2.IMWRITE_JPEG_QUALITY, 100])
data = real.read_bytes()
xmp = (
b"http://ns.adobe.com/xap/1.0/\x00"
b'<x:xmpmeta xmlns:x="adobe:ns:meta/"><rdf:RDF><rdf:Description>'
b"<Iptc4xmpExt:DigitalSourceType>http://cv.iptc.org/newscodes/digitalsourcetype/"
+ marker
+ b"</Iptc4xmpExt:DigitalSourceType></rdf:Description></rdf:RDF></x:xmpmeta>"
)
seg = b"\xff\xe1" + (len(xmp) + 2).to_bytes(2, "big") + xmp
path = tmp_path / name
path.write_bytes(data[:2] + seg + data[2:]) # splice APP1 right after SOI
return path
@pytest.mark.parametrize("marker", [b"trainedAlgorithmicMedia", b"AISystemUsed"])
def test_jpeg_strip_removes_iptc_marker_in_xmp(self, tmp_path: Path, marker: bytes):
"""Regression: the lossless JPEG strip must drop an AI-bearing APP1 XMP packet
when the AI signal is an IPTC ``digitalSourceType`` / 2025.1 field, not only a
C2PA or China-AIGC token. Before the fix these survived because the APP1 branch
of ``_jpeg_app_carries_ai`` checked only c2pa + AIGC markers, leaving the
Instagram/MidJourney/Meta 'Made with AI' XMP intact. Pixels stay bit-identical."""
import numpy as np
from remove_ai_watermarks import image_io
from remove_ai_watermarks.metadata import remove_ai_metadata
src = self._xmp_iptc_jpeg(tmp_path, "iptc-xmp.jpg", marker)
assert has_ai_metadata(src) # detected before
before = image_io.imread(str(src))
out = tmp_path / "clean.jpg"
remove_ai_metadata(src, out)
after = image_io.imread(str(out))
assert before is not None
assert after is not None
assert np.array_equal(before, after), "pixels changed: the DCT scan was re-encoded"
assert not has_ai_metadata(out), "IPTC AI marker in XMP survived the strip"
class TestC2paMarkerIn:
"""The C2PA presence check requires a JUMBF wrapper or the C2PA uuid box, so
@@ -235,6 +282,59 @@ class TestSamsungGenai:
p = self._samsung_jpeg(tmp_path, "stray.jpg", b'some other blob "genAIType":1 elsewhere')
assert samsung_genai(p) is None
def test_remove_strips_post_eoi_trailer(self, tmp_path: Path):
"""Regression: Galaxy AI appends ``PhotoEditor_Re_Edit_Data`` as a trailer AFTER
the JPEG EOI, so the verbatim scan copy in the lossless strip carried it through
(survived 0/8 on the corpus). The strip now truncates a Samsung AI trailer at EOI;
pixels stay bit-identical and a non-Samsung trailer is preserved."""
import cv2
import numpy as np
from remove_ai_watermarks import image_io
from remove_ai_watermarks.metadata import remove_ai_metadata
real = tmp_path / "real.jpg"
cv2.imwrite(str(real), np.full((32, 32, 3), 180, np.uint8), [cv2.IMWRITE_JPEG_QUALITY, 100])
src = tmp_path / "galaxy.jpg"
src.write_bytes(real.read_bytes() + b'PhotoEditor_Re_Edit_Data{"genAIType":1}')
assert samsung_genai(src) == 1 # detected before
before = image_io.imread(str(src))
out = tmp_path / "clean.jpg"
remove_ai_metadata(src, out)
after = image_io.imread(str(out))
assert before is not None
assert after is not None
assert np.array_equal(before, after), "pixels changed: the DCT scan was re-encoded"
assert samsung_genai(out) is None, "Samsung genAIType trailer survived the strip"
def test_non_samsung_trailer_preserved(self, tmp_path: Path):
"""A benign post-EOI trailer (e.g. an MPF block) must NOT be truncated."""
import cv2
import numpy as np
from remove_ai_watermarks.metadata import _strip_samsung_trailer
real = tmp_path / "real.jpg"
cv2.imwrite(str(real), np.full((16, 16, 3), 90, np.uint8))
tail = real.read_bytes() + b"MPF-benign-trailer-bytes"
assert _strip_samsung_trailer(tail) == tail
def test_detects_trailer_past_scan_window(self, tmp_path: Path):
"""Regression: the marker is a trailer AFTER the JPEG EOI, so on a multi-MB
photo it sits past the 512 KB quick-scan window. Detection must read the file
tail too, else it disagrees with removal (which reads the whole file). A random
1400x1400 q100 JPEG exceeds 512 KB; the marker is only in its post-EOI tail."""
import cv2
import numpy as np
real = tmp_path / "big.jpg"
big = np.random.default_rng(0).integers(0, 256, (1400, 1400, 3), dtype=np.uint8)
cv2.imwrite(str(real), big, [cv2.IMWRITE_JPEG_QUALITY, 100])
assert real.stat().st_size > 512 * 1024 # trailer will be past the quick-scan window
p = tmp_path / "galaxy_big.jpg"
p.write_bytes(real.read_bytes() + b'PhotoEditor_Re_Edit_Data{"genAIType":1}')
assert samsung_genai(p) == 1
def test_clean_image_is_none(self, tmp_clean_png):
assert samsung_genai(tmp_clean_png) is None
@@ -300,7 +400,6 @@ class TestGetAiMetadataRealSample:
[
b"trainedAlgorithmicMedia",
b"compositeSynthetic",
b"algorithmicMedia",
b"compositeWithTrainedAlgorithmicMedia",
],
)
@@ -311,6 +410,21 @@ def test_has_ai_metadata_detects_each_iptc_marker(tmp_path: Path, marker: bytes)
assert has_ai_metadata(path)
def test_bare_algorithmic_media_not_flagged_ai(tmp_path: Path):
"""Regression: the IPTC ``algorithmicMedia`` digitalSourceType is PROCEDURAL (an
algorithm not trained on sampled data), NOT AI/ML generation. It must NOT be flagged
-- flagging it made identify assert is_ai=high + has_invisible_target=True, which
would trigger a diffusion scrub of clean procedural content. It is a distinct token
from ``trainedAlgorithmicMedia``, so real 'Made with AI' labels are unaffected."""
path = tmp_path / "proc.jpg"
path.write_bytes(
b"\xff\xd8\xff\xe1<x:xmpmeta><Iptc4xmpExt:DigitalSourceType>"
b"http://cv.iptc.org/newscodes/digitalsourcetype/algorithmicMedia"
b"</Iptc4xmpExt:DigitalSourceType></x:xmpmeta>\xff\xd9"
)
assert not has_ai_metadata(path)
# ── SynthID-source detection (metadata proxy) ────────────────────────
@@ -920,6 +1034,18 @@ class TestAIGCLabel:
def test_has_ai_metadata_detects_raw_json_exif_form(self, tmp_path: Path):
assert has_ai_metadata(self._aigc_exif_jpeg(tmp_path))
def test_remove_strips_raw_json_exif_form(self, tmp_path: Path):
"""Regression: the TC260 AIGC ``{"AIGC":{...}}`` block Doubao embeds in EXIF
UserComment must be scrubbed on removal. Before the fix it survived because
``_scrub_ai_exif`` only touched Software/Make/Artist/ImageDescription in the
0th IFD, never UserComment in the Exif sub-IFD."""
from remove_ai_watermarks.metadata import aigc_label, remove_ai_metadata
out = tmp_path / "clean.jpg"
remove_ai_metadata(self._aigc_exif_jpeg(tmp_path), out)
assert aigc_label(out) is None
assert not has_ai_metadata(out)
def _aigc_bare_jpeg(self, tmp_path: Path, producer: str = "00119144030008867405X210002") -> Path:
"""Some China-served generators glue the TC260 label straight to its JSON
as a bare ``AIGC{...}`` blob inside a JPEG APP segment (no ``"AIGC":``
@@ -944,6 +1070,18 @@ class TestAIGCLabel:
def test_has_ai_metadata_detects_bare_aigc_jpeg_form(self, tmp_path: Path):
assert has_ai_metadata(self._aigc_bare_jpeg(tmp_path))
def test_remove_strips_bare_aigc_jpeg_form(self, tmp_path: Path):
"""Regression: a bare ``AIGC{...}`` blob in a non-standard JPEG APP segment
(APP9 here) is detected by aigc_label, so removal must drop that segment too.
Before the fix ``_jpeg_app_carries_ai`` only inspected APP11/APP1-XMP/APP13, so
the blob survived the lossless strip (detection<->removal parity break)."""
from remove_ai_watermarks.metadata import aigc_label, remove_ai_metadata
out = tmp_path / "clean.jpg"
remove_ai_metadata(self._aigc_bare_jpeg(tmp_path), out)
assert aigc_label(out) is None
assert not has_ai_metadata(out)
def test_bare_aigc_without_tc260_field_ignored(self, tmp_path: Path):
"""A bare ``AIGC{...}`` blob with no TC260 field must not false-positive."""
from remove_ai_watermarks.metadata import aigc_label
@@ -1192,6 +1330,27 @@ class TestLateProvenanceBox:
p.write_bytes(b"\x89PNG\r\n\x1a\n not an isobmff file")
assert scan_c2pa_region(p) == b""
def test_scan_c2pa_region_reads_largesize_uuid(self, tmp_path: Path):
"""A 64-bit largesize (size32 == 1) uuid box must be walked and collected."""
import struct
from remove_ai_watermarks.noai.isobmff import scan_c2pa_region
payload = b"LARGESIZE-C2PA-MANIFEST"
total = 16 + len(payload) # 4 (size32=1) + 4 (type) + 8 (largesize) + payload
uuid_box = struct.pack(">I", 1) + b"uuid" + struct.pack(">Q", total) + payload
p = tmp_path / "large.mp4"
p.write_bytes(_MP4_FTYP + uuid_box)
assert payload in scan_c2pa_region(p)
def test_scan_c2pa_region_caps_at_max_total(self, tmp_path: Path):
"""The collected payload is bounded by ``max_total`` (never unbounded)."""
from remove_ai_watermarks.noai.isobmff import scan_c2pa_region
p = tmp_path / "big.mp4"
p.write_bytes(_MP4_FTYP + _box(b"uuid", b"A" * 5000))
assert len(scan_c2pa_region(p, max_total=1000)) <= 1000
def test_front_placed_manifest_still_detected(self, tmp_path: Path):
# Regression: a faststart MP4 (manifest before mdat) is unaffected.
from remove_ai_watermarks.metadata import C2PA_UUID
+138
View File
@@ -321,6 +321,18 @@ class TestC2PADigitalSourceType:
assert "compositeWithTrainedAlgorithmicMedia" in info["source_type"]
assert "synthid_watermark" in info # AI-enhanced + OpenAI issuer
def test_composite_and_bare_algorithmic_cooccur_is_ai(self):
"""Regression: a manifest carrying BOTH ``compositeWithTrainedAlgorithmicMedia``
(AI-enhanced) and a bare procedural ``algorithmicMedia`` token must classify as
AI-enhanced. Before the reorder the bare-token elif fired first and returned
non-AI, dropping the composite AI signal (a false negative)."""
from remove_ai_watermarks.noai.c2pa import _populate_registry_fields
info: dict = {}
_populate_registry_fields(b"x compositeWithTrainedAlgorithmicMedia x algorithmicMedia x", info)
assert info.get("ai_source_kind") == "enhanced"
assert "compositeWithTrainedAlgorithmicMedia" in info["source_type"]
# ── ISOBMFF (AVIF / HEIF / JPEG-XL container stripping) ──────────────
@@ -400,6 +412,33 @@ class TestISOBMFF:
assert ifd[piexif.ImageIFD.Software].strip() == b""
assert ifd[piexif.ImageIFD.Make] == b"Canon"
def test_blank_aigc_block_in_exif(self):
"""Parity with the JPEG path: the China TC260 ``{"AIGC":{...}}`` block in EXIF
ImageDescription must be blanked on the ISOBMFF path too -- ``blank_ai_exif_tokens``
is the ONLY EXIF scrubber for HEIC/AVIF (``_scrub_ai_exif`` never runs there)."""
import piexif
aigc = b'{"AIGC":{"Label":"1","ContentProducer":"00119144030008867405X210002","ProduceID":"abc"}}'
data = self._avif_with_exif({piexif.ImageIFD.ImageDescription: aigc, piexif.ImageIFD.Make: b"Canon"})
out, blanked = blank_ai_exif_tokens(data)
assert blanked >= 1
assert len(out) == len(data) # same length -> box sizes / iloc stay valid
assert b'"AIGC"' not in out # TC260 block destroyed
assert b"Canon" in out # camera tag preserved
def test_blank_xai_signature_pair_in_exif(self):
"""Parity: the xAI/Grok ``Signature:`` blob + UUID ``Artist`` pair in EXIF is
dropped together on the ISOBMFF path too."""
import piexif
sig = b"Signature: " + b"A" * 80
art = b"12345678-1234-1234-1234-123456789012"
data = self._avif_with_exif({piexif.ImageIFD.ImageDescription: sig, piexif.ImageIFD.Artist: art})
out, blanked = blank_ai_exif_tokens(data)
assert blanked == 2 # both the signature and the UUID artist
assert len(out) == len(data)
assert b"Signature: AAAA" not in out
def test_blank_leaves_clean_exif_untouched(self):
import piexif
@@ -414,6 +453,105 @@ class TestISOBMFF:
assert out == FTYP + b"\x00\x00\x00\x0cmdat" + b"pixels!!"
class TestIterTopLevelBoxes:
"""The box walker's three size encodings and its underflow/overflow guards."""
def test_64bit_largesize(self):
from remove_ai_watermarks.noai.isobmff import _iter_top_level_boxes
# size32 == 1 -> a 64-bit largesize follows the type; total box length = 24.
box = struct.pack(">I", 1) + b"uuid" + struct.pack(">Q", 24) + b"payload!"
boxes = list(_iter_top_level_boxes(box))
assert len(boxes) == 1
start, end, btype, payload_off = boxes[0]
assert (start, end, btype, payload_off) == (0, 24, b"uuid", 16)
def test_size0_runs_to_eof(self):
from remove_ai_watermarks.noai.isobmff import _iter_top_level_boxes
box = struct.pack(">I", 0) + b"mdat" + b"tail-to-eof"
boxes = list(_iter_top_level_boxes(box))
assert len(boxes) == 1
start, end, btype, payload_off = boxes[0]
assert (start, end, btype, payload_off) == (0, len(box), b"mdat", 8)
def test_underflow_size_stops_safely(self):
from remove_ai_watermarks.noai.isobmff import _iter_top_level_boxes
# size (4) < the 8-byte header -> the guard returns without yielding a box.
assert list(_iter_top_level_boxes(struct.pack(">I", 4) + b"ftyp" + b"more")) == []
def test_overflow_size_stops_safely(self):
from remove_ai_watermarks.noai.isobmff import _iter_top_level_boxes
# size claims 999 but the buffer is far shorter -> guard returns, no partial box.
assert list(_iter_top_level_boxes(struct.pack(">I", 999) + b"uuid" + b"x")) == []
class TestBlankAiXmpPackets:
"""XMP-packet blanking: same-length overwrite only for AI-marked packets, and only
when the packet is fully delimited."""
AIMARK = b"trainedAlgorithmicMedia"
def test_ai_packet_blanked_same_length(self):
from remove_ai_watermarks.noai.isobmff import blank_ai_xmp_packets
packet = b'<?xpacket begin="x"?><x:xmpmeta>' + self.AIMARK + b'</x:xmpmeta><?xpacket end="w"?>'
data = b"boxhdr" + packet + b"tail"
out, n = blank_ai_xmp_packets(data)
assert n == 1
assert len(out) == len(data) # same length -> iloc offsets stay valid
assert self.AIMARK not in out
assert b"boxhdr" in out
assert b"tail" in out
def test_clean_packet_left_intact(self):
from remove_ai_watermarks.noai.isobmff import blank_ai_xmp_packets
packet = b'<?xpacket begin="x"?><x:xmpmeta>plain copyright</x:xmpmeta><?xpacket end="w"?>'
out, n = blank_ai_xmp_packets(packet)
assert n == 0
assert out == packet
def test_missing_end_delimiter_not_blanked(self):
from remove_ai_watermarks.noai.isobmff import blank_ai_xmp_packets
# No <?xpacket end?> -> the packet regex cannot match, so it is left unchanged.
data = b'<?xpacket begin="x"?><x:xmpmeta>' + self.AIMARK + b"</x:xmpmeta>"
out, n = blank_ai_xmp_packets(data)
assert n == 0
assert out == data
class TestC2paBufferScans:
"""The shared buffer-scan helpers (used by both the PNG caBX parser and the
format-agnostic binary scan). Data-driven off the registries so they stay valid
as vendors are added."""
def test_soft_binding_vendors_in(self):
from remove_ai_watermarks.noai.c2pa import C2PA_SOFT_BINDINGS, soft_binding_vendors_in
sig, name = next(iter(C2PA_SOFT_BINDINGS.items()))
assert name in soft_binding_vendors_in(b"...manifest..." + sig + b"...tail...")
assert soft_binding_vendors_in(b"") == []
assert soft_binding_vendors_in(b"no soft-binding assertion here") == []
def test_synthid_vendors_in_requires_synthid_issuer(self):
from remove_ai_watermarks.noai.c2pa import C2PA_ISSUERS, SYNTHID_C2PA_ISSUERS, synthid_vendors_in
syn_sig = next(s for s in C2PA_ISSUERS if s in SYNTHID_C2PA_ISSUERS)
non_sig = next(s for s in C2PA_ISSUERS if s not in SYNTHID_C2PA_ISSUERS)
assert C2PA_ISSUERS[syn_sig] in synthid_vendors_in(b"x" + syn_sig + b"x")
# an issuer that does NOT pair SynthID with C2PA must not be reported as one
assert C2PA_ISSUERS[non_sig] not in synthid_vendors_in(b"x" + non_sig + b"x")
def test_synthid_verdict_format(self):
from remove_ai_watermarks.noai.c2pa import synthid_verdict
assert synthid_verdict("Google LLC") == "likely present (Google LLC embeds SynthID with C2PA)"
class TestC2PAInvalidSignature:
"""A .png file that is not actually PNG-signed must read as clean, not crash."""
+34
View File
@@ -8,6 +8,40 @@ import pytest
from remove_ai_watermarks.region_eraser import boxes_to_mask, erase, lama_available, migan_available
class TestPaddedCropBox:
"""The padded bounding box that bounds the learned backends' ONNX working set."""
def test_empty_mask_returns_none(self):
from remove_ai_watermarks.region_eraser import _padded_crop_box
assert _padded_crop_box(np.zeros((100, 100), np.uint8), 100, 100, pad_frac=0.1, pad_min=8) is None
def test_pad_min_dominates_and_clamps_at_border(self):
from remove_ai_watermarks.region_eraser import _padded_crop_box
mask = np.zeros((100, 100), np.uint8)
mask[0:5, 0:5] = 255 # 5-px mark in the top-left corner
# pad = max(8, int(0.1*5)) = 8; x0 clamps to 0 (not -8), x1 = min(100, 4+1+8) = 13.
assert _padded_crop_box(mask, 100, 100, pad_frac=0.1, pad_min=8) == (0, 0, 13, 13)
def test_pad_frac_dominates_for_large_mark(self):
from remove_ai_watermarks.region_eraser import _padded_crop_box
mask = np.zeros((400, 400), np.uint8)
mask[100:300, 100:300] = 255 # 200-px span
# pad = max(8, int(0.2*200)) = 40; box = (100-40, .., 299+1+40, ..).
assert _padded_crop_box(mask, 400, 400, pad_frac=0.2, pad_min=8) == (60, 60, 340, 340)
def test_clamps_at_far_border(self):
from remove_ai_watermarks.region_eraser import _padded_crop_box
mask = np.zeros((50, 60), np.uint8)
mask[45:50, 55:60] = 255 # bottom-right corner
_x0, _y0, x1, y1 = _padded_crop_box(mask, 50, 60, pad_frac=0.1, pad_min=8)
assert x1 == 60 # clamped to w, no overflow
assert y1 == 50 # clamped to h, no overflow
class TestBoxesToMask:
def test_mask_set_inside_box(self):
mask = boxes_to_mask((100, 100), [(10, 20, 30, 40)], dilate=0)
+22
View File
@@ -49,6 +49,28 @@ class TestScan:
dets = reg.detect_marks(np.zeros((256, 256, 3), np.uint8), include_explicit=False)
assert not any(d.detected for d in dets)
@pytest.mark.parametrize("shape", [(1, 1, 3), (8, 8, 3), (15, 15, 3), (12, 300, 3), (300, 10, 3)])
def test_tiny_image_no_crash(self, shape):
"""Regression: an image whose short side is < 16 px (below the Gemini template
floor) must yield no detection, not crash. detect_marks/remove_auto_marks are
the public visible/all/batch path; a tiny thumbnail in a batch used to take the
whole auto pass down with an IndexError (empty candidate list dereference)."""
img = np.full(shape, 100, np.uint8)
assert not any(d.detected for d in reg.detect_marks(img, include_explicit=False))
result, removed = reg.remove_auto_marks(img, backend="cv2")
assert removed == []
assert result.shape == img.shape
@pytest.mark.parametrize("shape", [(0, 5), (5, 0), (0, 5, 4), (0, 0)])
def test_forced_remove_on_empty_array_no_crash(self, shape):
"""Regression: footprint_mask ran to_bgr (cvtColor) before any size check, so a
forced remove on a zero-size ndarray crashed (cv2.error on an empty Mat). detect
already guarded this; footprint_mask must too. Covers the text + gemini engines."""
empty = np.zeros(shape, np.uint8)
for key in ("doubao", "jimeng", "samsung", "gemini"):
_result, mask = reg.get_mark(key).remove(empty, force=True)
assert mask is None
class TestBackendResolution:
def test_auto_resolves_to_available_backend(self):
+3 -1
View File
@@ -14,7 +14,9 @@ class ImageIFD:
Artist: int
ImageDescription: int
class ExifIFD: ...
class ExifIFD:
UserComment: int
class GPSIFD: ...
def load(input_data: bytes | str, key_is_name: bool = ...) -> dict[str, Any]: ...