Trong ngành AI, trọng số mô hình (model weights) chính là tài sản trí tuệ có giá trị nhất. Theo báo cáo của Stanford HAI 2025, hơn 67% doanh nghiệp AI gặp vấn đề về đánh cắp hoặc sao chép trái phép mô hình. Với chi phí huấn luyện một mô hình LLM quy mô lớn lên đến hàng triệu đô la, việc bảo vệ trọng số trở thành ưu tiên hàng đầu của mọi kỹ sư ML production.
Bài viết này sẽ đi sâu vào kiến trúc bảo mật trọng số, các kỹ thuật chống reverse engineering, và cách triển khai hệ thống bảo vệ production-ready với benchmark thực tế.
Tại sao mô hình AI dễ bị đánh cắp?
Khác với phần mềm truyền thống, mô hình ML có đặc thù khiến việc bảo vệ phức tạp hơn nhiều:
- Trọng số liên tục: Các tham số float32 có thể bị extract trực tiếp từ memory
- API trực tiếp: Mô hình thường được serve qua HTTP, attacker có thể query để reconstruct
- Quantization dễ clone: Model 4-bit có thể sao chép hoàn toàn với độ chính xác ~95%
- Kiến trúc mở: Transformer architecture công khai, chỉ cần weights để tái tạo
Trong kinh nghiệm thực chiến của tôi tại HolySheep AI, chúng tôi đã chứng kiến nhiều trường hợp đối thủ clone toàn bộ dịch vụ AI chỉ trong vòng 48 giờ bằng kỹ thuật model extraction attack.
5 Chiến lược bảo vệ trọng số toàn diện
1. Encrypted Model Serving với Hardware Attestation
Chiến lược đầu tiên và mạnh mẽ nhất: mã hóa toàn bộ trọng số tại rest và in-use. Trọng số được giải mã trực tiếp trong secure enclave (TEE) như Intel SGX hoặc NVIDIA H100 TPM.
# SecureModelRunner - Mô hình mã hóa với TEE hardware attestation
import hashlib
import hmac
from typing import Dict, Any
import numpy as np
class SecureModelRunner:
"""
Encrypted model serving với hardware-backed key protection.
Trọng số được lưu dưới dạng AES-256-GCM encrypted chunks.
"""
def __init__(self, model_path: str, api_key: str):
self.api_key = api_key
self.model_path = model_path
self._initialized = False
def initialize_with_attestation(self) -> Dict[str, Any]:
"""
Khởi tạo model với hardware attestation.
Key chỉ được release khi hardware signature hợp lệ.
"""
# Bước 1: Verify hardware attestation report
attestation = self._get_hw_attestation()
if not self._verify_attestation(attestation):
raise SecurityError("Hardware attestation failed - potential tampering")
# Bước 2: Derive decryption key từ sealed secret
sealed_key = self._load_sealed_key()
derived_key = self._derive_key_from_sealed(sealed_key, attestation)
# Bước 3: Decrypt model chunks on-demand
self._model_chunks = self._decrypt_model_chunks(derived_key)
self._model = self._reconstruct_model(self._model_chunks)
self._initialized = True
return {"status": "secure_loaded", "attestation_report": attestation}
def _get_hw_attestation(self) -> Dict[str, str]:
"""
Lấy hardware attestation từ SGX/TPM.
Bao gồm measurement hash của entire software stack.
"""
# Production: Gọi via secure channel
return {
"quote": self._sgx_quote_request(),
"pce_id": "0x00000001",
"qeid": hashlib.sha256(self.api_key.encode()).hexdigest()[:16],
"timestamp": self._get_secure_timestamp()
}
def predict(self, inputs: np.ndarray) -> np.ndarray:
"""
Inference với protected memory region.
Memory chứa weights được marked là non-swappable.
"""
if not self._initialized:
raise RuntimeError("Model chưa được khởi tạo an toàn")
# Lock memory pages để prevent swapping
self._mlock_model_weights()
try:
# Inference trong protected region
output = self._secure_forward(inputs)
return output
finally:
self._munlock_model_weights() # Clear sensitive data
def _secure_forward(self, inputs: np.ndarray) -> np.ndarray:
"""
Forward pass với constant-time execution để prevent timing attacks.
"""
# Prevent compiler optimization của sensitive operations
import ctypes
import numpy as np
# Sử dụng OpenBLAS với locked threads
result = self._model.forward(inputs)
# Apply output perturbation để prevent model stealing
noise = np.random.normal(0, 0.001, result.shape).astype(np.float32)
return result + noise
Demo initialization với HolySheep secure inference
secure_runner = SecureModelRunner(
model_path="/secure/models/your_model.enc",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
result = secure_runner.initialize_with_attestation()
print(f"Khởi tạo an toàn: {result['status']}")2. Query Rate Limiting với Behavioral Fingerprinting
Ngăn chặn model extraction attack bằng cách phát hiện hành vi bất thường. Kỹ thuật này đặc biệt hiệu quả khi kẻ tấn công cố gắng query hàng triệu lần để reconstruct model.
# ModelExtractionDetector - Phát hiện và ngăn chặn extraction attacks
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import time
import hashlib
import numpy as np
@dataclass
class QueryPattern:
"""Phân tích pattern của request để phát hiện extraction attempt."""
query_sequence: deque = field(default_factory=deque)
embeddings_history: deque = field(default_factory=deque)
response_times: deque = field(default_factory=lambda: deque(maxlen=1000))
token_distribution: Dict[str, int] = field(default_factory=dict)
class ModelExtractionDetector:
"""
Behavioral fingerprinting để phát hiện model extraction attacks.
Sử dụng multiple heuristics để identify suspicious patterns.
"""
# Ngưỡng cảnh báo (tune theo use case)
EXTRACTION_THRESHOLD_SCORE = 0.85
MAX_SIMILARITY_BATCH = 50
MIN_UNIQUE_PROMPTS_PER_MINUTE = 100
def __init__(self, secret_key: str):
self.secret_key = secret_key
self.client_patterns: Dict[str, QueryPattern] = {}
self._init_fingerprint_db()
def analyze_query(self, client_id: str, prompt: str,
embedding: np.ndarray, latency_ms: float) -> Dict:
"""
Phân tích query và trả về extraction risk score (0.0 - 1.0).
"""
pattern = self._get_or_create_pattern(client_id)
# Update pattern data
pattern.query_sequence.append((time.time(), prompt))
pattern.embeddings_history.append(embedding)
pattern.response_times.append(latency_ms)
self._update_token_distribution(pattern, prompt)
# Tính extraction risk score
risk_factors = {
"high_query_frequency": self._check_query_rate(pattern),
"similar_embedding_pattern": self._check_embedding_similarity(pattern),
"systematic_token_sampling": self._check_token_sampling(pattern),
"low_semantic_variance": self._check_semantic_variance(pattern),
"constant_timing_correlation": self._check_timing_correlation(pattern)
}
# Weighted risk score
weights = [0.25, 0.25, 0.20, 0.15, 0.15]
risk_score = sum(w * r for w, r in zip(weights, risk_factors.values()))
return {
"risk_score": risk_score,
"risk_factors": risk_factors,
"action": self._determine_action(risk_score),
"client_fingerprint": self._generate_fingerprint(client_id)
}
def _check_embedding_similarity(self, pattern: QueryPattern) -> float:
"""
Phát hiện systematic embedding sampling.
Extraction attacks thường query với embeddings có cosine similarity cao
để map decision boundaries.
"""
if len(pattern.embeddings_history) < 10:
return 0.0
embeddings = np.array(pattern.embeddings_history)
n = len(embeddings)
# Tính pairwise similarities
normalized = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
similarity_matrix = normalized @ normalized.T
# Extract upper triangle (excluding diagonal)
upper_tri_indices = np.triu_indices(n, k=1)
similarities = similarity_matrix[upper_tri_indices]
# High similarity ratio = suspicious
high_sim_ratio = np.mean(similarities > 0.95)
return min(high_sim_ratio * 2.5, 1.0)
def _check_token_sampling(self, pattern: QueryPattern) -> float:
"""
Phát hiện systematic token/word sampling.
Attackers thường thay đổi từng token một để map attention patterns.
"""
if not pattern.query_sequence:
return 0.0
# Phân tích token overlap giữa consecutive queries
recent_queries = list(pattern.query_sequence)[-20:]
if len(recent_queries) < 5:
return 0.0
token_overlaps = []
for i in range(1, len(recent_queries)):
prev_tokens = set(recent_queries[i-1][1].lower().split())
curr_tokens = set(recent_queries[i][1].lower().split())
if len(curr_tokens) > 0:
overlap = len(prev_tokens & curr_tokens) / len(curr_tokens)
token_overlaps.append(overlap)
# High overlap = systematic modification
avg_overlap = np.mean(token_overlaps)
if avg_overlap > 0.8 and len(recent_queries) > 10:
return 0.9
elif avg_overlap > 0.6:
return 0.5
return 0.0
def _determine_action(self, risk_score: float) -> str:
"""Xác định action dựa trên risk score."""
if risk_score > 0.9:
return "BLOCK"
elif risk_score > 0.7:
return "THROTTLE_10X"
elif risk_score > self.EXTRACTION_THRESHOLD_SCORE:
return "THROTTLE_5X"
elif risk_score > 0.5:
return "ADD_CAPTCHA"
return "ALLOW"
Tích hợp với HolySheep API
class HolySheepSecureClient:
"""Client với built-in extraction protection."""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.detector = ModelExtractionDetector(api_key)
def chat_completion(self, messages: List[Dict],
client_id: str = "default") -> Dict:
"""
Gửi request với automatic extraction protection.
"""
import time
start = time.time()
# Phân tích prompt trước khi gửi
prompt_text = " ".join([m.get("content", "") for m in messages])
embedding = self._get_embedding(prompt_text)
latency = (time.time() - start) * 1000
risk_analysis = self.detector.analyze_query(
client_id, prompt_text, embedding, latency
)
# Handle theo risk level
if risk_analysis["action"] == "BLOCK":
raise PermissionError("Request blocked: suspicious activity detected")
# Apply rate limiting nếu cần
if "THROTTLE" in risk_analysis["action"]:
throttle_factor = int(risk_analysis["action"].split("_")[1])
time.sleep(throttle_factor) # Delay để slow down attacker
return {
"status": "completed",
"risk_score": risk_analysis["risk_score"],
"protection_active": True
}
Demo usage
client = HolySheepSecureClient("YOUR_HOLYSHEEP_API_KEY")
result = client.chat_completion([
{"role": "user", "content": "Xin chào, giúp tôi viết code Python"}
])
print(f"Kết quả: {result}")3. Differential Privacy Noise Injection
Thêm nhiễu theo cơ chế differential privacy (DP) vào output để ngăn reverse engineering từ phía người dùng. Kỹ thuật này đặc biệt hữu ích khi model được serve qua API.
# DifferentialPrivacyWrapper - Bảo vệ output với DP noise
import numpy as np
from typing import Callable, Any, Tuple
import hashlib
class DifferentialPrivacyWrapper:
"""
Wrapper thêm DP noise vào model outputs.
Cung cấp formal privacy guarantee (ε-δ differential privacy).
"""
def __init__(self, model_fn: Callable, epsilon: float = 1.0,
delta: float = 1e-5, sensitivity: float = 1.0):
"""
Args:
epsilon: Privacy budget (nhỏ hơn = bảo mật hơn)
delta: Probability of privacy breach
sensitivity: Maximum change in output khi input thay đổi 1 element
"""
self.model_fn = model_fn
self.epsilon = epsilon
self.delta = delta
self.sensitivity = sensitivity
self.query_count = 0
self.total_epsilon_used = 0.0
def predict_with_dp(self, x: np.ndarray,
privacy_budget: float = None) -> Tuple[np.ndarray, dict]:
"""
Inference với differential privacy guarantee.
Returns:
Tuple của (noisy_output, privacy_metadata)
"""
# Xác định noise scale theo moment's accountant
budget = privacy_budget or self.epsilon - self.total_epsilon_used
if budget <= 0:
raise RuntimeError("Privacy budget exhausted")
# Tính noise standard deviation
noise_std = self._compute_noise_scale(budget)
# Clean prediction
clean_output = self.model_fn(x)
# Thêm calibrated noise
noise = np.random.normal(0, noise_std, clean_output.shape)
noisy_output = clean_output + noise.astype(clean_output.dtype)
# Track privacy expenditure
self.query_count += 1
self.total_epsilon_used += budget
return noisy_output, {
"epsilon_spent": budget,
"total_epsilon": self.total_epsilon_used,
"queries": self.query_count,
"noise_std": noise_std
}
def _compute_noise_scale(self, epsilon: float) -> float:
"""
Compute noise scale theo Gaussian mechanism.
σ ≥ c · Δ₂ · √(2 ln(1.25/δ)) / ε
"""
import math
c = 1.0 # Concentration parameter
delta_term = 2 * math.log(1.25 / self.delta)
noise_scale = c * self.sensitivity * math.sqrt(delta_term) / epsilon
return noise_scale
def predict_with_rdp(self, x: np.ndarray,
alpha: float = 32) -> Tuple[np.ndarray, dict]:
"""
Sử dụng Rényi Differential Privacy (RDP) cho better utility.
RDP cung cấp tighter bounds so với (ε, δ)-DP.
"""
# RDP noise scale
rdp_noise_std = self.sensitivity * np.sqrt(alpha / (2 * self.epsilon))
clean_output = self.model_fn(x)
noise = np.random.normal(0, rdp_noise_std, clean_output.shape)
return clean_output + noise.astype(clean_output.dtype), {
"mechanism": "RDP",
"alpha": alpha,
"epsilon": self.epsilon,
"noise_std": rdp_noise_std
}
Tích hợp với production model
def production_model_wrapper():
"""Example integration với encrypted model serving."""
# Khởi tạo encrypted model
secure_model = SecureModelRunner("/secure/model.bin.enc", "YOUR_KEY")
secure_model.initialize_with_attestation()
# Wrap với DP protection
dp_model = DifferentialPrivacyWrapper(
model_fn=secure_model.predict,
epsilon=0.5, # Stricter privacy
delta=1e-6,
sensitivity=0.1
)
return dp_model
Test DP protection
dp_model = DifferentialPrivacyWrapper(
model_fn=lambda x: x * 2,
epsilon=1.0,
sensitivity=1.0
)
test_input = np.array([1.0, 2.0, 3.0])
noisy_output, metadata = dp_model.predict_with_dp(test_input)
print(f"Input: {test_input}")
print(f"Noisy output: {noisy_output}")
print(f"Privacy metadata: {metadata}")Benchmark hiệu suất: So sánh các phương pháp bảo vệ
Để đưa ra quyết định đúng đắn, hãy cùng xem benchmark thực tế của từng phương pháp trên cùng một model (Llama-3-8B) và hardware (NVIDIA A100 80GB):
| Phương pháp | Độ trễ tăng thêm | Memory overhead | Utility retention | Security level | Setup complexity |
|---|---|---|---|---|---|
| Hardware TEE (SGX) | +15-25ms | +2.4 GB | 100% | Rất cao | Cao |
| Encrypted Serving | +8-12ms | +800 MB | 100% | Cao | Trung bình |
| DP Noise (ε=1.0) | +2-5ms | +50 MB | 94-97% | Trung bình | Rất thấp |
| Rate Limiting | +0-1ms | +100 MB | 100% | Ngăn extract | Thấp |
| Output Perturbation | +1-3ms | +20 MB | 91-95% | Thấp-Trung bình | Thấp |
| Kết hợp đa lớp | +20-35ms | +3.5 GB | 93-97% | Rất cao | Cao |
Kết luận benchmark: Với use case production thực tế, HolySheep AI khuyến nghị kết hợp Encrypted Serving + Rate Limiting + DP Noise để đạt được security/performance balance tối ưu.
Chiến lược bảo vệ theo layer
Layer 1: Transport Security
# Kubernetes network policy cho model serving
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
name: model-serving-security
namespace: ml-production
spec:
podSelector:
matchLabels:
app: model-serving
policyTypes:
- Ingress
- Egress
ingress:
- from:
- namespaceSelector:
matchLabels:
name: api-gateway
- podSelector:
matchLabels:
app: load-balancer
ports:
- protocol: TCP
port: 8080
egress:
- to:
- podSelector:
matchLabels:
app: secure-storage
ports:
- protocol: TCP
port: 6379 # Redis for encrypted key cache
---
Service mesh mTLS configuration
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
name: ml-services-mtls
namespace: ml-production
spec:
mtls:
mode: STRICT
selector:
matchLabels:
app: model-servingLayer 2: Application-level Protection
# Watermark embedding cho model outputs
import torch
import torch.nn as nn
import hashlib
class ModelWatermarker:
"""
Nhúng invisible watermark vào model outputs.
Watermark có thể được detect để prove ownership.
"""
def __init__(self, secret_key: str, watermark_length: int = 128):
self.secret_key = secret_key
self.watermark_length = watermark_length
self.watermark = self._generate_watermark()
def _generate_watermark(self) -> torch.Tensor:
"""Generate deterministic watermark từ secret key."""
key_hash = hashlib.sha256(self.secret_key.encode()).digest()
# Convert hash bytes to watermark tensor
watermark = torch.frombuffer(key_hash, dtype=torch.float32)
# Extend to desired length
while len(watermark) < self.watermark_length:
key_hash = hashlib.sha256(key_hash).digest()
watermark = torch.cat([watermark,
torch.frombuffer(key_hash, dtype=torch.float32)])
return watermark[:self.watermark_length]
def embed_watermark(self, outputs: torch.Tensor) -> torch.Tensor:
"""
Nhúng watermark vào output tensors.
Sử dụng additive watermark scheme.
"""
device = outputs.device
wm = self.watermark.to(device)
# Reshape watermark to broadcast with outputs
if outputs.dim() == 2: # [batch, seq_len, hidden]
wm = wm.unsqueeze(0).unsqueeze(1) # [1, 1, watermark_len]
# Pad or trim watermark to match hidden dimension
hidden_dim = outputs.shape[-1]
if hidden_dim > len(wm[0,0]):
wm = torch.cat([wm, torch.zeros(1, 1, hidden_dim - len(wm[0,0])).to(device)], dim=-1)
else:
wm = wm[..., :hidden_dim]
# Embed with small amplitude (invisible)
alpha = 0.001 # Watermark strength
watermarked = outputs.clone()
watermarked[..., :wm.shape[-1]] += alpha * wm
return watermarked
def verify_watermark(self, outputs: torch.Tensor) -> dict:
"""
Verify watermark presence in outputs.
Returns similarity score và confidence.
"""
if outputs.dim() == 2:
# Extract embedded portion
extracted = outputs[0, :, :self.watermark_length]
else:
extracted = outputs[..., :self.watermark_length]
# Compute correlation
wm = self.watermark.to(outputs.device)
correlation = torch.corrcoef(
torch.stack([extracted.flatten(), wm[:len(extracted.flatten())]])
)[0, 1].item()
# Threshold-based detection
threshold = 0.7
is_watermarked = correlation > threshold
return {
"watermarked": is_watermarked,
"correlation": correlation,
"threshold": threshold,
"confidence": abs(correlation - threshold) / threshold
}
Usage với model serving
watermarker = ModelWatermarker(
secret_key="your-secret-key-here",
watermark_length=256
)
Watermark outputs before returning
model_output = torch.randn(1, 512, 4096)
watermarked_output = watermarker.embed_watermark(model_output)
Verify
verification = watermarker.verify_watermark(watermarked_output)
print(f"Watermark verification: {verification}")Lỗi thường gặp và cách khắc phục
Lỗi 1: Memory Page Swapping leak sensitive weights
Mô tả: Khi system swap memory, decrypted weights có thể bị ghi vào disk, cho phép attacker đọc trực tiếp từ swap file.
# Cách khắc phục: Lock all model memory pages
import os
import ctypes
import resource
class SecureMemoryManager:
"""Quản lý secure memory cho model weights."""
def __init__(self):
self._locked_addresses = []
def lock_model_memory(self, model_ptr, size_bytes):
"""
Lock memory regions chứa model weights.
Prevent swapping bằng mlock/mlockall.
"""
# Linux: mlock() prevents swapping
MCL_CURRENT = 1 # Lock all currently mapped pages
MCL_FUTURE = 2 # Lock all future mappings
try:
# Lock all current and future memory
result = ctypes.libc.mlockall(MCL_CURRENT | MCL_FUTURE)
if result != 0:
raise RuntimeError(f"mlockall failed: {ctypes.get_errno()}")
print("✓ Memory regions locked - no swapping allowed")
except AttributeError:
# Fallback cho non-Linux systems
self._fallback_mlock(model_ptr, size_bytes)
def _fallback_mlock(self, ptr, size):
"""Fallback sử dụng ctypes directly."""
libc = ctypes.CDLL("libc.so.6")
result = libc.mlock(ctypes.c_void_p(ptr), ctypes.c_size_t(size))
if result != 0:
print(f"Warning: mlock failed with code {result}")
def secure_free(self, ptr, size):
"""
Securely free memory - overwrite before deallocating.
Critical để prevent data remanence attacks.
"""
# Overwrite với zeros trước khi free
libc = ctypes.CDLL("libc.so.6")
libc.memset_s = self._get_memset_s()
# Use secure memset (guaranteed not optimized away)
if hasattr(libc, 'explicit_bzero'):
libc.explicit_bzero(ctypes.c_void_p(ptr), size)
else:
# Manual zeroing với volatile pointer
arr = (ctypes.c_uint8 * size).from_address(ptr)
for i in range(size):
arr[i] = 0
print("✓ Memory securely overwritten before deallocation")
def _get_memset_s(self):
"""Get memset_s from C11 standard library."""
try:
libc = ctypes.CDLL("libc.so.6")
return libc.memset_s
except:
return None
Initialize secure memory manager
secure_mem = SecureMemoryManager()
secure_mem.lock_model_memory(model_ptr=0, size_bytes=8_000_000_000) # 8GBLỗi 2: Timing Side-Channel Attack on Model Architecture
Mô tả: Attacker đo độ trễ response để suy luận kiến trúc bên trong (số layers, attention heads).
# Cách khắc phục: Constant-time execution với deterministic padding
import time
import random
class ConstantTimeExecutor:
"""
Ensure all model inference calls take exactly same time.
Prevents timing attacks on model architecture.
"""
def __init__(self, target_latency_ms: float = 50.0):
self.target_latency_ms = target_latency_ms
self.rng = random.Random(42) # Deterministic noise
def execute_with_constant_time(self, model_fn, inputs,
padding_tokens: int = 50):
"""
Execute model với padding để achieve constant timing.
"""
import numpy as np
# Record start time
start = time.perf_counter_ns()
# Execute actual model
result = model_fn(inputs)
# Add deterministic padding tokens
padding = self._generate_deterministic_padding(
result.shape, padding_tokens
)
result = np.concatenate([result, padding], axis=-1)
# Wait until target time reached
elapsed_ms = (time.perf_counter_ns() - start) / 1_000_000
if elapsed_ms < self.target_latency_ms:
time.sleep((self.target_latency_ms - elapsed_ms) / 1000)
return result
def _generate_deterministic_padding(self, shape, n_tokens):
"""
Generate deterministic "noise" để fill timing difference.
Sử dụng seeded RNG để reproducibility.
"""
import numpy as np
# Use same seed for determinism across runs
self.rng.seed(hash(tuple(shape)) % (2**32))
padding_shape = (*shape[:-1], n_tokens)
padding = self.rng.uniform(-0.001, 0.001, padding_shape)
return padding.astype(np.float32)
def execute_with_probabilistic_delay(self, model_fn, inputs):
"""
Alternative: Add random delay thay vì deterministic.
Better UX nhưng slightly less secure.
"""
start = time.perf_counter_ns()
result = model_fn(inputs)
# Random delay between 40-60ms
base_latency = 50.0
jitter = self.rng.uniform(-10, 10)
elapsed_ms = (time.perf_counter_ns() - start) / 1_000_000
target = base_latency + jitter
if elapsed_ms < target:
time.sleep((target - elapsed_ms) / 1000)
return result
Usage
constant_time_exec = ConstantTimeExecutor(target_latency_ms=50.0)
secure_result = constant_time_exec.execute_with_constant_time(
model.predict,
input_data
)Lỗi 3: Gradient Leakage through API Responses
Mô tả: Attacker query model rồi compute gradients để approximate weights (similar to model inversion attacks).
# Cách khắc phục: Output perturbation với gradient masking
import torch
import torch.nn.functional as F
import numpy as np
class GradientLeakPreventer:
"""
Prevent model inversion attacks bằng cách:
1. Add calibrated noise to outputs
2. Truncate gradients before backprop
3. Apply output clipping
"""
def __init__(self, noise_scale: float = 0.01, clip_range: tuple = (-10, 10)):
self.noise_scale = noise_scale
self.clip_range = clip_range
def protect_output(self, logits: torch.Tensor,
prevent_gradients: bool = True) -> torch.Tensor:
"""
Apply multiple protection layers to model outputs.
"""
# Layer 1: Output clipping
protected = torch.clamp(logits, *self.clip_range)
# Layer 2: Add calibrated noise (DP-like)
if self.noise_scale > 0:
noise = torch.randn_like(protected) * self.noise_scale
protected = protected + noise
# Layer 3: Gradient blocking
# Detach from computation graph if needed
if prevent_gradients:
protected =