diff --git a/data/README.md b/data/README.md index 59706b0..a4580d6 100644 --- a/data/README.md +++ b/data/README.md @@ -48,6 +48,8 @@ durable store is `data/research/` in the main checkout. commands, hashes, and oracle verdicts instead of committing another corpus copy. A small curated before-and-after example may live in `docs/images/` when it is part of the public documentation. + Local benchmark manifests may point to these canonical inputs or to an + untracked corpus; every artifact and reference must carry its SHA-256 digest. `evaluations/engine-selection/` is the bounded exception for ORIGINAL study inputs: its manifest tracks a prompt-matched OpenAI/Meta content matrix, not generated model outputs or watermark-oracle evidence. diff --git a/docs/benchmark-kernel.md b/docs/benchmark-kernel.md new file mode 100644 index 0000000..2a652ed --- /dev/null +++ b/docs/benchmark-kernel.md @@ -0,0 +1,84 @@ +# Watermark benchmark kernel + +`scripts/watermark_benchmark.py` is a development-only runner for reproducible +image-watermark experiments. It calls the existing local DWT-DCT and TrustMark +detectors through thin adapters. It does not add a runtime command, download a +corpus, or contact a provenance oracle or provider API. The optional TrustMark +package can fetch its official Adobe model weights when its local package cache +is incomplete; prepare that dependency before an offline run. + +The kernel answers three different questions and never merges their answers: + +- `detection` records whether one named adapter recognized its own signal. +- `removal` records that a removed-state artifact no longer produced a signal, + or that the signal remained. It always sets `certifies_erasure` to `false`. +- `fidelity` compares decoded pixels with an explicit reference. It says + nothing about whether an invisible watermark remains. + +In particular, `not_detected` means only that the selected detector did not +recognize its signal. It is not a general `clean` verdict. + +## Manifest + +Input is strict JSONL, one case per line. Relative paths resolve from the +manifest directory. The runner verifies every named file against its lowercase +SHA-256 digest before invoking any detector, rejects duplicate `case_id` values, +and rejects missing or extra fields. + +```json +{"schema_version":1,"case_id":"sample-removed","pair_id":"sample","media_type":"image","adapter":"dwt-dct","arm":"positive","state":"removed","path":"removed.png","sha256":"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef","reference_path":"clean.png","reference_sha256":"abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789","source_revision":"local-corpus@2026-09-05","transform":{"name":"remove","revision":"remove-ai-watermarks@abc123","parameters":{"method":"auto"}},"seed":7,"expected":"not_detected"} +``` + +The fixed fields are: + +| Field | Contract | +| --- | --- | +| `schema_version` | Integer `1`. | +| `case_id` | Unique case identity. | +| `pair_id` | Groups related clean, marked, attacked, and removed cases. | +| `media_type` | `image` in schema v1. | +| `adapter` | `dwt-dct` or `trustmark`. | +| `arm` | `positive`, `matched_negative`, `wrong_key`, or `hard_negative`. | +| `state` | `clean`, `marked`, `attacked`, or `removed`. | +| `path`, `sha256` | Artifact path and pinned content digest. | +| `reference_path`, `reference_sha256` | Both strings or both `null`; the fidelity reference. | +| `source_revision` | Corpus, generator, or acquisition revision. | +| `transform` | Exact `name`, `revision`, and JSON `parameters`. | +| `seed` | Integer or `null`; never an implicit random seed. | +| `expected` | `detected`, `not_detected`, or `unresolved`. | + +The arms prevent a positive-only benchmark from being mistaken for a detector +evaluation. A useful detector run needs known positives plus matched negatives, +wrong-key cases where the method supports keys, and difficult watermark-free +images. Transformations such as resize, crop, recompression, or removal should +be separate rows with exact parameters, not edits made outside the manifest. + +## Run and output + +Keep the report outside the repository, and choose a new output path for every +run: + +```bash +uv run python scripts/watermark_benchmark.py \ + /path/to/manifest.jsonl \ + --output /path/to/results/run-2026-09-05.jsonl +``` + +The runner refuses to overwrite an existing report. Every output row repeats +the case identity and records the repository commit, whether tracked files were +dirty, and SHA-256 digests of the benchmark kernel and adapter source. Detector +dependency absence is `unavailable`; an undecodable artifact or adapter +exception is `error`. Neither is counted as `not_detected`. + +Fidelity is calculated only when a reference is present, both decoded images +are 8-bit, and their dimensions match. Other bit depths remain explicitly +incomparable until their sample-range contract is defined. The result records +8-bit MAE, the fraction of changed pixel positions, and PSNR. Identical decoded +pixels produce `psnr_db: null` with +`psnr_status: unbounded_identical`, never a misleading numeric zero. Missing or +incomparable references remain explicit states rather than fabricated metrics. + +Schema v1 deliberately has no aggregate score, confidence interval, video +adapter, automatic attack generator, removal runner, or provider oracle. Add +those only after the case-level evidence is large enough to define and test the +corresponding statistical or media contract. diff --git a/docs/index.md b/docs/index.md index b9280c5..a5fb154 100644 --- a/docs/index.md +++ b/docs/index.md @@ -22,6 +22,7 @@ to run the tool. Use the maintainer references only when changing the code. | --- | --- | | [Module internals](module-internals.md) | Current architecture, invariants, and regression guards by module. | | [Development](development.md) | Environment setup, dependency recovery, CI behavior, and fixture policy. | +| [Watermark benchmark kernel](benchmark-kernel.md) | Reproducible local detection, removal-observation, and fidelity measurements. | | [Code provenance](code-provenance.md) | Required notices for licensed derivative work. | | [Verification plan](verification-plan.md) | Verification methods, completed measurements, and remaining validation gaps. | | [Release and distribution](release-and-distribution.md) | PyPI, Homebrew, Hugging Face Space, photo-classify Hub model, ComfyUI Registry, and the release workflow. | diff --git a/scripts/README.md b/scripts/README.md index fd49726..805be62 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -25,6 +25,7 @@ explicitly names a tracked canonical result. | `video_fidelity_probe.py` | Compare delivered video fidelity with its source. | | `visible_eval.py` | Benchmark registered visible-mark detectors. | | `visible_removal_audit.py` | Audit visible-removal results over a local corpus. | +| `watermark_benchmark.py` | Run hash-pinned image cases while keeping detection, removal observation, and fidelity separate. | ## Calibration and corpus preparation diff --git a/scripts/watermark_benchmark.py b/scripts/watermark_benchmark.py new file mode 100644 index 0000000..d23bb9b --- /dev/null +++ b/scripts/watermark_benchmark.py @@ -0,0 +1,609 @@ +#!/usr/bin/env python3 +"""Run reproducible, development-only watermark benchmark cases. + +The input is a strict JSONL manifest. Every row names one immutable artifact, +its evidence arm and processing state, the detector adapter, source and +transform revisions, an explicit seed, and an optional decoded-pixel reference. +Artifacts and references are hash-checked before any detector runs. + +The output keeps three questions separate: + +* ``detection`` records only what one named adapter recognized; +* ``removal`` records the post-removal observation without certifying erasure; +* ``fidelity`` measures decoded-pixel distance from an explicit reference. + +This is a maintainer tool, not an installed command. It calls no provenance +oracle or provider API, does not download corpora, and refuses to overwrite an +existing report. The optional TrustMark dependency may fetch its official model +weights when its local package cache is incomplete. + + uv run python scripts/watermark_benchmark.py manifest.jsonl --output report.jsonl +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import logging +import math +import os +import shutil +import subprocess +import sys +import tempfile +from collections import Counter +from dataclasses import dataclass +from functools import cache, lru_cache +from pathlib import Path +from typing import TYPE_CHECKING, Any, Literal, Protocol, cast + +if TYPE_CHECKING: + from collections.abc import Callable, Iterable, Mapping + + from numpy.typing import NDArray + +REPO = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO / "src")) + +from remove_ai_watermarks._internal.schema import require_schema_version # noqa: E402 + +log = logging.getLogger(__name__) + +SCHEMA_VERSION = 1 +Arm = Literal["positive", "matched_negative", "wrong_key", "hard_negative"] +State = Literal["clean", "marked", "attacked", "removed"] +ExpectedDetection = Literal["detected", "not_detected", "unresolved"] +DetectorStatus = Literal["detected", "not_detected", "unavailable", "error"] + +_ARMS: tuple[Arm, ...] = ("positive", "matched_negative", "wrong_key", "hard_negative") +_STATES: tuple[State, ...] = ("clean", "marked", "attacked", "removed") +_EXPECTED: tuple[ExpectedDetection, ...] = ("detected", "not_detected", "unresolved") +_FIELDS = { + "schema_version", + "case_id", + "pair_id", + "media_type", + "adapter", + "arm", + "state", + "path", + "sha256", + "reference_path", + "reference_sha256", + "source_revision", + "transform", + "seed", + "expected", +} +_TRANSFORM_FIELDS = {"name", "revision", "parameters"} + + +@dataclass(frozen=True) +class Transform: + """Exact transform identity carried by one benchmark case.""" + + name: str + revision: str + parameters: dict[str, Any] + + +@dataclass(frozen=True) +class BenchmarkCase: + """One validated manifest row with absolute, hash-checked paths.""" + + case_id: str + pair_id: str + media_type: Literal["image"] + adapter: str + arm: Arm + state: State + path: Path + sha256: str + reference_path: Path | None + reference_sha256: str | None + source_revision: str + transform: Transform + seed: int | None + expected: ExpectedDetection + + +@dataclass(frozen=True) +class DetectorOutcome: + """Provider-scoped detector observation, never a general clean verdict.""" + + status: DetectorStatus + label: str | None + error: str | None = None + + +class DetectorAdapter(Protocol): + """Small local detector seam used by the benchmark kernel.""" + + @property + def name(self) -> str: ... + + @property + def source_file(self) -> Path: ... + + @property + def available(self) -> bool: ... + + def detect(self, path: Path, image: NDArray[Any]) -> DetectorOutcome: ... + + +@dataclass(frozen=True) +class FunctionAdapter: + """Adapt one local optional detector without copying its implementation.""" + + name: str + source_file: Path + available: bool + detector: Callable[[Path, NDArray[Any]], str | None] + + def detect(self, path: Path, image: NDArray[Any]) -> DetectorOutcome: + label = self.detector(path, image) + return DetectorOutcome(status="detected" if label is not None else "not_detected", label=label) + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +@cache +def _code_sha256(path: Path) -> str: + return _sha256(path) + + +def _reject_json_constant(value: str) -> None: + raise ValueError(f"non-finite number {value}") + + +def _unique_json_object(pairs: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, value in pairs: + if key in result: + raise ValueError(f"duplicate field {key!r}") + result[key] = value + return result + + +def _nonempty_string(value: object, *, field: str, location: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{location}: {field} must be a non-empty string") + return value + + +def _digest(value: object, *, field: str, location: str) -> str: + digest = _nonempty_string(value, field=field, location=location) + if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest): + raise ValueError(f"{location}: {field} must be a lowercase sha256") + return digest + + +def _artifact( + path_value: object, + digest_value: object, + *, + manifest: Path, + prefix: str, + location: str, + digest_cache: dict[Path, str], +) -> tuple[Path, str]: + raw_path = _nonempty_string(path_value, field=f"{prefix}path", location=location) + expected_digest = _digest(digest_value, field=f"{prefix}sha256", location=location) + path = Path(raw_path) + if not path.is_absolute(): + path = manifest.parent / path + path = path.resolve() + if not path.is_file(): + raise ValueError(f"{location}: {prefix}path is not a file: {path}") + actual_digest = digest_cache.get(path) + if actual_digest is None: + actual_digest = _sha256(path) + digest_cache[path] = actual_digest + if actual_digest != expected_digest: + raise ValueError( + f"{location}: {prefix}sha256 mismatch for {path}: expected {expected_digest}, got {actual_digest}" + ) + return path, expected_digest + + +def _transform(value: object, *, location: str) -> Transform: + if not isinstance(value, dict): + raise ValueError(f"{location}: transform must contain exactly name, revision, and parameters") + transform = cast("dict[str, object]", value) + if set(transform) != _TRANSFORM_FIELDS: + raise ValueError(f"{location}: transform must contain exactly name, revision, and parameters") + name = _nonempty_string(transform["name"], field="transform.name", location=location) + revision = _nonempty_string(transform["revision"], field="transform.revision", location=location) + parameters = transform["parameters"] + if not isinstance(parameters, dict): + raise ValueError(f"{location}: transform.parameters must be an object") + return Transform(name=name, revision=revision, parameters=cast("dict[str, Any]", parameters)) + + +def _case( + raw: object, + *, + manifest: Path, + line_number: int, + digest_cache: dict[Path, str], +) -> BenchmarkCase: + location = f"{manifest}:{line_number}" + if not isinstance(raw, dict): + raise ValueError(f"{location}: expected exactly {', '.join(sorted(_FIELDS))}") + values = cast("dict[str, object]", raw) + if set(values) != _FIELDS: + raise ValueError(f"{location}: expected exactly {', '.join(sorted(_FIELDS))}") + try: + require_schema_version( + values["schema_version"], + contract="watermark benchmark", + supported=(SCHEMA_VERSION,), + ) + except ValueError as exc: + raise ValueError(f"{location}: {exc}") from exc + + media_type = values["media_type"] + if media_type != "image": + raise ValueError(f"{location}: media_type must be image in schema v{SCHEMA_VERSION}") + arm = values["arm"] + if arm not in _ARMS: + raise ValueError(f"{location}: arm must be one of {', '.join(_ARMS)}") + state = values["state"] + if state not in _STATES: + raise ValueError(f"{location}: state must be one of {', '.join(_STATES)}") + expected = values["expected"] + if expected not in _EXPECTED: + raise ValueError(f"{location}: expected must be one of {', '.join(_EXPECTED)}") + seed = values["seed"] + if seed is not None and type(seed) is not int: + raise ValueError(f"{location}: seed must be an integer or null") + + path, sha256 = _artifact( + values["path"], + values["sha256"], + manifest=manifest, + prefix="", + location=location, + digest_cache=digest_cache, + ) + reference_path_value = values["reference_path"] + reference_sha256_value = values["reference_sha256"] + if (reference_path_value is None) != (reference_sha256_value is None): + raise ValueError(f"{location}: reference_path and reference_sha256 must be provided together") + reference_path: Path | None = None + reference_sha256: str | None = None + if reference_path_value is not None: + reference_path, reference_sha256 = _artifact( + reference_path_value, + reference_sha256_value, + manifest=manifest, + prefix="reference_", + location=location, + digest_cache=digest_cache, + ) + + return BenchmarkCase( + case_id=_nonempty_string(values["case_id"], field="case_id", location=location), + pair_id=_nonempty_string(values["pair_id"], field="pair_id", location=location), + media_type="image", + adapter=_nonempty_string(values["adapter"], field="adapter", location=location), + arm=arm, + state=state, + path=path, + sha256=sha256, + reference_path=reference_path, + reference_sha256=reference_sha256, + source_revision=_nonempty_string(values["source_revision"], field="source_revision", location=location), + transform=_transform(values["transform"], location=location), + seed=seed, + expected=expected, + ) + + +def load_manifest(path: Path) -> list[BenchmarkCase]: + """Load a strict JSONL manifest and verify every artifact digest.""" + rows: list[BenchmarkCase] = [] + case_ids: set[str] = set() + digest_cache: dict[Path, str] = {} + with path.open(encoding="utf-8") as manifest: + for line_number, line in enumerate(manifest, 1): + if not line.strip(): + continue + try: + raw = json.loads( + line, + object_pairs_hook=_unique_json_object, + parse_constant=_reject_json_constant, + ) + except (json.JSONDecodeError, ValueError) as exc: + detail = exc.msg if isinstance(exc, json.JSONDecodeError) else str(exc) + raise ValueError(f"{path}:{line_number}: invalid JSON: {detail}") from exc + row = _case(raw, manifest=path, line_number=line_number, digest_cache=digest_cache) + if row.case_id in case_ids: + raise ValueError(f"{path}:{line_number}: duplicate case_id {row.case_id!r}") + case_ids.add(row.case_id) + rows.append(row) + if not rows: + raise ValueError(f"{path}: manifest is empty") + return rows + + +def _detect_dwt_dct(path: Path, image: NDArray[Any]) -> str | None: + from remove_ai_watermarks.image_io import to_bgr + from remove_ai_watermarks.invisible_watermark import detect_invisible_watermark + + return detect_invisible_watermark(path, image=to_bgr(image)) + + +def _detect_trustmark(path: Path, _image: NDArray[Any]) -> str | None: + from remove_ai_watermarks.trustmark_detector import detect_trustmark + + return detect_trustmark(path) + + +def default_adapters() -> dict[str, DetectorAdapter]: + """Return the first two local, open image-watermark adapters.""" + from remove_ai_watermarks import invisible_watermark, trustmark_detector + + dwt_source = Path(invisible_watermark.__file__).resolve() + trustmark_source = Path(trustmark_detector.__file__).resolve() + return { + "dwt-dct": FunctionAdapter( + name="dwt-dct", + source_file=dwt_source, + available=invisible_watermark.is_available(), + detector=_detect_dwt_dct, + ), + "trustmark": FunctionAdapter( + name="trustmark", + source_file=trustmark_source, + available=trustmark_detector.is_available(), + detector=_detect_trustmark, + ), + } + + +def repository_state() -> dict[str, str | bool]: + """Identify the repository commit and whether tracked files differ from it.""" + git_executable = shutil.which("git") + if git_executable is None: + raise RuntimeError("cannot resolve repository revision: git is not installed") + + def git(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( # noqa: S603 - resolved git executable and fixed repository cwd + [git_executable, *args], + cwd=REPO, + capture_output=True, + text=True, + check=False, + ) + + revision = git("rev-parse", "HEAD") + if revision.returncode != 0 or not revision.stdout.strip(): + raise RuntimeError(f"cannot resolve repository revision: {revision.stderr.strip()}") + tracked = git("status", "--porcelain", "--untracked-files=no") + if tracked.returncode != 0: + raise RuntimeError(f"cannot inspect repository state: {tracked.stderr.strip()}") + return {"commit": revision.stdout.strip(), "dirty": bool(tracked.stdout.strip())} + + +def _decode_image(path: Path) -> NDArray[Any] | None: + import cv2 + + from remove_ai_watermarks.image_io import imread + + return imread(path, cv2.IMREAD_UNCHANGED) + + +def _image_fidelity( + case: BenchmarkCase, + artifact: NDArray[Any] | None, + reference_decoder: Callable[[Path], NDArray[Any] | None], +) -> dict[str, Any]: + if case.reference_path is None: + return {"status": "not_measured", "reason": "reference_not_provided"} + + import numpy as np + + reference = reference_decoder(case.reference_path) + if artifact is None or reference is None: + return {"status": "not_measured", "reason": "decoded_pixels_unavailable"} + if artifact.shape != reference.shape: + return { + "status": "incomparable", + "reason": "shape_mismatch", + "artifact_shape": list(artifact.shape), + "reference_shape": list(reference.shape), + } + if artifact.dtype != np.uint8 or reference.dtype != np.uint8: + return { + "status": "incomparable", + "reason": "unsupported_dtype", + "artifact_dtype": str(artifact.dtype), + "reference_dtype": str(reference.dtype), + } + + delta = artifact.astype(np.float64) - reference.astype(np.float64) + absolute = np.abs(delta) + mse = float(np.mean(np.square(delta))) + changed: np.ndarray[Any, np.dtype[np.bool_]] = ( + artifact != reference if artifact.ndim == 2 else np.any(artifact != reference, axis=-1) + ) + identical = mse == 0.0 + return { + "status": "measured", + "identical": identical, + "mae_8bit": float(np.mean(absolute)), + "changed_fraction": float(np.mean(changed)), + "psnr_db": None if identical else 20.0 * math.log10(255.0) - 10.0 * math.log10(mse), + "psnr_status": "unbounded_identical" if identical else "measured", + } + + +def _detect(adapter: DetectorAdapter, path: Path, image: NDArray[Any]) -> DetectorOutcome: + try: + if not adapter.available: + return DetectorOutcome(status="unavailable", label=None) + return adapter.detect(path, image) + except Exception as exc: + log.warning("Benchmark adapter %s failed for %s: %s", adapter.name, path, exc) + return DetectorOutcome(status="error", label=None, error=f"{type(exc).__name__}: {exc}") + + +def _detection_record(outcome: DetectorOutcome, expected: ExpectedDetection) -> dict[str, Any]: + comparable = outcome.status in ("detected", "not_detected") and expected != "unresolved" + record: dict[str, Any] = { + "status": outcome.status, + "label": outcome.label, + "expected": expected, + "matches_expected": outcome.status == expected if comparable else None, + "positive_evidence": outcome.status == "detected", + } + if outcome.error is not None: + record["error"] = outcome.error + return record + + +def _removal_record(state: State, outcome: DetectorOutcome) -> dict[str, str | bool]: + if state != "removed": + status = "not_applicable" + elif outcome.status == "detected": + status = "signal_detected_after_removal" + elif outcome.status == "not_detected": + status = "no_recognized_signal_after_removal" + else: + status = "unmeasured" + return { + "attempted": state == "removed", + "status": status, + "certifies_erasure": False, + } + + +def evaluate_case( + case: BenchmarkCase, + *, + adapters: Mapping[str, DetectorAdapter], + repository: Mapping[str, str | bool], + reference_decoder: Callable[[Path], NDArray[Any] | None] = _decode_image, +) -> dict[str, Any]: + """Evaluate one case while keeping detector and fidelity claims distinct.""" + try: + adapter = adapters[case.adapter] + except KeyError as exc: + raise ValueError(f"unknown adapter {case.adapter!r}") from exc + artifact = _decode_image(case.path) + outcome = ( + _detect(adapter, case.path, artifact) + if artifact is not None + else DetectorOutcome(status="error", label=None, error="artifact could not be decoded") + ) + return { + "schema_version": SCHEMA_VERSION, + "case_id": case.case_id, + "pair_id": case.pair_id, + "media_type": case.media_type, + "adapter": case.adapter, + "arm": case.arm, + "state": case.state, + "artifact": { + "sha256": case.sha256, + "source_revision": case.source_revision, + }, + "reference": {"sha256": case.reference_sha256} if case.reference_sha256 is not None else None, + "transform": { + "name": case.transform.name, + "revision": case.transform.revision, + "parameters": case.transform.parameters, + "seed": case.seed, + }, + "run": { + "repository": dict(repository), + "kernel_source_sha256": _code_sha256(Path(__file__)), + "adapter_source_sha256": _code_sha256(adapter.source_file), + }, + "detection": _detection_record(outcome, case.expected), + "removal": _removal_record(case.state, outcome), + "fidelity": _image_fidelity(case, artifact, reference_decoder), + } + + +def write_jsonl(path: Path, records: Iterable[Mapping[str, Any]]) -> None: + """Publish strict JSONL atomically and refuse to replace an existing report.""" + if path.exists(): + raise FileExistsError(f"refusing to overwrite existing report: {path}") + path.parent.mkdir(parents=True, exist_ok=True) + temporary: Path | None = None + try: + with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as stream: + temporary = Path(stream.name) + for record in records: + stream.write(json.dumps(record, sort_keys=True, allow_nan=False)) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + try: + os.link(temporary, path) + except FileExistsError as exc: + raise FileExistsError(f"refusing to overwrite existing report: {path}") from exc + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + +def run_benchmark(manifest: Path, output: Path) -> Counter[str]: + """Evaluate a manifest with the built-in local adapters and write JSONL.""" + cases = load_manifest(manifest) + adapters = default_adapters() + unknown = sorted({case.adapter for case in cases} - adapters.keys()) + if unknown: + raise ValueError(f"unknown adapters: {', '.join(unknown)}") + repository = repository_state() + counts: Counter[str] = Counter() + reference_decoder = lru_cache(maxsize=8)(_decode_image) + + def records() -> Iterable[dict[str, Any]]: + for case in cases: + record = evaluate_case( + case, + adapters=adapters, + repository=repository, + reference_decoder=reference_decoder, + ) + counts[record["detection"]["status"]] += 1 + yield record + + write_jsonl(output, records()) + return counts + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("manifest", type=Path, help="strict benchmark JSONL manifest") + parser.add_argument("--output", type=Path, required=True, help="new JSONL result path; never overwritten") + args = parser.parse_args() + logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") + try: + counts = run_benchmark(args.manifest, args.output) + except (FileNotFoundError, OSError, RuntimeError, ValueError) as exc: + parser.error(str(exc)) + log.info( + "Wrote %s cases to %s: %s detected, %s not detected, %s unavailable, %s errors", + counts.total(), + args.output, + counts["detected"], + counts["not_detected"], + counts["unavailable"], + counts["error"], + ) + + +if __name__ == "__main__": + main() diff --git a/tests/test_script_help.py b/tests/test_script_help.py index 731b508..a8e9e69 100644 --- a/tests/test_script_help.py +++ b/tests/test_script_help.py @@ -19,6 +19,7 @@ ROOT = Path(__file__).resolve().parents[1] "visible_recall_sample.py", "visible_sheets.py", "registered_mark_calibrate.py", + "watermark_benchmark.py", ], ) def test_script_help_exits_cleanly(script: str) -> None: diff --git a/tests/test_watermark_benchmark.py b/tests/test_watermark_benchmark.py new file mode 100644 index 0000000..8fd857c --- /dev/null +++ b/tests/test_watermark_benchmark.py @@ -0,0 +1,390 @@ +"""Contracts for the development-only watermark benchmark kernel.""" + +from __future__ import annotations + +import hashlib +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any + +import numpy as np +import pytest +from PIL import Image + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "watermark_benchmark.py" +OFFICIAL_TRUSTMARK = Path(__file__).resolve().parents[1] / "data/fixtures/provenance/adobe-trustmark-p.png" +SPEC = importlib.util.spec_from_file_location("watermark_benchmark", SCRIPT) +assert SPEC is not None +assert SPEC.loader is not None +MODULE = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = MODULE +SPEC.loader.exec_module(MODULE) + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _image(path: Path, value: int) -> Path: + pixels = np.full((32, 40, 3), value, dtype=np.uint8) + Image.fromarray(pixels, "RGB").save(path) + return path + + +def _row( + artifact: Path, + *, + reference: Path | None = None, + case_id: str = "case-1", + arm: str = "positive", + state: str = "marked", + expected: str = "detected", +) -> dict[str, Any]: + return { + "schema_version": 1, + "case_id": case_id, + "pair_id": "pair-1", + "media_type": "image", + "adapter": "fake", + "arm": arm, + "state": state, + "path": artifact.name, + "sha256": _sha256(artifact), + "reference_path": reference.name if reference is not None else None, + "reference_sha256": _sha256(reference) if reference is not None else None, + "source_revision": "fixture-set@abc123", + "transform": { + "name": "identity", + "revision": "builtin-v1", + "parameters": {}, + }, + "seed": 7, + "expected": expected, + } + + +def _write_manifest(path: Path, rows: list[dict[str, Any]]) -> Path: + path.write_text("\n".join(json.dumps(row) for row in rows) + "\n", encoding="utf-8") + return path + + +class _FakeAdapter: + name = "fake" + source_file = SCRIPT + + def __init__(self, status: str = "detected", label: str | None = "fake mark") -> None: + self.status = status + self.label = label + self.calls: list[Path] = [] + self.images: list[Any] = [] + + @property + def available(self) -> bool: + return self.status != "unavailable" + + def detect(self, path: Path, image: Any) -> Any: + self.calls.append(path) + self.images.append(image) + return MODULE.DetectorOutcome(status=self.status, label=self.label) + + +def test_manifest_preserves_explicit_arms_states_and_revisions(tmp_path: Path) -> None: + artifact = _image(tmp_path / "marked.png", 100) + reference = _image(tmp_path / "clean.png", 90) + manifest = _write_manifest( + tmp_path / "manifest.jsonl", + [ + _row(artifact, reference=reference), + _row( + reference, + case_id="case-2", + arm="hard_negative", + state="clean", + expected="not_detected", + ), + ], + ) + + rows = MODULE.load_manifest(manifest) + + assert [row.arm for row in rows] == ["positive", "hard_negative"] + assert [row.state for row in rows] == ["marked", "clean"] + assert rows[0].path == artifact.resolve() + assert rows[0].reference_path == reference.resolve() + assert rows[0].source_revision == "fixture-set@abc123" + assert rows[0].transform.revision == "builtin-v1" + assert rows[0].seed == 7 + + +@pytest.mark.parametrize("arm", ["positive", "matched_negative", "wrong_key", "hard_negative"]) +def test_manifest_supports_every_evidence_arm(tmp_path: Path, arm: str) -> None: + artifact = _image(tmp_path / f"{arm}.png", 80) + expected = "detected" if arm == "positive" else "not_detected" + manifest = _write_manifest(tmp_path / "manifest.jsonl", [_row(artifact, arm=arm, expected=expected)]) + + assert MODULE.load_manifest(manifest)[0].arm == arm + + +def test_manifest_rejects_duplicate_case_ids(tmp_path: Path) -> None: + first = _image(tmp_path / "first.png", 1) + second = _image(tmp_path / "second.png", 2) + manifest = _write_manifest( + tmp_path / "manifest.jsonl", + [_row(first), _row(second)], + ) + + with pytest.raises(ValueError, match="duplicate case_id"): + MODULE.load_manifest(manifest) + + +def test_manifest_rejects_artifact_hash_drift(tmp_path: Path) -> None: + artifact = _image(tmp_path / "marked.png", 100) + row = _row(artifact) + row["sha256"] = "0" * 64 + manifest = _write_manifest(tmp_path / "manifest.jsonl", [row]) + + with pytest.raises(ValueError, match="sha256 mismatch"): + MODULE.load_manifest(manifest) + + +def test_manifest_requires_reference_path_and_hash_together(tmp_path: Path) -> None: + artifact = _image(tmp_path / "marked.png", 100) + row = _row(artifact) + row["reference_path"] = "clean.png" + manifest = _write_manifest(tmp_path / "manifest.jsonl", [row]) + + with pytest.raises(ValueError, match="reference_path and reference_sha256"): + MODULE.load_manifest(manifest) + + +def test_manifest_rejects_unknown_fields(tmp_path: Path) -> None: + artifact = _image(tmp_path / "marked.png", 100) + row = _row(artifact) + row["implicit_default"] = True + manifest = _write_manifest(tmp_path / "manifest.jsonl", [row]) + + with pytest.raises(ValueError, match="expected exactly"): + MODULE.load_manifest(manifest) + + +@pytest.mark.parametrize( + ("manifest_text", "message"), + [ + ('{"schema_version":1,"schema_version":1}\n', "duplicate field"), + ('{"schema_version":1,"seed":NaN}\n', "non-finite number"), + ], +) +def test_manifest_rejects_non_strict_json(tmp_path: Path, manifest_text: str, message: str) -> None: + manifest = tmp_path / "manifest.jsonl" + manifest.write_text(manifest_text, encoding="utf-8") + + with pytest.raises(ValueError, match=message): + MODULE.load_manifest(manifest) + + +def test_evaluation_keeps_detection_removal_and_fidelity_separate(tmp_path: Path) -> None: + reference = _image(tmp_path / "clean.png", 90) + removed = _image(tmp_path / "removed.png", 100) + manifest = _write_manifest( + tmp_path / "manifest.jsonl", + [_row(removed, reference=reference, state="removed", expected="not_detected")], + ) + case = MODULE.load_manifest(manifest)[0] + adapter = _FakeAdapter(status="not_detected", label=None) + + record = MODULE.evaluate_case( + case, + adapters={"fake": adapter}, + repository={"commit": "abc123", "dirty": False}, + ) + + assert adapter.calls == [removed.resolve()] + assert adapter.images[0].shape == (32, 40, 3) + assert record["detection"] == { + "status": "not_detected", + "label": None, + "expected": "not_detected", + "matches_expected": True, + "positive_evidence": False, + } + assert record["removal"] == { + "attempted": True, + "status": "no_recognized_signal_after_removal", + "certifies_erasure": False, + } + assert record["fidelity"]["status"] == "measured" + assert record["fidelity"]["identical"] is False + assert record["fidelity"]["mae_8bit"] == pytest.approx(10.0) + assert record["fidelity"]["changed_fraction"] == pytest.approx(1.0) + assert record["fidelity"]["psnr_db"] is not None + assert "clean" not in record + + +def test_identical_pair_marks_psnr_as_unbounded_not_zero(tmp_path: Path) -> None: + artifact = _image(tmp_path / "same.png", 90) + manifest = _write_manifest( + tmp_path / "manifest.jsonl", + [_row(artifact, reference=artifact, state="clean", expected="not_detected")], + ) + + record = MODULE.evaluate_case( + MODULE.load_manifest(manifest)[0], + adapters={"fake": _FakeAdapter(status="not_detected", label=None)}, + repository={"commit": "abc123", "dirty": False}, + ) + + assert record["fidelity"] == { + "status": "measured", + "identical": True, + "mae_8bit": 0.0, + "changed_fraction": 0.0, + "psnr_db": None, + "psnr_status": "unbounded_identical", + } + + +def test_missing_reference_is_explicitly_unmeasured(tmp_path: Path) -> None: + artifact = _image(tmp_path / "marked.png", 100) + manifest = _write_manifest(tmp_path / "manifest.jsonl", [_row(artifact)]) + + record = MODULE.evaluate_case( + MODULE.load_manifest(manifest)[0], + adapters={"fake": _FakeAdapter()}, + repository={"commit": "abc123", "dirty": True}, + ) + + assert record["fidelity"] == {"status": "not_measured", "reason": "reference_not_provided"} + assert record["run"]["repository"] == {"commit": "abc123", "dirty": True} + + +def test_unavailable_adapter_is_not_reported_as_negative(tmp_path: Path) -> None: + artifact = _image(tmp_path / "marked.png", 100) + manifest = _write_manifest(tmp_path / "manifest.jsonl", [_row(artifact)]) + adapter = _FakeAdapter(status="unavailable", label=None) + + record = MODULE.evaluate_case( + MODULE.load_manifest(manifest)[0], + adapters={"fake": adapter}, + repository={"commit": "abc123", "dirty": False}, + ) + + assert record["detection"]["status"] == "unavailable" + assert record["detection"]["matches_expected"] is None + assert adapter.calls == [] + + +def test_undecodable_artifact_is_error_not_negative(tmp_path: Path) -> None: + artifact = tmp_path / "broken.png" + artifact.write_bytes(b"not an image") + manifest = _write_manifest(tmp_path / "manifest.jsonl", [_row(artifact)]) + adapter = _FakeAdapter(status="not_detected", label=None) + + record = MODULE.evaluate_case( + MODULE.load_manifest(manifest)[0], + adapters={"fake": adapter}, + repository={"commit": "abc123", "dirty": False}, + ) + + assert record["detection"]["status"] == "error" + assert record["detection"]["matches_expected"] is None + assert record["detection"]["error"] == "artifact could not be decoded" + assert adapter.calls == [] + + +def test_detector_disagreement_is_recorded_not_reinterpreted(tmp_path: Path) -> None: + artifact = _image(tmp_path / "marked.png", 100) + manifest = _write_manifest(tmp_path / "manifest.jsonl", [_row(artifact, expected="detected")]) + + record = MODULE.evaluate_case( + MODULE.load_manifest(manifest)[0], + adapters={"fake": _FakeAdapter(status="not_detected", label=None)}, + repository={"commit": "abc123", "dirty": False}, + ) + + assert record["detection"]["status"] == "not_detected" + assert record["detection"]["matches_expected"] is False + + +def test_shape_mismatch_is_incomparable(tmp_path: Path) -> None: + reference = _image(tmp_path / "clean.png", 90) + artifact = tmp_path / "different-shape.png" + Image.fromarray(np.full((16, 20, 3), 90, dtype=np.uint8), "RGB").save(artifact) + manifest = _write_manifest(tmp_path / "manifest.jsonl", [_row(artifact, reference=reference)]) + + record = MODULE.evaluate_case( + MODULE.load_manifest(manifest)[0], + adapters={"fake": _FakeAdapter()}, + repository={"commit": "abc123", "dirty": False}, + ) + + assert record["fidelity"] == { + "status": "incomparable", + "reason": "shape_mismatch", + "artifact_shape": [16, 20, 3], + "reference_shape": [32, 40, 3], + } + + +def test_non_8_bit_pair_is_incomparable(tmp_path: Path) -> None: + reference = tmp_path / "clean.png" + artifact = tmp_path / "removed.png" + Image.fromarray(np.full((16, 20), 900, dtype=np.uint16)).save(reference) + Image.fromarray(np.full((16, 20), 901, dtype=np.uint16)).save(artifact) + manifest = _write_manifest(tmp_path / "manifest.jsonl", [_row(artifact, reference=reference)]) + + record = MODULE.evaluate_case( + MODULE.load_manifest(manifest)[0], + adapters={"fake": _FakeAdapter()}, + repository={"commit": "abc123", "dirty": False}, + ) + + assert record["fidelity"] == { + "status": "incomparable", + "reason": "unsupported_dtype", + "artifact_dtype": "uint16", + "reference_dtype": "uint16", + } + + +def test_jsonl_writer_refuses_overwrite_and_emits_strict_json(tmp_path: Path) -> None: + output = tmp_path / "report.jsonl" + records = [{"schema_version": 1, "value": None}] + + MODULE.write_jsonl(output, records) + assert output.read_text(encoding="utf-8") == '{"schema_version": 1, "value": null}\n' + + with pytest.raises(FileExistsError): + MODULE.write_jsonl(output, records) + + +def test_builtin_adapters_name_their_exact_source_files() -> None: + adapters = MODULE.default_adapters() + + assert set(adapters) == {"dwt-dct", "trustmark"} + assert adapters["dwt-dct"].source_file.name == "invisible_watermark.py" + assert adapters["trustmark"].source_file.name == "trustmark_detector.py" + + +def test_run_benchmark_writes_official_trustmark_result(tmp_path: Path) -> None: + if not MODULE.default_adapters()["trustmark"].available: + pytest.skip("trustmark not installed") + row = _row(OFFICIAL_TRUSTMARK) + row.update( + { + "adapter": "trustmark", + "path": str(OFFICIAL_TRUSTMARK), + "source_revision": "adobe/trustmark@0ed40cbe8188f664fd9cbbeacd969807de27440a", + } + ) + manifest = _write_manifest(tmp_path / "manifest.jsonl", [row]) + output = tmp_path / "result.jsonl" + + counts = MODULE.run_benchmark(manifest, output) + record = json.loads(output.read_text(encoding="utf-8")) + + assert counts == {"detected": 1} + assert record["detection"]["status"] == "detected" + assert record["detection"]["label"] == "Adobe TrustMark (variant P, schema 1)" + assert record["case_id"] == "case-1"