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

feat: model-free keyed-Gumbel (EXP) same-key detector (#190)

Browse Source 
Implements the replay test of ARBI's keyed-Gumbel technical report
(u = PRF(Hash(key, window), token); exact Gamma-tail p-value; repeated-window
masking) as a stdlib-only detector (detect_gumbel.py) — no GPU, model, or
logits — wired into the TextDetector registry, /capabilities and the
iterative Layer B rewrite loop (--gumbel-key; priority gumbel > markllm >
lexical divergence) with a gumbel.before/after/cleared report.

Also cites the ARBI article in the README bibliography (renamed from
References) and names EXP/Gumbel in the open-LLM vendor row. Same-key-only:
valid against the same key, tokenizer, and PRF layout used at generation;
not a vendor oracle. Key is never logged.

Co-authored-by: guillaumemeyer <guillaumemeyer@users.noreply.github.com>
This commit is contained in:
Guillaume Meyer (The Opinionated Man)
2026-08-19 12:49:11 -07:00
committed by GitHub
co-authored by guillaumemeyer
parent 8c9ff345ed
commit 9dda608a86
7 changed files with 940 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service/scripts/detect_gumbel.py
+308
View File
@@ -0,0 +1,308 @@
#!/usr/bin/env python3
"""Model-free keyed-Gumbel (Aaronson EXP) text-watermark detector.
Implements the detection arithmetic of the ARBI keyed-Gumbel technical report
(Sections 2-3): replay the keyed sampler's noise from the text alone and test
whether the observed tokens look like winners of keyed draws.
seed = Hash(key, last H tokens) H = 4 by default
u_t = PRF(seed, token_t) -> replayable from text alone
S = sum_t -log(1 - u_t) ~ Gamma(counted, 1) under the null
p = P(Gamma(counted, 1) >= S) exact for integer shape
Repeated context windows are masked (the generator falls back to ordinary
randomness on recurrence; the detector applies the same skip rule), so reused
windows contribute no evidence. Positions with fewer than H preceding tokens
have no full window and are not counted either.
Stdlib-only: the p-value uses the exact Poisson-sum identity for an integer
Gamma shape (upper regularized incomplete gamma), so scipy is never required.
Honesty caveat (same as the MarkLLM harness): detection is a *same-key replay*
— it is valid only against the same key, tokenizer, and PRF layout used at
generation (self-hosted engines such as arbi-serve). A negative result
establishes nothing: unwatermarked text, another provider's key, and human
text all sit at chance. This detector is not a vendor oracle.
The default PRF layout here is HMAC-SHA256 over packed token ids. It is a
clean-room, auditable instantiation of the scheme, not bit-compatible with any
specific engine's kernel: for exact replay against a real engine, pass its
token ids (--tokens) and, if the engine uses a different PRF, reimplement the
two functions in this module accordingly.
"""
from __future__ import annotations
import argparse
import hashlib
import hmac
import json
import math
import os
import re
import struct
import sys
from collections.abc import Sequence
from pathlib import Path
from typing import Any
sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import emit_json, eprint, read_text_input
DEFAULT_WINDOW = 4
DEFAULT_THRESHOLD = 1e-6
_ID_PACK = struct.Struct(">Q")
_SIMPLE_TOKEN_RE = re.compile(r"[A-Za-z0-9]+")
def _token_id(tok: str) -> int:
"""Deterministic 64-bit id for a simple-tokenizer token string."""
return int.from_bytes(hashlib.sha256(tok.encode("utf-8")).digest()[:8], "big")
def tokenize_simple(text: str) -> list[int]:
"""Deterministic word/run tokenizer -> stable token ids.
Convenience path for testing and quick checks. Exact replay against a real
engine requires the engine's own tokenizer: use --tokens with its ids.
"""
return [_token_id(t) for t in _SIMPLE_TOKEN_RE.findall(text.lower())]
def load_token_ids(raw: str) -> list[int]:
"""Parse a token-id input: a JSON array, or one integer per line."""
stripped = raw.strip()
if stripped.startswith("["):
data = json.loads(stripped)
if not isinstance(data, list) or not all(isinstance(x, int) for x in data):
raise ValueError("token-id JSON must be an array of integers")
return data
ids: list[int] = []
for raw_line in stripped.splitlines():
line = raw_line.strip()
if not line:
continue
ids.append(int(line, 0))
return ids
def _normalize_key(raw: str) -> bytes:
"""Key -> bytes: 0x<hex> decodes to raw bytes, anything else is UTF-8."""
s = raw.strip()
if s.startswith("0x") or s.startswith("0X"):
hexpart = s[2:]
if (
not hexpart
or len(hexpart) % 2
or any(c not in "0123456789abcdefABCDEF" for c in hexpart)
):
raise ValueError("invalid hex key (expected 0x followed by even-length hex)")
return bytes.fromhex(hexpart)
return s.encode("utf-8")
def _seed(key: bytes, window: tuple[int, ...]) -> bytes:
"""Context-window seed: HMAC-SHA256(key, packed window)."""
packed = b"".join(_ID_PACK.pack(t) for t in window)
return hmac.new(key, packed, hashlib.sha256).digest()
def _uniform(seed: bytes, token_id: int) -> float:
"""Per-candidate uniform in (0, 1): HMAC-SHA256(seed, token id)."""
digest = hmac.new(seed, _ID_PACK.pack(token_id), hashlib.sha256).digest()
return (int.from_bytes(digest[:8], "big") + 0.5) / (1 << 64)
def _poisson_survival(s: float, n: int) -> float:
"""P(Gamma(n, 1) >= s) = e^{-s} * sum_{k=0}^{n-1} s^k / k! (exact for integer n).
Computed via logsumexp so large statistics never overflow or NaN.
"""
if n <= 0 or s <= 0.0:
return 1.0
lns = math.log(s)
log_term = 0.0 # k = 0 term is s^0 / 0! = 1
maxv = 0.0
for k in range(1, n):
log_term += lns - math.log(k)
if log_term > maxv:
maxv = log_term
log_term = 0.0
acc = math.exp(-maxv) # k = 0 term
for k in range(1, n):
log_term += lns - math.log(k)
acc += math.exp(log_term - maxv)
p = math.exp(-s + maxv + math.log(acc))
if p < 0.0:
return 0.0
if p > 1.0:
return 1.0
return p
def detect_token_ids(
token_ids: Sequence[int],
key: str | bytes,
*,
window: int = DEFAULT_WINDOW,
threshold: float = DEFAULT_THRESHOLD,
mask_repeated: bool = True,
) -> dict[str, Any]:
"""Run the keyed-Gumbel replay test over a sequence of token ids.
mask_repeated mirrors the generator's repeated-window masking: a context
window is counted only on its first occurrence (the generator fell back to
ordinary randomness on recurrence, so later occurrences carry no signal).
Positions with fewer than window preceding tokens are never counted.
"""
if window < 1:
raise ValueError("window must be >= 1")
if not 0.0 < threshold < 1.0:
raise ValueError("threshold must be in (0, 1)")
key_bytes = _normalize_key(key) if isinstance(key, str) else bytes(key)
ids = list(token_ids)
for t in ids:
if not isinstance(t, int) or isinstance(t, bool) or t < 0 or t > (1 << 64) - 1:
raise ValueError(f"token id out of range: {t!r}")
total = len(ids)
statistic = 0.0
counted = 0
seen_windows: set[tuple[int, ...]] = set()
skipped_repeated = 0
for t in range(window, total):
win = tuple(ids[t - window : t])
if mask_repeated:
if win in seen_windows:
skipped_repeated += 1
continue
seen_windows.add(win)
u = _uniform(_seed(key_bytes, win), ids[t])
statistic += -math.log1p(-u)
counted += 1
skipped_no_context = min(window, total)
report: dict[str, Any] = {
"detector": "gumbel",
"scheme": "exp",
"vendor": "self-hosted",
"available": True,
"window": window,
"threshold": threshold,
"tokens_total": total,
"skipped_no_context": skipped_no_context,
"skipped_repeated": skipped_repeated,
"counted": counted,
}
if counted == 0:
report["is_watermarked"] = False
report["p_value"] = 1.0
report["score"] = 0.0
report["note"] = "no verifiable token positions (text too short for a full context window)"
return report
p = _poisson_survival(statistic, counted)
report["statistic"] = round(statistic, 6)
report["p_value"] = p
report["score"] = round(-math.log10(p) if p > 0.0 else 300.0, 6)
report["is_watermarked"] = p < threshold
report["note"] = (
"same-key replay of the keyed-Gumbel (Aaronson EXP) watermark; valid only "
"against the same key, tokenizer, and PRF layout used at generation"
)
return report
def detect_text(
text: str,
key: str | bytes,
*,
window: int = DEFAULT_WINDOW,
threshold: float = DEFAULT_THRESHOLD,
) -> dict[str, Any]:
"""Run the replay test over plain text via the deterministic tokenizer."""
return detect_token_ids(tokenize_simple(text), key, window=window, threshold=threshold)
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("path", nargs="?", default="-", help="Text file, or - for stdin")
(
p.add_argument(
"--tokens",
action="store_true",
help="Treat the input as token ids (JSON array or one integer per line) "
"instead of text — required for exact replay with an engine's tokenizer",
),
)
(
p.add_argument(
"--key",
default=os.environ.get("WATERMARKS_GUMBEL_KEY"),
help="Watermark key (0x<hex> or a string); default: $WATERMARKS_GUMBEL_KEY. "
"Preferred via env — keys on argv are visible in ps/history.",
),
)
(
p.add_argument(
"--window",
type=int,
default=DEFAULT_WINDOW,
help=f"Context window size H in tokens (default: {DEFAULT_WINDOW})",
),
)
(
p.add_argument(
"--threshold",
type=float,
default=DEFAULT_THRESHOLD,
help=f"p-value threshold for is_watermarked (default: {DEFAULT_THRESHOLD:g})",
),
)
p.add_argument("--json", action="store_true", help="Emit the report as JSON on stdout")
(
p.add_argument(
"--force-text",
action="store_true",
help="Process input even when it looks like a binary container",
),
)
return p
def main() -> int:
args = build_parser().parse_args()
if not args.key:
eprint("error: no watermark key (pass --key or set WATERMARKS_GUMBEL_KEY)")
return 2
raw = read_text_input(args.path, allow_binary=args.force_text)
try:
if args.tokens:
report = detect_token_ids(
load_token_ids(raw),
args.key,
window=args.window,
threshold=args.threshold,
)
else:
report = detect_text(raw, args.key, window=args.window, threshold=args.threshold)
except (ValueError, json.JSONDecodeError) as e:
eprint(f"error: {e}")
return 2
if args.json:
emit_json(report)
else:
print(
f"keyed-Gumbel (EXP) detection: watermarked={report['is_watermarked']} "
f"p={report['p_value']:.3g} (threshold {report['threshold']:g}) "
f"counted={report['counted']}/{report['tokens_total']} tokens, skipped "
f"{report['skipped_no_context'] + report['skipped_repeated']} "
f"(no-context {report['skipped_no_context']}, repeated-window "
f"{report['skipped_repeated']})"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
service/scripts/rewrite_text.py
+59 -13
View File
@@ -19,12 +19,13 @@ Rewriting is iterative and evaluation-driven: each loop generates
--candidates (default 1) variants, evaluates each, and stops as soon as an
attempt passes watermark detection; --max-loops (default 1) caps how many
evaluation rounds run before the best-effort variant is returned
(WATERMARKS_REWRITE_LOOPS). The evaluator is chosen by priority: MarkLLM
same-config detection (when --markllm-scheme is passed) or, when no detector
is configured, bigram-Jaccard lexical divergence (no pass/fail verdict — all
(WATERMARKS_REWRITE_LOOPS). The evaluator is chosen by priority: keyed-Gumbel
same-key replay (when --gumbel-key / WATERMARKS_GUMBEL_KEY is set), else
MarkLLM same-config detection (--markllm-scheme), else, when no detector is
configured, bigram-Jaccard lexical divergence (no pass/fail verdict — all
attempts are generated and the most diverged one is selected). A vendor-detector
seam (Google's retired SynthID-text detector) is reserved ahead of MarkLLM
should a vendor endpoint return.
seam (Google's retired SynthID-text detector) is reserved ahead of the
same-config detectors should a vendor endpoint return.
Security notes:
- Only http(s) endpoints are accepted; redirects are refused outright so an
@@ -49,7 +50,7 @@ from urllib.parse import urlparse
sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import cleaned_path, eprint, read_text_input, write_text_output
from text_detectors import MarkLLMTextDetector
from text_detectors import GumbelTextDetector, MarkLLMTextDetector
from text_unicode import clean_text
DEFAULT_MARKLLM_MODEL = "facebook/opt-1.3b"
@@ -205,16 +206,20 @@ def _safe_detect(detector: object, text: str) -> dict:
def _pick_evaluator(
markllm_detector: MarkLLMTextDetector | None,
gumbel_detector: GumbelTextDetector | None,
) -> tuple[str, object | None]:
"""Pick the evaluator that drives the iterative rewrite loop.
Priority: MarkLLM same-config detection (when the caller passed
--markllm-scheme) > bigram-Jaccard lexical divergence (fallback with no
pass/fail verdict). A vendor-detector seam (Google's SynthID-text detector,
retired Aug 2026) is reserved ahead of MarkLLM should a vendor endpoint
return; it only needs available()/detect()/name per the TextDetector
protocol in text_detectors.py.
Priority: keyed-Gumbel same-key replay (when the caller passed
--gumbel-key) > MarkLLM same-config detection (--markllm-scheme) >
bigram-Jaccard lexical divergence (fallback with no pass/fail verdict).
A vendor-detector seam (Google's SynthID-text detector, retired Aug 2026)
is reserved ahead of both should a vendor endpoint return; it only needs
available()/detect()/name per the TextDetector protocol in
text_detectors.py.
"""
if gumbel_detector is not None:
return "gumbel", gumbel_detector
if markllm_detector is not None:
return "markllm", markllm_detector
return "lexical-divergence", None
@@ -356,6 +361,7 @@ def rewrite(
markllm_dir: str | None = None,
markllm_model: str | None = None,
markllm_timeout: float = 180.0,
gumbel_key: str | None = None,
) -> tuple[str, dict]:
prompt = build_prompt(strength, text, lang=lang, original_lang=original_lang)
info: dict = {
@@ -387,6 +393,15 @@ def rewrite(
eprint(f"markllm verification unavailable: {markllm['before']['error']}")
info["markllm"] = markllm
gumbel: dict | None = None
gumbel_detector: GumbelTextDetector | None = None
if gumbel_key:
gumbel_detector = GumbelTextDetector(key=gumbel_key)
gumbel = {"before": _safe_detect(gumbel_detector, text)}
if not gumbel["before"]["available"]:
eprint(f"gumbel verification unavailable: {gumbel['before']['error']}")
info["gumbel"] = gumbel
if backend == "print-prompt":
info["mode"] = "print-prompt"
if candidates > 1:
@@ -404,7 +419,7 @@ def rewrite(
n_loops = max(1, max_loops)
info["candidates"] = n_cands
info["max_loops"] = n_loops
evaluator_name, evaluator = _pick_evaluator(markllm_detector)
evaluator_name, evaluator = _pick_evaluator(markllm_detector, gumbel_detector)
info["evaluator"] = evaluator_name
# Iterative rewrite: each loop generates --candidates variants and
@@ -522,6 +537,28 @@ def rewrite(
"keys used at generation; it does not certify a vendor detector."
)
if gumbel:
assert gumbel_detector is not None # set together with gumbel above
if evaluator_name == "gumbel":
# The loop already scored the selected attempt; reuse the verdict
# instead of paying another replay.
after = rec["evaluation"]
else:
after = _safe_detect(gumbel_detector, out)
gumbel["after"] = after
before = gumbel["before"]
if before.get("available") and after.get("available"):
gumbel["cleared"] = bool(
before.get("is_watermarked") and not after.get("is_watermarked")
)
else:
gumbel["cleared"] = None
gumbel["note"] = (
"Keyed-Gumbel detection is a same-key replay: valid only with the "
"same key, tokenizer, and PRF layout used at generation; it does "
"not certify a vendor detector."
)
eprint(
f"note: evaluator={evaluator_name} attempts={len(attempts)}/"
f"{n_cands * n_loops} loops={n_loops} passed={passed}"
@@ -623,6 +660,14 @@ def build_parser() -> argparse.ArgumentParser:
default=float(_env("WATERMARKS_MARKLLM_TIMEOUT", "180.0")),
help="Timeout per MarkLLM detection call (default: 180.0)",
)
p.add_argument(
"--gumbel-key",
default=_env("WATERMARKS_GUMBEL_KEY"),
help="Secret key for keyed-Gumbel (Aaronson EXP) same-key replay "
"detection; drives the iterative rewrite loop as the evaluator when "
"set (default: $WATERMARKS_GUMBEL_KEY). Preferred via env — keys on "
"argv are visible in ps/history; never logged.",
)
p.add_argument(
"--force-text",
action="store_true",
@@ -661,6 +706,7 @@ def main() -> int:
markllm_dir=args.markllm_dir,
markllm_model=args.markllm_model,
markllm_timeout=args.markllm_timeout,
gumbel_key=args.gumbel_key,
)
except (urllib.error.URLError, TimeoutError, RuntimeError) as e:
eprint(f"rewrite failed: {e}")
service/scripts/text_detectors.py
+102 -4
View File
@@ -17,6 +17,10 @@ Detectors:
- markllm — research harness (KGW / SynthID schemes) via
detect_text_watermark.py, activated by MARKLLM_DIR. Same-config-only
detection; not a vendor oracle.
- gumbel — model-free same-key replay of the keyed-Gumbel (Aaronson EXP)
scheme (detect_gumbel.py), activated by WATERMARKS_GUMBEL_KEY. Stdlib-only;
valid only against the same key, tokenizer, and PRF layout used at
generation (self-hosted engines such as arbi-serve); not a vendor oracle.
- claude-text — placeholder for Anthropic's announced text-watermark
detection API. Reports unavailable until a public endpoint exists; the
interface it must implement is already defined here.
@@ -40,6 +44,8 @@ from collections.abc import Callable
from pathlib import Path
from typing import Any, Protocol
from detect_gumbel import DEFAULT_THRESHOLD, DEFAULT_WINDOW, detect_text
DEFAULT_MARKLLM_SCHEME = "kgw"
DEFAULT_MARKLLM_TIMEOUT = 600.0
@@ -251,6 +257,75 @@ class MarkLLMTextDetector:
return payload
# ---------------------------------------------------------------------------
# Keyed-Gumbel (Aaronson EXP) — model-free same-key replay
# ---------------------------------------------------------------------------
class GumbelTextDetector:
"""Same-key replay of the keyed-Gumbel (Aaronson EXP) text watermark.
Model-free: implements the detection arithmetic of the ARBI keyed-Gumbel
technical report (Sections 2-3) — replay u = PRF(Hash(key, window), token)
from the text alone and test the Gamma tail — so it needs no GPU, model,
or logits. Detection is valid only against the SAME key, tokenizer, and
PRF layout used at generation (self-hosted engines such as arbi-serve);
it is not a vendor oracle. Key from WATERMARKS_GUMBEL_KEY (env) or the
constructor override.
"""
name = "gumbel"
vendor = "self-hosted"
def __init__(
self,
*,
key: str | None = None,
window: int | None = None,
threshold: float | None = None,
) -> None:
self._key = key
self._window = window
self._threshold = threshold
def _key_env(self) -> str | None:
if self._key:
return self._key
return os.environ.get("WATERMARKS_GUMBEL_KEY", "").strip() or None
def available(self) -> bool:
return self._key_env() is not None
def detect(self, text: str) -> dict[str, Any]:
key = self._key_env()
report: dict[str, Any] = {
"detector": self.name,
"scheme": "exp",
"vendor": self.vendor,
"available": False,
}
if key is None:
report["error"] = "WATERMARKS_GUMBEL_KEY not set"
return report
try:
payload = detect_text(
text,
key,
window=self._window or DEFAULT_WINDOW,
threshold=self._threshold or DEFAULT_THRESHOLD,
)
except Exception as e: # fail-soft contract: never raise
report["error"] = f"keyed-Gumbel detection failed: {e}"
return report
payload["detector"] = self.name
payload["note"] = (
"same-key replay of the keyed-Gumbel (Aaronson EXP) watermark: valid "
"only with the same key, tokenizer, and PRF layout used at generation; "
"not a vendor detector."
)
return payload
# ---------------------------------------------------------------------------
# Claude (Anthropic) — announced detector API, not yet public
# ---------------------------------------------------------------------------
@@ -290,11 +365,17 @@ class ClaudeTextDetector:
def all_detectors(
markllm: MarkLLMTextDetector | None = None, *, include_markllm: bool = True
markllm: MarkLLMTextDetector | None = None,
*,
include_markllm: bool = True,
gumbel: GumbelTextDetector | None = None,
include_gumbel: bool = True,
) -> list[TextDetector]:
detectors: list[TextDetector] = []
if include_markllm:
detectors.append(markllm or MarkLLMTextDetector())
if include_gumbel:
detectors.append(gumbel or GumbelTextDetector())
detectors.append(ClaudeTextDetector())
return detectors
@@ -309,14 +390,24 @@ def run_all_text_detectors(
*,
markllm: MarkLLMTextDetector | None = None,
include_markllm: bool = True,
gumbel: GumbelTextDetector | None = None,
include_gumbel: bool = True,
) -> list[dict[str, Any]]:
"""Run every detector (including unavailable ones, with reasons).
markllm injects a caller-parameterized MarkLLM detector (e.g. one
driven by rewrite_text.py CLI flags); pass include_markllm=False to
exclude the MarkLLM harness entirely.
exclude the MarkLLM harness entirely. Same for gumbel.
"""
return [d.detect(text) for d in all_detectors(markllm, include_markllm=include_markllm)]
return [
d.detect(text)
for d in all_detectors(
markllm,
include_markllm=include_markllm,
gumbel=gumbel,
include_gumbel=include_gumbel,
)
]
def run_text_detectors(
@@ -324,10 +415,17 @@ def run_text_detectors(
*,
markllm: MarkLLMTextDetector | None = None,
include_markllm: bool = True,
gumbel: GumbelTextDetector | None = None,
include_gumbel: bool = True,
) -> list[dict[str, Any]]:
"""Run only the detectors that are configured and usable."""
return [
d.detect(text)
for d in all_detectors(markllm, include_markllm=include_markllm)
for d in all_detectors(
markllm,
include_markllm=include_markllm,
gumbel=gumbel,
include_gumbel=include_gumbel,
)
if d.available()
]