Vergiftiging van federated learning (training pipeline)

┌─────────────┐     Model updates      ┌──────────────────┐
│ Participant 1│ ───────────────────── │                  │
│ (honest)     │                       │  Aggregation     │
├─────────────┤     Model updates      │  Server          │
│ Participant 2│ ───────────────────── │                  │
│ (honest)     │                       │  FedAvg /        │
├─────────────┤     POISONED updates   │  Robust Agg      │
│ Participant 3│ ───────────────────── │                  │
│ (MALICIOUS)  │                       │  ──── Global ──▶ │
├─────────────┤     Model updates      │       Model      │
│ Participant N│ ───────────────────── │                  │
│ (honest)     │                       └──────────────────┘
└─────────────┘

def federated_average(participant_updates: list[dict], weights: list[float]) -> dict:
    """FedAvg: gewogen gemiddelde van modelupdates van deelnemers.
    Kwetsbaar voor vergiftiging omdat het alle updates evenveel vertrouwt."""
    global_update = {}
    for key in participant_updates[0]:
        global_update[key] = sum(
            w * update[key] for w, update in zip(weights, participant_updates)
        ) / sum(weights)
    return global_update

def byzantine_poisoning_attack(
    malicious_model,
    target_behavior_data,
    global_model_state,
    num_participants: int,
    scaling_factor: float = 10.0,
):
    """Vervaardig een vergiftigde update die het globale model richting doelgedrag verschuift."""
    # Stap 1: Bereken gradiënt richting doelgedrag
    malicious_model.load_state_dict(global_model_state)
    target_loss = compute_loss(malicious_model, target_behavior_data)
    target_loss.backward()
 
    # Stap 2: Bereken de update-delta
    poisoned_update = {}
    for name, param in malicious_model.named_parameters():
        clean_delta = param.data - global_model_state[name]
        target_delta = -param.grad  # gradient descent richting doel
 
        # Meng schoon gedrag met vergiftigd doel
        poisoned_update[name] = clean_delta + scaling_factor * target_delta
 
    # Stap 3: Schaal om de aggregatie te domineren
    # Met N deelnemers zorgt schalen met N ervoor dat de vergiftigde update
    # zwaarder weegt dan alle eerlijke deelnemers samen
    for name in poisoned_update:
        poisoned_update[name] *= num_participants
 
    return poisoned_update

def model_replacement_attack(
    target_model_state: dict,
    global_model_state: dict,
    num_participants: int,
    learning_rate: float,
):
    """Bereken een update die, na FedAvg, het globale model vervangt
    door het doelmodel van de aanvaller."""
    replacement_update = {}
    for name in target_model_state:
        # Na FedAvg: new_global = global + lr * avg(updates)
        # We willen: new_global = target_model
        # Dus: update = (target - global) * num_participants / lr
        replacement_update[name] = (
            (target_model_state[name] - global_model_state[name])
            * num_participants / learning_rate
        )
    return replacement_update

def constrained_backdoor_update(
    backdoored_model, global_model, norm_bound: float
):
    """Backdoor-update beperkt om norm-gebaseerde detectie te passeren."""
    update = {}
    for name in backdoored_model.state_dict():
        update[name] = backdoored_model.state_dict()[name] - global_model.state_dict()[name]
 
    # Bereken update-norm
    total_norm = torch.sqrt(sum(
        (update[name] ** 2).sum() for name in update
    ))
 
    # Schaal indien nodig naar beneden om binnen de detectiedrempel te blijven
    if total_norm > norm_bound:
        scale = norm_bound / total_norm
        for name in update:
            update[name] *= scale
 
    return update

def krum_aggregation(updates: list[dict], num_byzantine: int) -> dict:
    """Krum: selecteer de update die het dichtst bij de meerderheid ligt.
    Byzantijns-robuust voor < 50% kwaadwillende deelnemers."""
    n = len(updates)
    scores = []
 
    for i in range(n):
        # Bereken afstand tot alle andere updates
        distances = []
        for j in range(n):
            if i != j:
                dist = sum(
                    ((updates[i][k] - updates[j][k]) ** 2).sum()
                    for k in updates[i]
                )
                distances.append(dist.item())
 
        # Score = som van de n - num_byzantine - 2 dichtstbijzijnde afstanden
        distances.sort()
        score = sum(distances[:n - num_byzantine - 2])
        scores.append(score)
 
    # Selecteer de update met de laagste score (dichtst bij de meerderheid)
    best_idx = scores.index(min(scores))
    return updates[best_idx]