Beveiliging van methoden voor attributie van trainingsdata

"""
Berekening en beveiligingsanalyse van influence functions.
Demonstreert de berekening en beperkingen van influence functions
voor attributie van trainingsdata.
"""
import numpy as np
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class InfluenceResult:
    """Resultaat van de berekening van een influence function."""
    test_index: int
    training_index: int
    influence_score: float
    is_helpful: bool  # Positieve invloed = behulpzaam
    computation_method: str
    confidence: float
 
def compute_influence_approximation(
    training_gradients: np.ndarray,
    test_gradient: np.ndarray,
    hessian_inverse_approx: np.ndarray,
) -> np.ndarray:
    """
    Bereken de benadering van de influence function.
 
    De invloed van trainingsvoorbeeld z_i op het testverlies bij z_test is:
        I(z_i, z_test) = -grad_test^T @ H^{-1} @ grad_i
 
    waarbij H de Hessiaan van het trainingsverlies is en grad_i, grad_test
    de gradiënten per voorbeeld zijn.
 
    Args:
        training_gradients: Vorm (num_train, param_dim).
        test_gradient: Vorm (param_dim,).
        hessian_inverse_approx: Vorm (param_dim, param_dim).
            Benadering van de inverse Hessiaan.
 
    Returns:
        Invloedsscores voor elk trainingsvoorbeeld.
    """
    # ihvp = H^{-1} @ grad_test (inverse Hessiaan-vectorproduct)
    ihvp = hessian_inverse_approx @ test_gradient
 
    # Invloed = -grad_train_i^T @ ihvp voor elk trainingsvoorbeeld
    influences = -training_gradients @ ihvp
 
    return influences
 
def stochastic_hessian_inverse(
    training_gradients: np.ndarray,
    damping: float = 0.01,
    num_iterations: int = 100,
    scale: float = 1.0,
) -> np.ndarray:
    """
    Bereken een benadering van de inverse Hessiaan met behulp van
    de Neumann-reeksbenadering.
 
    H^{-1} ≈ (1/damping) * sum_{j=0}^{J} (I - H/damping)^j
 
    Dit is goedkoper dan directe inversie, maar introduceert
    benaderingsfouten die de nauwkeurigheid van de attributie beïnvloeden.
    """
    n, d = training_gradients.shape
 
    # Benader H met het uitwendige product van de gradiënten
    # H ≈ (1/n) * sum_i grad_i @ grad_i^T + damping * I
    H_approx = (training_gradients.T @ training_gradients) / n
    H_approx += damping * np.eye(d)
 
    # Directe inversie voor kleine dimensies (demonstratie)
    # Gebruik in de praktijk iteratieve methoden voor grote modellen
    try:
        H_inv = np.linalg.inv(H_approx) * scale
    except np.linalg.LinAlgError:
        H_inv = np.linalg.pinv(H_approx) * scale
 
    return H_inv
 
# Demonstratie
np.random.seed(42)
num_train, param_dim = 100, 32
 
# Simuleer gradiënten per voorbeeld
train_grads = np.random.randn(num_train, param_dim) * 0.01
test_grad = np.random.randn(param_dim) * 0.01
 
# Bereken invloedsscores
H_inv = stochastic_hessian_inverse(train_grads, damping=0.01)
influences = compute_influence_approximation(train_grads, test_grad, H_inv)
 
# Rapporteer de meest invloedrijke trainingsvoorbeelden
top_helpful = np.argsort(influences)[:5]  # Meest negatief = meest behulpzaam
top_harmful = np.argsort(influences)[-5:][::-1]  # Meest positief = meest schadelijk
 
print("Most helpful training examples (reduce test loss):")
for idx in top_helpful:
    print(f"  Training example {idx}: influence = {influences[idx]:.6f}")
 
print("\nMost harmful training examples (increase test loss):")
for idx in top_harmful:
    print(f"  Training example {idx}: influence = {influences[idx]:.6f}")

"""
Het uitbuiten van de beperkingen van influence functions.
Demonstreert hoe een aanvaller trainingsvoorbeelden kan samenstellen die
detectie op basis van influence functions ontwijken.
"""
import numpy as np
 
def craft_influence_evading_poison(
    clean_gradients: np.ndarray,
    target_gradient: np.ndarray,
    hessian_inverse: np.ndarray,
    detection_threshold: float = 0.1,
    max_attempts: int = 100,
    seed: int = 42,
) -> Optional[np.ndarray]:
    """
    Stel een vergiftigd trainingsvoorbeeld samen waarvan de gradiënt
    detectie via influence functions ontwijkt, terwijl het nog steeds het gewenste
    effect op het modelgedrag heeft.
 
    Strategie: ontleed de vergiftigde gradiënt in een component
    die door influence functions wordt gedetecteerd en een component die
    dat niet wordt (in de nulruimte van H^{-1} @ test_gradient ligt).
 
    Args:
        clean_gradients: Gradiënten van schone trainingsvoorbeelden.
        target_gradient: De gradiëntrichting die we willen injecteren.
        hessian_inverse: Benaderde inverse Hessiaan.
        detection_threshold: Maximale invloedsscore om detectie te vermijden.
 
    Returns:
        Een vergiftigde gradiënt die invloedsdetectie ontwijkt, of None
        als ontwijking niet mogelijk is binnen de beperkingen.
    """
    rng = np.random.default_rng(seed)
 
    # De detectierichting: H^{-1} @ test_grad
    # Invloed = -poison_grad^T @ H^{-1} @ test_grad
    # Om te ontwijken: maak poison_grad orthogonaal aan H^{-1} @ test_grad
    # terwijl het uitgelijnd blijft met target_gradient
 
    test_grad_mean = clean_gradients.mean(axis=0)
    detection_direction = hessian_inverse @ test_grad_mean
    detection_direction /= np.linalg.norm(detection_direction) + 1e-10
 
    # Projecteer de doelgradiënt om de detecteerbare component te verwijderen
    target_norm = target_gradient / (np.linalg.norm(target_gradient) + 1e-10)
    detectable_component = np.dot(target_norm, detection_direction) * detection_direction
    evasive_gradient = target_norm - detectable_component
 
    # Verifieer de ontwijking
    if np.linalg.norm(evasive_gradient) < 1e-10:
        return None  # Kan niet ontwijken (doel is uitgelijnd met detectierichting)
 
    evasive_gradient /= np.linalg.norm(evasive_gradient)
 
    # Schaal om een vergelijkbare magnitude te hebben als de schone gradiënten
    clean_norm = np.mean([np.linalg.norm(g) for g in clean_gradients])
    evasive_gradient *= clean_norm
 
    # Verifieer dat de invloedsscore onder de drempel ligt
    influence = -evasive_gradient @ hessian_inverse @ test_grad_mean
    if abs(influence) < detection_threshold:
        return evasive_gradient
 
    return None
 
# Demonstratie
np.random.seed(42)
n, d = 100, 32
clean_grads = np.random.randn(n, d) * 0.01
target_grad = np.random.randn(d) * 0.01  # Wat we willen injecteren
 
H_inv = stochastic_hessian_inverse(clean_grads, damping=0.01)
 
evasive = craft_influence_evading_poison(
    clean_grads, target_grad, H_inv, detection_threshold=0.001
)
 
if evasive is not None:
    # Controleer de invloedsscore
    test_grad = clean_grads.mean(axis=0)
    influence = -evasive @ H_inv @ test_grad
    alignment = np.dot(
        evasive / np.linalg.norm(evasive),
        target_grad / np.linalg.norm(target_grad)
    )
    print(f"Evasive gradient crafted:")
    print(f"  Influence score: {influence:.6f} (below threshold)")
    print(f"  Alignment with target: {alignment:.3f}")
    print(f"  Norm: {np.linalg.norm(evasive):.4f} (matches clean: {np.mean([np.linalg.norm(g) for g in clean_grads]):.4f})")
else:
    print("Could not craft an evasive gradient (target aligned with detection)")

"""
Implementatie van een membership-inference-aanval.
Test of specifieke voorbeelden tijdens de modeltraining zijn gebruikt
op basis van verschillen in modelgedrag.
"""
import numpy as np
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class MembershipInferenceResult:
    """Resultaat van een membership-inference-aanval."""
    example_id: str
    predicted_member: bool
    confidence: float
    loss_value: float
    loss_threshold: float
    method: str
 
def loss_based_membership_inference(
    target_losses: np.ndarray,
    reference_losses: np.ndarray,
    threshold_percentile: float = 50.0,
) -> tuple[np.ndarray, float]:
    """
    Eenvoudige op verlies gebaseerde membership inference.
 
    Het kerninzicht: trainingsvoorbeelden hebben doorgaans een lager verlies
    dan niet-trainingsvoorbeelden, omdat het model is geoptimaliseerd
    om het verlies op de trainingsset te verminderen.
 
    Args:
        target_losses: Verlies per voorbeeld voor te classificeren voorbeelden.
        reference_losses: Verlies voor bekende niet-lid-voorbeelden.
        threshold_percentile: Percentiel van het referentieverlies dat
            wordt gebruikt als beslissingsdrempel.
 
    Returns:
        (predictions, threshold) — booleaanse voorspellingen en de gebruikte drempel.
    """
    threshold = np.percentile(reference_losses, threshold_percentile)
    predictions = target_losses < threshold  # Laag verlies = waarschijnlijk lid
    return predictions, float(threshold)
 
def calibrated_membership_inference(
    target_losses: np.ndarray,
    shadow_member_losses: np.ndarray,
    shadow_nonmember_losses: np.ndarray,
) -> np.ndarray:
    """
    Gekalibreerde membership inference met behulp van shadow-model-statistieken.
 
    Gebruikt de verliesdistributie van een shadow-model (getraind
    op vergelijkbare data) om de lidmaatschapsbeslissing te kalibreren.
    Leden en niet-leden hebben verschillende verliesdistributies,
    en het shadow-model helpt deze distributies te schatten.
 
    Gebaseerd op de methodologie van Carlini et al. 2022,
    "Membership Inference Attacks From First Principles."
    """
    # Pas Gaussische distributies aan voor verlies van leden en niet-leden
    member_mean = np.mean(shadow_member_losses)
    member_std = np.std(shadow_member_losses) + 1e-10
    nonmember_mean = np.mean(shadow_nonmember_losses)
    nonmember_std = np.std(shadow_nonmember_losses) + 1e-10
 
    # Bereken de likelihood-ratio voor elk doelvoorbeeld
    member_likelihood = np.exp(
        -0.5 * ((target_losses - member_mean) / member_std) ** 2
    ) / member_std
    nonmember_likelihood = np.exp(
        -0.5 * ((target_losses - nonmember_mean) / nonmember_std) ** 2
    ) / nonmember_std
 
    # Lidmaatschapskans via de regel van Bayes (uitgaande van een 50/50-prior)
    membership_prob = member_likelihood / (
        member_likelihood + nonmember_likelihood + 1e-10
    )
 
    return membership_prob
 
def evaluate_membership_inference(
    predictions: np.ndarray,
    ground_truth: np.ndarray,
) -> dict:
    """Evalueer de nauwkeurigheid van een membership-inference-aanval."""
    tp = np.sum(predictions & ground_truth)
    fp = np.sum(predictions & ~ground_truth)
    fn = np.sum(~predictions & ground_truth)
    tn = np.sum(~predictions & ~ground_truth)
 
    precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0
    recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0
    accuracy = (tp + tn) / len(predictions) if len(predictions) > 0 else 0.0
 
    # True positive rate bij lage false positive rate (beveiligingsrelevante metriek)
    # In het binaire geval is dit slechts een proxy
    fpr = fp / (fp + tn) if (fp + tn) > 0 else 0.0
    tpr = tp / (tp + fn) if (tp + fn) > 0 else 0.0
 
    return {
        "accuracy": float(accuracy),
        "precision": float(precision),
        "recall": float(recall),
        "true_positive_rate": float(tpr),
        "false_positive_rate": float(fpr),
        "advantage": float(tpr - fpr),  # TPR - FPR (willekeurig = 0)
    }
 
# Demonstratie
np.random.seed(42)
 
# Simuleer verlies: leden hebben een lager verlies dan niet-leden
member_losses = np.random.normal(1.5, 0.3, 200)      # Trainingsdata
nonmember_losses = np.random.normal(2.0, 0.4, 200)    # Niet in de training
 
# Shadow-model-verlies voor kalibratie
shadow_member = np.random.normal(1.6, 0.35, 500)
shadow_nonmember = np.random.normal(2.1, 0.45, 500)
 
# Meng leden en niet-leden als doel
target_losses = np.concatenate([member_losses[:100], nonmember_losses[:100]])
ground_truth = np.concatenate([
    np.ones(100, dtype=bool),
    np.zeros(100, dtype=bool),
])
 
# Eenvoudige op verlies gebaseerde aanval
simple_preds, threshold = loss_based_membership_inference(
    target_losses, nonmember_losses[100:], threshold_percentile=50
)
simple_eval = evaluate_membership_inference(simple_preds, ground_truth)
 
# Gekalibreerde aanval
membership_probs = calibrated_membership_inference(
    target_losses, shadow_member, shadow_nonmember
)
calibrated_preds = membership_probs > 0.5
calibrated_eval = evaluate_membership_inference(calibrated_preds, ground_truth)
 
print("Loss-based MI attack:")
print(f"  Accuracy: {simple_eval['accuracy']:.1%}")
print(f"  Advantage: {simple_eval['advantage']:.3f}")
 
print("\nCalibrated MI attack:")
print(f"  Accuracy: {calibrated_eval['accuracy']:.1%}")
print(f"  Advantage: {calibrated_eval['advantage']:.3f}")

"""
Integriteitsverificatie van dataherkomst.
Implementeert een manipulatiebestendige herkomstketen voor trainingsdata.
"""
import hashlib
import json
import time
from dataclasses import dataclass, field
 
@dataclass
class ProvenanceRecord:
    """Een enkel record in een dataherkomstketen."""
    record_id: str
    data_hash: str
    source: str
    transformation: str
    timestamp: float
    previous_hash: str
    actor: str
 
def create_provenance_chain(
    records: list[dict],
) -> list[ProvenanceRecord]:
    """
    Creëer een manipulatiebestendige herkomstketen (vergelijkbaar met een blockchain).
    Elk record bevat de hash van het voorgaande record, waardoor het
    onmogelijk wordt om eerdere records te wijzigen zonder detectie.
    """
    chain = []
    previous_hash = "genesis"
 
    for record_data in records:
        record_content = json.dumps(record_data, sort_keys=True)
        data_hash = hashlib.sha256(record_content.encode()).hexdigest()
 
        chain_input = f"{data_hash}:{previous_hash}:{record_data.get('timestamp', 0)}"
        record_id = hashlib.sha256(chain_input.encode()).hexdigest()[:16]
 
        record = ProvenanceRecord(
            record_id=record_id,
            data_hash=data_hash,
            source=record_data.get("source", "unknown"),
            transformation=record_data.get("transformation", "none"),
            timestamp=record_data.get("timestamp", time.time()),
            previous_hash=previous_hash,
            actor=record_data.get("actor", "system"),
        )
 
        chain.append(record)
        previous_hash = record_id
 
    return chain
 
def verify_provenance_chain(
    chain: list[ProvenanceRecord],
) -> dict:
    """Verifieer de integriteit van een herkomstketen."""
    issues = []
 
    for i, record in enumerate(chain):
        # Verifieer de ketenkoppeling
        if i == 0:
            if record.previous_hash != "genesis":
                issues.append(f"Record {i}: invalid genesis (expected 'genesis')")
        else:
            if record.previous_hash != chain[i - 1].record_id:
                issues.append(
                    f"Record {i}: broken chain link "
                    f"(expected {chain[i - 1].record_id}, got {record.previous_hash})"
                )
 
        # Verifieer de temporele ordening
        if i > 0 and record.timestamp < chain[i - 1].timestamp:
            issues.append(f"Record {i}: timestamp out of order")
 
    return {
        "chain_length": len(chain),
        "is_valid": len(issues) == 0,
        "issues": issues,
    }
 
# Demonstratie
records = [
    {"source": "wikipedia", "transformation": "crawl", "timestamp": 1000.0, "actor": "crawler"},
    {"source": "wikipedia", "transformation": "dedup", "timestamp": 1001.0, "actor": "pipeline"},
    {"source": "wikipedia", "transformation": "filter", "timestamp": 1002.0, "actor": "pipeline"},
    {"source": "wikipedia", "transformation": "tokenize", "timestamp": 1003.0, "actor": "pipeline"},
]
 
chain = create_provenance_chain(records)
valid_result = verify_provenance_chain(chain)
print(f"Valid chain: {valid_result['is_valid']}")
 
# Manipuleer de keten (wijzig een tussenliggend record)
tampered_chain = list(chain)
tampered_chain[1] = ProvenanceRecord(
    record_id="tampered_id",
    data_hash="fake_hash",
    source="malicious_source",
    transformation="poison",
    timestamp=1001.0,
    previous_hash=chain[0].record_id,
    actor="attacker",
)
tampered_result = verify_provenance_chain(tampered_chain)
print(f"Tampered chain: {tampered_result['is_valid']}")
print(f"Issues: {tampered_result['issues']}")