#!/usr/bin/env python3 """Stormtape public AudioSeal verifier / Öffentlicher Stormtape-AudioSeal-Prüfer. Stormtape (stormtape.com) marks its synthetic speech segments with an imperceptible AudioSeal watermark carrying the 16-bit payload 0x5354 (ASCII "ST"). This standalone script lets anyone verify that marking in downloaded audio — independently of Stormtape's own infrastructure. Deutsch: Stormtape markiert seine synthetischen Sprachsegmente mit einem unhörbaren AudioSeal-Wasserzeichen (16-Bit-Payload 0x5354, ASCII "ST"). Dieses eigenständige Skript erlaubt jedem, diese Markierung unabhängig zu prüfen. Usage / Nutzung: pip install audioseal torch soundfile numpy # Published file outputs (Shorts, downloads) — marking starts at sample 0 python stormtape-audioseal-verifier.py [more.wav ...] # Excerpt cut out of the LIVESTREAM — start offset is unknown python stormtape-audioseal-verifier.py --livestream Notes: - Input should be PCM WAV (convert other formats first, e.g. with ffmpeg: ``ffmpeg -i input.m4a -ac 1 -c:a pcm_s16le out.wav``). - Detection runs at the AudioSeal-native 16 kHz; other sample rates are resampled here with a windowed-sinc kernel before detection. - A detection score close to 1.0 on speech segments plus the payload "ST" identifies Stormtape-marked speech. Music-only passages are unmarked by design; scores there stay low. - The detector is the official Meta AudioSeal 16-bit detector loaded via the ``audioseal`` package. No Stormtape secret is required — detectability by third parties is the point of the marking (Art. 50(2) EU AI Act). Why ``--livestream`` exists (measured by Stormtape on 2026-08-01, real capture): the AudioSeal detector is NOT shift-invariant. The network runs at 16 kHz with a total stride of 320 samples, so an excerpt is only decoded reliably when its first sample sits on the same frame grid as the marking. At 48 kHz that grid has a period of 960 samples (20 ms) = 3 (resampling 48→16) × 320 (model frame). A file output always starts where its marking starts, so the phase is unambiguous and the default mode is sufficient. An excerpt cut from the livestream starts at an arbitrary point, so ``--livestream`` tries all 960 start phases and reports the first window that verifies. Evidence: from a non-detecting position, shifting by exactly 669 samples (bringing the phase to 0) yields score 0.9939 with the correct payload, while shifts of 300, 600 or 960 samples all stay undetected. ``--livestream`` additionally probes several start points inside the excerpt, not just its beginning (measured 2026-08-02 on a 30-minute capture). Stormtape marks an announced window in overlapping blocks; where two blocks cross-fade, two watermarks sit on top of each other and the detector reports a score near 1.0 while the payload does not always decode. Observed: no decode at 802.0 s, clear hits 1.5 s before and after. Probing only the first five seconds could therefore report a correctly marked capture as unverified — a false negative in the very tool third parties use to check the claim. Stormtape marks only announced speech windows, padded by 2 s, so the music bed stays unmarked — a hit therefore proves synthetic speech at that spot, not merely "somewhere in the stream". This file is integrity-pinned in Stormtape's operations tooling; its SHA-256 is published in the audit trail. """ from __future__ import annotations import argparse import json import math import os import sys # --- Thread limit / Threadgrenze ------------------------------------------- # # This MUST run before torch is imported. Afterwards the thread pools already # exist and these variables have no effect. # # Without a limit, PyTorch spawns as many compute threads as the machine has # logical cores. On 2026-08-02 that made an otherwise idle workstation stop # responding during a detection run. This script runs on YOUR machine, so it # stays modest by default. Raise it if you want it faster: # # STORMTAPE_VERIFY_THREADS=8 python stormtape-audioseal-verifier.py ... # # The limit changes only how fast detection runs, never its result. # # Deutsch: Muss VOR dem Import von torch laufen -- danach sind die Threadpools # schon aufgebaut und die Variablen wirkungslos. Ohne Grenze spannt PyTorch so # viele Straenge auf, wie der Rechner logische Kerne hat. Die Grenze aendert nur # die Geschwindigkeit, nie das Ergebnis. try: STORMTAPE_VERIFY_THREADS = max(1, int(os.environ.get("STORMTAPE_VERIFY_THREADS", "4"))) except ValueError: # Ein unlesbarer Wert darf das Werkzeug nicht abbrechen lassen -- der # Prueferfolg haengt nicht an der Threadzahl. STORMTAPE_VERIFY_THREADS = 4 for _thread_var in ( "OMP_NUM_THREADS", "MKL_NUM_THREADS", "OPENBLAS_NUM_THREADS", "NUMEXPR_NUM_THREADS", "TORCH_NUM_THREADS", ): # setdefault, nicht ueberschreiben: Wer die Variablen selbst gesetzt hat, # meint es so. os.environ.setdefault(_thread_var, str(STORMTAPE_VERIFY_THREADS)) # Wartende OpenMP-Straenge sollen den Prozessor abgeben statt zu kreisen. os.environ.setdefault("OMP_WAIT_POLICY", "PASSIVE") os.environ.setdefault("KMP_BLOCKTIME", "0") STORMTAPE_PAYLOAD_HEX = "5354" STORMTAPE_PAYLOAD_BITS = format(int(STORMTAPE_PAYLOAD_HEX, 16), "016b") MARKING_DOMAIN_HZ = 16000 STRICT_THRESHOLD = 0.99 PRESENCE_THRESHOLD = 0.5 MODEL_FRAME_16K = 320 # AudioSeal total stride at 16 kHz LIVESTREAM_WINDOW_SECONDS = 5.0 # Abstand der Startpunkte innerhalb eines Ausschnitts. 1,5 s genuegt: in der # Messung vom 02.08.2026 lagen nicht entschluesselnde Stellen jeweils weniger # als 1,5 s von einem klaren Treffer entfernt. LIVESTREAM_PROBE_STEP_SECONDS = 1.5 # Obergrenze, damit ein sehr langer Mitschnitt nicht stundenlang rechnet. Jeder # Startpunkt kostet eine volle Phasensuche ueber 960 Positionen. LIVESTREAM_MAX_PROBES = 12 _LOWPASS_WIDTH = 64 _ROLLOFF = 0.9475937167399596 _KAISER_BETA = 14.769656459379492 def resample_to_16k(torch, tensor, orig_freq: int): """Windowed-sinc polyphase resample of (batch, channel, time) to 16 kHz.""" if orig_freq == MARKING_DOMAIN_HZ: return tensor gcd = math.gcd(orig_freq, MARKING_DOMAIN_HZ) orig = orig_freq // gcd new = MARKING_DOMAIN_HZ // gcd base_freq = min(orig, new) * _ROLLOFF width = math.ceil(_LOWPASS_WIDTH * orig / base_freq) idx = torch.arange(-width, width + orig, dtype=torch.float64)[None, None] / orig t = torch.arange(0, -new, -1, dtype=torch.float64)[:, None, None] / new + idx t = (t * base_freq).clamp_(-_LOWPASS_WIDTH, _LOWPASS_WIDTH) window = torch.i0( _KAISER_BETA * torch.sqrt((1 - (t / _LOWPASS_WIDTH) ** 2).clamp_(min=0.0)) ) / torch.i0(torch.tensor(_KAISER_BETA, dtype=torch.float64)) t = t * math.pi kernel = torch.where(t == 0, torch.tensor(1.0, dtype=torch.float64), torch.sin(t) / t) kernel = kernel * window * (base_freq / orig) batch, channels, length = tensor.shape x = tensor.to(torch.float64).reshape(batch * channels, 1, length) x = torch.nn.functional.pad(x, (width, width + orig)) out = torch.nn.functional.conv1d(x, kernel, stride=orig) out = out.transpose(1, 2).reshape(batch * channels, 1, -1) out = out[..., : int(math.ceil(new * length / orig))] return out.reshape(batch, channels, -1).to(tensor.dtype) def verify_file(torch, detector, path: str) -> dict: import numpy as np import soundfile as sf audio, sample_rate = sf.read(path, dtype="float32", always_2d=True) if audio.shape[1] > 1: audio = audio.mean(axis=1, keepdims=True) if not np.isfinite(audio).all(): raise RuntimeError("audio contains non-finite samples") if audio.shape[0] < sample_rate // 4: raise RuntimeError("audio shorter than 0.25 s cannot be verified reliably") tensor = torch.from_numpy(audio.T.copy()).unsqueeze(0) with torch.inference_mode(): probe = resample_to_16k(torch, tensor, int(sample_rate)) result, message = detector.detect_watermark(probe) score = float(result.item() if hasattr(result, "item") else result) bits = "".join( "1" if float(bit) >= 0.5 else "0" for bit in message.reshape(-1).tolist() ) payload_matches = bits == STORMTAPE_PAYLOAD_BITS return { "file": path, "sampleRate": int(sample_rate), "detectionScore": round(score, 6), "detectedBits": bits, "expectedBits": STORMTAPE_PAYLOAD_BITS, "expectedPayloadAscii": "ST", "payloadMatches": payload_matches, "watermarkPresent": math.isfinite(score) and score >= PRESENCE_THRESHOLD, "stormtapeMarkVerified": math.isfinite(score) and score >= STRICT_THRESHOLD and payload_matches, } def _decode(detector, probe, torch) -> tuple: with torch.inference_mode(): result, message = detector.detect_watermark(probe) score = float(result.item() if hasattr(result, "item") else result) bits = "".join( "1" if float(bit) >= 0.5 else "0" for bit in message.reshape(-1).tolist() ) return score, bits def verify_livestream_excerpt(torch, detector, path: str) -> dict: """Excerpt from the continuous stream: search every start phase. The excerpt begins at an arbitrary sample, so its frame grid generally does not match the marking's. All ``ratio * 320`` phases are tried; the first window that clears the strict threshold with the exact payload is reported. Music-only excerpts legitimately yield no hit. """ import numpy as np import soundfile as sf audio, sample_rate = sf.read(path, dtype="float32", always_2d=True) if audio.shape[1] > 1: audio = audio.mean(axis=1, keepdims=True) if not np.isfinite(audio).all(): raise RuntimeError("audio contains non-finite samples") sample_rate = int(sample_rate) gcd = math.gcd(sample_rate, MARKING_DOMAIN_HZ) ratio = sample_rate // gcd period = ratio * MODEL_FRAME_16K window = int(LIVESTREAM_WINDOW_SECONDS * sample_rate) window16 = int(LIVESTREAM_WINDOW_SECONDS * MARKING_DOMAIN_HZ) if audio.shape[0] < window + period: raise RuntimeError( f"livestream mode needs at least " f"{(window + period) / sample_rate:.1f} s of audio" ) mono = audio[:, 0] best_score = 0.0 best_bits = "" checked = 0 # Nicht nur den Anfang des Ausschnitts pruefen, sondern mehrere Startpunkte. # # Warum (gemessen 2026-08-02 an einem 30-Minuten-Mitschnitt): Stormtape # markiert ein angemeldetes Fenster in ueberlappenden Bloecken. In der # Ueberblendung zweier Bloecke liegen zwei Wasserzeichen uebereinander; der # Detektor meldet dort zwar einen Wert nahe 1,0, entschluesselt die Kennung # aber nicht immer. Beispiel aus der Messung: bei 802,0 s kein Nachweis, # 1,5 s davor und danach jeweils klarer Treffer. # # Prueft man nur den Ausschnittsanfang, kann ein voellig korrekt markierter # Mitschnitt als "nicht verifiziert" gemeldet werden - ein falsches Negativ # ausgerechnet in dem Werkzeug, mit dem Dritte unsere Zusicherung # nachpruefen sollen. Deshalb wird der Ausschnitt in Schritten abgesucht, # bis ein Startpunkt verifiziert. step = int(LIVESTREAM_PROBE_STEP_SECONDS * sample_rate) last_start = max(0, audio.shape[0] - window - period) starts = list(range(0, last_start + 1, step)) or [0] starts = starts[:LIVESTREAM_MAX_PROBES] for start in starts: for sub in range(ratio): span = mono[start + sub : start + sub + window + period] if span.shape[0] < window + period: continue tensor = torch.from_numpy(span[None, None, :].copy()) with torch.inference_mode(): span16 = resample_to_16k(torch, tensor, sample_rate) for frame in range(MODEL_FRAME_16K): if frame + window16 > span16.shape[-1]: break checked += 1 score, bits = _decode(detector, span16[..., frame : frame + window16], torch) if score >= STRICT_THRESHOLD and bits == STORMTAPE_PAYLOAD_BITS: return { "file": path, "mode": "livestream", "sampleRate": sample_rate, "detectionScore": round(score, 6), "detectedBits": bits, "expectedBits": STORMTAPE_PAYLOAD_BITS, "expectedPayloadAscii": "ST", "payloadMatches": True, "watermarkPresent": True, "stormtapeMarkVerified": True, "hitAtSeconds": round((start + sub + frame * ratio) / sample_rate, 6), "probesChecked": starts.index(start) + 1, "phasesChecked": checked, } if score > best_score: best_score, best_bits = score, bits return { "file": path, "mode": "livestream", "sampleRate": sample_rate, "detectionScore": round(best_score, 6), "detectedBits": best_bits, "expectedBits": STORMTAPE_PAYLOAD_BITS, "expectedPayloadAscii": "ST", "payloadMatches": best_bits == STORMTAPE_PAYLOAD_BITS, "watermarkPresent": best_score >= PRESENCE_THRESHOLD, "stormtapeMarkVerified": False, "hitAtSeconds": None, "phasesChecked": checked, } def main() -> int: parser = argparse.ArgumentParser( description="Verify Stormtape's AudioSeal speech marking in audio files." ) parser.add_argument("files", nargs="+", help="PCM WAV file(s) to verify") parser.add_argument( "--livestream", action="store_true", help="excerpt cut from the livestream: search all start phases " "(see module docstring); needs >= 5 s containing an announcement", ) args = parser.parse_args() try: import torch from audioseal import AudioSeal except ImportError as error: print( json.dumps( { "error": f"missing dependency: {error.name}", "hint": "pip install audioseal torch soundfile numpy", } ) ) return 2 # Belt and braces: the environment variables above only work if nothing # imported torch earlier. Setting the limit in torch itself covers that. # set_num_interop_threads raises once work has begun -- then the main limit # is already in place and that is enough. try: torch.set_num_threads(STORMTAPE_VERIFY_THREADS) torch.set_num_interop_threads(1) except Exception: # noqa: BLE001 try: torch.set_num_threads(STORMTAPE_VERIFY_THREADS) except Exception: # noqa: BLE001 pass detector = AudioSeal.load_detector("audioseal_detector_16bits") detector.eval() failures = 0 for path in args.files: try: report = ( verify_livestream_excerpt(torch, detector, path) if args.livestream else verify_file(torch, detector, path) ) except Exception as error: # noqa: BLE001 - report and continue report = {"file": path, "error": str(error)} failures += 1 print(json.dumps(report, ensure_ascii=False, sort_keys=True)) return 1 if failures else 0 if __name__ == "__main__": raise SystemExit(main())