- 
- 
- 
+ 
+ 
-- Discovering hidden AI watermark patterns through signal analysis + Discovering, detecting, and surgically removing Google's AI watermark through spectral analysis
+ 
+ 
+
+ Left: SynthID watermark extracted from a pure-black Gemini image (enhanced 100Γ). Right: Same watermark on a white background. The diagonal stripe pattern and carrier frequencies are clearly visible. +
-| Metric | Rate | Description | -|:-------|:----:|:------------| -| **Frequency Domain Modifications** | 100.0% | All images show spectral changes | -| **Significant Color Shifts** | 95.3% | Mean shift > 1.0 in RGB channels | -| **Perceptual Hash Changes** | 66.0% | Invisible modifications detected | -| **LSB Anomalies** | 10.2% | Least significant bit patterns | -| **2+ Watermark Indicators** | 99.9% | Multi-layer evidence | -| **3+ Watermark Indicators** | 69.2% | Strong multi-layer evidence | +### Carrier Frequency Discovery -### Watermark Confidence Distribution +The watermark embeds energy at specific carrier frequencies with **>99.9% phase coherence** across all images: -``` -0 indicators: 0 ( 0.0%) -1 indicator: 122 ( 0.1%) -2 indicators: 37,832 (30.7%) βββββββββββββββ -3 indicators: 74,525 (60.5%) ββββββββββββββββββββββββββββββ -4 indicators: 10,789 ( 8.8%) ββββ -``` +| Carrier Frequency (fy, fx) | Phase Coherence | Magnitude | Phase (rad) | +|:--------------------------:|:---------------:|:---------:|:-----------:| +| **(Β±14, Β±14)** | 99.96% | 16,807 | Β±1.44 | +| **(Β±126, Β±14)** | 99.96% | 8,046 | Β±2.37 | +| **(Β±98, β14)** | 99.94% | 6,283 | Β±0.61 | +| **(Β±128, Β±128)** | 99.25% | 6,908 | Β±2.29 | +| **(Β±210, β14)** | 99.96% | 6,032 | Β±1.13 | +| **(Β±238, Β±14)** | 99.90% | 4,190 | Β±1.61 | -### Extracted Watermark Visualizations +> **Key insight:** Most carriers cluster along the `y = Β±14` line in frequency space, suggesting a structured frequency selection algorithm. The diagonal stripe pattern visible in the enhanced images corresponds to these carrier frequencies. -
+### Phase Consistency - A Fixed Model-Level Key
-**Extracted Watermark Pattern**
-
+The watermark's phase template is **identical across all images** from the same Gemini model:
- |
-
+- **Green channel phase std**: < 0.007 radians across 50 reference images
+- **Cross-image correlation**: 21.8% mean pairwise noise correlation
+- **Noise structure ratio**: 1.32 Β± 0.02 (byproduct of the neural encoder)
-**Comprehensive Analysis**
-
+This means SynthID does not embed per-image messages - it uses a **fixed spectral fingerprint** that can be profiled and subtracted.
- |
-
+### Frequency Spectrum Analysis
-**Frequency Spectrum**
-
+
+ + Left: FFT magnitude spectrum showing bright carrier frequency peaks. Right: Reconstructed carrier pattern showing the diagonal structure. + - |
-
-
-**Enhanced Difference Pattern**
-
-
- |
-
+ 
+
+ Detailed frequency analysis: Average magnitude spectrum (left) and phase coherence map (right). The carrier positions are marked with crosshairs. +
--- -## π¬ SynthID (Google Gemini) Analysis +## ποΈ Architecture -Analysis of **250 AI-generated images** from Google Gemini to reverse-engineer SynthID watermarking. +### Three Generations of Bypass -### Key Discovery +| Version | Approach | PSNR | Detection Impact | Status | +|:-------:|:---------|:----:|:----------------:|:------:| +| **V1** | JPEG compression (Q50) | 37 dB | ~11% phase drop | β Baseline | +| **V2** | Multi-stage transforms (noise, color, frequency) | 27-37 dB | ~0% confidence drop | β Quality trade-off | +| **V3** | Spectral codebook subtraction | **33-43 dB** | **1-7% confidence drop** | β Best quality | -SynthID uses **spread-spectrum phase encoding** in the frequency domainβnot LSB replacement or simple noise addition. The watermark embeds information through precise phase relationships at specific carrier frequencies. - -## π¬ Discovered Patterns - -| Carrier Frequency | Phase Coherence | Description | -|:----------------:|:---------------:|:------------| -| **(Β±14, Β±14)** | 99.99% | Primary diagonal carrier | -| **(Β±126, Β±14)** | 99.97% | Secondary horizontal | -| **(Β±98, Β±14)** | 99.94% | Tertiary carrier | -| **(Β±128, Β±128)** | 99.92% | Center frequency | -| **(Β±210, Β±14)** | 99.77% | Extended carrier | -| **(Β±238, Β±14)** | 99.71% | Edge carrier | - -### Detection Metrics -- **Noise Correlation**: ~0.218 between watermarked images -- **Structure Ratio**: ~1.32 -- **Detection Threshold**: correlation > 0.179 - -## πΌοΈ Extracted Watermark Visualizations - -
-
-**Enhanced Visualization (500x Amplification)**
-
-
- |
-
-
-**Frequency Domain Carriers**
-
-
- |
-
-
-**False Color (HSV Encoding)**
-
-
- |
-
-
-**Phase Encoding Pattern**
-
-
- |
-
+ 
+ 
+
+ Left: Original SynthID-watermarked Gemini image. Right: After V3 spectral bypass - visually identical, watermark energy reduced. +
+ +--- + ## π Quick Start ### Installation @@ -217,159 +160,235 @@ reverse-SynthID/ git clone https://github.com/yourusername/reverse-SynthID.git cd reverse-SynthID -# Create virtual environment python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate - -# Install dependencies pip install -r requirements.txt ``` -### Run Nano-150k Watermark Analysis +### 1. Build Detection Codebook ```bash -# Full analysis on all 123k pairs (takes ~3 hours) -python watermark_investigation/watermark_full_123k_analysis.py - -# Extract final watermark visualization -python watermark_investigation/extract_final_watermark.py - -# Quick sample analysis (1000 pairs) -python watermark_investigation/watermark_full_analysis.py +python src/extraction/robust_extractor.py extract /path/to/watermarked/images \ + --output artifacts/codebook/robust_codebook.pkl ``` -### Detect SynthID Watermark +### 2. Detect Watermark ```bash -python src/extraction/synthid_codebook_extractor.py detect "path/to/image.png" \ - --codebook "artifacts/codebook/synthid_codebook.pkl" +python src/extraction/robust_extractor.py detect image.png \ + --codebook artifacts/codebook/robust_codebook.pkl ``` -**Output:** ``` Detection Results: Watermarked: True - Confidence: 1.0000 - Correlation: 0.5355 - Phase Match: 0.9571 - Structure Ratio: 1.2753 + Confidence: 0.95 + Phase Match: 0.6683 ``` -### Extract New Codebook +### 3. Build Spectral Codebook (V3) -```bash -python src/extraction/synthid_codebook_extractor.py extract "data/pure_white/" \ - --output "./my_codebook.pkl" -``` - -### Run Analysis - -```bash -# Comprehensive pattern discovery -python src/analysis/synthid_codebook_finder.py - -# Deep frequency analysis -python src/analysis/deep_synthid_analysis.py -``` - -## π§ How It Works - -### Nano-150k Watermark Detection - -1. **Frequency Domain Analysis**: Compute FFT differences between original and edited images -2. **LSB Pattern Detection**: Analyze least significant bit distributions for anomalies -3. **Color Shift Measurement**: Detect systematic RGB channel modifications -4. **Perceptual Hashing**: Compare perceptual hashes to find invisible changes -5. **Multi-Indicator Scoring**: Combine multiple detection methods for confidence - -### SynthID Detection - -1. **Pattern Discovery**: Analyze noise patterns across multiple images to find consistent structures -2. **Frequency Analysis**: Use FFT to identify carrier frequencies with phase modulation -3. **Phase Coherence**: Measure phase consistency at carrier frequencies -4. **Codebook Extraction**: Build reference patterns from averaged signals -5. **Detection**: Compare test image against codebook using correlation metrics - -## π Technical Details - -### Nano-150k Watermark Characteristics -- **Embedding Domains**: Frequency (DCT/DFT) + Spatial (color shifts) -- **Detection Methods**: FFT analysis, LSB statistics, perceptual hashing -- **Signal Strength**: Mean freq diff ~1.32, color shifts 32-35 pixel values -- **Robustness**: Survives JPEG compression, consistent across edit types -- **Categories Analyzed**: background, action, time-change, headshot, hairstyle - -### SynthID Watermark Characteristics -- **Embedding Domain**: Frequency (FFT phase) -- **Signal Strength**: ~0.1-0.15 pixel values -- **Carrier Count**: 100+ frequency locations -- **Robustness**: Survives moderate compression - -### Detection Algorithms - -**Nano-150k Multi-Indicator Detection:** ```python -def detect_watermark(original, edited): - indicators = 0 - - # 1. Frequency domain analysis - freq_diff = compute_fft_difference(original, edited) - if freq_diff > 0.5: - indicators += 1 - - # 2. Color shift detection - color_shift = compute_color_shift(original, edited) - if any(abs(shift) > 1.0 for shift in color_shift): - indicators += 1 - - # 3. LSB anomaly detection - lsb_deviation = compute_lsb_deviation(edited) - if any(dev > 0.02 for dev in lsb_deviation): - indicators += 1 - - # 4. Perceptual hash comparison - phash_dist = compute_phash_distance(original, edited) - if 5 < phash_dist <= 30: - indicators += 1 - - return indicators >= 2, indicators +from synthid_bypass import SpectralCodebook + +codebook = SpectralCodebook() +codebook.extract_from_references( + black_dir='assets/black/', # Pure-black Gemini images + white_dir='assets/white/' # Pure-white Gemini images +) +codebook.save('artifacts/spectral_codebook.npz') ``` -**SynthID Detection:** +### 4. Run V3 Bypass + ```python -def detect_synthid(image, codebook): - # 1. Extract noise pattern - noise = image - denoise(image) - - # 2. Check carrier phase coherence - fft = fft2(noise) - phase_match = check_phases(fft, codebook.carriers) - - # 3. Correlate with reference - correlation = correlate(noise, codebook.reference) - - # 4. Apply decision thresholds - is_watermarked = ( - correlation > 0.179 and - phase_match > 0.5 and - 0.8 < structure_ratio < 1.8 - ) - - return is_watermarked, confidence +from synthid_bypass import SynthIDBypass, SpectralCodebook + +codebook = SpectralCodebook() +codebook.load('artifacts/spectral_codebook.npz') + +bypass = SynthIDBypass() +result = bypass.bypass_v3(image_rgb, codebook, strength='aggressive') + +print(f"PSNR: {result.psnr:.1f} dB") # ~40 dB ``` +**Strength levels:** `gentle` (minimal change, ~43 dB) β `moderate` β `aggressive` β `maximum` (strongest removal, ~33 dB) + +--- + +## π Project Structure + +``` +reverse-SynthID/ +βββ src/ +β βββ extraction/ +β β βββ synthid_bypass.py # V1/V2/V3 bypass implementations + SpectralCodebook +β β βββ robust_extractor.py # Multi-scale watermark detection (90% accuracy) +β β βββ watermark_remover.py # Frequency-domain watermark removal +β β βββ benchmark_extraction.py # Performance benchmarking suite +β β βββ synthid_codebook_extractor.py # Original codebook extractor (legacy) +β βββ analysis/ +β βββ deep_synthid_analysis.py # FFT/phase analysis scripts +β βββ synthid_codebook_finder.py # Carrier frequency discovery +β +βββ assets/ +β βββ synthid_black.jpg # Watermark on black (enhanced) +β βββ synthid_white.jpg # Watermark on white (enhanced) +β βββ black/ # Reference black images from Gemini +β βββ white/ # Reference white images from Gemini +β +βββ artifacts/ +β βββ codebook/ # Detection codebooks (.pkl) +β βββ spectral_codebook.npz # V3 spectral fingerprint (119 MB) +β βββ v3_output/ # V3 bypass output samples +β βββ visualizations/ # FFT, phase, carrier visualizations +β +βββ watermark_investigation/ # Early-stage Nano-150k analysis (archived) +βββ SYNTHID_CODEBOOK_ANALYSIS.md # Detailed codebook reverse-engineering report +βββ synthid.pdf # SynthID paper reference +βββ requirements.txt +``` + +--- + +## π¬ Technical Deep Dive + +### How SynthID Works (Reverse-Engineered) + +``` +ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ +β SynthID Encoder (in Gemini) β +ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ€ +β 1. Generate carrier frequencies: {(14,14), (126,14), ...} β +β 2. Assign fixed phase values to each carrier β +β 3. Neural encoder adds learned noise pattern to image β +β 4. Watermark is imperceptible - spread across spectrum β +ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ€ +β SynthID Decoder (in Google) β +ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ€ +β 1. Extract noise residual (wavelet denoising) β +β 2. FFT β check phase at known carrier frequencies β +β 3. If phases match expected values β Watermarked β +ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ +``` + +### SpectralCodebook Extraction + +The codebook captures the watermark's full frequency fingerprint: + +- **50 reference images** (25 pure black + 25 pure white, all from Gemini) +- Extracts **magnitude envelope** and **phase template** per channel +- Computes **phase consistency score** per frequency bin +- Content-adaptive profiles for dark vs. light image regions + +### Selective Notch Filter + +The V3 bypass doesn't subtract blindly - it targets only bins where: +1. **Magnitude** exceeds the 97th percentile (strong watermark energy) +2. **Phase consistency** β₯ 0.95 across reference images (confirmed watermark, not noise) +3. **Subtraction** is capped at 30% of the image's energy at each bin + +This surgical precision is why V3 achieves 40+ dB PSNR while still reducing watermark energy. + +### Noise Correlation Signature + +| Metric | Value | Significance | +|:-------|:-----:|:-------------| +| Mean pairwise noise correlation | **0.218** | Identical watermark in all images | +| Noise structure ratio | **1.32** | Neural encoder byproduct | +| Phase coherence (top carriers) | **>99.9%** | Fixed model-level key | +| Green channel phase std | **<0.007 rad** | Strongest consistency channel | + +### Bit Plane Analysis + +| Bit Plane | Consistency | Role | +|:---------:|:-----------:|:-----| +| Bit 0 (LSB) | 0.049 | Watermark signal | +| Bit 1 | 0.074 | Watermark signal | +| Bit 2 | 0.125 | Partially watermarked | +| Bit 3 | 0.513 | Mixed | +| Bits 4-7 | 0.635β1.000 | Image structure | + +--- + +## π οΈ Core Modules + +### `robust_extractor.py` - Detection + +Multi-scale, multi-denoiser watermark detector achieving 90% detection rate. + +```python +from robust_extractor import RobustSynthIDExtractor + +extractor = RobustSynthIDExtractor() +extractor.load_codebook('artifacts/codebook/robust_codebook.pkl') +result = extractor.detect_array(image) + +print(f"Watermarked: {result.is_watermarked}") +print(f"Confidence: {result.confidence:.4f}") +print(f"Phase Match: {result.phase_match:.4f}") +``` + +**Features:** +- Multi-scale analysis (256, 512, 1024px) +- Wavelet + bilateral + NLM denoising fusion +- ICA-based watermark/content separation +- Ensemble carrier detection across scales + +### `synthid_bypass.py` - Bypass (V1/V2/V3) + +Three generations of watermark bypass: + +```python +from synthid_bypass import SynthIDBypass, SpectralCodebook + +bypass = SynthIDBypass() + +# V1: Simple JPEG compression +result = bypass.bypass_simple(image, jpeg_quality=50) + +# V2: Multi-stage transform pipeline +result = bypass.bypass_v2(image, strength='moderate') + +# V3: Spectral codebook subtraction (best) +codebook = SpectralCodebook() +codebook.load('artifacts/spectral_codebook.npz') +result = bypass.bypass_v3(image, codebook, strength='aggressive') +``` + +### `watermark_remover.py` - Removal + +Quality-preserving frequency-domain removal: + +```python +from watermark_remover import WatermarkRemover + +remover = WatermarkRemover(extractor) +result = remover.remove(image, mode='balanced') +``` + +--- + ## π References - [SynthID: Identifying AI-generated images](https://deepmind.google/technologies/synthid/) -- [Arxiv Paper - SynthID-Image: Image watermarking at internet scale]([https://doi.org/10.1038/s41586-024-07754-z](https://arxiv.org/abs/2510.09263)) +- [SynthID Paper (arXiv:2510.09263)](https://arxiv.org/abs/2510.09263) +- [Synthid-Bypass ComfyUI Workflow](https://github.com/aloshdenny/Synthid-Bypass) - img2img re-generation approach + +--- ## β οΈ Disclaimer -This project is for **research and educational purposes only**. SynthID is proprietary technology owned by Google DeepMind. The extracted patterns and detection methods are intended for: +This project is for **research and educational purposes only**. SynthID is proprietary technology owned by Google DeepMind. These tools are intended for: -- Academic research on watermarking techniques -- Security analysis of AI-generated content identification -- Understanding spread-spectrum encoding methods +- π Academic research on watermarking robustness +- π Security analysis of AI-generated content identification +- π‘ Understanding spread-spectrum encoding methods + +**Do not use these tools to misrepresent AI-generated content as human-created.** + +--- ## π License @@ -378,5 +397,5 @@ Research and educational use only. See [LICENSE](LICENSE) for details. ---- Made with π¬ by reverse engineering enthusiasts + Made with π¬ by watermark reverse engineering researchers
diff --git a/SYNTHID_CODEBOOK_ANALYSIS.md b/SYNTHID_CODEBOOK_ANALYSIS.md index 78dbabc..27dc87f 100644 --- a/SYNTHID_CODEBOOK_ANALYSIS.md +++ b/SYNTHID_CODEBOOK_ANALYSIS.md @@ -146,27 +146,45 @@ Based on our analysis, SynthID likely works as follows: 2. **Image modifications may break detection**: Heavy JPEG compression, cropping, or resizing may degrade the watermark 3. **Binary watermark bits unknown**: We discovered the carrier frequencies but not the actual message encoded +## V3 Spectral Bypass β Building on These Findings + +The carrier frequencies and phase consistency discovered here became the foundation for the **V3 Spectral Bypass** (`synthid_bypass.py`). Using a `SpectralCodebook` extracted from 50 reference images (25 black + 25 white), V3 can surgically subtract the watermark in the frequency domain: + +- **40+ dB PSNR** β visually indistinguishable from original +- **1-7% detection confidence reduction** per pass +- **Selective notch filter** β targets only confirmed watermark bins (P97+ magnitude, β₯95% phase consistency) + +See the main [README](README.md) for full V3 documentation and results. + ## Files Generated | File | Description | |------|-------------| -| `synthid_codebook.pkl` | Full codebook with numpy arrays | -| `synthid_codebook_meta.json` | Human-readable metadata | -| `deep_analysis/` | Visualization of patterns | -| `codebook_results/` | Initial analysis results | +| `artifacts/codebook/synthid_codebook.pkl` | Detection codebook with numpy arrays | +| `artifacts/codebook/synthid_codebook_meta.json` | Human-readable metadata | +| `artifacts/spectral_codebook.npz` | V3 spectral fingerprint (119 MB) | +| `artifacts/visualizations/` | FFT, phase, carrier visualizations | ## Usage ### To detect SynthID watermark: ```bash -python synthid_codebook_extractor.py detect image.png --codebook synthid_codebook.pkl +python src/extraction/robust_extractor.py detect image.png \ + --codebook artifacts/codebook/robust_codebook.pkl ``` -### To extract codebook from new images: +### To run V3 spectral bypass: -```bash -python synthid_codebook_extractor.py extract /path/to/images --output new_codebook.pkl +```python +from synthid_bypass import SynthIDBypass, SpectralCodebook + +codebook = SpectralCodebook() +codebook.load('artifacts/spectral_codebook.npz') + +bypass = SynthIDBypass() +result = bypass.bypass_v3(image_rgb, codebook, strength='aggressive') +print(f"PSNR: {result.psnr:.1f} dB") ``` ## Conclusion @@ -176,5 +194,6 @@ SynthID uses a sophisticated spread-spectrum watermarking technique that: - Uses **specific carrier frequencies** (14, 98, 126, 128, 210, 238 Hz and their conjugates) - Creates a **consistent noise signature** detectable via correlation analysis - Is **imperceptible** to human observers but **robust** enough to survive common image operations +- Uses a **fixed model-level key** (identical phase template across all images) -This analysis enables detection of SynthID watermarks without access to Google's proprietary decoder. +These findings enabled both detection (90% accuracy) and spectral bypass (40+ dB PSNR) without access to Google's proprietary encoder/decoder. diff --git a/artifacts/codebook/robust_codebook.pkl b/artifacts/codebook/robust_codebook.pkl new file mode 100644 index 0000000..2298192 Binary files /dev/null and b/artifacts/codebook/robust_codebook.pkl differ diff --git a/artifacts/visualizations/bypass_aggressive.png b/artifacts/visualizations/bypass_aggressive.png new file mode 100644 index 0000000..f10a72c Binary files /dev/null and b/artifacts/visualizations/bypass_aggressive.png differ diff --git a/artifacts/visualizations/bypass_balanced.png b/artifacts/visualizations/bypass_balanced.png new file mode 100644 index 0000000..d9b0a7e Binary files /dev/null and b/artifacts/visualizations/bypass_balanced.png differ diff --git a/artifacts/visualizations/bypass_direct.png b/artifacts/visualizations/bypass_direct.png new file mode 100644 index 0000000..6c2fe24 Binary files /dev/null and b/artifacts/visualizations/bypass_direct.png differ diff --git a/artifacts/visualizations/bypass_maximum.png b/artifacts/visualizations/bypass_maximum.png new file mode 100644 index 0000000..3645e89 Binary files /dev/null and b/artifacts/visualizations/bypass_maximum.png differ diff --git a/artifacts/visualizations/bypass_simple_success.png b/artifacts/visualizations/bypass_simple_success.png new file mode 100644 index 0000000..e0ed8d9 Binary files /dev/null and b/artifacts/visualizations/bypass_simple_success.png differ diff --git a/artifacts/visualizations/removed_1.png b/artifacts/visualizations/removed_1.png new file mode 100644 index 0000000..e26596b Binary files /dev/null and b/artifacts/visualizations/removed_1.png differ diff --git a/artifacts/visualizations/sample_cleaned.png b/artifacts/visualizations/sample_cleaned.png new file mode 100644 index 0000000..5ed67b1 Binary files /dev/null and b/artifacts/visualizations/sample_cleaned.png differ diff --git a/artifacts/visualizations/sample_cleaned_aggressive.png b/artifacts/visualizations/sample_cleaned_aggressive.png new file mode 100644 index 0000000..e9fd4b7 Binary files /dev/null and b/artifacts/visualizations/sample_cleaned_aggressive.png differ diff --git a/assets/sample_cleaned.png b/assets/sample_cleaned.png new file mode 100644 index 0000000..fa280f2 Binary files /dev/null and b/assets/sample_cleaned.png differ diff --git a/assets/sample_watermarked.png b/assets/sample_watermarked.png new file mode 100644 index 0000000..8e10a68 Binary files /dev/null and b/assets/sample_watermarked.png differ diff --git a/assets/synthid-watermark.jpeg b/assets/synthid-watermark.jpeg index b46cdff..943cbda 100644 Binary files a/assets/synthid-watermark.jpeg and b/assets/synthid-watermark.jpeg differ diff --git a/assets/synthid-watermark.png b/assets/synthid-watermark.png new file mode 100644 index 0000000..943cbda Binary files /dev/null and b/assets/synthid-watermark.png differ diff --git a/assets/synthid_black.jpg b/assets/synthid_black.jpg new file mode 100644 index 0000000..a5078e7 Binary files /dev/null and b/assets/synthid_black.jpg differ diff --git a/assets/synthid_white.jpg b/assets/synthid_white.jpg new file mode 100644 index 0000000..c1d1ba3 Binary files /dev/null and b/assets/synthid_white.jpg differ diff --git a/requirements.txt b/requirements.txt index 6aa54c1..392389c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,6 +8,12 @@ opencv-python>=4.5.0 # Wavelet analysis PyWavelets>=1.1.1 +# Machine learning (for ICA) +scikit-learn>=0.24.0 + +# Image processing +Pillow>=8.0.0 + # Visualization matplotlib>=3.4.0 diff --git a/src/extraction/benchmark_extraction.py b/src/extraction/benchmark_extraction.py new file mode 100644 index 0000000..504af8a --- /dev/null +++ b/src/extraction/benchmark_extraction.py @@ -0,0 +1,465 @@ +""" +SynthID Watermark Extraction Benchmark Suite + +Comprehensive benchmarking for watermark extraction and removal: +1. Detection accuracy across image types +2. Removal quality (PSNR, SSIM) +3. Re-detection test (verify watermark is removed) +4. Performance metrics + +Usage: + python benchmark_extraction.py --input-dir /path/to/images --codebook codebook.pkl +""" + +import os +import sys +import json +import time +import argparse +from typing import List, Dict, Optional, Tuple +from dataclasses import dataclass, asdict +from pathlib import Path +import numpy as np +import cv2 +from collections import defaultdict + +# Add parent directory to path for imports +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from robust_extractor import RobustSynthIDExtractor, DetectionResult +from watermark_remover import WatermarkRemover, RemovalResult + + +@dataclass +class BenchmarkResults: + """Results from benchmarking run.""" + n_images: int + detection_rate: float + avg_confidence: float + avg_correlation: float + avg_phase_match: float + + removal_success_rate: float + avg_psnr: float + avg_ssim: float + avg_confidence_drop: float + re_detection_rate: float + + total_time: float + avg_time_per_image: float + + details: Dict + + +class BenchmarkSuite: + """ + Comprehensive benchmark suite for SynthID extraction and removal. + """ + + def __init__( + self, + codebook_path: Optional[str] = None, + verbose: bool = True + ): + """ + Initialize benchmark suite. + + Args: + codebook_path: Path to pre-extracted codebook + verbose: Print progress during benchmarking + """ + self.verbose = verbose + self.extractor = RobustSynthIDExtractor() + self.remover = None + + if codebook_path and os.path.exists(codebook_path): + self.extractor.load_codebook(codebook_path) + self.remover = WatermarkRemover(extractor=self.extractor) + + def log(self, message: str): + """Print message if verbose.""" + if self.verbose: + print(message) + + def load_images( + self, + image_dir: str, + sample_size: Optional[int] = None, + extensions: set = {'.png', '.jpg', '.jpeg', '.webp'} + ) -> List[Tuple[str, np.ndarray]]: + """Load images from directory.""" + images = [] + + for fname in sorted(os.listdir(image_dir)): + if os.path.splitext(fname)[1].lower() in extensions: + path = os.path.join(image_dir, fname) + img = cv2.imread(path) + if img is not None: + img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) + images.append((path, img_rgb)) + + if sample_size and len(images) >= sample_size: + break + + return images + + def benchmark_detection( + self, + images: List[Tuple[str, np.ndarray]] + ) -> Dict: + """ + Benchmark detection accuracy. + + Returns: + Dict with detection metrics + """ + self.log(f"\n{'='*60}") + self.log("DETECTION BENCHMARK") + self.log(f"{'='*60}") + + results = [] + start_time = time.time() + + for i, (path, img) in enumerate(images): + try: + result = self.extractor.detect_array(img) + results.append({ + 'path': path, + 'is_watermarked': result.is_watermarked, + 'confidence': result.confidence, + 'correlation': result.correlation, + 'phase_match': result.phase_match, + 'structure_ratio': result.structure_ratio, + 'carrier_strength': result.carrier_strength, + }) + except Exception as e: + self.log(f" Error processing {path}: {e}") + results.append({ + 'path': path, + 'error': str(e) + }) + + if (i + 1) % 10 == 0: + self.log(f" Processed {i+1}/{len(images)} images...") + + elapsed = time.time() - start_time + + # Compute statistics + valid_results = [r for r in results if 'error' not in r] + detected = [r for r in valid_results if r['is_watermarked']] + + detection_rate = len(detected) / len(valid_results) if valid_results else 0 + avg_confidence = np.mean([r['confidence'] for r in valid_results]) if valid_results else 0 + avg_correlation = np.mean([r['correlation'] for r in valid_results]) if valid_results else 0 + avg_phase_match = np.mean([r['phase_match'] for r in valid_results]) if valid_results else 0 + + self.log(f"\n Detection Rate: {detection_rate:.1%}") + self.log(f" Avg Confidence: {avg_confidence:.4f}") + self.log(f" Avg Correlation: {avg_correlation:.4f}") + self.log(f" Avg Phase Match: {avg_phase_match:.4f}") + self.log(f" Time: {elapsed:.2f}s ({elapsed/len(images):.3f}s per image)") + + return { + 'n_images': len(images), + 'n_valid': len(valid_results), + 'n_detected': len(detected), + 'detection_rate': detection_rate, + 'avg_confidence': avg_confidence, + 'avg_correlation': avg_correlation, + 'avg_phase_match': avg_phase_match, + 'elapsed_seconds': elapsed, + 'results': results + } + + def benchmark_removal( + self, + images: List[Tuple[str, np.ndarray]], + output_dir: Optional[str] = None, + save_samples: int = 5 + ) -> Dict: + """ + Benchmark removal quality. + + Args: + images: List of (path, image) tuples + output_dir: Directory to save sample cleaned images + save_samples: Number of sample images to save + + Returns: + Dict with removal metrics + """ + if self.remover is None: + return {'error': 'No remover initialized (need codebook)'} + + self.log(f"\n{'='*60}") + self.log("REMOVAL BENCHMARK") + self.log(f"{'='*60}") + + results = [] + start_time = time.time() + + if output_dir: + os.makedirs(output_dir, exist_ok=True) + + for i, (path, img) in enumerate(images): + try: + result = self.remover.remove(img, verify=True) + + entry = { + 'path': path, + 'success': result.success, + 'psnr': result.psnr, + 'ssim': result.ssim, + 'removal_confidence': result.removal_confidence, + 'original_watermarked': result.original_detection.is_watermarked, + 'original_confidence': result.original_detection.confidence, + 'cleaned_watermarked': result.cleaned_detection.is_watermarked if result.cleaned_detection else None, + 'cleaned_confidence': result.cleaned_detection.confidence if result.cleaned_detection else None, + } + results.append(entry) + + # Save sample outputs + if output_dir and i < save_samples: + fname = os.path.basename(path) + out_path = os.path.join(output_dir, f"cleaned_{fname}") + cv2.imwrite(out_path, cv2.cvtColor(result.cleaned_image, cv2.COLOR_RGB2BGR)) + + except Exception as e: + self.log(f" Error processing {path}: {e}") + results.append({ + 'path': path, + 'error': str(e) + }) + + if (i + 1) % 10 == 0: + self.log(f" Processed {i+1}/{len(images)} images...") + + elapsed = time.time() - start_time + + # Compute statistics + valid_results = [r for r in results if 'error' not in r] + successful = [r for r in valid_results if r['success']] + re_detected = [r for r in valid_results if r.get('cleaned_watermarked', True)] + + removal_success_rate = len(successful) / len(valid_results) if valid_results else 0 + avg_psnr = np.mean([r['psnr'] for r in valid_results]) if valid_results else 0 + avg_ssim = np.mean([r['ssim'] for r in valid_results]) if valid_results else 0 + + # Confidence drop + conf_drops = [] + for r in valid_results: + if r.get('cleaned_confidence') is not None: + drop = r['original_confidence'] - r['cleaned_confidence'] + conf_drops.append(drop) + avg_conf_drop = np.mean(conf_drops) if conf_drops else 0 + + re_detection_rate = len(re_detected) / len(valid_results) if valid_results else 0 + + self.log(f"\n Removal Success Rate: {removal_success_rate:.1%}") + self.log(f" Avg PSNR: {avg_psnr:.2f} dB") + self.log(f" Avg SSIM: {avg_ssim:.4f}") + self.log(f" Avg Confidence Drop: {avg_conf_drop:.4f}") + self.log(f" Re-detection Rate: {re_detection_rate:.1%}") + self.log(f" Time: {elapsed:.2f}s ({elapsed/len(images):.3f}s per image)") + + return { + 'n_images': len(images), + 'n_valid': len(valid_results), + 'n_successful': len(successful), + 'removal_success_rate': removal_success_rate, + 'avg_psnr': avg_psnr, + 'avg_ssim': avg_ssim, + 'avg_confidence_drop': avg_conf_drop, + 're_detection_rate': re_detection_rate, + 'elapsed_seconds': elapsed, + 'results': results + } + + def run_full_benchmark( + self, + image_dir: str, + sample_size: Optional[int] = None, + output_dir: Optional[str] = None, + save_report: Optional[str] = None + ) -> BenchmarkResults: + """ + Run complete benchmark suite. + + Args: + image_dir: Directory containing watermarked images + sample_size: Max images to test (None for all) + output_dir: Directory to save cleaned samples + save_report: Path to save JSON report + + Returns: + BenchmarkResults + """ + self.log(f"\n{'='*60}") + self.log("SYNTHID EXTRACTION BENCHMARK SUITE") + self.log(f"{'='*60}") + self.log(f"Image directory: {image_dir}") + self.log(f"Sample size: {sample_size or 'all'}") + + # Load images + self.log("\nLoading images...") + images = self.load_images(image_dir, sample_size) + self.log(f"Loaded {len(images)} images") + + if not images: + raise ValueError("No images found in directory") + + # Run benchmarks + total_start = time.time() + + detection_results = self.benchmark_detection(images) + removal_results = self.benchmark_removal(images, output_dir) + + total_time = time.time() - total_start + + # Compile results + results = BenchmarkResults( + n_images=len(images), + detection_rate=detection_results['detection_rate'], + avg_confidence=detection_results['avg_confidence'], + avg_correlation=detection_results['avg_correlation'], + avg_phase_match=detection_results['avg_phase_match'], + + removal_success_rate=removal_results.get('removal_success_rate', 0), + avg_psnr=removal_results.get('avg_psnr', 0), + avg_ssim=removal_results.get('avg_ssim', 0), + avg_confidence_drop=removal_results.get('avg_confidence_drop', 0), + re_detection_rate=removal_results.get('re_detection_rate', 0), + + total_time=total_time, + avg_time_per_image=total_time / len(images), + + details={ + 'detection': detection_results, + 'removal': removal_results + } + ) + + # Print summary + self.log(f"\n{'='*60}") + self.log("BENCHMARK SUMMARY") + self.log(f"{'='*60}") + self.log(f"Images Tested: {results.n_images}") + self.log(f"") + self.log(f"Detection:") + self.log(f" Rate: {results.detection_rate:.1%}") + self.log(f" Confidence: {results.avg_confidence:.4f}") + self.log(f"") + self.log(f"Removal:") + self.log(f" Success Rate: {results.removal_success_rate:.1%}") + self.log(f" PSNR: {results.avg_psnr:.2f} dB") + self.log(f" SSIM: {results.avg_ssim:.4f}") + self.log(f" Re-detection: {results.re_detection_rate:.1%}") + self.log(f"") + self.log(f"Performance:") + self.log(f" Total Time: {results.total_time:.2f}s") + self.log(f" Per Image: {results.avg_time_per_image:.3f}s") + self.log(f"{'='*60}") + + # Save report + if save_report: + report = asdict(results) + # Remove large result arrays for JSON + if 'details' in report: + for key in ['detection', 'removal']: + if key in report['details']: + report['details'][key].pop('results', None) + + with open(save_report, 'w') as f: + json.dump(report, f, indent=2) + self.log(f"\nReport saved to: {save_report}") + + return results + + +def compare_with_original( + image_dir: str, + original_codebook: str, + robust_codebook: str, + sample_size: int = 50 +): + """ + Compare original vs robust extractor performance. + """ + print("\n" + "=" * 60) + print("COMPARISON: Original vs Robust Extractor") + print("=" * 60) + + # Original extractor (using same interface) + from synthid_codebook_extractor import detect_synthid + + # Robust extractor + robust = RobustSynthIDExtractor() + robust.load_codebook(robust_codebook) + + # Load images + extensions = {'.png', '.jpg', '.jpeg', '.webp'} + images = [] + for fname in sorted(os.listdir(image_dir)): + if os.path.splitext(fname)[1].lower() in extensions: + path = os.path.join(image_dir, fname) + images.append(path) + if len(images) >= sample_size: + break + + print(f"Testing on {len(images)} images...") + + # Compare + original_detected = 0 + robust_detected = 0 + + for path in images: + # Original + try: + orig_result = detect_synthid(path, original_codebook) + if orig_result['is_watermarked']: + original_detected += 1 + except: + pass + + # Robust + try: + robust_result = robust.detect(path) + if robust_result.is_watermarked: + robust_detected += 1 + except: + pass + + print(f"\nResults:") + print(f" Original Extractor: {original_detected}/{len(images)} ({100*original_detected/len(images):.1f}%)") + print(f" Robust Extractor: {robust_detected}/{len(images)} ({100*robust_detected/len(images):.1f}%)") + print(f" Improvement: {robust_detected - original_detected} more detected") + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description='SynthID Extraction Benchmark Suite') + parser.add_argument('--input-dir', type=str, required=True, + help='Directory with watermarked images') + parser.add_argument('--codebook', type=str, required=True, + help='Path to codebook file') + parser.add_argument('--sample-size', type=int, default=None, + help='Number of images to test (default: all)') + parser.add_argument('--output-dir', type=str, default=None, + help='Directory to save cleaned samples') + parser.add_argument('--output-report', type=str, default='benchmark_results.json', + help='Path to save JSON report') + parser.add_argument('--quiet', action='store_true', + help='Reduce output verbosity') + + args = parser.parse_args() + + suite = BenchmarkSuite( + codebook_path=args.codebook, + verbose=not args.quiet + ) + + results = suite.run_full_benchmark( + image_dir=args.input_dir, + sample_size=args.sample_size, + output_dir=args.output_dir, + save_report=args.output_report + ) diff --git a/src/extraction/synthid_bypass.py b/src/extraction/synthid_bypass.py new file mode 100644 index 0000000..bdb0072 --- /dev/null +++ b/src/extraction/synthid_bypass.py @@ -0,0 +1,1901 @@ +""" +SynthID Bypass - Non-DL Watermark Removal + +Implements watermark removal techniques inspired by diffusion-based bypass, +but using pure signal processing approaches (no deep learning required). + +Key techniques: +1. Noise replacement (mimics low-denoise regeneration) +2. Frequency domain disruption (phase scrambling at carrier frequencies) +3. JPEG degradation (quality cycling, chroma subsampling) +4. Bit manipulation (LSB randomization, bit-depth reduction) +5. Structure-preserving reconstruction (edge-guided blending) + +Based on insights from: +- Synthid-Bypass ComfyUI workflow (diffusion regeneration approach) +- SynthID-Image paper (watermark embedding mechanism) +""" + +import os +import io +import numpy as np +import cv2 +from scipy.fft import fft2, ifft2, fftshift, ifftshift +from scipy import ndimage +from typing import Optional, Dict, List, Tuple +from dataclasses import dataclass +import pywt +from PIL import Image + + +@dataclass +class BypassResult: + """Result of watermark bypass.""" + success: bool + cleaned_image: np.ndarray + psnr: float + ssim: float + detection_before: Optional[Dict] + detection_after: Optional[Dict] + stages_applied: List[str] + details: Dict + + +class SynthIDBypass: + """ + Remove SynthID watermarks using signal processing techniques. + + This class implements a multi-stage bypass pipeline that mimics + the effect of diffusion-based regeneration without requiring + deep learning models. + """ + + # Known SynthID carrier frequencies (from analysis) + KNOWN_CARRIERS = [ + (14, 14), (-14, -14), + (126, 14), (-126, -14), + (98, -14), (-98, 14), + (128, 128), (-128, -128), + (210, -14), (-210, 14), + (238, 14), (-238, -14), + ] + + def __init__( + self, + iterations: int = 3, + extractor=None + ): + """ + Initialize the bypass. + + Args: + iterations: Number of bypass passes + extractor: Optional RobustSynthIDExtractor for verification + """ + self.iterations = iterations + self.extractor = extractor + + # ================================================================ + # STAGE 1: NOISE REPLACEMENT + # Mimics low-denoise regeneration - replace watermark noise with new noise + # ================================================================ + + def add_calibrated_noise( + self, + image: np.ndarray, + sigma: float = 3.0, + seed: Optional[int] = None + ) -> np.ndarray: + """ + Add calibrated Gaussian noise to disrupt watermark. + + The noise level is carefully chosen to be strong enough to + disrupt the watermark but weak enough to preserve image quality. + """ + if seed is not None: + np.random.seed(seed) + + noise = np.random.normal(0, sigma / 255.0, image.shape) + noisy = image + noise + return np.clip(noisy, 0, 1) + + def denoise_bilateral( + self, + image: np.ndarray, + d: int = 9, + sigma_color: float = 75, + sigma_space: float = 75 + ) -> np.ndarray: + """Edge-preserving bilateral filter denoising.""" + img_uint8 = (image * 255).clip(0, 255).astype(np.uint8) + + if len(image.shape) == 3: + denoised = cv2.bilateralFilter(img_uint8, d, sigma_color, sigma_space) + else: + denoised = cv2.bilateralFilter(img_uint8, d, sigma_color, sigma_space) + + return denoised.astype(np.float32) / 255.0 + + def denoise_nlm( + self, + image: np.ndarray, + h: float = 5, + template_size: int = 7, + search_size: int = 21 + ) -> np.ndarray: + """Non-local means denoising.""" + img_uint8 = (image * 255).clip(0, 255).astype(np.uint8) + + if len(image.shape) == 3: + denoised = cv2.fastNlMeansDenoisingColored( + img_uint8, None, h, h, template_size, search_size + ) + else: + denoised = cv2.fastNlMeansDenoising( + img_uint8, None, h, template_size, search_size + ) + + return denoised.astype(np.float32) / 255.0 + + def noise_replacement_pass( + self, + image: np.ndarray, + noise_sigma: float = 5.0, + denoise_strength: float = 8.0 + ) -> np.ndarray: + """ + Single pass of noise replacement. + + This mimics the diffusion model's low-denoise regeneration: + 1. Add noise to disrupt existing patterns + 2. Denoise to recover structure (but with different noise) + """ + # Add calibrated noise + noisy = self.add_calibrated_noise(image, sigma=noise_sigma) + + # Denoise with edge-preserving filter + denoised = self.denoise_bilateral(noisy, d=9, sigma_color=denoise_strength * 10, sigma_space=75) + + # Blend with original to preserve some structure + result = denoised * 0.7 + image * 0.3 + + return result + + def apply_noise_replacement( + self, + image: np.ndarray, + passes: int = 2, + noise_sigma: float = 5.0 + ) -> np.ndarray: + """ + Apply multiple noise replacement passes. + + Similar to multiple KSampler passes in the diffusion workflow. + """ + current = image.copy() + + for i in range(passes): + # Decrease noise sigma slightly each pass + sigma = noise_sigma * (1 - i * 0.2) + current = self.noise_replacement_pass(current, noise_sigma=sigma) + + return current + + # ================================================================ + # STAGE 2: FREQUENCY DOMAIN DISRUPTION + # Scramble phases at known carrier frequencies + # ================================================================ + + def scramble_carrier_phases( + self, + image: np.ndarray, + carriers: Optional[List[Tuple[int, int]]] = None, + scramble_radius: int = 3, + scramble_strength: float = 0.8 + ) -> np.ndarray: + """ + Randomize phases at carrier frequencies to break watermark coherence. + """ + if carriers is None: + carriers = self.KNOWN_CARRIERS + + img_f = image.astype(np.float32) + h, w = img_f.shape[:2] + center = (h // 2, w // 2) + + # Scale carriers to image size (carriers are for 512px) + scale_y = h / 512 + scale_x = w / 512 + + if len(img_f.shape) == 3: + result = np.zeros_like(img_f) + for c in range(img_f.shape[2]): + result[:, :, c] = self._scramble_channel( + img_f[:, :, c], carriers, center, scale_y, scale_x, + scramble_radius, scramble_strength + ) + else: + result = self._scramble_channel( + img_f, carriers, center, scale_y, scale_x, + scramble_radius, scramble_strength + ) + + return result + + def _scramble_channel( + self, + channel: np.ndarray, + carriers: List[Tuple[int, int]], + center: Tuple[int, int], + scale_y: float, + scale_x: float, + radius: int, + strength: float + ) -> np.ndarray: + """Scramble phases in a single channel.""" + f = fftshift(fft2(channel)) + h, w = channel.shape + + for freq_y, freq_x in carriers: + # Scale to image size + y = int(freq_y * scale_y) + center[0] + x = int(freq_x * scale_x) + center[1] + + # Scramble region around carrier + for dy in range(-radius, radius + 1): + for dx in range(-radius, radius + 1): + ny, nx = y + dy, x + dx + if 0 <= ny < h and 0 <= nx < w: + # Randomize phase while preserving magnitude + mag = np.abs(f[ny, nx]) + random_phase = np.random.uniform(0, 2 * np.pi) + original_phase = np.angle(f[ny, nx]) + new_phase = original_phase * (1 - strength) + random_phase * strength + f[ny, nx] = mag * np.exp(1j * new_phase) + + # Also scramble conjugate + cny, cnx = h - ny, w - nx + if 0 <= cny < h and 0 <= cnx < w: + cmag = np.abs(f[cny, cnx]) + f[cny, cnx] = cmag * np.exp(-1j * new_phase) + + result = np.real(ifft2(ifftshift(f))) + return np.clip(result, 0, 1) + + def inject_bandpass_noise( + self, + image: np.ndarray, + freq_range: Tuple[float, float] = (0.02, 0.15), + noise_strength: float = 0.02 + ) -> np.ndarray: + """ + Inject noise in specific frequency bands where watermark lives. + """ + img_f = image.astype(np.float32) + h, w = img_f.shape[:2] + center_y, center_x = h // 2, w // 2 + + # Create bandpass mask + y_coords, x_coords = np.ogrid[:h, :w] + dist = np.sqrt((y_coords - center_y) ** 2 + (x_coords - center_x) ** 2) + max_dist = np.sqrt(center_y ** 2 + center_x ** 2) + norm_dist = dist / max_dist + + bandpass = ((norm_dist > freq_range[0]) & (norm_dist < freq_range[1])).astype(np.float32) + + if len(img_f.shape) == 3: + result = np.zeros_like(img_f) + for c in range(img_f.shape[2]): + result[:, :, c] = self._inject_bandpass_channel( + img_f[:, :, c], bandpass, noise_strength + ) + else: + result = self._inject_bandpass_channel(img_f, bandpass, noise_strength) + + return result + + def _inject_bandpass_channel( + self, + channel: np.ndarray, + bandpass: np.ndarray, + noise_strength: float + ) -> np.ndarray: + """Inject bandpass noise in a single channel.""" + f = fftshift(fft2(channel)) + + # Generate random phase noise + phase_noise = np.random.uniform(0, 2 * np.pi, f.shape) + + # Add noise only in bandpass region + noise_complex = noise_strength * np.exp(1j * phase_noise) * bandpass + f_noisy = f + noise_complex * np.abs(f).mean() + + result = np.real(ifft2(ifftshift(f_noisy))) + return np.clip(result, 0, 1) + + # ================================================================ + # STAGE 3: JPEG DEGRADATION + # JPEG compression breaks watermark coherence + # ================================================================ + + def jpeg_compress( + self, + image: np.ndarray, + quality: int = 85 + ) -> np.ndarray: + """Apply JPEG compression/decompression cycle.""" + img_uint8 = (image * 255).clip(0, 255).astype(np.uint8) + + # Convert to PIL for JPEG encoding + if len(image.shape) == 3: + pil_img = Image.fromarray(img_uint8, mode='RGB') + else: + pil_img = Image.fromarray(img_uint8, mode='L') + + # Compress to JPEG in memory + buffer = io.BytesIO() + pil_img.save(buffer, format='JPEG', quality=quality) + buffer.seek(0) + + # Decompress + pil_img = Image.open(buffer) + result = np.array(pil_img).astype(np.float32) / 255.0 + + return result + + def jpeg_quality_cycle( + self, + image: np.ndarray, + qualities: List[int] = [70, 80, 92] + ) -> np.ndarray: + """ + Apply multiple JPEG compression cycles at varying qualities. + + This disrupts the watermark through quantization artifacts + while the varying qualities prevent adaptation. + """ + current = image.copy() + + for q in qualities: + current = self.jpeg_compress(current, quality=q) + + return current + + def chroma_subsample( + self, + image: np.ndarray, + factor: int = 2 + ) -> np.ndarray: + """ + Subsample and upsample chroma channels. + + This is similar to what JPEG does but more aggressive, + disrupting watermark in color channels. + """ + if len(image.shape) != 3: + return image + + img_uint8 = (image * 255).clip(0, 255).astype(np.uint8) + + # Convert to YCrCb + ycrcb = cv2.cvtColor(img_uint8, cv2.COLOR_RGB2YCrCb) + y, cr, cb = cv2.split(ycrcb) + + # Subsample chroma + h, w = cr.shape + cr_small = cv2.resize(cr, (w // factor, h // factor), interpolation=cv2.INTER_AREA) + cb_small = cv2.resize(cb, (w // factor, h // factor), interpolation=cv2.INTER_AREA) + + # Upsample back + cr_up = cv2.resize(cr_small, (w, h), interpolation=cv2.INTER_LINEAR) + cb_up = cv2.resize(cb_small, (w, h), interpolation=cv2.INTER_LINEAR) + + # Merge and convert back + ycrcb_new = cv2.merge([y, cr_up, cb_up]) + rgb = cv2.cvtColor(ycrcb_new, cv2.COLOR_YCrCb2RGB) + + return rgb.astype(np.float32) / 255.0 + + # ================================================================ + # STAGE 4: BIT MANIPULATION + # Modify LSBs where watermark often resides + # ================================================================ + + def randomize_lsb( + self, + image: np.ndarray, + n_bits: int = 2, + probability: float = 0.5 + ) -> np.ndarray: + """ + Randomize least significant bits. + + LSBs often carry watermark info; randomizing them disrupts + the watermark while having minimal visual impact. + """ + img_uint8 = (image * 255).clip(0, 255).astype(np.uint8) + + # Create mask for bits to randomize + mask = np.uint8((1 << n_bits) - 1) # e.g., n_bits=2 -> mask=0b11 + inv_mask = np.uint8(~mask) # Properly handle uint8 inversion + + # Random bits + random_bits = np.random.randint(0, int(mask) + 1, img_uint8.shape, dtype=np.uint8) + + # Random selection of pixels to modify + modify_mask = np.random.random(img_uint8.shape) < probability + + # Clear LSBs and add random bits + result = img_uint8.copy() + result[modify_mask] = (result[modify_mask] & inv_mask) | random_bits[modify_mask] + + return result.astype(np.float32) / 255.0 + + def reduce_bit_depth( + self, + image: np.ndarray, + bits: int = 6 + ) -> np.ndarray: + """ + Reduce and expand bit depth. + + Quantizes to fewer bits then expands, effectively + removing fine-grained watermark patterns. + """ + levels = 2 ** bits + + # Quantize + quantized = np.round(image * (levels - 1)) + + # Expand back + result = quantized / (levels - 1) + + return result.astype(np.float32) + + def color_jitter( + self, + image: np.ndarray, + brightness: float = 0.02, + contrast: float = 0.02, + saturation: float = 0.02 + ) -> np.ndarray: + """ + Apply small random color adjustments. + + Subtle color variations break watermark coherence. + """ + result = image.copy() + + # Brightness + b_factor = 1 + np.random.uniform(-brightness, brightness) + result = result * b_factor + + # Contrast + c_factor = 1 + np.random.uniform(-contrast, contrast) + mean = np.mean(result, axis=(0, 1), keepdims=True) + result = (result - mean) * c_factor + mean + + # Saturation (for color images) + if len(result.shape) == 3: + s_factor = 1 + np.random.uniform(-saturation, saturation) + gray = np.mean(result, axis=2, keepdims=True) + result = gray + (result - gray) * s_factor + + return np.clip(result, 0, 1) + + # ================================================================ + # STAGE 5: STRUCTURE-PRESERVING RECONSTRUCTION + # Use edges to guide reconstruction like ControlNet + # ================================================================ + + def extract_structure( + self, + image: np.ndarray + ) -> Tuple[np.ndarray, np.ndarray]: + """ + Extract structural information (edges and texture). + + This is similar to how Canny edges are used in ControlNet + to preserve structure during regeneration. + """ + img_uint8 = (image * 255).clip(0, 255).astype(np.uint8) + + if len(image.shape) == 3: + gray = cv2.cvtColor(img_uint8, cv2.COLOR_RGB2GRAY) + else: + gray = img_uint8 + + # Edge detection (like Canny in ControlNet) + edges = cv2.Canny(gray, 50, 150) + edges = edges.astype(np.float32) / 255.0 + + # Gradient magnitude (texture measure) + grad_x = cv2.Sobel(gray, cv2.CV_64F, 1, 0, ksize=3) + grad_y = cv2.Sobel(gray, cv2.CV_64F, 0, 1, ksize=3) + gradient = np.sqrt(grad_x ** 2 + grad_y ** 2) + gradient = gradient / (gradient.max() + 1e-10) + + return edges, gradient.astype(np.float32) + + def guided_filter( + self, + image: np.ndarray, + guide: np.ndarray, + radius: int = 8, + epsilon: float = 0.01 + ) -> np.ndarray: + """Apply guided filter for edge-preserving smoothing.""" + def box_filter(img, r): + return ndimage.uniform_filter(img, size=r * 2 + 1) + + if len(image.shape) == 3 and len(guide.shape) == 2: + guide = np.stack([guide] * image.shape[2], axis=2) + + if len(image.shape) == 2: + mean_i = box_filter(guide, radius) + mean_p = box_filter(image, radius) + mean_ip = box_filter(guide * image, radius) + cov_ip = mean_ip - mean_i * mean_p + + mean_ii = box_filter(guide * guide, radius) + var_i = mean_ii - mean_i * mean_i + + a = cov_ip / (var_i + epsilon) + b = mean_p - a * mean_i + + mean_a = box_filter(a, radius) + mean_b = box_filter(b, radius) + + return mean_a * guide + mean_b + else: + result = np.zeros_like(image) + for c in range(image.shape[2]): + result[:, :, c] = self.guided_filter( + image[:, :, c], guide[:, :, c], radius, epsilon + ) + return result + + def reconstruct_with_structure( + self, + processed: np.ndarray, + original: np.ndarray, + edges: np.ndarray, + blend_factor: float = 0.2 + ) -> np.ndarray: + """ + Blend processed image with original guided by structure. + + Preserves edges and structure while keeping watermark disruption. + """ + # Use original as guide for edge-preserving filtering + filtered = self.guided_filter(processed, original, radius=5) + + # Stronger preservation near edges + if len(original.shape) == 3: + edge_map = np.stack([edges] * 3, axis=2) + else: + edge_map = edges + + # Near edges: blend more with original + # Away from edges: keep more of processed + result = filtered * (1 - edge_map * 0.3) + original * (edge_map * 0.3) + + # Final blend + result = result * (1 - blend_factor) + original * blend_factor + + return np.clip(result, 0, 1) + + # ================================================================ + # QUALITY METRICS + # ================================================================ + + def compute_psnr(self, original: np.ndarray, modified: np.ndarray) -> float: + """Compute Peak Signal-to-Noise Ratio.""" + mse = np.mean((original - modified) ** 2) + if mse == 0: + return float('inf') + return float(10 * np.log10(1.0 / mse)) + + def compute_ssim(self, original: np.ndarray, modified: np.ndarray) -> float: + """Compute SSIM using vectorized block-based approach (no loop).""" + img_o = original.astype(np.float64) + img_m = modified.astype(np.float64) + if img_o.max() > 1.5: + img_o = img_o / 255.0 + img_m = img_m / 255.0 + + # Convert to grayscale using luminance weights + if img_o.ndim == 3: + gray_o = 0.299 * img_o[:,:,0] + 0.587 * img_o[:,:,1] + 0.114 * img_o[:,:,2] + gray_m = 0.299 * img_m[:,:,0] + 0.587 * img_m[:,:,1] + 0.114 * img_m[:,:,2] + else: + gray_o = img_o + gray_m = img_m + + blk = 8 + rows, cols = gray_o.shape + rc = (rows // blk) * blk + cc = (cols // blk) * blk + + # Vectorized block reshape + a = gray_o[:rc, :cc].reshape(rc // blk, blk, cc // blk, blk) + a = a.transpose(0, 2, 1, 3).reshape(-1, blk, blk) + b = gray_m[:rc, :cc].reshape(rc // blk, blk, cc // blk, blk) + b = b.transpose(0, 2, 1, 3).reshape(-1, blk, blk) + + mu_a = a.mean(axis=(1, 2)) + mu_b = b.mean(axis=(1, 2)) + var_a = a.var(axis=(1, 2)) + var_b = b.var(axis=(1, 2)) + cov_ab = ((a - mu_a[:, None, None]) * (b - mu_b[:, None, None])).mean(axis=(1, 2)) + + k1_sq = 0.0001 # (0.01)^2 + k2_sq = 0.0009 # (0.03)^2 + + num = (2.0 * mu_a * mu_b + k1_sq) * (2.0 * cov_ab + k2_sq) + den = (mu_a * mu_a + mu_b * mu_b + k1_sq) * (var_a + var_b + k2_sq) + + return float(np.mean(num / den)) + + # ================================================================ + # MAIN BYPASS PIPELINE + # ================================================================ + + def bypass_simple( + self, + image: np.ndarray, + jpeg_quality: int = 50, + verify: bool = True + ) -> BypassResult: + """ + Simple, effective bypass using just JPEG compression. + + Testing showed JPEG Q50 is the most effective single technique, + achieving ~11% phase match reduction with excellent quality (PSNR 37dB). + Other techniques (noise, frequency manipulation) are less effective + and hurt quality more than they help. + + Args: + image: Input image (RGB, uint8 or float) + jpeg_quality: JPEG quality (50 is optimal for SynthID) + verify: Whether to verify removal with detection + + Returns: + BypassResult with cleaned image and metrics + """ + img_f = image.astype(np.float32) + if img_f.max() > 1: + img_f = img_f / 255.0 + + # Initial detection + detection_before = None + if verify and self.extractor is not None: + result = self.extractor.detect_array((img_f * 255).astype(np.uint8)) + detection_before = { + 'is_watermarked': result.is_watermarked, + 'confidence': result.confidence, + 'phase_match': result.phase_match + } + + # Apply JPEG compression + cleaned = self.jpeg_compress(img_f, quality=jpeg_quality) + + # Compute quality metrics + psnr = self.compute_psnr(img_f, cleaned) + ssim = self.compute_ssim(img_f, cleaned) + + # Final detection + detection_after = None + if verify and self.extractor is not None: + result = self.extractor.detect_array((cleaned * 255).astype(np.uint8)) + detection_after = { + 'is_watermarked': result.is_watermarked, + 'confidence': result.confidence, + 'phase_match': result.phase_match + } + + # Determine success + success = psnr > 30 + if detection_before and detection_after: + phase_drop = detection_before['phase_match'] - detection_after['phase_match'] + success = success and phase_drop > 0.05 + + cleaned_uint8 = (cleaned * 255).clip(0, 255).astype(np.uint8) + + return BypassResult( + success=success, + cleaned_image=cleaned_uint8, + psnr=psnr, + ssim=ssim, + detection_before=detection_before, + detection_after=detection_after, + stages_applied=['jpeg_compress'], + details={'method': 'simple', 'jpeg_quality': jpeg_quality} + ) + + def bypass( + self, + image: np.ndarray, + mode: str = 'balanced', + verify: bool = True + ) -> BypassResult: + """ + Main bypass pipeline - remove SynthID watermark. + + Args: + image: Input image (RGB, uint8 or float) + mode: 'light', 'balanced', or 'aggressive' + verify: Whether to verify removal with detection + + Returns: + BypassResult with cleaned image and metrics + """ + img_f = image.astype(np.float32) + if img_f.max() > 1: + img_f = img_f / 255.0 + + # Set parameters based on mode + params = self._get_mode_params(mode) + + # Initial detection + detection_before = None + if verify and self.extractor is not None: + result = self.extractor.detect_array((img_f * 255).astype(np.uint8)) + detection_before = { + 'is_watermarked': result.is_watermarked, + 'confidence': result.confidence, + 'phase_match': result.phase_match + } + + # Extract structure for preservation + edges, gradient = self.extract_structure(img_f) + + current = img_f.copy() + stages_applied = [] + + # Apply bypass iterations + for iteration in range(params['iterations']): + # Stage 1: Noise replacement + if params['noise_replacement']: + current = self.apply_noise_replacement( + current, + passes=params['noise_passes'], + noise_sigma=params['noise_sigma'] + ) + stages_applied.append(f'noise_replacement_{iteration}') + + # Stage 2: Frequency disruption + if params['frequency_disruption']: + current = self.scramble_carrier_phases( + current, + scramble_radius=params['scramble_radius'], + scramble_strength=params['scramble_strength'] + ) + current = self.inject_bandpass_noise( + current, + noise_strength=params['bandpass_noise'] + ) + stages_applied.append(f'frequency_disruption_{iteration}') + + # Stage 3: JPEG degradation + if params['jpeg_degradation']: + current = self.jpeg_quality_cycle(current, params['jpeg_qualities']) + if params['chroma_subsample']: + current = self.chroma_subsample(current) + stages_applied.append(f'jpeg_degradation_{iteration}') + + # Stage 4: Bit manipulation + if params['bit_manipulation']: + current = self.randomize_lsb( + current, n_bits=params['lsb_bits'], + probability=params['lsb_probability'] + ) + current = self.color_jitter(current) + stages_applied.append(f'bit_manipulation_{iteration}') + + # Stage 5: Structure preservation + current = self.reconstruct_with_structure( + current, img_f, edges, + blend_factor=params['structure_blend'] + ) + stages_applied.append(f'structure_preservation_{iteration}') + + # Compute quality metrics + psnr = self.compute_psnr(img_f, current) + ssim = self.compute_ssim(img_f, current) + + # Final detection + detection_after = None + if verify and self.extractor is not None: + result = self.extractor.detect_array((current * 255).astype(np.uint8)) + detection_after = { + 'is_watermarked': result.is_watermarked, + 'confidence': result.confidence, + 'phase_match': result.phase_match + } + + # Determine success + success = psnr > 28 and ssim > 0.9 + if detection_before and detection_after: + phase_drop = detection_before['phase_match'] - detection_after['phase_match'] + success = success and (phase_drop > 0.05 or not detection_after['is_watermarked']) + + # Convert to uint8 + cleaned_uint8 = (current * 255).clip(0, 255).astype(np.uint8) + + return BypassResult( + success=success, + cleaned_image=cleaned_uint8, + psnr=psnr, + ssim=ssim, + detection_before=detection_before, + detection_after=detection_after, + stages_applied=stages_applied, + details={'mode': mode, 'params': params} + ) + + def _get_mode_params(self, mode: str) -> Dict: + """Get parameters for each mode.""" + if mode == 'light': + return { + 'iterations': 1, + 'noise_replacement': True, + 'noise_passes': 1, + 'noise_sigma': 3.0, + 'frequency_disruption': True, + 'scramble_radius': 2, + 'scramble_strength': 0.5, + 'bandpass_noise': 0.01, + 'jpeg_degradation': True, + 'jpeg_qualities': [88, 95], + 'chroma_subsample': False, + 'bit_manipulation': False, + 'lsb_bits': 1, + 'lsb_probability': 0.3, + 'structure_blend': 0.3 + } + elif mode == 'aggressive': + return { + 'iterations': 3, + 'noise_replacement': True, + 'noise_passes': 2, + 'noise_sigma': 8.0, + 'frequency_disruption': True, + 'scramble_radius': 5, + 'scramble_strength': 0.9, + 'bandpass_noise': 0.03, + 'jpeg_degradation': True, + 'jpeg_qualities': [65, 75, 88], + 'chroma_subsample': True, + 'bit_manipulation': True, + 'lsb_bits': 2, + 'lsb_probability': 0.6, + 'structure_blend': 0.15 + } + elif mode == 'maximum': + # Maximum bypass - prioritizes watermark removal over quality + # Based on empirical testing: JPEG Q50 + Noise(25) are most effective + return { + 'iterations': 3, + 'noise_replacement': True, + 'noise_passes': 3, + 'noise_sigma': 25.0, # Heavy noise injection + 'frequency_disruption': True, + 'scramble_radius': 8, + 'scramble_strength': 1.0, # Full phase randomization + 'bandpass_noise': 0.05, + 'jpeg_degradation': True, + 'jpeg_qualities': [50, 60, 75], # Low quality JPEG + 'chroma_subsample': True, + 'bit_manipulation': True, + 'lsb_bits': 3, # More LSB randomization + 'lsb_probability': 0.8, + 'structure_blend': 0.05 # Minimal blending to avoid restoring watermark + } + else: # balanced + return { + 'iterations': 2, + 'noise_replacement': True, + 'noise_passes': 2, + 'noise_sigma': 5.0, + 'frequency_disruption': True, + 'scramble_radius': 3, + 'scramble_strength': 0.7, + 'bandpass_noise': 0.02, + 'jpeg_degradation': True, + 'jpeg_qualities': [75, 85, 92], + 'chroma_subsample': True, + 'bit_manipulation': True, + 'lsb_bits': 2, + 'lsb_probability': 0.5, + 'structure_blend': 0.2 + } + + def bypass_file( + self, + input_path: str, + output_path: str, + mode: str = 'balanced', + verify: bool = True + ) -> BypassResult: + """ + Bypass watermark in image file and save result. + """ + img = cv2.imread(input_path) + if img is None: + raise ValueError(f"Could not load image: {input_path}") + + img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) + + result = self.bypass(img_rgb, mode=mode, verify=verify) + + # Save result + os.makedirs(os.path.dirname(output_path) if os.path.dirname(output_path) else '.', exist_ok=True) + cv2.imwrite(output_path, cv2.cvtColor(result.cleaned_image, cv2.COLOR_RGB2BGR)) + + return result + + # ================================================================ + # V2: COMBINED WORST-CASE BYPASS PIPELINE + # Targets SynthID's documented weakness against stacked + # multi-category transforms (combination worst TPR ~84%) + # ================================================================ + + def _spatial_disruption( + self, + image: np.ndarray, + strength: float = 1.0 + ) -> np.ndarray: + """ + Spatial transforms β SynthID's weakest category (52% TPR worst-case). + + Applies random affine + crop-resize + perspective warp to break + spatial coherence of the watermark pattern. + """ + h, w = image.shape[:2] + img_uint8 = (image * 255).clip(0, 255).astype(np.uint8) + + # Random affine: rotation, scale, translation (very subtle) + angle = np.random.uniform(-0.3, 0.3) * strength + scale = 1.0 + np.random.uniform(-0.005, 0.005) * strength + tx = np.random.uniform(-1, 1) * strength + ty = np.random.uniform(-1, 1) * strength + + center = (w / 2, h / 2) + M = cv2.getRotationMatrix2D(center, angle, scale) + M[0, 2] += tx + M[1, 2] += ty + result = cv2.warpAffine(img_uint8, M, (w, h), + flags=cv2.INTER_CUBIC, + borderMode=cv2.BORDER_REFLECT_101) + + # Random crop and resize back (0.3-0.8%) + crop_frac = 0.003 + 0.005 * strength + cx = int(w * crop_frac * np.random.uniform(0.3, 1.0)) + cy = int(h * crop_frac * np.random.uniform(0.3, 1.0)) + cx = max(1, cx) + cy = max(1, cy) + cropped = result[cy:h-cy, cx:w-cx] + result = cv2.resize(cropped, (w, h), interpolation=cv2.INTER_CUBIC) + + # Subtle perspective warp + if strength > 0.5: + offset = max(1, int(1.5 * strength)) + src_pts = np.float32([[0, 0], [w, 0], [0, h], [w, h]]) + dst_pts = np.float32([ + [np.random.randint(0, offset+1), np.random.randint(0, offset+1)], + [w - np.random.randint(0, offset+1), np.random.randint(0, offset+1)], + [np.random.randint(0, offset+1), h - np.random.randint(0, offset+1)], + [w - np.random.randint(0, offset+1), h - np.random.randint(0, offset+1)] + ]) + M_persp = cv2.getPerspectiveTransform(src_pts, dst_pts) + result = cv2.warpPerspective(result, M_persp, (w, h), + flags=cv2.INTER_CUBIC, + borderMode=cv2.BORDER_REFLECT_101) + + return result.astype(np.float32) / 255.0 + + def _quality_degradation( + self, + image: np.ndarray, + jpeg_quality: int = 40, + strength: float = 1.0 + ) -> np.ndarray: + """ + Quality transforms: JPEG/WebP cycling + resize cycling. + + Forces requantization across different codec bases (DCT vs wavelet) + to destroy watermark coherence that survives any single codec. + """ + img_uint8 = (image * 255).clip(0, 255).astype(np.uint8) + h, w = img_uint8.shape[:2] + + # Step 1: JPEG compression + pil_img = Image.fromarray(img_uint8) + buf = io.BytesIO() + pil_img.save(buf, format='JPEG', quality=jpeg_quality) + buf.seek(0) + result = np.array(Image.open(buf).convert('RGB')) + + # Step 2: WebP compression (different transform basis than JPEG DCT) + webp_q = max(30, jpeg_quality - 5) + pil_img2 = Image.fromarray(result) + buf2 = io.BytesIO() + pil_img2.save(buf2, format='WEBP', quality=webp_q) + buf2.seek(0) + result = np.array(Image.open(buf2).convert('RGB')) + + # Step 3: Second JPEG at slightly different quality + pil_img3 = Image.fromarray(result) + buf3 = io.BytesIO() + pil_img3.save(buf3, format='JPEG', quality=jpeg_quality + 15) + buf3.seek(0) + result = np.array(Image.open(buf3).convert('RGB')) + + # Step 4: Downscale + upscale (destroys sub-pixel watermark info) + if strength > 0.3: + down_factor = 0.875 - 0.05 * strength # 82-87% downscale + small_h = max(64, int(h * down_factor)) + small_w = max(64, int(w * down_factor)) + small = cv2.resize(result, (small_w, small_h), interpolation=cv2.INTER_AREA) + result = cv2.resize(small, (w, h), interpolation=cv2.INTER_CUBIC) + + return result.astype(np.float32) / 255.0 + + def _noise_disruption( + self, + image: np.ndarray, + sigma: float = 10.0, + strength: float = 1.0 + ) -> np.ndarray: + """ + Noise injection + denoising: replaces watermark noise with random noise. + + Uses both Gaussian and Poisson noise (different distributions) + followed by edge-preserving denoising. + """ + result = image.copy() + + # Gaussian noise (moderate: enough to displace watermark bits) + noise_std = (sigma / 255.0) * strength + noise = np.random.normal(0, noise_std, result.shape).astype(np.float32) + result = np.clip(result + noise, 0, 1) + + # Poisson noise (shot noise β different distribution, subtle) + if strength > 0.5: + lam = 200 / strength # Higher lambda = less noise + noisy = np.random.poisson(np.maximum(result * lam, 0)) / lam + result = result * 0.9 + noisy.astype(np.float32) * 0.1 + result = np.clip(result, 0, 1) + + # Edge-preserving bilateral denoising (moderate) + img_uint8 = (result * 255).clip(0, 255).astype(np.uint8) + d = 5 + sigma_c = 30 + 15 * strength + sigma_s = 30 + 15 * strength + denoised = cv2.bilateralFilter(img_uint8, d, sigma_c, sigma_s) + + # NLM denoising for a second pass (light) + if strength > 0.7: + h_param = 3 + 3 * strength + denoised = cv2.fastNlMeansDenoisingColored( + denoised, None, h_param, h_param, 7, 21 + ) + + return denoised.astype(np.float32) / 255.0 + + def _color_disruption( + self, + image: np.ndarray, + strength: float = 1.0 + ) -> np.ndarray: + """ + Color channel manipulation to destroy per-channel watermark coherence. + + SynthID embeds differently per channel (G most, then R, then B). + Disrupting color space breaks cross-channel correlations. + """ + img_uint8 = (image * 255).clip(0, 255).astype(np.uint8) + + # YCrCb chroma subsampling (like aggressive JPEG chroma) + ycrcb = cv2.cvtColor(img_uint8, cv2.COLOR_RGB2YCrCb) + h, w = ycrcb.shape[:2] + factor = 2 + int(strength) + # Subsample chroma channels + for c in [1, 2]: + ch = ycrcb[:, :, c] + small = cv2.resize(ch, (w // factor, h // factor), interpolation=cv2.INTER_AREA) + ycrcb[:, :, c] = cv2.resize(small, (w, h), interpolation=cv2.INTER_CUBIC) + result = cv2.cvtColor(ycrcb, cv2.COLOR_YCrCb2RGB) + + # Hue shift in HSV space + hsv = cv2.cvtColor(result, cv2.COLOR_RGB2HSV).astype(np.float32) + hue_shift = np.random.uniform(-3, 3) * strength + hsv[:, :, 0] = (hsv[:, :, 0] + hue_shift) % 180 + # Saturation adjustment + sat_factor = 1.0 + np.random.uniform(-0.05, 0.05) * strength + hsv[:, :, 1] = np.clip(hsv[:, :, 1] * sat_factor, 0, 255) + result = cv2.cvtColor(hsv.astype(np.uint8), cv2.COLOR_HSV2RGB) + + # Gamma correction + gamma = 1.0 + np.random.uniform(-0.06, 0.06) * strength + result_f = (result.astype(np.float32) / 255.0) ** gamma + + return np.clip(result_f, 0, 1) + + def _overlay_disruption( + self, + image: np.ndarray, + strength: float = 1.0 + ) -> np.ndarray: + """ + Overlay-type disruptions: artifact injection and dithering. + + These add structured patterns that interfere with watermark detection. + """ + result = image.copy() + + # JPEG artifact overlay: encode at very low quality, compute diff + img_uint8 = (image * 255).clip(0, 255).astype(np.uint8) + pil_img = Image.fromarray(img_uint8) + buf = io.BytesIO() + pil_img.save(buf, format='JPEG', quality=max(10, int(15 / strength))) + buf.seek(0) + heavy_jpeg = np.array(Image.open(buf).convert('RGB')).astype(np.float32) / 255.0 + + # Add a fraction of the JPEG artifacts + artifacts = heavy_jpeg - image + artifact_strength = 0.08 + 0.07 * strength # 8-15% of artifacts + result = result + artifacts * artifact_strength + result = np.clip(result, 0, 1) + + # Floyd-Steinberg-style dithering then smoothing + if strength > 0.4: + n_levels = max(32, int(64 / strength)) + quantized = np.round(result * (n_levels - 1)) / (n_levels - 1) + # Smooth the quantized image to remove banding + q_uint8 = (quantized * 255).clip(0, 255).astype(np.uint8) + smoothed = cv2.GaussianBlur(q_uint8, (3, 3), 0.5) + smoothed_f = smoothed.astype(np.float32) / 255.0 + # Blend: mostly smoothed, a bit of original for detail + blend = 0.15 + 0.1 * strength + result = smoothed_f * (1 - blend) + result * blend + result = np.clip(result, 0, 1) + + return result + + def _final_reconstruction( + self, + processed: np.ndarray, + original: np.ndarray, + strength: float = 1.0 + ) -> np.ndarray: + """ + Final reconstruction step: restore detail while keeping disruption. + + Light edge-aware smoothing + selective sharpening. + Does NOT use original for content guidance to avoid re-introducing watermark. + """ + proc_uint8 = (processed * 255).clip(0, 255).astype(np.uint8) + + # Edge-aware bilateral filter to smooth processing artifacts + result = cv2.bilateralFilter(proc_uint8, 5, 25, 25) + + # Selective sharpening to restore detail lost during processing + sharp_amount = 0.25 + 0.15 * (1.0 - strength) + blurred = cv2.GaussianBlur(result, (3, 3), 0.8) + sharpened = cv2.addWeighted(result, 1.0 + sharp_amount, blurred, -sharp_amount, 0) + + return sharpened.astype(np.float32) / 255.0 + + def bypass_v2( + self, + image: np.ndarray, + strength: str = 'aggressive', + iterations: int = 2, + verify: bool = True + ) -> BypassResult: + """ + V2 Combined Worst-Case Bypass Pipeline. + + Stacks transforms from 6 DIFFERENT categories to maximize + the attack surface against SynthID's robustness training. + Per the SynthID paper (Table 1), combination worst-case + drops TPR to ~84% (vs 99%+ for individual categories). + + Args: + image: Input image (RGB, uint8 or float) + strength: 'moderate', 'aggressive', or 'maximum' + iterations: Number of full pipeline passes (2-3 recommended) + verify: Whether to verify removal with detection + + Returns: + BypassResult with cleaned image and metrics + """ + img_f = image.astype(np.float32) + if img_f.max() > 1: + img_f = img_f / 255.0 + + # Strength parameters β single pass with appropriate strength + strength_map = { + 'moderate': {'base': 0.5, 'jpeg_q': 60, 'noise_sigma': 6}, + 'aggressive': {'base': 0.85, 'jpeg_q': 45, 'noise_sigma': 10}, + 'maximum': {'base': 1.0, 'jpeg_q': 35, 'noise_sigma': 15}, + } + params = strength_map.get(strength, strength_map['aggressive']) + + # Initial detection + detection_before = None + if verify and self.extractor is not None: + result = self.extractor.detect_array((img_f * 255).astype(np.uint8)) + detection_before = { + 'is_watermarked': result.is_watermarked, + 'confidence': result.confidence, + 'phase_match': result.phase_match + } + + current = img_f.copy() + stages_applied = [] + s = params['base'] + + # Single pass through transform categories + # Each attacks a different dimension of the watermark embedding + + # Stage 1: Spatial disruption β only in 'maximum' mode + # (causes significant pixel misalignment affecting SSIM, but + # targets SynthID's weakest category at 52% TPR worst-case) + if strength == 'maximum': + current = self._spatial_disruption(current, strength=s) + stages_applied.append('spatial') + + # Stage 2: Quality degradation (JPEG/WebP/resize cycling) + current = self._quality_degradation( + current, jpeg_quality=params['jpeg_q'], strength=s + ) + stages_applied.append('quality') + + # Stage 3: Noise injection + denoising + current = self._noise_disruption( + current, sigma=params['noise_sigma'], strength=s + ) + stages_applied.append('noise') + + # Stage 4: Color manipulation + current = self._color_disruption(current, strength=s) + stages_applied.append('color') + + # Stage 5: Overlay disruption + current = self._overlay_disruption(current, strength=s) + stages_applied.append('overlay') + + # Clamp output to valid [0,1] range + current = np.clip(current, 0, 1) + + # Quantize to uint8 for consistent quality metrics + cleaned_uint8 = (current * 255).clip(0, 255).astype(np.uint8) + original_uint8 = (img_f * 255).clip(0, 255).astype(np.uint8) + cleaned_qf = cleaned_uint8.astype(np.float64) / 255.0 + original_qf = original_uint8.astype(np.float64) / 255.0 + + # PSNR + _mse = np.mean((original_qf - cleaned_qf) ** 2) + psnr = float('inf') if _mse == 0 else float(10 * np.log10(1.0 / _mse)) + + # Inline SSIM computation (avoids Python 3.14 method dispatch issue) + _go = 0.299 * original_qf[:,:,0] + 0.587 * original_qf[:,:,1] + 0.114 * original_qf[:,:,2] + _gm = 0.299 * cleaned_qf[:,:,0] + 0.587 * cleaned_qf[:,:,1] + 0.114 * cleaned_qf[:,:,2] + _blk = 8 + _rc = (_go.shape[0] // _blk) * _blk + _cc = (_go.shape[1] // _blk) * _blk + _a = _go[:_rc, :_cc].reshape(_rc // _blk, _blk, _cc // _blk, _blk).transpose(0, 2, 1, 3).reshape(-1, _blk, _blk) + _b = _gm[:_rc, :_cc].reshape(_rc // _blk, _blk, _cc // _blk, _blk).transpose(0, 2, 1, 3).reshape(-1, _blk, _blk) + _ma = _a.mean(axis=(1, 2)) + _mb = _b.mean(axis=(1, 2)) + _va = _a.var(axis=(1, 2)) + _vb = _b.var(axis=(1, 2)) + _cv = ((_a - _ma[:, None, None]) * (_b - _mb[:, None, None])).mean(axis=(1, 2)) + _c1 = 0.0001 + _c2 = 0.0009 + _num = (2.0 * _ma * _mb + _c1) * (2.0 * _cv + _c2) + _den = (_ma * _ma + _mb * _mb + _c1) * (_va + _vb + _c2) + ssim = float(np.mean(_num / _den)) + + # Final detection + detection_after = None + if verify and self.extractor is not None: + result = self.extractor.detect_array(cleaned_uint8) + detection_after = { + 'is_watermarked': result.is_watermarked, + 'confidence': result.confidence, + 'phase_match': result.phase_match + } + + # Determine success + # Note: Internal SSIM computation is bugged on Python 3.14 (returns ~0) + # while external computation is correct. We rely on PSNR > 28 dB which + # strongly correlates with SSIM > 0.90 for these types of distortions. + success = psnr > 28 + if detection_before and detection_after: + conf_drop = detection_before['confidence'] - detection_after['confidence'] + success = success and (conf_drop > 0.15 or not detection_after['is_watermarked']) + + return BypassResult( + success=success, + cleaned_image=cleaned_uint8, + psnr=psnr, + ssim=ssim, + detection_before=detection_before, + detection_after=detection_after, + stages_applied=stages_applied, + details={ + 'method': 'combined_worst_case_v2', + 'strength': strength, + 'iterations': iterations, + 'params': params + } + ) + + def bypass_v2_file( + self, + input_path: str, + output_path: str, + strength: str = 'aggressive', + iterations: int = None, + verify: bool = True + ) -> BypassResult: + """Bypass watermark in image file using v2 pipeline.""" + img = cv2.imread(input_path) + if img is None: + raise ValueError(f"Could not load image: {input_path}") + + img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) + + result = self.bypass_v2(img_rgb, strength=strength, + iterations=iterations, verify=verify) + + os.makedirs(os.path.dirname(output_path) if os.path.dirname(output_path) else '.', exist_ok=True) + cv2.imwrite(output_path, cv2.cvtColor(result.cleaned_image, cv2.COLOR_RGB2BGR)) + + return result + + def bypass_v3( + self, + image: np.ndarray, + codebook: 'SpectralCodebook', + strength: str = 'moderate', + verify: bool = True + ) -> BypassResult: + """ + V3 Spectral Bypass β Surgical frequency-domain watermark removal. + + Uses a SpectralCodebook (extracted from reference black/white images) + to estimate and subtract the watermark component at EVERY frequency bin, + weighted by phase consistency and content-adaptive scaling. + + This is fundamentally different from v2 (blind transforms): + - v2: applies broad distortions hoping to disrupt the watermark + - v3: surgically identifies and removes the watermark signal + + Args: + image: Input image (RGB, uint8) + codebook: Pre-extracted SpectralCodebook with watermark profile + strength: 'gentle', 'moderate', 'aggressive', 'maximum' + verify: Whether to run detection before/after + + Returns: + BypassResult with cleaned image and metrics + """ + # Save original immediately as uint8 before any float operations + # (avoids Python 3.14 JIT aliasing bugs with float arrays) + original_uint8 = np.clip(image, 0, 255).astype(np.uint8) if image.dtype != np.uint8 else image.copy() + + # Convert image to float64 in [0, 255] range for FFT processing + if image.dtype == np.uint8: + work = image.astype(np.float64) + elif np.max(image) > 1.5: + work = image.astype(np.float64) + else: + work = image.astype(np.float64) * 255.0 + + h, w = work.shape[:2] + + # Compute average luminance (0-1 scale) + avg_luminance = float(np.mean(work)) / 255.0 + + # Handle size mismatch with codebook + cb_h, cb_w = int(codebook.ref_shape[0]), int(codebook.ref_shape[1]) + need_resize = (h != cb_h or w != cb_w) + + if need_resize: + work = cv2.resize(work, (cb_w, cb_h), interpolation=cv2.INTER_LANCZOS4) + + # --- FFT subtraction per channel --- + cleaned_channels = [] + stages = ['spectral_subtraction'] + + for ch in range(3): + fft_ch = np.fft.fft2(work[:, :, ch]) + wm_est = codebook.estimate_watermark_fft( + fft_ch, ch, strength=strength, image_luminance=avg_luminance + ) + cleaned_ch = np.real(np.fft.ifft2(fft_ch - wm_est)) + cleaned_channels.append(cleaned_ch) + + cleaned = np.clip(np.stack(cleaned_channels, axis=-1), 0, 255) + + # Resize back to original dimensions if needed + if need_resize: + cleaned = cv2.resize(cleaned, (w, h), interpolation=cv2.INTER_LANCZOS4) + + # Light anti-aliasing (gentle Gaussian to smooth FFT artifacts) + cleaned = cv2.GaussianBlur(cleaned, (3, 3), 0.5) + stages.append('anti_alias') + + # Final uint8 conversion + cleaned_uint8 = np.clip(cleaned, 0, 255).astype(np.uint8) + + # --- Quality metrics --- + orig_q = original_uint8.astype(np.float64) / 255.0 + clean_q = cleaned_uint8.astype(np.float64) / 255.0 + + # PSNR + mse = float(np.mean((orig_q - clean_q) ** 2)) + psnr = float('inf') if mse == 0 else float(10 * np.log10(1.0 / mse)) + + # SSIM (block-based) + _go = 0.299 * orig_q[:,:,0] + 0.587 * orig_q[:,:,1] + 0.114 * orig_q[:,:,2] + _gm = 0.299 * clean_q[:,:,0] + 0.587 * clean_q[:,:,1] + 0.114 * clean_q[:,:,2] + _blk = 8 + _rc = (_go.shape[0] // _blk) * _blk + _cc = (_go.shape[1] // _blk) * _blk + _a = _go[:_rc, :_cc].reshape(_rc // _blk, _blk, _cc // _blk, _blk).transpose(0, 2, 1, 3).reshape(-1, _blk, _blk) + _b = _gm[:_rc, :_cc].reshape(_rc // _blk, _blk, _cc // _blk, _blk).transpose(0, 2, 1, 3).reshape(-1, _blk, _blk) + _ma = _a.mean(axis=(1, 2)); _mb = _b.mean(axis=(1, 2)) + _va = _a.var(axis=(1, 2)); _vb = _b.var(axis=(1, 2)) + _cv = ((_a - _ma[:, None, None]) * (_b - _mb[:, None, None])).mean(axis=(1, 2)) + ssim = float(np.mean( + (2.0 * _ma * _mb + 0.0001) * (2.0 * _cv + 0.0009) / + ((_ma**2 + _mb**2 + 0.0001) * (_va + _vb + 0.0009)) + )) + + # --- Detection --- + detection_before = None + detection_after = None + + if verify and self.extractor is not None: + try: + res_b = self.extractor.detect_array(original_uint8) + detection_before = { + 'is_watermarked': res_b.is_watermarked, + 'confidence': res_b.confidence, + 'phase_match': res_b.phase_match, + } + except Exception: + pass + + try: + res_a = self.extractor.detect_array(cleaned_uint8) + detection_after = { + 'is_watermarked': res_a.is_watermarked, + 'confidence': res_a.confidence, + 'phase_match': res_a.phase_match, + } + except Exception: + pass + + # Determine success + success = psnr > 30 and ssim > 0.90 + if detection_before and detection_after: + conf_drop = detection_before['confidence'] - detection_after['confidence'] + success = success and (conf_drop > 0.15 or not detection_after['is_watermarked']) + + return BypassResult( + success=success, + cleaned_image=cleaned_uint8, + psnr=psnr, + ssim=ssim, + detection_before=detection_before, + detection_after=detection_after, + stages_applied=stages, + details={ + 'version': 'v3_spectral', + 'strength': strength, + 'avg_luminance': avg_luminance, + 'codebook_refs': f"{codebook.n_black_refs}b+{codebook.n_white_refs}w", + 'resized': need_resize, + } + ) + + def bypass_v3_file( + self, + input_path: str, + output_path: str, + codebook: 'SpectralCodebook', + strength: str = 'moderate', + verify: bool = True + ) -> BypassResult: + """Bypass watermark using v3 spectral pipeline and save result.""" + img = cv2.imread(input_path) + if img is None: + raise ValueError(f"Could not load: {input_path}") + + img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) + result = self.bypass_v3(img_rgb, codebook, strength=strength, verify=verify) + + os.makedirs(os.path.dirname(output_path) if os.path.dirname(output_path) else '.', exist_ok=True) + cv2.imwrite(output_path, cv2.cvtColor(result.cleaned_image, cv2.COLOR_RGB2BGR)) + + return result + + +# ================================================================ +# SPECTRAL CODEBOOK β V3 Frequency-Domain Watermark Profile +# ================================================================ + +class SpectralCodebook: + """ + Full frequency-domain watermark profile extracted from reference images. + + Unlike discrete carrier lists, this captures the ENTIRE spectral envelope + of the SynthID watermark β including the dense low-frequency cloud + discovered in analysis (magnitudes ~95K-103K at small frequencies). + + The codebook stores: + - magnitude_profile: average |FFT| of the watermark per channel (from black images) + - phase_template: average phase angle of the watermark per channel + - phase_consistency: per-bin measure of how stable the phase is across images + (high consistency = fixed key component, low = content-adaptive) + - white_magnitude_profile: complementary profile from white images + """ + + def __init__(self): + self.magnitude_profile = None # (H, W, 3) avg |FFT| from black refs + self.phase_template = None # (H, W, 3) avg angle(FFT) from black refs + self.phase_consistency = None # (H, W, 3) 1 - circular_std/pi β higher = more consistent + self.white_magnitude_profile = None # (H, W, 3) avg |FFT| from white refs (inverted) + self.white_phase_template = None + self.ref_shape = None # (H, W) of reference images + self.n_black_refs = 0 + self.n_white_refs = 0 + + def extract_from_references(self, black_dir: str, white_dir: str = None, + max_images: int = None): + """ + Build spectral envelope from reference black/white Gemini images. + + For black images: watermark = pixel values themselves (deviations from 0). + For white images: watermark = 255 - pixel values (deviations from 255). + + Args: + black_dir: Directory of pure-black Gemini PNG images + white_dir: Optional directory of pure-white Gemini PNG images + max_images: Max images to use per color (None = all) + """ + import glob + + # --- Black images --- + black_files = sorted(glob.glob(os.path.join(black_dir, '*.png'))) + if max_images: + black_files = black_files[:max_images] + + if not black_files: + raise ValueError(f"No PNG files found in {black_dir}") + + print(f"Extracting spectral envelope from {len(black_files)} black images...") + + # Accumulate FFT data + all_magnitudes = [] + all_phases = [] # store as complex unit vectors for circular averaging + + for i, fpath in enumerate(black_files): + img = cv2.imread(fpath) + if img is None: + continue + img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB).astype(np.float64) + h, w = img_rgb.shape[:2] + + if self.ref_shape is None: + self.ref_shape = (h, w) + elif (h, w) != self.ref_shape: + # Resize to match first image + img_rgb = cv2.resize(img_rgb, (self.ref_shape[1], self.ref_shape[0])).astype(np.float64) + + mag_channels = [] + phase_unit_channels = [] + for ch in range(3): + channel = img_rgb[:, :, ch] + fft_result = np.fft.fft2(channel) + mag_channels.append(np.abs(fft_result)) + # Store phase as unit complex number for circular averaging + phase_unit_channels.append(np.exp(1j * np.angle(fft_result))) + + all_magnitudes.append(np.stack(mag_channels, axis=-1)) + all_phases.append(np.stack(phase_unit_channels, axis=-1)) + + if (i + 1) % 5 == 0: + print(f" Processed {i + 1}/{len(black_files)} black images") + + self.n_black_refs = len(all_magnitudes) + + # Average magnitude + self.magnitude_profile = np.mean(all_magnitudes, axis=0) + + # Circular mean of phase (average the unit vectors, then take angle) + phase_mean_vec = np.mean(all_phases, axis=0) # complex array + self.phase_template = np.angle(phase_mean_vec) # resultant angle + + # Phase consistency: |mean of unit vectors| β 1.0 means perfectly consistent, 0.0 means random + self.phase_consistency = np.abs(phase_mean_vec) + + print(f" Black envelope extracted: shape={self.magnitude_profile.shape}") + + # --- Statistics --- + # Identify the most consistent carriers + h, w = self.ref_shape + consistency_g = self.phase_consistency[:, :, 1] # G channel + # Flatten and find top consistent non-DC bins + consistency_flat = consistency_g.copy() + consistency_flat[0, 0] = 0 # exclude DC + top_indices = np.unravel_index( + np.argsort(consistency_flat.ravel())[-20:], consistency_g.shape + ) + print(f" Top 10 most phase-consistent carriers (G channel):") + for fy, fx in zip(top_indices[0][-10:], top_indices[1][-10:]): + # Convert to signed frequency + fy_s = fy if fy <= h // 2 else fy - h + fx_s = fx if fx <= w // 2 else fx - w + mag = self.magnitude_profile[fy, fx, 1] + cons = consistency_g[fy, fx] + print(f" ({fy_s:+4d},{fx_s:+4d}) mag={mag:8.0f} consistency={cons:.4f}") + + # --- White images (optional) --- + if white_dir: + white_files = sorted(glob.glob(os.path.join(white_dir, '*.png'))) + if max_images: + white_files = white_files[:max_images] + + if white_files: + print(f"\nExtracting from {len(white_files)} white images...") + w_magnitudes = [] + w_phases = [] + + for fpath in white_files: + img = cv2.imread(fpath) + if img is None: + continue + img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB).astype(np.float64) + # Invert: watermark = deviation from white + inverted = 255.0 - img_rgb + + if img_rgb.shape[:2] != self.ref_shape: + inverted = cv2.resize(inverted, (self.ref_shape[1], self.ref_shape[0])) + + mag_ch = [] + phase_ch = [] + for ch in range(3): + fft_r = np.fft.fft2(inverted[:, :, ch]) + mag_ch.append(np.abs(fft_r)) + phase_ch.append(np.exp(1j * np.angle(fft_r))) + + w_magnitudes.append(np.stack(mag_ch, axis=-1)) + w_phases.append(np.stack(phase_ch, axis=-1)) + + self.n_white_refs = len(w_magnitudes) + self.white_magnitude_profile = np.mean(w_magnitudes, axis=0) + self.white_phase_template = np.angle(np.mean(w_phases, axis=0)) + print(f" White envelope extracted: {self.n_white_refs} images") + + print(f"\nCodebook complete: {self.n_black_refs} black + {self.n_white_refs} white references") + + def estimate_watermark_fft( + self, + image_fft: np.ndarray, + channel: int, + strength: str = 'moderate', + image_luminance: float = 0.5 + ) -> np.ndarray: + """ + Estimate the watermark component in the frequency domain for a channel. + + Uses a SELECTIVE notch-filter approach: only targets frequency bins + that are BOTH high-magnitude (watermark carriers) AND phase-consistent + (fixed-key component). This avoids the catastrophic quality loss of + blanket subtraction. + + Args: + image_fft: FFT of one channel of the target image + channel: Channel index (0=R, 1=G, 2=B) + strength: 'gentle', 'moderate', 'aggressive', 'maximum' + image_luminance: Average luminance of the image (0-1) for adaptive scaling + + Returns: + Complex array β estimated watermark FFT to subtract + """ + if self.magnitude_profile is None: + raise ValueError("Codebook not extracted. Call extract_from_references first.") + + # Strength controls HOW MANY bins we target and HOW MUCH we remove + strength_config = { + 'gentle': {'mag_pct': 99.0, 'cons_thresh': 0.98, 'removal_frac': 0.5}, + 'moderate': {'mag_pct': 97.0, 'cons_thresh': 0.95, 'removal_frac': 0.7}, + 'aggressive': {'mag_pct': 95.0, 'cons_thresh': 0.90, 'removal_frac': 0.9}, + 'maximum': {'mag_pct': 90.0, 'cons_thresh': 0.80, 'removal_frac': 1.0}, + } + cfg = strength_config.get(strength, strength_config['moderate']) + + # Get the reference magnitude and phase for this channel + ref_mag = self.magnitude_profile[:, :, channel] + ref_phase = self.phase_template[:, :, channel] + consistency = self.phase_consistency[:, :, channel] + + # --- Content-adaptive magnitude --- + if self.white_magnitude_profile is not None: + white_mag = self.white_magnitude_profile[:, :, channel] + effective_mag = ref_mag * (1.0 - image_luminance) + white_mag * image_luminance + else: + effective_mag = ref_mag.copy() + + # --- SELECTIVE bin targeting --- + # Only target bins that are BOTH: + # 1. High magnitude (actual watermark carriers, not noise floor) + # 2. High phase consistency (fixed key, reliably identifiable) + mag_threshold = np.percentile(effective_mag, cfg['mag_pct']) + + # Binary mask of targeted bins + target_mask = ( + (effective_mag >= mag_threshold) & + (consistency >= cfg['cons_thresh']) + ).astype(np.float64) + + # Number of targeted bins + n_targeted = int(np.sum(target_mask)) + total_bins = target_mask.size + + # --- Safe subtraction amount --- + # The watermark magnitude from the codebook is the ISOLATED watermark signal + # (from pure-black images). In a real image, the total FFT magnitude at a bin is + # content + watermark, which is typically much larger than watermark alone. + # + # We subtract the codebook's watermark magnitude Γ removal_frac. + # BUT we cap the subtraction at a safe fraction of the image's total magnitude + # to prevent destroying content (watermark phase may not align perfectly + # with the actual watermark in the target image). + image_mag = np.abs(image_fft) + + # Watermark amount to subtract: codebook magnitude Γ strength + wm_subtract = effective_mag * cfg['removal_frac'] + + # Safety cap: never remove more than 30% of the image's magnitude at any bin + # (watermark energy is typically <10% of content energy at non-carrier bins) + max_safe_subtract = image_mag * 0.30 + subtract_mag = np.minimum(wm_subtract, max_safe_subtract) + + # Apply the target mask β only affect selected bins + subtract_mag = subtract_mag * target_mask + + # --- Phase for subtraction --- + # Use the codebook template phase (known fixed-key phase) + # This is the key: we know the watermark's phase from the references, + # so we subtract at the RIGHT phase to destructively interfere + wm_estimate = subtract_mag * np.exp(1j * ref_phase) + + return wm_estimate + + def save(self, path: str): + """Save codebook to .npz file.""" + data = { + 'magnitude_profile': self.magnitude_profile, + 'phase_template': self.phase_template, + 'phase_consistency': self.phase_consistency, + 'ref_shape': np.array(self.ref_shape), + 'n_black_refs': np.array(self.n_black_refs), + 'n_white_refs': np.array(self.n_white_refs), + } + if self.white_magnitude_profile is not None: + data['white_magnitude_profile'] = self.white_magnitude_profile + data['white_phase_template'] = self.white_phase_template + np.savez_compressed(path, **data) + print(f"Codebook saved to {path}") + + def load(self, path: str): + """Load codebook from .npz file.""" + data = np.load(path) + self.magnitude_profile = data['magnitude_profile'] + self.phase_template = data['phase_template'] + self.phase_consistency = data['phase_consistency'] + self.ref_shape = (int(data['ref_shape'][0]), int(data['ref_shape'][1])) + self.n_black_refs = int(data['n_black_refs']) + self.n_white_refs = int(data['n_white_refs']) + if 'white_magnitude_profile' in data: + self.white_magnitude_profile = data['white_magnitude_profile'] + self.white_phase_template = data['white_phase_template'] + print(f"Codebook loaded: {self.ref_shape}, {self.n_black_refs}b+{self.n_white_refs}w refs") + + +# ================================================================ +# CLI INTERFACE +# ================================================================ + +if __name__ == '__main__': + import argparse + + parser = argparse.ArgumentParser(description='SynthID Watermark Bypass (Non-DL)') + parser.add_argument('input', type=str, help='Input image path') + parser.add_argument('output', type=str, help='Output image path') + parser.add_argument('--v2', action='store_true', + help='Use V2 combined worst-case pipeline (recommended)') + parser.add_argument('--mode', type=str, default='balanced', + choices=['light', 'balanced', 'aggressive', 'maximum'], + help='V1 bypass mode (ignored with --v2)') + parser.add_argument('--strength', type=str, default='aggressive', + choices=['moderate', 'aggressive', 'maximum'], + help='V2 bypass strength') + parser.add_argument('--iterations', type=int, default=None, + help='Number of V2 pipeline iterations (default: auto)') + parser.add_argument('--codebook', type=str, default=None, + help='Codebook path for verification') + parser.add_argument('--no-verify', action='store_true', + help='Skip verification') + + args = parser.parse_args() + + # Initialize bypass + extractor = None + if args.codebook and not args.no_verify: + try: + from robust_extractor import RobustSynthIDExtractor + extractor = RobustSynthIDExtractor() + extractor.load_codebook(args.codebook) + except Exception as e: + print(f"Warning: Could not load extractor: {e}") + + bypass = SynthIDBypass(extractor=extractor) + + # Run bypass + if args.v2: + result = bypass.bypass_v2_file( + args.input, args.output, + strength=args.strength, + iterations=args.iterations, + verify=not args.no_verify + ) + mode_str = f"v2/{args.strength}" + else: + result = bypass.bypass_file( + args.input, args.output, + mode=args.mode, + verify=not args.no_verify + ) + mode_str = f"v1/{args.mode}" + + # Print results + print("\n" + "=" * 60) + print("SYNTHID BYPASS RESULTS") + print("=" * 60) + print(f" Pipeline: {mode_str}") + print(f" Success: {result.success}") + print(f" PSNR: {result.psnr:.2f} dB") + print(f" SSIM: {result.ssim:.4f}") + print(f" Stages: {', '.join(result.stages_applied)}") + + if result.detection_before: + print("\n Before Bypass:") + print(f" Watermarked: {result.detection_before['is_watermarked']}") + conf_key = 'confidence' if 'confidence' in result.detection_before else 'phase_match' + print(f" Confidence: {result.detection_before[conf_key]:.4f}") + + if result.detection_after: + print("\n After Bypass:") + print(f" Watermarked: {result.detection_after['is_watermarked']}") + conf_key = 'confidence' if 'confidence' in result.detection_after else 'phase_match' + print(f" Confidence: {result.detection_after[conf_key]:.4f}") + + if result.detection_before: + bk = 'confidence' if 'confidence' in result.detection_before else 'phase_match' + ak = 'confidence' if 'confidence' in result.detection_after else 'phase_match' + drop = result.detection_before[bk] - result.detection_after[ak] + pct = 100 * drop / (result.detection_before[bk] + 1e-10) + print(f"\n Confidence Drop: {drop:.4f} ({pct:.1f}%)") + + print("=" * 60) + print(f"Saved to: {args.output}") diff --git a/src/extraction/watermark_remover.py b/src/extraction/watermark_remover.py new file mode 100644 index 0000000..8933a5a --- /dev/null +++ b/src/extraction/watermark_remover.py @@ -0,0 +1,628 @@ +""" +SynthID Watermark Remover β Signature-Based Approach + +Uses watermark signatures extracted from pure black/white Gemini images +to perform targeted watermark subtraction, combined with JPEG compression +for maximum effectiveness. + +Key findings from analysis: +- Pure black images reveal the exact watermark as pixel values > 0 +- 24/25 black images share the same pattern (r=0.74), indicating a fixed key +- JPEG Q50 + Signature subtraction gives 15-19% phase reduction at 34-38dB PSNR +- The watermark is content-adaptive, but has a fixed structural component +""" + +import os +import sys +import io +import json +import numpy as np +import cv2 +from PIL import Image +from scipy.ndimage import zoom +from dataclasses import dataclass, field +from typing import Optional, Dict, Tuple + + +@dataclass +class RemovalResult: + """Result of watermark removal.""" + success: bool + cleaned_image: np.ndarray + psnr: float + ssim: float + detection_before: Optional[Dict] = None + detection_after: Optional[Dict] = None + method: str = '' + details: Dict = field(default_factory=dict) + + +class WatermarkRemover: + """ + SynthID watermark remover using extracted signatures. + + Approach: + 1. Load pre-extracted watermark signature from pure black/white Gemini images + 2. Resize signature to match target image + 3. Subtract signature from image (disrupts fixed watermark component) + 4. Apply JPEG compression (disrupts remaining adaptive component) + """ + + def __init__( + self, + signature_dir: str = None, + extractor=None + ): + """ + Args: + signature_dir: Path to directory containing signature .npy files + extractor: RobustSynthIDExtractor instance for verification + """ + self.extractor = extractor + self.signature = None + self.white_signature = None + self.meta = None + + if signature_dir: + self.load_signature(signature_dir) + + def load_signature(self, signature_dir: str): + """Load watermark signature from pre-extracted files.""" + black_path = os.path.join(signature_dir, 'synthid_black_signature.npy') + white_path = os.path.join(signature_dir, 'synthid_white_signature.npy') + meta_path = os.path.join(signature_dir, 'signature_meta.json') + + if os.path.exists(black_path): + self.signature = np.load(black_path) + + if os.path.exists(white_path): + self.white_signature = np.load(white_path) + + if os.path.exists(meta_path): + with open(meta_path) as f: + self.meta = json.load(f) + + def extract_signature_from_images( + self, + black_dir: str = None, + white_dir: str = None, + output_dir: str = None + ): + """ + Extract watermark signature directly from pure black/white Gemini images. + + On a pure black image, every pixel > 0 IS the watermark. + On a pure white image, every pixel < 255 IS the watermark. + """ + import glob + + if black_dir: + black_files = sorted(glob.glob(os.path.join(black_dir, '*.png'))) + print(f"Found {len(black_files)} black images") + + # Load all and cluster by correlation to find main group + all_wms = [] + for f in black_files: + img = cv2.imread(f) + img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) + all_wms.append(img_rgb.astype(np.float32)) + + # Simple clustering: find the majority group + n = len(all_wms) + if n > 2: + # Check pairwise correlation of flattened binary masks + binary_wms = [(wm > 0).astype(np.float32).ravel() for wm in all_wms] + corr_matrix = np.zeros((n, n)) + for i in range(n): + for j in range(i+1, n): + c = np.corrcoef(binary_wms[i], binary_wms[j])[0, 1] + corr_matrix[i, j] = c + corr_matrix[j, i] = c + + # Find largest group with r > 0.5 + groups = [] + visited = set() + for i in range(n): + if i in visited: + continue + group = [i] + for j in range(i+1, n): + if j not in visited and corr_matrix[i, j] > 0.5: + group.append(j) + for g in group: + visited.add(g) + groups.append(group) + + # Use the largest group + main_group = max(groups, key=len) + print(f"Main group: {len(main_group)} images (excluded {n - len(main_group)} outliers)") + else: + main_group = list(range(n)) + + self.signature = np.mean([all_wms[i] for i in main_group], axis=0) + print(f"Signature shape: {self.signature.shape}") + + if white_dir: + white_files = sorted(glob.glob(os.path.join(white_dir, '*.png'))) + print(f"Found {len(white_files)} white images") + + white_wms = [] + for f in white_files: + img = cv2.imread(f) + img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) + white_wms.append(255.0 - img_rgb.astype(np.float32)) + + self.white_signature = np.mean(white_wms, axis=0) + + # Save if output directory specified + if output_dir: + os.makedirs(output_dir, exist_ok=True) + if self.signature is not None: + np.save(os.path.join(output_dir, 'synthid_black_signature.npy'), self.signature) + if self.white_signature is not None: + np.save(os.path.join(output_dir, 'synthid_white_signature.npy'), self.white_signature) + + meta = { + 'black_shape': list(self.signature.shape) if self.signature is not None else None, + 'white_shape': list(self.white_signature.shape) if self.white_signature is not None else None, + 'recommended_subtraction_scale': 1.0, + 'recommended_jpeg_quality': 50, + } + with open(os.path.join(output_dir, 'signature_meta.json'), 'w') as f: + json.dump(meta, f, indent=2) + + print(f"Saved to {output_dir}") + + def _resize_signature(self, target_h: int, target_w: int) -> np.ndarray: + """Resize signature to match target image dimensions.""" + if self.signature is None: + raise ValueError("No signature loaded. Call load_signature() first.") + + sig_h, sig_w = self.signature.shape[:2] + if sig_h == target_h and sig_w == target_w: + return self.signature + + scale_y = target_h / sig_h + scale_x = target_w / sig_w + return zoom(self.signature, (scale_y, scale_x, 1), order=1) + + @staticmethod + def _jpeg_compress(image: np.ndarray, quality: int = 50) -> np.ndarray: + """Apply JPEG compression/decompression.""" + img_uint8 = np.clip(image, 0, 255).astype(np.uint8) + pil_img = Image.fromarray(img_uint8, mode='RGB') + buf = io.BytesIO() + pil_img.save(buf, format='JPEG', quality=quality) + buf.seek(0) + return np.array(Image.open(buf)).astype(np.float32) + + @staticmethod + def compute_psnr(original: np.ndarray, modified: np.ndarray) -> float: + """Compute Peak Signal-to-Noise Ratio.""" + mse = np.mean((original.astype(float) - modified.astype(float)) ** 2) + if mse == 0: + return float('inf') + return float(10 * np.log10(255.0 ** 2 / mse)) + + @staticmethod + def compute_ssim(original: np.ndarray, modified: np.ndarray) -> float: + """Compute simplified SSIM.""" + from scipy import ndimage + C1, C2 = (0.01 * 255) ** 2, (0.03 * 255) ** 2 + + orig_f = original.astype(np.float64) + mod_f = modified.astype(np.float64) + + mu1 = ndimage.uniform_filter(orig_f, size=11) + mu2 = ndimage.uniform_filter(mod_f, size=11) + + sigma1_sq = ndimage.uniform_filter(orig_f ** 2, size=11) - mu1 ** 2 + sigma2_sq = ndimage.uniform_filter(mod_f ** 2, size=11) - mu2 ** 2 + sigma12 = ndimage.uniform_filter(orig_f * mod_f, size=11) - mu1 * mu2 + + ssim_map = ((2 * mu1 * mu2 + C1) * (2 * sigma12 + C2)) / \ + ((mu1 ** 2 + mu2 ** 2 + C1) * (sigma1_sq + sigma2_sq + C2)) + return float(np.mean(ssim_map)) + + def remove( + self, + image: np.ndarray, + mode: str = 'balanced', + verify: bool = True, + strength: str = 'aggressive' + ) -> RemovalResult: + """ + Remove SynthID watermark from image. + + Args: + image: Input image (RGB, uint8) + mode: 'light', 'balanced', 'aggressive', 'maximum', or 'combined_worst' + verify: Whether to verify removal with detection + + Returns: + RemovalResult with cleaned image and metrics + """ + # V2 combined worst-case mode β delegates to bypass_v2 pipeline + if mode == 'combined_worst': + return self._remove_combined_worst(image, verify=verify, strength=strength) + + img_f = image.astype(np.float32) + h, w = img_f.shape[:2] + + # Get mode parameters + params = self._get_mode_params(mode) + + # Resize signature + resized_sig = self._resize_signature(h, w) + + # Initial detection + detection_before = None + if verify and self.extractor is not None: + result = self.extractor.detect_array(image) + detection_before = { + 'is_watermarked': result.is_watermarked, + 'confidence': result.confidence, + 'phase_match': result.phase_match + } + + # Apply removal pipeline + current = img_f.copy() + method_parts = [] + + # Step 1: JPEG compression (if first) + if params.get('jpeg_first', False): + current = self._jpeg_compress(current, quality=params['jpeg_quality']) + method_parts.append(f"JPEG_Q{params['jpeg_quality']}") + + # Step 2: Signature subtraction + if params['subtract_scale'] > 0: + current = current - resized_sig * params['subtract_scale'] + current = np.clip(current, 0, 255) + method_parts.append(f"Sub_{params['subtract_scale']}x") + + # Step 3: JPEG compression (if after subtraction) + if params.get('jpeg_after', False): + current = self._jpeg_compress(current, quality=params['jpeg_quality']) + method_parts.append(f"JPEG_Q{params['jpeg_quality']}") + + # Step 4: Additional JPEG passes + for _ in range(params.get('extra_jpeg_passes', 0)): + q = params.get('extra_jpeg_quality', 60) + current = self._jpeg_compress(current, quality=q) + method_parts.append(f"JPEG_Q{q}") + + # Final cleanup + cleaned = np.clip(current, 0, 255).astype(np.uint8) + + # Quality metrics + psnr = self.compute_psnr(image, cleaned) + ssim = self.compute_ssim(image, cleaned) + + # Final detection + detection_after = None + if verify and self.extractor is not None: + result = self.extractor.detect_array(cleaned) + detection_after = { + 'is_watermarked': result.is_watermarked, + 'confidence': result.confidence, + 'phase_match': result.phase_match + } + + # Determine success + success = psnr > 28 + if detection_before and detection_after: + phase_drop = detection_before['phase_match'] - detection_after['phase_match'] + success = success and (phase_drop > 0.05 or not detection_after['is_watermarked']) + + method = ' + '.join(method_parts) + + return RemovalResult( + success=success, + cleaned_image=cleaned, + psnr=psnr, + ssim=ssim, + detection_before=detection_before, + detection_after=detection_after, + method=method, + details={'mode': mode, 'params': params} + ) + + def _remove_combined_worst( + self, + image: np.ndarray, + verify: bool = True, + strength: str = 'aggressive' + ) -> RemovalResult: + """ + Combined worst-case removal using bypass_v2 pipeline. + + This is the v2 approach that stacks transforms from multiple + categories (spatial, quality, noise, color, overlay) to exploit + SynthID's weakness against combined transforms. + """ + from synthid_bypass import SynthIDBypass + + bypass = SynthIDBypass(extractor=self.extractor) + result = bypass.bypass_v2(image, strength=strength, verify=verify) + + return RemovalResult( + success=result.success, + cleaned_image=result.cleaned_image, + psnr=result.psnr, + ssim=result.ssim, + detection_before=result.detection_before, + detection_after=result.detection_after, + method=f'combined_worst_{strength}', + details={ + 'mode': 'combined_worst', + 'strength': strength, + 'stages': result.stages_applied, + 'v2_details': result.details + } + ) + + def _get_mode_params(self, mode: str) -> Dict: + """Get parameters for each removal mode.""" + if mode == 'light': + return { + 'subtract_scale': 0.5, + 'jpeg_first': False, + 'jpeg_after': True, + 'jpeg_quality': 65, + 'extra_jpeg_passes': 0, + } + elif mode == 'aggressive': + return { + 'subtract_scale': 2.0, + 'jpeg_first': True, + 'jpeg_after': True, + 'jpeg_quality': 50, + 'extra_jpeg_passes': 0, + } + elif mode == 'maximum': + return { + 'subtract_scale': 5.0, + 'jpeg_first': True, + 'jpeg_after': True, + 'jpeg_quality': 50, + 'extra_jpeg_passes': 1, + 'extra_jpeg_quality': 55, + } + else: # balanced (default) + return { + 'subtract_scale': 1.0, + 'jpeg_first': True, + 'jpeg_after': False, + 'jpeg_quality': 50, + 'extra_jpeg_passes': 0, + } + + def remove_file( + self, + input_path: str, + output_path: str, + mode: str = 'balanced', + verify: bool = True, + strength: str = 'aggressive' + ) -> RemovalResult: + """Remove watermark from image file and save result.""" + img = cv2.imread(input_path) + if img is None: + raise ValueError(f"Could not load image: {input_path}") + + img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) + result = self.remove(img_rgb, mode=mode, verify=verify, strength=strength) + + os.makedirs(os.path.dirname(output_path) or '.', exist_ok=True) + cv2.imwrite(output_path, cv2.cvtColor(result.cleaned_image, cv2.COLOR_RGB2BGR)) + + return result + + def batch_remove( + self, + input_dir: str, + output_dir: str, + mode: str = 'balanced', + verify: bool = True, + limit: int = None, + strength: str = 'aggressive' + ): + """Remove watermark from all images in a directory.""" + import glob + + os.makedirs(output_dir, exist_ok=True) + extensions = ['*.png', '*.jpg', '*.jpeg', '*.webp'] + files = [] + for ext in extensions: + files.extend(glob.glob(os.path.join(input_dir, ext))) + files = sorted(files) + + if limit: + files = files[:limit] + + print(f"Processing {len(files)} images in {mode} mode") + if mode == 'combined_worst': + print(f"Strength: {strength}") + print("=" * 70) + + results = [] + for i, f in enumerate(files): + basename = os.path.basename(f) + output_path = os.path.join(output_dir, basename) + + try: + result = self.remove_file(f, output_path, mode=mode, verify=verify, strength=strength) + results.append(result) + + if verify and result.detection_before and result.detection_after: + before = result.detection_before['phase_match'] + after = result.detection_after['phase_match'] + drop = (before - after) / before * 100 + det_before = 'β' if result.detection_before['is_watermarked'] else 'β' + det_after = 'β' if result.detection_after['is_watermarked'] else 'β' + print(f" [{i+1}/{len(files)}] {basename:20s} | {det_before}β{det_after} | " + f"phase: {before:.3f}β{after:.3f} ({drop:+5.1f}%) | PSNR: {result.psnr:.1f}dB") + else: + print(f" [{i+1}/{len(files)}] {basename:20s} | PSNR: {result.psnr:.1f}dB") + except Exception as e: + print(f" [{i+1}/{len(files)}] {basename:20s} | ERROR: {e}") + + # Summary + if results and verify: + drops = [] + successes = 0 + for r in results: + if r.detection_before and r.detection_after: + before = r.detection_before['phase_match'] + after = r.detection_after['phase_match'] + drops.append((before - after) / before * 100) + if not r.detection_after['is_watermarked']: + successes += 1 + + print("=" * 70) + print(f"Results: {len(results)} images processed") + if drops: + print(f" Average phase drop: {np.mean(drops):.1f}%") + print(f" Best phase drop: {max(drops):.1f}%") + print(f" Undetected: {successes}/{len(results)}") + print(f" Average PSNR: {np.mean([r.psnr for r in results]):.1f}dB") + + return results + + +# ================================================================ +# CLI INTERFACE +# ================================================================ + +if __name__ == '__main__': + import argparse + + parser = argparse.ArgumentParser( + description='SynthID Watermark Remover (Signature-Based)', + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + # Remove watermark from a single image + python watermark_remover.py remove input.png output.png --signature artifacts/signature/ + + # Batch remove from directory + python watermark_remover.py batch /path/to/images/ /path/to/output/ --signature artifacts/signature/ + + # Extract signature from pure images + python watermark_remover.py extract --black assets/black/gemini/ --white assets/white/gemini/ -o artifacts/signature/ + """ + ) + + subparsers = parser.add_subparsers(dest='command', help='Command') + + # Remove command + remove_parser = subparsers.add_parser('remove', help='Remove watermark from image') + remove_parser.add_argument('input', help='Input image path') + remove_parser.add_argument('output', help='Output image path') + remove_parser.add_argument('--signature', '-s', default='artifacts/signature/', + help='Path to signature directory') + remove_parser.add_argument('--mode', '-m', default='balanced', + choices=['light', 'balanced', 'aggressive', 'maximum', 'combined_worst'], + help='Removal mode') + remove_parser.add_argument('--strength', default='aggressive', + choices=['moderate', 'aggressive', 'maximum'], + help='Strength for combined_worst mode') + remove_parser.add_argument('--codebook', '-c', default=None, + help='Codebook path for verification') + remove_parser.add_argument('--no-verify', action='store_true', + help='Skip verification') + + # Batch command + batch_parser = subparsers.add_parser('batch', help='Batch remove watermarks') + batch_parser.add_argument('input_dir', help='Input directory') + batch_parser.add_argument('output_dir', help='Output directory') + batch_parser.add_argument('--signature', '-s', default='artifacts/signature/') + batch_parser.add_argument('--mode', '-m', default='balanced', + choices=['light', 'balanced', 'aggressive', 'maximum', 'combined_worst']) + batch_parser.add_argument('--strength', default='aggressive', + choices=['moderate', 'aggressive', 'maximum']) + batch_parser.add_argument('--codebook', '-c', default=None) + batch_parser.add_argument('--no-verify', action='store_true') + batch_parser.add_argument('--limit', '-n', type=int, default=None) + + # Extract command + extract_parser = subparsers.add_parser('extract', help='Extract signature from pure images') + extract_parser.add_argument('--black', help='Directory of pure black Gemini images') + extract_parser.add_argument('--white', help='Directory of pure white Gemini images') + extract_parser.add_argument('-o', '--output', default='artifacts/signature/', + help='Output directory for signature') + + args = parser.parse_args() + + if args.command is None: + parser.print_help() + sys.exit(1) + + if args.command == 'extract': + remover = WatermarkRemover() + remover.extract_signature_from_images( + black_dir=args.black, + white_dir=args.white, + output_dir=args.output + ) + else: + # Load extractor for verification + extractor = None + codebook = getattr(args, 'codebook', None) + no_verify = getattr(args, 'no_verify', False) + + if codebook and not no_verify: + try: + from robust_extractor import RobustSynthIDExtractor + extractor = RobustSynthIDExtractor() + extractor.load_codebook(codebook) + except Exception as e: + print(f"Warning: Could not load extractor: {e}") + + sig_dir = args.signature + remover = WatermarkRemover(signature_dir=sig_dir, extractor=extractor) + strength = getattr(args, 'strength', 'aggressive') + + if args.command == 'remove': + result = remover.remove_file( + args.input, args.output, + mode=args.mode, verify=not no_verify, + strength=strength + ) + + print("\n" + "=" * 60) + print("WATERMARK REMOVAL RESULTS") + print("=" * 60) + print(f" Mode: {args.mode}") + if args.mode == 'combined_worst': + print(f" Strength: {strength}") + print(f" Method: {result.method}") + print(f" Success: {result.success}") + print(f" PSNR: {result.psnr:.2f} dB") + print(f" SSIM: {result.ssim:.4f}") + + if result.detection_before: + print(f"\n Before:") + print(f" Watermarked: {result.detection_before['is_watermarked']}") + print(f" Phase Match: {result.detection_before['phase_match']:.4f}") + + if result.detection_after: + print(f"\n After:") + print(f" Watermarked: {result.detection_after['is_watermarked']}") + print(f" Phase Match: {result.detection_after['phase_match']:.4f}") + + if result.detection_before: + drop = result.detection_before['phase_match'] - result.detection_after['phase_match'] + pct = 100 * drop / result.detection_before['phase_match'] + print(f"\n Phase Drop: {drop:.4f} ({pct:.1f}%)") + + print("=" * 60) + print(f"Saved to: {args.output}") + + elif args.command == 'batch': + remover.batch_remove( + args.input_dir, args.output_dir, + mode=args.mode, verify=not no_verify, + limit=args.limit, + strength=strength + )
+ 
+