mirror of
https://github.com/guillaumemeyer/watermarks-remover.git
synced 2026-08-22 13:11:57 +02:00
feat: recursively inspect and clean embedded raster data URIs in SVGs, HTML, and Markdown (#87) (#88)
This commit is contained in:
Poorvith M P
@@ -4,18 +4,38 @@ Formats: SVG, PDF (best-effort), DOCX, ODT, HTML, Markdown frontmatter.
|
||||
Stdlib-first; PDF prefers optional exiftool/c2patool when present.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import io
|
||||
import posixpath
|
||||
import re
|
||||
import subprocess
|
||||
import urllib.parse
|
||||
import zipfile
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from common import classify_finding_confidence, safe_arg, safe_write_bytes, safe_write_text, subprocess_preexec_fn, which
|
||||
from image_meta import AI_META_HINTS, C2PA_MARKERS, run_optional_tools
|
||||
from common import (
|
||||
classify_finding_confidence,
|
||||
safe_arg,
|
||||
safe_write_bytes,
|
||||
safe_write_text,
|
||||
subprocess_preexec_fn,
|
||||
which,
|
||||
)
|
||||
from image_meta import (
|
||||
AI_META_HINTS,
|
||||
C2PA_MARKERS,
|
||||
detect_format as detect_image_format,
|
||||
inspect_isobmff,
|
||||
inspect_jpeg,
|
||||
inspect_png,
|
||||
inspect_webp,
|
||||
run_optional_tools,
|
||||
strip_isobmff,
|
||||
strip_jpeg,
|
||||
strip_png,
|
||||
strip_webp,
|
||||
)
|
||||
|
||||
# Frontmatter / meta keys that often carry AI provenance
|
||||
AI_FRONTMATTER_KEYS = frozenset(
|
||||
@@ -146,6 +166,133 @@ def _blob_hits(blob: bytes) -> tuple[bool, bool, list[str]]:
|
||||
return has_c2pa, has_ai or has_c2pa, findings[:30]
|
||||
|
||||
|
||||
RE_DATA_IMAGE_URI = re.compile(
|
||||
r"data:image\/(?P<mime>[a-zA-Z0-9\+\-\.]+)(?P<params>;[^\s\"'\)<>]+)?,(?P<payload>[A-Za-z0-9+/=\s%]+)",
|
||||
re.I,
|
||||
)
|
||||
|
||||
|
||||
def _inspect_embedded_data_uris(text: str) -> tuple[bool, bool, list[str]]:
|
||||
has_c2pa = False
|
||||
has_ai = False
|
||||
findings: list[str] = []
|
||||
|
||||
for m in RE_DATA_IMAGE_URI.finditer(text):
|
||||
mime = m.group("mime").lower()
|
||||
params = (m.group("params") or "").lower()
|
||||
payload = m.group("payload")
|
||||
is_b64 = "base64" in params
|
||||
|
||||
try:
|
||||
if is_b64:
|
||||
raw_b64 = re.sub(r"\s+", "", payload)
|
||||
pad = len(raw_b64) % 4
|
||||
if pad:
|
||||
raw_b64 += "=" * (4 - pad)
|
||||
data = base64.b64decode(raw_b64)
|
||||
else:
|
||||
data = urllib.parse.unquote_to_bytes(payload)
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if not data:
|
||||
continue
|
||||
|
||||
fmt = detect_image_format(data)
|
||||
if fmt == "png":
|
||||
sub_c2pa, sub_ai, sub_findings = inspect_png(data)
|
||||
elif fmt == "jpeg":
|
||||
sub_c2pa, sub_ai, sub_findings = inspect_jpeg(data)
|
||||
elif fmt == "webp":
|
||||
sub_c2pa, sub_ai, sub_findings = inspect_webp(data)
|
||||
elif fmt in ("avif", "heic"):
|
||||
sub_c2pa, sub_ai, sub_findings = inspect_isobmff(data, fmt)
|
||||
elif "svg" in mime or data.lstrip().startswith(b"<"):
|
||||
sub_c2pa, sub_ai, sub_findings, _ = inspect_svg(data)
|
||||
else:
|
||||
sub_c2pa, sub_ai, sub_findings = _blob_hits(data)
|
||||
|
||||
if sub_c2pa:
|
||||
has_c2pa = True
|
||||
if sub_ai or sub_c2pa:
|
||||
has_ai = True
|
||||
for f in sub_findings:
|
||||
findings.append(f"embedded data:image/{mime}: {f}")
|
||||
|
||||
return has_c2pa, has_ai, findings
|
||||
|
||||
|
||||
def _clean_embedded_data_uris(
|
||||
text: str, *, strip_all_metadata: bool = True
|
||||
) -> tuple[str, list[str]]:
|
||||
actions: list[str] = []
|
||||
|
||||
def _replace_uri(m: re.Match[str]) -> str:
|
||||
full_match = m.group(0)
|
||||
mime = m.group("mime")
|
||||
params = m.group("params") or ""
|
||||
payload = m.group("payload")
|
||||
is_b64 = "base64" in params.lower()
|
||||
|
||||
try:
|
||||
if is_b64:
|
||||
raw_b64 = re.sub(r"\s+", "", payload)
|
||||
pad = len(raw_b64) % 4
|
||||
if pad:
|
||||
raw_b64 += "=" * (4 - pad)
|
||||
data = base64.b64decode(raw_b64)
|
||||
else:
|
||||
data = urllib.parse.unquote_to_bytes(payload)
|
||||
except Exception:
|
||||
return full_match
|
||||
|
||||
if not data:
|
||||
return full_match
|
||||
|
||||
fmt = detect_image_format(data)
|
||||
sub_actions: list[str] = []
|
||||
cleaned_bytes = data
|
||||
|
||||
try:
|
||||
if fmt == "png":
|
||||
cleaned_bytes, sub_actions = strip_png(
|
||||
data, strip_all_text=strip_all_metadata
|
||||
)
|
||||
elif fmt == "jpeg":
|
||||
cleaned_bytes, sub_actions = strip_jpeg(
|
||||
data, strip_all_app=strip_all_metadata
|
||||
)
|
||||
elif fmt == "webp":
|
||||
cleaned_bytes, sub_actions = strip_webp(
|
||||
data, strip_all_metadata=strip_all_metadata
|
||||
)
|
||||
elif fmt in ("avif", "heic"):
|
||||
cleaned_bytes, sub_actions = strip_isobmff(
|
||||
data, fmt, strip_all_metadata=strip_all_metadata
|
||||
)
|
||||
elif "svg" in mime.lower() or data.lstrip().startswith(b"<"):
|
||||
cleaned_bytes, sub_actions = clean_svg(data)
|
||||
except Exception:
|
||||
return full_match
|
||||
|
||||
if not any("drop" in a.lower() for a in sub_actions) or cleaned_bytes == data:
|
||||
return full_match
|
||||
|
||||
actions.append(
|
||||
f"cleaned embedded data:image/{mime} ({', '.join(sub_actions[:2])})"
|
||||
)
|
||||
|
||||
if is_b64:
|
||||
new_b64 = base64.b64encode(cleaned_bytes).decode("ascii")
|
||||
return f"data:image/{mime}{params},{new_b64}"
|
||||
else:
|
||||
new_payload = urllib.parse.quote_from_bytes(cleaned_bytes)
|
||||
return f"data:image/{mime}{params},{new_payload}"
|
||||
|
||||
out = RE_DATA_IMAGE_URI.sub(_replace_uri, text)
|
||||
return out, actions
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Markdown frontmatter
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -170,76 +317,91 @@ def _parse_simple_yaml_keys(block: str) -> list[tuple[str, str, int]]:
|
||||
def inspect_markdown(text: str) -> tuple[bool, bool, list[str], dict]:
|
||||
findings: list[str] = []
|
||||
has_ai = False
|
||||
m = _FM_RE.match(text)
|
||||
if not m:
|
||||
return False, False, [], {"has_frontmatter": False}
|
||||
block = m.group(1)
|
||||
has_fm = False
|
||||
keys = []
|
||||
for key, _line, _i in _parse_simple_yaml_keys(block):
|
||||
keys.append(key)
|
||||
if key.lower() in AI_FRONTMATTER_KEYS or AI_META_NAME_RE.search(key):
|
||||
has_ai = True
|
||||
findings.append(f"frontmatter key: {key}")
|
||||
# also check value
|
||||
val = _line.split(":", 1)[1] if ":" in _line else ""
|
||||
if AI_META_NAME_RE.search(val):
|
||||
has_ai = True
|
||||
findings.append(f"frontmatter value hit on {key}")
|
||||
c2pa = any("c2pa" in f.lower() or "content" in f.lower() for f in findings)
|
||||
return c2pa, has_ai, findings, {"has_frontmatter": True, "keys": keys}
|
||||
m = _FM_RE.match(text)
|
||||
if m:
|
||||
has_fm = True
|
||||
block = m.group(1)
|
||||
for key, _line, _i in _parse_simple_yaml_keys(block):
|
||||
keys.append(key)
|
||||
if key.lower() in AI_FRONTMATTER_KEYS or AI_META_NAME_RE.search(key):
|
||||
has_ai = True
|
||||
findings.append(f"frontmatter key: {key}")
|
||||
# also check value
|
||||
val = _line.split(":", 1)[1] if ":" in _line else ""
|
||||
if AI_META_NAME_RE.search(val):
|
||||
has_ai = True
|
||||
findings.append(f"frontmatter value hit on {key}")
|
||||
|
||||
uri_c2pa, uri_ai, uri_findings = _inspect_embedded_data_uris(text)
|
||||
if uri_c2pa:
|
||||
has_ai = True
|
||||
if uri_ai:
|
||||
has_ai = True
|
||||
findings.extend(uri_findings)
|
||||
|
||||
c2pa = uri_c2pa or any("c2pa" in f.lower() or "content" in f.lower() for f in findings)
|
||||
return c2pa, has_ai, findings, {"has_frontmatter": has_fm, "keys": keys}
|
||||
|
||||
|
||||
def clean_markdown(text: str) -> tuple[str, list[str]]:
|
||||
actions: list[str] = []
|
||||
m = _FM_RE.match(text)
|
||||
if not m:
|
||||
return text, ["no YAML frontmatter"]
|
||||
block = m.group(1)
|
||||
body = text[m.end() :]
|
||||
kept: list[str] = []
|
||||
dropping = False # inside the nested block of a dropped top-level key
|
||||
for line in block.splitlines():
|
||||
stripped = line.strip()
|
||||
if m:
|
||||
block = m.group(1)
|
||||
body = text[m.end() :]
|
||||
kept: list[str] = []
|
||||
dropping = False # inside the nested block of a dropped top-level key
|
||||
for line in block.splitlines():
|
||||
stripped = line.strip()
|
||||
|
||||
# Blank lines and comments belong to whichever block we are inside.
|
||||
if not stripped or stripped.startswith("#"):
|
||||
if not dropping:
|
||||
# Blank lines and comments belong to whichever block we are inside.
|
||||
if not stripped or stripped.startswith("#"):
|
||||
if not dropping:
|
||||
kept.append(line)
|
||||
continue
|
||||
|
||||
# Continuation lines (nested mappings, list items) follow their parent.
|
||||
if line[0] in (" ", "\t", "-"):
|
||||
if not dropping:
|
||||
kept.append(line)
|
||||
continue
|
||||
|
||||
km = re.match(r"^([A-Za-z0-9_.-]+)\s*:", line)
|
||||
if not km:
|
||||
dropping = False
|
||||
kept.append(line)
|
||||
continue
|
||||
continue
|
||||
|
||||
# Continuation lines (nested mappings, list items) follow their parent.
|
||||
if line[0] in (" ", "\t", "-"):
|
||||
if not dropping:
|
||||
kept.append(line)
|
||||
continue
|
||||
key = km.group(1)
|
||||
val = line.split(":", 1)[1] if ":" in line else ""
|
||||
if key.lower() in AI_FRONTMATTER_KEYS or AI_META_NAME_RE.search(key):
|
||||
actions.append(f"drop frontmatter key: {key}")
|
||||
dropping = True
|
||||
continue
|
||||
if AI_META_NAME_RE.search(val):
|
||||
actions.append(f"drop frontmatter key (value hit): {key}")
|
||||
dropping = True
|
||||
continue
|
||||
|
||||
km = re.match(r"^([A-Za-z0-9_.-]+)\s*:", line)
|
||||
if not km:
|
||||
dropping = False
|
||||
kept.append(line)
|
||||
continue
|
||||
|
||||
key = km.group(1)
|
||||
val = line.split(":", 1)[1] if ":" in line else ""
|
||||
if key.lower() in AI_FRONTMATTER_KEYS or AI_META_NAME_RE.search(key):
|
||||
actions.append(f"drop frontmatter key: {key}")
|
||||
dropping = True
|
||||
continue
|
||||
if AI_META_NAME_RE.search(val):
|
||||
actions.append(f"drop frontmatter key (value hit): {key}")
|
||||
dropping = True
|
||||
continue
|
||||
|
||||
dropping = False
|
||||
kept.append(line)
|
||||
if not actions:
|
||||
actions.append("no AI frontmatter keys removed")
|
||||
new_block = "\n".join(kept).strip("\n")
|
||||
if new_block:
|
||||
out = f"---\n{new_block}\n---\n{body}"
|
||||
new_block = "\n".join(kept).strip("\n")
|
||||
if new_block:
|
||||
out = f"---\n{new_block}\n---\n{body}"
|
||||
else:
|
||||
out = body.lstrip("\n")
|
||||
actions.append("removed empty frontmatter block")
|
||||
else:
|
||||
out = body.lstrip("\n")
|
||||
actions.append("removed empty frontmatter block")
|
||||
out = text
|
||||
|
||||
out, uri_actions = _clean_embedded_data_uris(out)
|
||||
if uri_actions:
|
||||
actions.extend(uri_actions)
|
||||
|
||||
if not actions:
|
||||
actions.append("no AI frontmatter keys or embedded data URIs removed")
|
||||
return out, actions
|
||||
|
||||
|
||||
@@ -313,6 +475,14 @@ def inspect_html(text: str) -> tuple[bool, bool, list[str], dict]:
|
||||
for m in re.finditer(r"\bdata-ai[\w-]*\s*=\s*[\"'][^\"']*[\"']", text, re.I):
|
||||
has_ai = True
|
||||
findings.append(f"attr: {m.group(0)[:80]}")
|
||||
|
||||
uri_c2pa, uri_ai, uri_findings = _inspect_embedded_data_uris(text)
|
||||
if uri_c2pa:
|
||||
has_c2pa = True
|
||||
if uri_ai:
|
||||
has_ai = True
|
||||
findings.extend(uri_findings)
|
||||
|
||||
return has_c2pa, has_ai, findings, {}
|
||||
|
||||
|
||||
@@ -346,6 +516,11 @@ def clean_html(text: str) -> tuple[str, list[str]]:
|
||||
if n:
|
||||
actions.append(f"drop data-ai* attributes x{n}")
|
||||
out = out2
|
||||
|
||||
out, uri_actions = _clean_embedded_data_uris(out)
|
||||
if uri_actions:
|
||||
actions.extend(uri_actions)
|
||||
|
||||
if not actions:
|
||||
actions.append("no HTML AI meta removed")
|
||||
return out, actions
|
||||
@@ -369,6 +544,13 @@ def inspect_svg(data: bytes) -> tuple[bool, bool, list[str], dict]:
|
||||
findings.append("XMP/RDF-like content in SVG")
|
||||
if re.search(r"c2pa|jumbf", text, re.I):
|
||||
has_c2pa = True
|
||||
|
||||
uri_c2pa, uri_ai, uri_findings = _inspect_embedded_data_uris(text)
|
||||
if uri_c2pa:
|
||||
has_c2pa = True
|
||||
if uri_ai:
|
||||
has_ai = True
|
||||
findings.extend(uri_findings)
|
||||
except Exception as e:
|
||||
findings.append(f"svg decode note: {e}")
|
||||
return has_c2pa, has_ai or has_c2pa, findings, {}
|
||||
@@ -406,6 +588,12 @@ def clean_svg(data: bytes) -> tuple[bytes, list[str]]:
|
||||
return body
|
||||
|
||||
text = re.sub(r"<!--.*?-->", _cmt, text, flags=re.DOTALL)
|
||||
|
||||
# Clean embedded data URIs
|
||||
text, uri_actions = _clean_embedded_data_uris(text)
|
||||
if uri_actions:
|
||||
actions.extend(uri_actions)
|
||||
|
||||
if not actions:
|
||||
# still strip generator attribute on root if present
|
||||
new, n = re.subn(
|
||||
|
||||
@@ -6,12 +6,12 @@
|
||||
| Stylometric AI cadence / burstiness / n-grams (zero-LLM) | Statistical variance & cadence scoring | `score_stylometry.py`, `inspect_text.py --stylometry`, `audit_dir.py --check-stylometry` | None (detection only) | Yes (calibrated score + phrase spans) |
|
||||
| Statistical text watermark (SynthID-class / Kirchenbauer) | Multi-pass paraphrase / humanize / back-translate / structural | Agent Layer B + optional `rewrite_text.py` | Meaning/style drift | No without vendor key/detector; **MarkLLM harness** (`detect_text_watermark.py`) verifies a specific scheme config before/after |
|
||||
| C2PA on PNG/JPEG/WebP/AVIF/HEIC | Drop APP11 / PNG `caBX` / RIFF `C2PA` / ISOBMFF `jumb` & `uuid` / exiftool | `clean_image.py` | Loses provenance metadata | Yes |
|
||||
| SVG metadata / XMP | Drop `<metadata>`, xmpmeta | `clean_file.py` | Loses SVG metadata | Yes (re-inspect) |
|
||||
| SVG metadata / XMP / embedded data URIs | Drop `<metadata>`, xmpmeta; clean embedded data URIs | `clean_file.py` | Loses SVG metadata; cleans embedded rasters | Yes (re-inspect) |
|
||||
| PDF XMP / info | exiftool `-all=` preferred | `clean_file.py` | Loses PDF metadata; degraded without exiftool | Partial |
|
||||
| DOCX props / customXml | Rewrite OOXML zip | `clean_file.py` | Loses doc properties | Yes |
|
||||
| ODT meta:generator | Scrub `meta.xml` | `clean_file.py` | Loses generator tag | Yes |
|
||||
| HTML generator / JSON-LD provenance | Strip tags | `clean_file.py` | Loses meta | Yes |
|
||||
| Markdown AI frontmatter keys | Drop keys | `clean_file.py` | Loses YAML keys | Yes |
|
||||
| HTML generator / JSON-LD / embedded data URIs | Strip tags; clean embedded data URIs | `clean_file.py` | Loses meta; cleans embedded rasters | Yes |
|
||||
| Markdown AI frontmatter keys / embedded data URIs | Drop keys; clean embedded data URIs | `clean_file.py` | Loses YAML keys; cleans embedded rasters | Yes |
|
||||
| Pixel image watermark (SynthID-media / StegaStamp / Tree-Ring / StableSignature) | CtrlRegen regeneration (external backend) | `clean_ctrlregen.py` / `clean_image.py --remove-pixel ctrlregen` | Regenerates pixels; heavy compute; detail drift at higher strength | No without official detector; reverse-SynthID score is a local surrogate; **MarkDiffusion same-scheme harness** (`markdiffusion_harness.py detect`) verifies a Tree-Ring-class scheme config before/after |
|
||||
| Pixel image watermark (Tree-Ring-class) | DiffusionPurification regeneration (external MarkDiffusion backend) | `clean_image.py --remove-pixel diffusion` | Blind regeneration; more drift than CtrlRegen; heavy compute | Same-scheme only via the MarkDiffusion harness (not a vendor-detector oracle) |
|
||||
| Audio / video watermarks (SynthID-media) | — | Out of scope | — | — |
|
||||
|
||||
@@ -0,0 +1,153 @@
|
||||
"""Tests for embedded data URI inspection and cleaning in SVG, HTML, and Markdown."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPTS = ROOT / "service" / "scripts"
|
||||
sys.path.insert(0, str(ROOT))
|
||||
sys.path.insert(0, str(SCRIPTS))
|
||||
|
||||
from container_meta import ( # noqa: E402
|
||||
clean_html,
|
||||
clean_markdown,
|
||||
clean_svg,
|
||||
inspect_html,
|
||||
inspect_markdown,
|
||||
inspect_svg,
|
||||
)
|
||||
from tests.test_clean_image import _minimal_jpeg_with_app11, _minimal_png_with_text # noqa: E402
|
||||
|
||||
|
||||
def test_svg_embedded_png_c2pa_cleaned():
|
||||
png_c2pa = _minimal_png_with_text()
|
||||
png_b64 = base64.b64encode(png_c2pa).decode("ascii")
|
||||
|
||||
svg_data = f"""<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink">
|
||||
<image width="100" height="100" xlink:href="data:image/png;base64,{png_b64}" />
|
||||
</svg>""".encode("utf-8")
|
||||
|
||||
# Inspect
|
||||
has_c2pa, has_ai, findings, _ = inspect_svg(svg_data)
|
||||
assert has_c2pa is True
|
||||
assert has_ai is True
|
||||
assert any("embedded data:image/png" in f for f in findings)
|
||||
|
||||
# Clean
|
||||
cleaned_bytes, actions = clean_svg(svg_data)
|
||||
assert any("cleaned embedded data:image/png" in a for a in actions)
|
||||
|
||||
# Re-inspect
|
||||
has_c2pa_after, has_ai_after, findings_after, _ = inspect_svg(cleaned_bytes)
|
||||
assert has_c2pa_after is False
|
||||
assert has_ai_after is False
|
||||
assert b"c2pa" not in cleaned_bytes.lower()
|
||||
|
||||
|
||||
def test_svg_embedded_base64_with_newlines():
|
||||
png_c2pa = _minimal_png_with_text()
|
||||
png_b64 = base64.b64encode(png_c2pa).decode("ascii")
|
||||
# Split base64 across newlines as common in formatted SVGs
|
||||
multiline_b64 = png_b64[:20] + "\n \r\n " + png_b64[20:]
|
||||
|
||||
svg_data = f"""<svg xmlns="http://www.w3.org/2000/svg">
|
||||
<image href="data:image/png;base64,{multiline_b64}" />
|
||||
</svg>""".encode("utf-8")
|
||||
|
||||
cleaned_bytes, actions = clean_svg(svg_data)
|
||||
assert any("cleaned embedded data:image/png" in a for a in actions)
|
||||
|
||||
has_c2pa, _, _, _ = inspect_svg(cleaned_bytes)
|
||||
assert has_c2pa is False
|
||||
|
||||
|
||||
def test_html_embedded_jpeg_c2pa_cleaned():
|
||||
jpeg_c2pa = _minimal_jpeg_with_app11()
|
||||
jpeg_b64 = base64.b64encode(jpeg_c2pa).decode("ascii")
|
||||
|
||||
html_text = f"""<!DOCTYPE html>
|
||||
<html>
|
||||
<head><title>Test</title></head>
|
||||
<body>
|
||||
<img src="data:image/jpeg;base64,{jpeg_b64}" alt="test" />
|
||||
</body>
|
||||
</html>"""
|
||||
|
||||
# Inspect
|
||||
has_c2pa, has_ai, findings, _ = inspect_html(html_text)
|
||||
assert has_c2pa is True
|
||||
assert has_ai is True
|
||||
assert any("embedded data:image/jpeg" in f for f in findings)
|
||||
|
||||
# Clean
|
||||
cleaned_text, actions = clean_html(html_text)
|
||||
assert any("cleaned embedded data:image/jpeg" in a for a in actions)
|
||||
assert "c2pa-manifest-fake" not in cleaned_text
|
||||
|
||||
# Re-inspect
|
||||
has_c2pa_after, has_ai_after, _, _ = inspect_html(cleaned_text)
|
||||
assert has_c2pa_after is False
|
||||
assert has_ai_after is False
|
||||
|
||||
|
||||
def test_markdown_embedded_data_uri_cleaned():
|
||||
png_c2pa = _minimal_png_with_text()
|
||||
png_b64 = base64.b64encode(png_c2pa).decode("ascii")
|
||||
|
||||
md_text = f"""---
|
||||
title: Sample Article
|
||||
author: Dev
|
||||
---
|
||||
|
||||
Here is an image:
|
||||
|
||||
"""
|
||||
|
||||
has_c2pa, has_ai, findings, _ = inspect_markdown(md_text)
|
||||
assert has_c2pa is True
|
||||
assert any("embedded data:image/png" in f for f in findings)
|
||||
|
||||
cleaned_text, actions = clean_markdown(md_text)
|
||||
assert any("cleaned embedded data:image/png" in a for a in actions)
|
||||
assert "c2pa" not in cleaned_text.lower()
|
||||
|
||||
|
||||
def test_already_clean_embedded_image_no_op():
|
||||
# PNG with no metadata
|
||||
clean_png = (
|
||||
b"\x89PNG\r\n\x1a\n"
|
||||
+ b"\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x02\x00\x00\x00\x90wS\xde"
|
||||
+ b"\x00\x00\x00\x0cIDATx\x9cc\xf8\xff\xff?\x00\x05\xfe\x02\xfe\xa7V\xfe"
|
||||
+ b"\x00\x00\x00\x00IEND\xaeB`\x82"
|
||||
)
|
||||
png_b64 = base64.b64encode(clean_png).decode("ascii")
|
||||
html_text = f'<img src="data:image/png;base64,{png_b64}">'
|
||||
|
||||
cleaned_text, actions = clean_html(html_text)
|
||||
# The string must remain exactly identical
|
||||
assert cleaned_text == html_text
|
||||
assert "no HTML AI meta removed" in actions
|
||||
|
||||
|
||||
def test_corrupted_data_uri_graceful_fallback():
|
||||
# Corrupted base64 that shouldn't crash the file cleaner
|
||||
bad_html = '<img src="data:image/png;base64,!!!NOT_VALID_BASE64###">'
|
||||
cleaned_text, actions = clean_html(bad_html)
|
||||
assert cleaned_text == bad_html
|
||||
|
||||
|
||||
def test_nested_svg_data_uri_cleaned():
|
||||
nested_svg = '<svg><metadata><ai:GeneratedBy>DALL-E</ai:GeneratedBy></metadata><rect width="10" height="10"/></svg>'
|
||||
nested_b64 = base64.b64encode(nested_svg.encode("utf-8")).decode("ascii")
|
||||
|
||||
parent_html = f'<img src="data:image/svg+xml;base64,{nested_b64}">'
|
||||
|
||||
has_c2pa, has_ai, findings, _ = inspect_html(parent_html)
|
||||
assert has_ai is True
|
||||
|
||||
cleaned_html, actions = clean_html(parent_html)
|
||||
assert any("cleaned embedded data:image/svg+xml" in a for a in actions)
|
||||
assert "DALL-E" not in base64.b64decode(cleaned_html.split("base64,")[1].split('"')[0]).decode("utf-8")
|
||||
Reference in New Issue
Block a user