Practical implementation: end-to-end watermarking flow

Consider an image generator (like Stable Diffusion) compliant with the AI Act:

import hashlib
from datetime import datetime
from diffusers import StableDiffusionPipeline

class AIWatermarkGenerator:
    def __init__(self, model_name, watermark_strength=0.15):
        self.model = StableDiffusionPipeline.from_pretrained(model_name)
        self.watermark_strength = watermark_strength
        self.model_id = model_name

    def generate_with_watermark(self, prompt, seed=42):
        """Generate image and apply watermark compliant."""
        # 1. Set seed for reproducibility
        self.model.set_progress_bar_config(disable=True)

        # 2. Generate image
        with torch.cuda.amp.autocast():
            image = self.model(prompt, generator=torch.Generator().manual_seed(seed)).images[0]

        # 3. Create watermark payload
        timestamp = datetime.utcnow().isoformat() + "Z"
        payload = {
            "timestamp": timestamp,
            "model": self.model_id,
            "seed": seed,
            "prompt_hash": hashlib.sha256(prompt.encode()).hexdigest()[:16]
        }

        # 4. Apply watermark (DCT in this example)
        watermarked = self.apply_dct_watermark(image, payload)

        # 5. Return marked image + metadata
        return {
            "image": watermarked,
            "metadata": payload,
            "verifiable": True
        }

    def apply_dct_watermark(self, image, payload):
        """Apply watermark in the DCT domain."""
        import cv2
        img_array = np.array(image)
        # DCT transform
        dct = cv2.dct(np.float32(img_array) / 255.0)
        # Modify mid-range coefficients
        watermark_bits = self._payload_to_bits(payload)
        strength = self.watermark_strength
        for i, bit in enumerate(watermark_bits):
            idx = (i % dct.shape[0], (i // dct.shape[0]) % dct.shape[1])
            dct[idx] += (-1 if bit == 0 else 1) * strength
        # Inverse DCT
        marked = cv2.idct(dct)
        return Image.fromarray((marked * 255).astype(np.uint8))

    def _payload_to_bits(self, payload):
        """Serialize payload to bitstring."""
        import json
        s = json.dumps(payload)
        return ''.join(format(ord(c), '08b') for c in s)
    

This stack: generates image, embeds watermark invisible in the DCT domain, returns marked image + certificate. At verification time, a user/media can extract the watermark and validate authenticity.

Watermark detection and verification

The client-side verification extracts the watermark and compares it against the registered certificate:

class WatermarkVerifier:
    """Extracts and verifies watermark from an image."""

    def extract_watermark(self, marked_image):
        """Extract payload from DCT watermark."""
        import cv2
        img_array = np.array(marked_image)
        dct = cv2.dct(np.float32(img_array) / 255.0)

        # Extract bitstring from coefficients
        bits = []
        payload_size = 256  # bytes
        for i in range(payload_size * 8):
            idx = (i % dct.shape[0], (i // dct.shape[0]) % dct.shape[1])
            bit = 1 if dct[idx] > 0 else 0
            bits.append(bit)

        # Convert bits to string
        payload_str = ''.join(chr(int(''.join(map(str, bits[i:i+8])), 2))
                               for i in range(0, len(bits), 8))
        return json.loads(payload_str)

    def verify_against_certificate(self, extracted_payload, blockchain_certificate):
        """Verify that extracted payload matches certificate."""
        checks = {
            "timestamp_match": extracted_payload["timestamp"] == blockchain_certificate["timestamp"],
            "model_match": extracted_payload["model"] == blockchain_certificate["model_id"],
            "seed_match": extracted_payload["seed"] == blockchain_certificate.get("seed"),
            "signature_valid": self._verify_signature(blockchain_certificate)
        }

        verification_result = all(checks.values())
        return {
            "verified": verification_result,
            "checks": checks,
            "confidence": sum(checks.values()) / len(checks)
        }

    def _verify_signature(self, cert):
        """Verify ECDSA signature of the certificate."""
        # Call to smart contract / centralized EU database
        pass
    

Trade-offs: performance, quality, and robustness

Adding watermarking creates tensions between three dimensions:

• Performance: the stronger the watermark, the more CPU/GPU cycles required (DCT is O(n²), DWT is O(n log n));
• Visual quality: invisible watermarks require care. Too-high strength degrades the image (PSNR < 40dB is perceptible);
• Robustness: resists JPEG compression better if watermark is in mid-frequencies (trade-off with invisibility).

Indicative benchmarks (1000-image dataset, Stable Diffusion 1.5):

The latent-space solution is most promising for EU compliance: low overhead, good robustness, invisible post-hoc.

Open problems and ongoing research

• Adversarial removal: "watermark-stripper" models (GAN-based) can remove the watermark without significant visual quality degradation. Countering them requires adaptive or multi-layer watermarks;
• Collisions and forgery: an attacker could create a fake blockchain certificate. Requires integration with verified identity system (e.g., decentralized DIDs);
• Performance on video and audio: video watermarking (per-frame + audio sync) and audio watermarking (perceptually imperceptible) remain open challenges at scale;
• Global standardization: EU, USA (NIST), China push different standards. An asset generated in the EU might not verify in the USA.

Implementation best practices

1. Use latent-space watermarking if the generative model supports it (lower latency, maximum robustness);
2. Fall back to DCT if latent-space unavailable (good trade-off for static images);
3. Do not republish watermark payload in response (avoid reverse engineering); share only "it is watermarked, verify at [official EU URL]";
4. Register blockchain certificates within 24 hours of generation (EU temporal compliance);
5. Periodic robustness tests with JPEG, blur, crop, rotation (at least monthly on a sample);
6. Audit documentation of removal attempts (maintain logs of failed verifications);
7. Offline backup of certificate signing private key (disaster recovery).

Frequently asked questions (FAQ)

Is the watermark really invisible?

It depends on strength and technique. An average human eye does not perceive DCT watermarks at strength < 0.1 (PSNR > 40dB). However, sophisticated forensic analysis (steganalysis) can detect the presence of a watermark even if it cannot extract the payload. For EU compliance, "invisible" means "imperceptible to human visual inspection", not "computationally undetectable".

Can I generate content without watermark if the user requests it?

No, according to the EU AI Act. Even on explicit request, the system must watermark. Absence of watermark on a synthetic asset is regulatory violation. However, for internal/private archive use, you can document the exception (e.g., "generated for internal R&D, not for distribution").

Which LLM model does this site use?

This site uses on-premise Ollama for local AI processing (RAG, analysis). No proprietary external cloud models are integrated. If you implement watermarking on a generative text LLM, note that open-source models (Llama 3, Mistral 7B) are already on self-hosted instances and do not require API-side watermarking (the cloud provider applies the watermark, if compliant).

How much does it cost to implement watermarking for an existing product?

Adding latent-space watermarking to a Stable Diffusion generator: < 2 days of engineering (module already available in diffusers library). DCT on existing images: 1 day. Blockchain certificate + verification backend: 3-5 days + infrastructure. For a medium-sized company, total budget < €15k.

Does the watermark survive downsampling/upsampling?

Depends on technique. Latent-space: yes, almost always (because it enters the generative process). DCT: only partially (because resampling modifies the frequency grid). If robustness to resizing is critical, use spread-spectrum watermarking with multi-tone modulation (complex but very robust).