Watermark Removal Techniques

import hashlib
import torch
 
def watermarked_sampling(logits, prev_token, watermark_key, bias=2.0):
    """Add bias to 'green list' tokens during sampling.
    The green list is determined by a hash of the previous token
    and the watermark key."""
    # Deterministic green/red split based on previous token + key
    seed = int(hashlib.sha256(
        f"{watermark_key}:{prev_token}".encode()
    ).hexdigest()[:8], 16)
    rng = torch.Generator().manual_seed(seed)
 
    vocab_size = logits.shape[-1]
    perm = torch.randperm(vocab_size, generator=rng)
    green_tokens = perm[:vocab_size // 2]
 
    # Bias logits toward green tokens
    watermarked_logits = logits.clone()
    watermarked_logits[green_tokens] += bias
 
    return watermarked_logits
 
def detect_watermark(text, tokenizer, watermark_key, threshold=0.5):
    """Detect watermark by measuring green token fraction."""
    tokens = tokenizer.encode(text)
    green_count = 0
 
    for i in range(1, len(tokens)):
        seed = int(hashlib.sha256(
            f"{watermark_key}:{tokens[i-1]}".encode()
        ).hexdigest()[:8], 16)
        rng = torch.Generator().manual_seed(seed)
        perm = torch.randperm(tokenizer.vocab_size, generator=rng)
        green_tokens = set(perm[:tokenizer.vocab_size // 2].tolist())
 
        if tokens[i] in green_tokens:
            green_count += 1
 
    green_fraction = green_count / (len(tokens) - 1)
    # Unwatermarked text: ~50% green tokens
    # Watermarked text: significantly > 50% green tokens
    z_score = (green_fraction - 0.5) / (0.5 / (len(tokens) - 1) ** 0.5)
    return z_score > threshold, z_score, green_fraction

def paraphrase_removal(watermarked_text, paraphrase_model, tokenizer):
    """Remove watermark by paraphrasing with a non-watermarked model."""
    prompt = f"""Rewrite the following text to convey the same meaning
using different words and sentence structures. Preserve all factual
content and technical accuracy.
 
Original text:
{watermarked_text}
 
Rewritten text:"""
 
    # The paraphrase model generates new tokens without the watermark bias
    paraphrased = paraphrase_model.generate(
        tokenizer(prompt, return_tensors="pt").input_ids,
        max_new_tokens=len(watermarked_text.split()) * 2,
        temperature=0.7,
    )
    return tokenizer.decode(paraphrased[0], skip_special_tokens=True)

def token_substitution_removal(text, tokenizer, watermark_key, model):
    """Remove watermark by selectively replacing green tokens
    with semantically equivalent red tokens."""
    tokens = tokenizer.encode(text)
    modified_tokens = list(tokens)
 
    for i in range(1, len(tokens)):
        # Determine if current token is green
        green_set = get_green_tokens(tokens[i-1], watermark_key, tokenizer.vocab_size)
 
        if tokens[i] in green_set:
            # Find semantically similar red token
            # Get model's probability distribution at this position
            context = torch.tensor([modified_tokens[:i]])
            logits = model(context).logits[0, -1]
 
            # Find highest-probability red token
            red_set = set(range(tokenizer.vocab_size)) - green_set
            red_probs = [(t, logits[t].item()) for t in red_set]
            red_probs.sort(key=lambda x: x[1], reverse=True)
 
            best_red_token = red_probs[0][0]
            # Only substitute if the red token is reasonably probable
            if logits[best_red_token] > logits[tokens[i]] - 3.0:
                modified_tokens[i] = best_red_token
 
    return tokenizer.decode(modified_tokens)

def embedding_perturbation_removal(text, model, tokenizer, noise_scale=0.1):
    """Remove embedding-level watermarks by adding noise
    to token embeddings and re-decoding."""
    tokens = tokenizer(text, return_tensors="pt")
    embeddings = model.get_input_embeddings()(tokens.input_ids)
 
    # Add calibrated noise to disrupt watermark signal
    noise = torch.randn_like(embeddings) * noise_scale
    perturbed_embeddings = embeddings + noise
 
    # Re-decode from perturbed embeddings
    with torch.no_grad():
        outputs = model(inputs_embeds=perturbed_embeddings)
        new_tokens = outputs.logits.argmax(dim=-1)
 
    return tokenizer.decode(new_tokens[0], skip_special_tokens=True)