Privacy Attacks & Differential Privacy for LLMs

# Measuring memorization: perplexity-based extraction detection
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
import numpy as np

def measure_memorization(
 model_name: str,
 candidate_texts: list[str],
 device: str = "cuda",
) -> list[dict]:
 """Score candidate texts by likelihood under the model.
 High-likelihood (low-perplexity) texts are more likely memorized."""
 tokenizer = AutoTokenizer.from_pretrained(model_name)
 model = AutoModelForCausalLM.from_pretrained(model_name).to(device)
 model.eval()

 results = []
 for text in candidate_texts:
 inputs = tokenizer(text, return_tensors="pt").to(device)
 with torch.no_grad():
 outputs = model(**inputs, labels=inputs["input_ids"])
 loss = outputs.loss.item()
 perplexity = np.exp(loss)

 results.append({
 "text_prefix": text[:80] + "...",
 "perplexity": round(perplexity, 2),
 "loss": round(loss, 4),
 "num_tokens": inputs["input_ids"].shape[1],
 })

 # Sort by perplexity (lowest = most likely memorized)
 results.sort(key=lambda x: x["perplexity"])
 return results

# Example: check if specific sequences are memorized
candidates = [
 "The quick brown fox jumps over the lazy dog.",
 # In a real attack, these would be suspected training data
 "def fibonacci(n):\n if n <= 1:\n return n",
]
# results = measure_memorization("gpt2", candidates)

# End-to-end privacy pipeline: scrub, train with DP, filter outputs
from dataclasses import dataclass
from typing import Optional

@dataclass
class PrivacyConfig:
 # Preprocessing
 scrub_pii: bool = True
 deduplicate: bool = True
 min_token_count: int = 3 # Remove near-unique sequences

 # Training
 use_dp_sgd: bool = True
 target_epsilon: float = 8.0
 target_delta: float = 1e-5
 max_grad_norm: float = 1.0

 # Output filtering
 scan_outputs_for_pii: bool = True
 block_verbatim_training_matches: bool = True
 verbatim_threshold: int = 50 # Char length for match detection

def privacy_aware_pipeline(config: PrivacyConfig, raw_data: list[str]):
 """Orchestrate a privacy-preserving fine-tuning pipeline."""
 # Stage 1: Data preprocessing
 data = raw_data
 if config.scrub_pii:
 data = [scrub_pii(text) for text in data]
 print(f"PII scrubbing: processed {len(data)} samples")

 if config.deduplicate:
 data = list(set(data)) # Simplified; use MinHash in production
 print(f"After dedup: {len(data)} samples")

 # Stage 2: DP fine-tuning (pseudocode)
 if config.use_dp_sgd:
 print(f"Training with DP-SGD: epsilon={config.target_epsilon}")
 # ... Opacus training loop from Section 3 ...
 else:
 print("Training WITHOUT differential privacy (not recommended)")

 # Stage 3: Output filter setup
 if config.scan_outputs_for_pii:
 print("Output PII scanner: ENABLED")
 if config.block_verbatim_training_matches:
 print(f"Verbatim match blocking: threshold={config.verbatim_threshold}")

 return data # Return processed dataset

config = PrivacyConfig(target_epsilon=8.0)
# processed = privacy_aware_pipeline(config, raw_training_data)

# Membership Inference Attack: loss-threshold method
import torch
import numpy as np
from transformers import AutoModelForCausalLM, AutoTokenizer
from sklearn.metrics import roc_auc_score, precision_recall_curve

def compute_per_sample_loss(
 model, tokenizer, texts: list[str], device: str = "cuda",
) -> np.ndarray:
 """Compute per-sample cross-entropy loss."""
 model.eval()
 losses = []
 for text in texts:
 inputs = tokenizer(
 text, return_tensors="pt", truncation=True, max_length=512,
 ).to(device)
 with torch.no_grad():
 outputs = model(**inputs, labels=inputs["input_ids"])
 losses.append(outputs.loss.item())
 return np.array(losses)

def membership_inference_attack(
 model_name: str,
 member_texts: list[str], # Known training data samples
 nonmember_texts: list[str], # Known non-training data samples
 device: str = "cuda",
) -> dict:
 """Run a loss-threshold membership inference attack."""
 tokenizer = AutoTokenizer.from_pretrained(model_name)
 model = AutoModelForCausalLM.from_pretrained(model_name).to(device)

 member_losses = compute_per_sample_loss(
 model, tokenizer, member_texts, device
 )
 nonmember_losses = compute_per_sample_loss(
 model, tokenizer, nonmember_texts, device
 )

 # Lower loss = more likely a member
 # Negate losses so higher score = more likely member
 scores = np.concatenate([-member_losses, -nonmember_losses])
 labels = np.concatenate([
 np.ones(len(member_losses)),
 np.zeros(len(nonmember_losses)),
 ])

 auc = roc_auc_score(labels, scores)
 precision, recall, _ = precision_recall_curve(labels, scores)

 return {
 "auc_roc": round(auc, 4),
 "member_mean_loss": round(float(member_losses.mean()), 4),
 "nonmember_mean_loss": round(float(nonmember_losses.mean()), 4),
 "loss_gap": round(
 float(nonmember_losses.mean() - member_losses.mean()), 4
 ),
 }

# DP-SGD fine-tuning with Opacus
import torch
from torch.utils.data import DataLoader
from transformers import AutoModelForSequenceClassification, AutoTokenizer
from opacus import PrivacyEngine
from opacus.validators import ModuleValidator

model_name = "distilbert-base-uncased"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(
 model_name, num_labels=2,
)

# Opacus requires models to pass validation
# (e.g., BatchNorm must be replaced with GroupNorm)
model = ModuleValidator.fix(model)
errors = ModuleValidator.validate(model, strict=False)
assert len(errors) == 0, f"Model validation failed: {errors}"

optimizer = torch.optim.AdamW(model.parameters(), lr=5e-5)

# Privacy parameters
MAX_GRAD_NORM = 1.0 # Per-sample gradient clipping bound
EPSILON = 8.0 # Target privacy budget
DELTA = 1e-5 # Should be < 1/N (training set size)
EPOCHS = 3
BATCH_SIZE = 32

# Wrap model, optimizer, and dataloader with PrivacyEngine
privacy_engine = PrivacyEngine()
model, optimizer, train_loader = privacy_engine.make_private_with_epsilon(
 module=model,
 optimizer=optimizer,
 data_loader=train_loader, # Your DataLoader
 epochs=EPOCHS,
 target_epsilon=EPSILON,
 target_delta=DELTA,
 max_grad_norm=MAX_GRAD_NORM,
)

print(f"Using sigma={optimizer.noise_multiplier:.4f} "
 f"for ({EPSILON}, {DELTA})-DP over {EPOCHS} epochs")

# Training loop (standard PyTorch, Opacus handles DP transparently)
for epoch in range(EPOCHS):
 model.train()
 total_loss = 0
 for batch in train_loader:
 optimizer.zero_grad()
 outputs = model(**batch)
 loss = outputs.loss
 loss.backward()
 optimizer.step()
 total_loss += loss.item()

 # Check actual privacy spent so far
 eps = privacy_engine.get_epsilon(delta=DELTA)
 print(f"Epoch {epoch+1}: loss={total_loss/len(train_loader):.4f}, "
 f"epsilon_spent={eps:.2f}")

# PII detection and scrubbing pipeline
import re
from dataclasses import dataclass

@dataclass
class PIIMatch:
 category: str
 text: str
 start: int
 end: int
 confidence: float

PII_PATTERNS = {
 "email": (
 r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", 0.95
 ),
 "phone_us": (
 r"(?:\+1[-.])?\(?\d{3}\)?[-.]?\d{3}[-.]?\d{4}", 0.85
 ),
 "ssn": (
 r"\b\d{3}-\d{2}-\d{4}\b", 0.95
 ),
 "credit_card": (
 r"\b(?:\d{4}[-\s]?){3}\d{4}\b", 0.90
 ),
 "ip_address": (
 r"\b(?:\d{1,3}\.){3}\d{1,3}\b", 0.70
 ),
}

def detect_pii(text: str) -> list[PIIMatch]:
 """Detect PII using pattern matching."""
 matches = []
 for category, (pattern, confidence) in PII_PATTERNS.items():
 for match in re.finditer(pattern, text):
 matches.append(PIIMatch(
 category=category,
 text=match.group(),
 start=match.start(),
 end=match.end(),
 confidence=confidence,
 ))
 return matches

def scrub_pii(text: str, replacement: str = "[REDACTED]") -> str:
 """Replace detected PII with redaction markers."""
 matches = detect_pii(text)
 # Process matches in reverse order to preserve positions
 for match in sorted(matches, key=lambda m: m.start, reverse=True):
 tag = f"[{match.category.upper()}_REDACTED]"
 text = text[:match.start] + tag + text[match.end:]
 return text

# Example
sample = "Contact John at john.doe@example.com or 555-123-4567"
print(scrub_pii(sample))
# "Contact John at [EMAIL_REDACTED] or [PHONE_US_REDACTED]"