External/watermarks-remover
1
0
Fork 0
mirror of https://github.com/guillaumemeyer/watermarks-remover.git synced 2026-08-22 13:11:57 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
360c8e8e4d4ffb405d8a90d838917a3558207c2c
watermarks-remover/tests/test_av_meta.py
T
Poorvith M PGitHubGuillaume Meyer
 275a087a07 feat: audio/video AI/C2PA metadata stripping (MP4/MOV, WAV, MP3) (#139) 
The file-cleaners layer covered 15 formats -- all image, document, or
text -- and zero audio/video. That gap gets more expensive every month:
Sora, Veo, ElevenLabs, and Suno all embed provenance through the same
mechanisms image generators do, just in different containers.

New av_meta.py adds inspect/clean for:
- MP4/MOV/M4A/M4V: top-level C2PA (jumb/c2pa box) and XMP (uuid box)
  detection/stripping reuse inspect_isobmff()/strip_isobmff() from
  image_meta.py unchanged -- that's exactly the mechanism the C2PA spec
  defines for ISOBMFF-family containers, already proven for AVIF/HEIC.
  moov/udta (where generator/tool tags live) is handled separately since
  it's MP4-specific.
- WAV: RIFF LIST INFO chunk + embedded id3 chunk.
- MP3: ID3v2 frames, per-frame for v2.3/v2.4, whole-tag fallback for
  v2.2 (3-byte frame IDs are detected but not decomposed, so a partial
  rewrite is never attempted there).

Every box/chunk/frame is either kept byte-identical or dropped whole --
nothing does a partial in-place rewrite of a payload, so a container can
never come out semantically mangled. Default strip_all_metadata=True
matches this project's existing default (privacy-first: drop everything,
--keep-non-ai-metadata narrows to only AI-flagged content), same as the
image cleaners.

Wired through the full dispatch stack so the feature isn't a half
integration: format_dispatch.py (new "av" Kind), inspect_file.py /
clean_file.py (--as av), audit_lib.py (so audit_dir.py's CI/SARIF path
and the pre-commit hooks from #135 both cover audio/video too), and
server.py (HTTP /inspect and /clean).

Closes #134

Co-authored-by: Guillaume Meyer (The Opinionated Man) <1385518+guillaumemeyer@users.noreply.github.com>
2026-08-18 07:47:43 -07:00

352 lines
12 KiB
Python
Raw Blame History

"""Tests for av_meta.py (MP4/MOV, WAV, MP3 AI/C2PA provenance metadata)."""
from __future__ import annotations
import struct
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = ROOT / "service" / "scripts"
sys.path.insert(0, str(SCRIPTS))
from av_meta import (
clean_av,
detect_av_format,
inspect_av,
)
# ---------------------------------------------------------------------------
# Fixture builders
# ---------------------------------------------------------------------------
def _isobmff_box(fourcc: bytes, payload: bytes) -> bytes:
return struct.pack(">I", len(payload) + 8) + fourcc + payload
def _mp4(*top_level_boxes: bytes) -> bytes:
ftyp = _isobmff_box(b"ftyp", b"isom" + struct.pack(">I", 0) + b"isomiso2mp41")
return ftyp + b"".join(top_level_boxes)
def _moov_with_udta(udta_payload: bytes) -> bytes:
mvhd = _isobmff_box(b"mvhd", b"\x00" * 20)
udta = _isobmff_box(b"udta", udta_payload)
return _isobmff_box(b"moov", mvhd + udta)
XMP_UUID_HEX = bytes.fromhex("be7acfcb97a942e89c71999491e3afac")
def _mp4_with_xmp(xmp_text: bytes) -> bytes:
uuid_box = _isobmff_box(b"uuid", XMP_UUID_HEX + xmp_text)
mdat = _isobmff_box(b"mdat", b"\x00" * 16)
return _mp4(uuid_box, mdat)
def _mp4_with_udta_tag(tag_text: bytes) -> bytes:
moov = _moov_with_udta(
b"\xa9too" + struct.pack(">I", len(tag_text) + 4) + b"\x00\x00\x00\x00" + tag_text
)
mdat = _isobmff_box(b"mdat", b"\x00" * 16)
return _mp4(moov, mdat)
def _riff_chunk(cid: bytes, payload: bytes) -> bytes:
pad = b"\x00" if len(payload) & 1 else b""
return cid + struct.pack("<I", len(payload)) + payload + pad
def _wav(*chunks: bytes) -> bytes:
body = b"WAVE" + b"".join(chunks)
return b"RIFF" + struct.pack("<I", len(body)) + body
def _wav_fmt_chunk() -> bytes:
return _riff_chunk(b"fmt ", struct.pack("<HHIIHH", 1, 1, 44100, 88200, 2, 16))
def _wav_data_chunk(n: int = 8) -> bytes:
return _riff_chunk(b"data", b"\x01" * n)
def _wav_list_info(text: bytes) -> bytes:
isft = _riff_chunk(b"ISFT", text + (b"\x00" if len(text) % 2 == 0 else b""))
return _riff_chunk(b"LIST", b"INFO" + isft)
def _id3v2_size_bytes(n: int) -> bytes:
return bytes([(n >> 21) & 0x7F, (n >> 14) & 0x7F, (n >> 7) & 0x7F, n & 0x7F])
def _id3v2_frame(frame_id: bytes, payload: bytes, *, major: int = 3) -> bytes:
size = _id3v2_size_bytes(len(payload)) if major == 4 else struct.pack(">I", len(payload))
return frame_id + size + b"\x00\x00" + payload
def _mp3(*frames: bytes, major: int = 3) -> bytes:
body = b"".join(frames)
header = b"ID3" + bytes([major, 0, 0]) + _id3v2_size_bytes(len(body))
audio = bytes([0xFF, 0xFB, 0x90, 0x00]) * 4 # placeholder MPEG frame-sync bytes
return header + body + audio
# ---------------------------------------------------------------------------
# detect_av_format
# ---------------------------------------------------------------------------
def test_detect_mp4():
assert detect_av_format(_mp4()) == "mp4"
def test_detect_wav():
assert detect_av_format(_wav(_wav_fmt_chunk())) == "wav"
def test_detect_mp3_with_id3():
assert detect_av_format(_mp3(_id3v2_frame(b"TIT2", b"\x00Song"))) == "mp3"
def test_detect_mp3_frame_sync_only():
data = bytes([0xFF, 0xFB, 0x90, 0x00]) * 10
assert detect_av_format(data) == "mp3"
def test_detect_unknown():
assert detect_av_format(b"not a known av container") == "unknown"
# ---------------------------------------------------------------------------
# MP4 / MOV
# ---------------------------------------------------------------------------
def test_mp4_xmp_uuid_detected_and_stripped(tmp_path):
data = _mp4_with_xmp(b"Generated by AI toolchain")
src = tmp_path / "clip.mp4"
src.write_bytes(data)
report = inspect_av(src)
assert report.format == "mp4"
assert report.has_ai_metadata is True
assert any("uuid" in f.lower() for f in report.findings)
dest = tmp_path / "clip.cleaned.mp4"
result = clean_av(src, dest, strip_all_metadata=True)
assert result["still_has_ai_metadata"] is False
assert b"Generated by AI" not in dest.read_bytes()
assert any("uuid" in a.lower() for a in result["actions"])
def test_mp4_moov_udta_generator_tag_detected_and_stripped(tmp_path):
data = _mp4_with_udta_tag(b"ElevenLabs AI Voice Generator")
src = tmp_path / "clip.mp4"
src.write_bytes(data)
report = inspect_av(src)
assert report.has_ai_metadata is False # ElevenLabs isn't in the flat hint list
# Use an explicit AI hint that IS in the flat list to prove detection works.
data2 = _mp4_with_udta_tag(b"Generated by AI")
src2 = tmp_path / "clip2.mp4"
src2.write_bytes(data2)
report2 = inspect_av(src2)
assert report2.has_ai_metadata is True
assert any("udta" in f for f in report2.findings)
dest = tmp_path / "clip2.cleaned.mp4"
result = clean_av(src2, dest, strip_all_metadata=True)
assert b"Generated by AI" not in dest.read_bytes()
assert result["still_has_ai_metadata"] is False
def test_mp4_udta_stripped_by_default_even_without_ai_hint(tmp_path):
"""Default strip_all_metadata=True strips udta regardless of hint match,
matching this project's existing default behaviour for image metadata."""
data = _mp4_with_udta_tag(b"Adobe Premiere Pro 2026")
src = tmp_path / "clip.mp4"
src.write_bytes(data)
dest = tmp_path / "clip.cleaned.mp4"
result = clean_av(src, dest, strip_all_metadata=True)
assert b"Adobe Premiere Pro 2026" not in dest.read_bytes()
assert any("udta" in a for a in result["actions"])
def test_mp4_keep_non_ai_metadata_preserves_unflagged_udta(tmp_path):
data = _mp4_with_udta_tag(b"Adobe Premiere Pro 2026")
src = tmp_path / "clip.mp4"
src.write_bytes(data)
dest = tmp_path / "clip.cleaned.mp4"
result = clean_av(src, dest, strip_all_metadata=False)
assert b"Adobe Premiere Pro 2026" in dest.read_bytes()
assert not any("udta" in a for a in result["actions"])
def test_mp4_clean_file_is_idempotent_when_already_clean(tmp_path):
data = _mp4()
src = tmp_path / "clean.mp4"
src.write_bytes(data)
dest = tmp_path / "clean.cleaned.mp4"
result = clean_av(src, dest, strip_all_metadata=True)
assert result["still_has_ai_metadata"] is False
assert result["still_has_c2pa"] is False
# ---------------------------------------------------------------------------
# WAV
# ---------------------------------------------------------------------------
def test_wav_list_info_ai_hint_detected_and_stripped(tmp_path):
data = _wav(_wav_fmt_chunk(), _wav_list_info(b"Generated by AI"), _wav_data_chunk(8))
src = tmp_path / "voice.wav"
src.write_bytes(data)
report = inspect_av(src)
assert report.format == "wav"
assert report.has_ai_metadata is True
assert any("LIST INFO" in f for f in report.findings)
dest = tmp_path / "voice.cleaned.wav"
result = clean_av(src, dest, strip_all_metadata=True)
cleaned = dest.read_bytes()
assert b"Generated by AI" not in cleaned
assert result["still_has_ai_metadata"] is False
def test_wav_audio_data_untouched(tmp_path):
audio = bytes(range(256)) * 4
data = _wav(_wav_fmt_chunk(), _wav_list_info(b"Generated by AI"), _riff_chunk(b"data", audio))
src = tmp_path / "voice.wav"
src.write_bytes(data)
dest = tmp_path / "voice.cleaned.wav"
clean_av(src, dest, strip_all_metadata=True)
cleaned = dest.read_bytes()
assert audio in cleaned
# RIFF size field must match the new (shorter) total.
riff_size = struct.unpack("<I", cleaned[4:8])[0]
assert riff_size == len(cleaned) - 8
def test_wav_clean_file_already_clean_no_changes(tmp_path):
audio = b"\x01\x02\x03\x04" * 4
data = _wav(_wav_fmt_chunk(), _riff_chunk(b"data", audio))
src = tmp_path / "clean.wav"
src.write_bytes(data)
dest = tmp_path / "clean.cleaned.wav"
result = clean_av(src, dest, strip_all_metadata=True)
assert "no WAV metadata chunks removed" in result["actions"][0]
assert dest.read_bytes() == data
def test_wav_keep_non_ai_metadata_preserves_unflagged_info(tmp_path):
data = _wav(_wav_fmt_chunk(), _wav_list_info(b"Adobe Audition"), _wav_data_chunk())
src = tmp_path / "voice.wav"
src.write_bytes(data)
dest = tmp_path / "voice.cleaned.wav"
clean_av(src, dest, strip_all_metadata=False)
assert b"Adobe Audition" in dest.read_bytes()
# ---------------------------------------------------------------------------
# MP3 (ID3v2)
# ---------------------------------------------------------------------------
def test_mp3_id3v23_ai_hint_detected(tmp_path):
data = _mp3(
_id3v2_frame(b"TIT2", b"\x00My Track"),
_id3v2_frame(b"TSSE", b"\x00Generated by AI"),
)
src = tmp_path / "song.mp3"
src.write_bytes(data)
report = inspect_av(src)
assert report.format == "mp3"
assert report.has_ai_metadata is True
assert any("TSSE" in f for f in report.findings)
def test_mp3_strip_all_drops_whole_tag(tmp_path):
data = _mp3(
_id3v2_frame(b"TIT2", b"\x00My Track"),
_id3v2_frame(b"TSSE", b"\x00Generated by AI"),
)
src = tmp_path / "song.mp3"
src.write_bytes(data)
dest = tmp_path / "song.cleaned.mp3"
result = clean_av(src, dest, strip_all_metadata=True)
cleaned = dest.read_bytes()
assert b"My Track" not in cleaned
assert b"Generated by AI" not in cleaned
assert result["still_has_ai_metadata"] is False
# The MPEG frame-sync audio placeholder bytes survive untouched.
assert cleaned.endswith(bytes([0xFF, 0xFB, 0x90, 0x00]) * 4)
def test_mp3_keep_mode_drops_only_flagged_frame(tmp_path):
data = _mp3(
_id3v2_frame(b"TIT2", b"\x00My Track"),
_id3v2_frame(b"TSSE", b"\x00Generated by AI"),
)
src = tmp_path / "song.mp3"
src.write_bytes(data)
dest = tmp_path / "song.cleaned.mp3"
result = clean_av(src, dest, strip_all_metadata=False)
cleaned = dest.read_bytes()
assert b"My Track" in cleaned # legitimate tag survives
assert b"Generated by AI" not in cleaned # flagged frame is gone
assert any("TSSE" in a for a in result["actions"])
def test_mp3_id3v24_syncsafe_frame_size_round_trip(tmp_path):
data = _mp3(
_id3v2_frame(b"TIT2", b"\x00My Track", major=4),
_id3v2_frame(b"TSSE", b"\x00Generated by AI", major=4),
major=4,
)
src = tmp_path / "song.mp3"
src.write_bytes(data)
report = inspect_av(src)
assert report.has_ai_metadata is True
dest = tmp_path / "song.cleaned.mp3"
result = clean_av(src, dest, strip_all_metadata=False)
cleaned = dest.read_bytes()
assert b"My Track" in cleaned
assert b"Generated by AI" not in cleaned
# Re-inspecting the cleaned, rewritten v2.4 tag must still parse correctly.
after = inspect_av(dest)
assert after.has_ai_metadata is False
assert result["still_has_ai_metadata"] is False
def test_mp3_no_id3_tag_clean_noop(tmp_path):
data = bytes([0xFF, 0xFB, 0x90, 0x00]) * 20
src = tmp_path / "raw.mp3"
src.write_bytes(data)
dest = tmp_path / "raw.cleaned.mp3"
result = clean_av(src, dest, strip_all_metadata=True)
assert dest.read_bytes() == data
assert result["still_has_ai_metadata"] is False
def test_mp3_id3v22_falls_back_to_whole_tag_scan_and_drop(tmp_path):
# v2.2: 3-byte frame IDs, no per-frame decomposition -- whole-tag handling.
body = b"TT2\x00\x00\x10\x00Generated by AI"
header = b"ID3" + bytes([2, 0, 0]) + _id3v2_size_bytes(len(body))
data = header + body + bytes([0xFF, 0xFB, 0x90, 0x00]) * 4
src = tmp_path / "old.mp3"
src.write_bytes(data)
report = inspect_av(src)
assert report.has_ai_metadata is True
dest = tmp_path / "old.cleaned.mp3"
result = clean_av(src, dest, strip_all_metadata=False)
cleaned = dest.read_bytes()
assert b"Generated by AI" not in cleaned
assert any("ID3v2.2" in a for a in result["actions"])
Reference in New Issue View Git Blame Copy Permalink