Federated Learning Security

Gevorderd18 min lezenBijgewerkt op 2026-03-20

Beveiligingsaanvallen op federated-learning-systemen, waaronder modelvergiftiging, data-inferentie en exploitatie van Byzantijnse fouten.

infrastructure federated-learning model-poisoning privacy

Overzicht

Federated learning (FL) stelt meerdere partijen in staat om gezamenlijk een gedeeld model te trainen zonder ruwe data te centraliseren. In plaats van data naar een centrale server te sturen, traint elke deelnemer een lokaal model op de eigen data en stuurt alleen modelupdates (gradiënten of gewichten) naar een aggregatieserver. Hoewel deze architectuur is ontworpen om privacy te beschermen, introduceert ze een eigen verzameling kwetsbaarheden die traditionele gecentraliseerde training niet kent.

De kernuitdaging op het gebied van beveiliging in federated learning is vertrouwen. De aggregatieserver moet erop vertrouwen dat de client-updates eerlijk zijn, en clients moeten erop vertrouwen dat de server de aggregatie correct uitvoert. Geen van beide aannames houdt stand in adversariële omgevingen. Eén enkele kwaadwillende client kan vergiftigde updates injecteren die het gedrag van het globale model verschuiven, en een nieuwsgierige server kan privé trainingsdata reconstrueren uit waargenomen gradiënten. Dit zijn geen theoretische zorgen — ze zijn herhaaldelijk aangetoond in onderzoek en zijn direct relevant voor elke organisatie die federated learning inzet in de zorg, financiën of organisatie-overstijgende AI-initiatieven.

Dit artikel behandelt de belangrijkste aanvalsklassen tegen federated-learning-systemen, biedt werkende exploitcode en evalueert de effectiviteit van de huidige verdedigingen. De hier beschreven aanvallen sluiten aan op de MITRE ATLAS-technieken voor compromittering van de ML-supplychain (AML.T0010) en modelvergiftiging (AML.T0020).

Architectuur van Federated Learning

Hoe Federated Averaging werkt

Het dominante federated-learning-algoritme is Federated Averaging (FedAvg), geïntroduceerd door McMahan et al. in "Communication-Efficient Learning of Deep Networks from Decentralized Data" (2017). Het protocol verloopt in rondes:

De server stuurt het huidige globale model naar een subset van clients
Elke client traint het model gedurende meerdere epochs op lokale data
Clients sturen bijgewerkte modelparameters (of gradiënten) terug naar de server
De server aggregeert de updates, doorgaans via gewogen middeling
Het proces herhaalt zich tot convergentie

import torch
import torch.nn as nn
from typing import Dict, List, Tuple
import copy
 
class FederatedServer:
    """Simplified federated averaging server."""
 
    def __init__(self, global_model: nn.Module):
        self.global_model = global_model
        self.round_number = 0
 
    def aggregate(
        self, client_updates: List[Dict[str, torch.Tensor]], weights: List[float]
    ) -> None:
        """Aggregate client updates using weighted averaging."""
        total_weight = sum(weights)
        global_state = self.global_model.state_dict()
 
        # Initialiseer geaggregeerde state met nullen
        aggregated = {
            key: torch.zeros_like(param) for key, param in global_state.items()
        }
 
        # Gewogen gemiddelde van alle client-updates
        for update, weight in zip(client_updates, weights):
            for key in aggregated:
                aggregated[key] += (weight / total_weight) * update[key]
 
        self.global_model.load_state_dict(aggregated)
        self.round_number += 1
 
    def distribute_model(self) -> Dict[str, torch.Tensor]:
        """Send current global model to clients."""
        return copy.deepcopy(self.global_model.state_dict())

Kaart van het aanvalsoppervlak

De federated-learning-architectuur stelt aanvalsoppervlakken bloot bij elk component:

Component	Aanvalsoppervlak	Voorbeeldaanvallen
Client-apparaten	Lokale modeltraining	Datavergiftiging, modelvervanging
Communicatiekanaal	Transmissie van gradiënten/gewichten	Gradiëntonderschepping, man-in-the-middle
Aggregatieserver	Combinatielogica van updates	Byzantijnse aanvallen, aggregatiemanipulatie
Globaal model	Gedeelde parameters	Backdoor-injectie, modelinversie
Clientselectie	Deelnameprotocol	Sybil-aanvallen, free-riding

Het fundamentele probleem is dat de server niet kan verifiëren of de update van een client is afgeleid van legitieme training op eerlijke data. Deze verificatiekloof is de grondoorzaak van de meeste federated-learning-aanvallen.

Modelvergiftigingsaanvallen

Ongerichte vergiftiging

Ongerichte vergiftiging heeft als doel de algehele modelprestaties te degraderen. Een kwaadwillende client stuurt updates die het globale model weg van convergentie duwen. De eenvoudigste aanpak is het sturen van willekeurige ruis of geïnverteerde gradiënten.

import torch
import torch.nn as nn
from typing import Dict
 
class UntargetedPoisoner:
    """Untargeted model poisoning — degrades global model accuracy."""
 
    def __init__(self, model: nn.Module, amplification_factor: float = 10.0):
        self.model = model
        self.amplification_factor = amplification_factor
 
    def generate_poisoned_update(
        self, honest_update: Dict[str, torch.Tensor]
    ) -> Dict[str, torch.Tensor]:
        """Generate a poisoned update by negating and amplifying honest gradients."""
        poisoned = {}
        for key, param in honest_update.items():
            # Keer de gradiëntrichting om en versterk
            poisoned[key] = -self.amplification_factor * param
        return poisoned
 
    def generate_noise_update(
        self, reference_update: Dict[str, torch.Tensor]
    ) -> Dict[str, torch.Tensor]:
        """Generate random noise update scaled to match legitimate update norms."""
        poisoned = {}
        for key, param in reference_update.items():
            noise = torch.randn_like(param)
            # Schaal de ruis om overeen te komen met de norm van de legitieme update
            scale = param.norm() / (noise.norm() + 1e-8)
            poisoned[key] = noise * scale
        return poisoned

De amplificatiefactor is cruciaal. Als de kwaadwillende update te groot is, zullen norm-gebaseerde verdedigingen deze clippen of afwijzen. Is hij te klein, dan heeft de aanval een verwaarloosbaar effect wanneer hij gemiddeld wordt met eerlijke updates. Onderzoek van Fang et al. in "Local Model Poisoning Attacks to Byzantine-Robust Federated Learning" (USENIX Security 2020) toonde aan dat adaptieve aanvallers die de aggregatieregel kennen, updates kunnen vervaardigen die detectie ontwijken terwijl ze de schade maximaliseren.

Gerichte backdoor-aanvallen

Gerichte aanvallen zijn geavanceerder en gevaarlijker. In plaats van de algehele prestaties te degraderen, injecteert de aanvaller een backdoor die ervoor zorgt dat het model specifieke inputs verkeerd classificeert, terwijl het de normale nauwkeurigheid op schone data behoudt. Bagdasaryan et al. demonstreerden dit in "How To Back Door Federated Learning" (AISTATS 2020).

De aanval werkt door een lokaal model te trainen op vergiftigde data die een triggerpatroon bevat, en de resulterende update vervolgens op te schalen om middeling met eerlijke clients te overwinnen:

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, Dataset
from typing import Dict, Optional, Tuple
import copy
 
class BackdoorPoisoner:
    """Targeted backdoor attack against federated learning."""
 
    def __init__(
        self,
        model: nn.Module,
        target_label: int,
        trigger_pattern: torch.Tensor,
        trigger_position: Tuple[int, int] = (0, 0),
        scaling_factor: float = 1.0,
    ):
        self.model = model
        self.target_label = target_label
        self.trigger_pattern = trigger_pattern
        self.trigger_position = trigger_position
        self.scaling_factor = scaling_factor
 
    def apply_trigger(self, images: torch.Tensor) -> torch.Tensor:
        """Stamp a trigger pattern onto images."""
        triggered = images.clone()
        h, w = self.trigger_pattern.shape[-2:]
        y, x = self.trigger_position
        triggered[:, :, y : y + h, x : x + w] = self.trigger_pattern
        return triggered
 
    def create_poisoned_batch(
        self, images: torch.Tensor, labels: torch.Tensor, poison_ratio: float = 0.5
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        """Create a mixed batch of clean and poisoned samples."""
        batch_size = images.size(0)
        num_poison = int(batch_size * poison_ratio)
 
        poisoned_images = images.clone()
        poisoned_labels = labels.clone()
 
        # Pas de trigger toe en wijzig labels voor de vergiftigde subset
        poisoned_images[:num_poison] = self.apply_trigger(images[:num_poison])
        poisoned_labels[:num_poison] = self.target_label
 
        return poisoned_images, poisoned_labels
 
    def train_and_scale(
        self,
        global_model_state: Dict[str, torch.Tensor],
        poisoned_loader: DataLoader,
        num_epochs: int = 5,
        lr: float = 0.01,
        num_total_clients: int = 10,
    ) -> Dict[str, torch.Tensor]:
        """Train on poisoned data and scale update to survive aggregation."""
        local_model = copy.deepcopy(self.model)
        local_model.load_state_dict(global_model_state)
 
        optimizer = torch.optim.SGD(local_model.parameters(), lr=lr)
        criterion = nn.CrossEntropyLoss()
 
        local_model.train()
        for epoch in range(num_epochs):
            for images, labels in poisoned_loader:
                poisoned_images, poisoned_labels = self.create_poisoned_batch(
                    images, labels
                )
                optimizer.zero_grad()
                outputs = local_model(poisoned_images)
                loss = criterion(outputs, poisoned_labels)
                loss.backward()
                optimizer.step()
 
        # Schaal de update zodat hij de aggregatie domineert
        # Na het middelen van n clients is de bijdrage van de vergiftigde update 1/n
        # Schalen met n zorgt ervoor dat de backdoor de middeling overleeft
        poisoned_state = local_model.state_dict()
        scaled_update = {}
        for key in global_model_state:
            delta = poisoned_state[key] - global_model_state[key]
            scaled_update[key] = (
                global_model_state[key]
                + delta * num_total_clients * self.scaling_factor
            )
 
        return scaled_update

De schaalfactor van num_total_clients is het kerninzicht: nadat FedAvg deelt door het aantal clients, behoudt de backdoor-update zijn oorspronkelijke magnitude. In de praktijk gebruiken aanvallers mogelijk kleinere schaalfactoren om anomaliedetectie te ontwijken, waarbij ze een langzamer maar onopvallender injectieproces accepteren dat meerdere rondes vergt.

Gradiënt-gebaseerde datareconstructie

Deep Leakage from Gradients

Zelfs zonder vergiftiging lekt het loutere delen van gradiënten al informatie over privé trainingsdata. Zhu et al. demonstreerden in "Deep Leakage from Gradients" (NeurIPS 2019) dat een waarnemer die de gradiëntupdate van een client ziet, de trainingsdata kan reconstrueren die deze heeft voortgebracht.

De aanval werkt door een dummy-input te optimaliseren zodat deze gradiënten produceert die overeenkomen met de waargenomen gradiënten:

import torch
import torch.nn as nn
from typing import List, Tuple, Optional
 
class GradientLeakageAttack:
    """Reconstruct private training data from observed gradients."""
 
    def __init__(self, model: nn.Module, device: str = "cpu"):
        self.model = model
        self.device = device
 
    def attack(
        self,
        observed_gradients: List[torch.Tensor],
        input_shape: Tuple[int, ...],
        num_classes: int,
        num_iterations: int = 300,
        lr: float = 1.0,
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        """
        Reconstruct training data from observed gradients.
 
        Args:
            observed_gradients: List of gradient tensors observed from the client
            input_shape: Shape of the input data (e.g., (3, 32, 32) for CIFAR-10)
            num_classes: Number of output classes
            num_iterations: Optimization iterations
            lr: Learning rate for reconstruction
 
        Returns:
            Tuple of (reconstructed_data, reconstructed_labels)
        """
        # Initialiseer willekeurige dummy-data en labels
        dummy_data = torch.randn(1, *input_shape, device=self.device, requires_grad=True)
        dummy_label = torch.randn(1, num_classes, device=self.device, requires_grad=True)
 
        optimizer = torch.optim.LBFGS([dummy_data, dummy_label], lr=lr)
        criterion = nn.CrossEntropyLoss()
 
        history = []
 
        for iteration in range(num_iterations):
            def closure():
                optimizer.zero_grad()
                self.model.zero_grad()
 
                # Forward pass met dummy-data
                pred = self.model(dummy_data)
                dummy_loss = criterion(pred, torch.softmax(dummy_label, dim=-1))
 
                # Bereken gradiënten van de dummy-data
                dummy_gradients = torch.autograd.grad(
                    dummy_loss, self.model.parameters(), create_graph=True
                )
 
                # Minimaliseer de afstand tussen dummy- en waargenomen gradiënten
                gradient_diff = sum(
                    ((dg - og) ** 2).sum()
                    for dg, og in zip(dummy_gradients, observed_gradients)
                )
 
                gradient_diff.backward()
                return gradient_diff
 
            loss = optimizer.step(closure)
            if iteration % 50 == 0:
                history.append(loss.item())
 
        reconstructed_label = torch.argmax(dummy_label, dim=-1)
        return dummy_data.detach(), reconstructed_label.detach()

Deze aanval is opmerkelijk effectief tegen kleine batchgroottes. Wanneer een client traint op één enkel sample, kan de reconstructie bijna perfect zijn voor beelddata. Naarmate de batchgrootte toeneemt, neemt de reconstructiekwaliteit af, maar gedeeltelijk informatielek blijft bestaan. Geiping et al. breidden dit uit in "Inverting Gradients - How Easy Is It to Break Privacy in Federated Learning?" (NeurIPS 2020) met cosinus-similariteitsverlies en regularisatie die de reconstructiekwaliteit aanzienlijk verbeterden.

Membership inference in federated-omgevingen

Naast volledige datareconstructie kan een tegenstander bepalen of een specifiek datapunt in de trainingsset van een client is gebruikt. Deze membership-inference-aanval wordt uitgevoerd door waar te nemen hoe het verlies van het model op een doelpunt verandert nadat de update van een client is verwerkt:

import torch
import torch.nn as nn
from typing import Dict, List
 
class FederatedMembershipInference:
    """Determine if a specific sample was in a client's training data."""
 
    def __init__(self, model: nn.Module, threshold: float = 0.5):
        self.model = model
        self.threshold = threshold
 
    def compute_loss_change(
        self,
        target_sample: torch.Tensor,
        target_label: torch.Tensor,
        model_before: Dict[str, torch.Tensor],
        model_after: Dict[str, torch.Tensor],
    ) -> float:
        """Measure how much the loss on a target sample changed after a client update."""
        criterion = nn.CrossEntropyLoss()
 
        # Verlies vóór de client-update
        self.model.load_state_dict(model_before)
        self.model.eval()
        with torch.no_grad():
            loss_before = criterion(
                self.model(target_sample.unsqueeze(0)), target_label.unsqueeze(0)
            ).item()
 
        # Verlies na de client-update
        self.model.load_state_dict(model_after)
        self.model.eval()
        with torch.no_grad():
            loss_after = criterion(
                self.model(target_sample.unsqueeze(0)), target_label.unsqueeze(0)
            ).item()
 
        return loss_before - loss_after
 
    def infer_membership(
        self,
        target_samples: List[torch.Tensor],
        target_labels: List[torch.Tensor],
        model_before: Dict[str, torch.Tensor],
        model_after: Dict[str, torch.Tensor],
    ) -> List[bool]:
        """Infer membership for multiple samples."""
        results = []
        for sample, label in zip(target_samples, target_labels):
            loss_change = self.compute_loss_change(
                sample, label, model_before, model_after
            )
            # Een grote positieve verliesverandering wijst erop dat het sample in de trainingsset zat
            results.append(loss_change > self.threshold)
        return results

Sybil- en free-rider-aanvallen

Sybil-aanvallen

Bij een Sybil-aanval creëert één enkele tegenstander meerdere valse client-identiteiten om disproportionele invloed te verkrijgen over de aggregatie. Als de aanvaller k van de n clients beheerst, krijgen zijn vergiftigde updates k/n gewicht in het gemiddelde in plaats van 1/n.

import torch
import torch.nn as nn
from typing import Dict, List
import copy
 
class SybilAttacker:
    """Launch a Sybil attack by controlling multiple federated clients."""
 
    def __init__(
        self,
        num_sybil_clients: int,
        poisoned_update_fn,  # callable die een vergiftigde update genereert
    ):
        self.num_sybil_clients = num_sybil_clients
        self.poisoned_update_fn = poisoned_update_fn
 
    def generate_sybil_updates(
        self,
        global_model_state: Dict[str, torch.Tensor],
        add_noise: bool = True,
        noise_scale: float = 0.01,
    ) -> List[Dict[str, torch.Tensor]]:
        """Generate multiple slightly varied poisoned updates to avoid detection."""
        base_update = self.poisoned_update_fn(global_model_state)
        updates = []
 
        for i in range(self.num_sybil_clients):
            if add_noise and i > 0:
                # Voeg kleine willekeurige verstoringen toe zodat updates niet identiek zijn
                noisy_update = {}
                for key, param in base_update.items():
                    noise = torch.randn_like(param) * noise_scale * param.abs().mean()
                    noisy_update[key] = param + noise
                updates.append(noisy_update)
            else:
                updates.append(copy.deepcopy(base_update))
 
        return updates

Verdedigen tegen Sybil-aanvallen vereist robuuste clientauthenticatie en reputatiesystemen. Proof-of-work, hardware-attestatie en het scoren van bijdragekwaliteit zijn allemaal voorgesteld, maar elk kent aanzienlijke beperkingen in organisatie-overstijgende omgevingen waar deelnemers mogelijk geen gevestigde vertrouwensrelaties hebben.

Free-rider-aanvallen

Free-riders nemen deel aan federated learning om het getrainde globale model te ontvangen zonder zinvolle updates bij te dragen. Ze kunnen nul-updates, willekeurige ruis sturen, of het globale model simpelweg terugkaatsen. Hoewel dit het model niet direct schaadt, verwatert het de kwaliteit van de aggregatie en schendt het de coöperatieve aanname waarvan federated learning afhankelijk is.

Byzantijns-bestendige aggregatie

Robuuste aggregatieregels

Standaard FedAvg is uiterst kwetsbaar voor zelfs één enkele kwaadwillende client. Byzantijns-bestendige aggregatieregels streven ernaar correcte resultaten te produceren, zelfs wanneer een fractie van de clients adversarieel is.

import torch
from typing import Dict, List
 
class RobustAggregator:
    """Byzantine-resilient aggregation methods for federated learning."""
 
    @staticmethod
    def coordinate_wise_median(
        updates: List[Dict[str, torch.Tensor]],
    ) -> Dict[str, torch.Tensor]:
        """Coordinate-wise median aggregation (Yin et al., 2018)."""
        keys = updates[0].keys()
        aggregated = {}
 
        for key in keys:
            stacked = torch.stack([u[key] for u in updates])
            aggregated[key] = torch.median(stacked, dim=0).values
 
        return aggregated
 
    @staticmethod
    def trimmed_mean(
        updates: List[Dict[str, torch.Tensor]], trim_ratio: float = 0.1
    ) -> Dict[str, torch.Tensor]:
        """Trimmed mean — remove extreme values before averaging."""
        keys = updates[0].keys()
        aggregated = {}
        n = len(updates)
        trim_count = int(n * trim_ratio)
 
        for key in keys:
            stacked = torch.stack([u[key] for u in updates])
            sorted_updates, _ = torch.sort(stacked, dim=0)
            # Verwijder de bovenste en onderste trim_count waarden
            trimmed = sorted_updates[trim_count : n - trim_count]
            aggregated[key] = trimmed.mean(dim=0)
 
        return aggregated
 
    @staticmethod
    def krum(
        updates: List[Dict[str, torch.Tensor]], num_byzantine: int
    ) -> Dict[str, torch.Tensor]:
        """
        Multi-Krum aggregation (Blanchard et al., 2017).
        Selects the update closest to its neighbors.
        """
        n = len(updates)
        num_select = n - num_byzantine - 2
 
        # Plat de updates uit voor afstandsberekening
        flat_updates = []
        for update in updates:
            flat = torch.cat([param.flatten() for param in update.values()])
            flat_updates.append(flat)
 
        # Bereken paarsgewijze afstanden
        distances = torch.zeros(n, n)
        for i in range(n):
            for j in range(i + 1, n):
                dist = (flat_updates[i] - flat_updates[j]).norm()
                distances[i][j] = dist
                distances[j][i] = dist
 
        # Voor elke update: som de afstanden tot de dichtstbijzijnde buren
        scores = []
        for i in range(n):
            sorted_dists, _ = torch.sort(distances[i])
            # Som van afstanden tot de num_select dichtstbijzijnde buren (zonder zichzelf)
            score = sorted_dists[1 : num_select + 1].sum()
            scores.append(score)
 
        # Selecteer de update met de laagste score
        best_idx = int(torch.argmin(torch.tensor(scores)))
        return updates[best_idx]

Onderzoek heeft aangetoond dat geen van deze verdedigingen voldoende is tegen adaptieve aanvallers. Fang et al. (USENIX Security 2020) demonstreerden aanvallen die vergiftigde updates vervaardigen die specifiek zijn ontworpen om Krum, trimmed mean en mediaan-aggregatie te ontwijken, door dicht bij de verdeling van eerlijke updates te blijven en tegelijk het model in de door de aanvaller gewenste richting te verschuiven.

Differential privacy in federated learning

Differential privacy (DP) biedt een wiskundige garantie op informatielek. In federated learning kan DP worden toegepast op clientniveau (lokale DP) of op serverniveau (centrale DP). De standaardaanpak clipt gradiëntnormen en voegt gekalibreerde Gaussiaanse ruis toe:

import torch
from typing import Dict
 
class DifferentiallyPrivateAggregator:
    """Federated aggregation with differential privacy guarantees."""
 
    def __init__(
        self,
        clip_norm: float = 1.0,
        noise_multiplier: float = 1.0,
        num_clients_per_round: int = 10,
    ):
        self.clip_norm = clip_norm
        self.noise_multiplier = noise_multiplier
        self.num_clients_per_round = num_clients_per_round
 
    def clip_update(
        self, update: Dict[str, torch.Tensor]
    ) -> Dict[str, torch.Tensor]:
        """Clip update to bounded L2 norm."""
        flat = torch.cat([param.flatten() for param in update.values()])
        total_norm = flat.norm()
 
        clip_factor = min(1.0, self.clip_norm / (total_norm + 1e-8))
        return {key: param * clip_factor for key, param in update.items()}
 
    def aggregate_with_dp(
        self, updates: list[Dict[str, torch.Tensor]]
    ) -> Dict[str, torch.Tensor]:
        """Aggregate with clipping and noise addition."""
        clipped_updates = [self.clip_update(u) for u in updates]
 
        # Middel de geclipte updates
        keys = clipped_updates[0].keys()
        aggregated = {}
        n = len(clipped_updates)
 
        for key in keys:
            summed = sum(u[key] for u in clipped_updates)
            averaged = summed / n
 
            # Voeg gekalibreerde Gaussiaanse ruis toe
            noise_std = (
                self.clip_norm * self.noise_multiplier / n
            )
            noise = torch.randn_like(averaged) * noise_std
            aggregated[key] = averaged + noise
 
        return aggregated

De privacy-bruikbaarheidsafweging is de centrale spanning: sterkere privacy (meer ruis, strakkere clipping) verlaagt de nauwkeurigheid van het model. Googles inzet van federated learning voor de voorspelling van het volgende woord in Gboard toonde aan dat bruikbare modellen getraind kunnen worden met betekenisvolle DP-garanties, maar de vereiste privacybudgetten zijn vaak groter dan wat theoretici van differential privacy als werkelijk beschermend zouden beschouwen.

Secure aggregation-protocollen

Secure aggregation stelt de server in staat om de som van client-updates te berekenen zonder enige individuele update te zien. Dit voorkomt de eerder beschreven gradiëntlek-aanvallen. Het protocol gebruikt doorgaans cryptografische technieken zoals secret sharing of homomorfe encryptie:

import secrets
import hashlib
from typing import List, Tuple
 
class SimpleSecretSharing:
    """
    Simplified additive secret sharing for secure aggregation.
    In production, use established libraries like TF Federated or PySyft.
    """
 
    @staticmethod
    def create_shares(
        value: int, num_shares: int, modulus: int = 2**32
    ) -> List[int]:
        """Split a value into n additive shares."""
        shares = [secrets.randbelow(modulus) for _ in range(num_shares - 1)]
        # De laatste share zorgt ervoor dat alle shares optellen tot de oorspronkelijke waarde
        last_share = (value - sum(shares)) % modulus
        shares.append(last_share)
        return shares
 
    @staticmethod
    def reconstruct(shares: List[int], modulus: int = 2**32) -> int:
        """Reconstruct value from all shares."""
        return sum(shares) % modulus
 
    @staticmethod
    def generate_pairwise_masks(
        client_id: int, peer_ids: List[int], seed_base: bytes, dimension: int
    ) -> List[int]:
        """Generate pairwise canceling masks between clients."""
        masks = [0] * dimension
        for peer_id in peer_ids:
            # Deterministische seed uit het geordende paar
            pair = (min(client_id, peer_id), max(client_id, peer_id))
            seed = hashlib.sha256(seed_base + str(pair).encode()).digest()
            # Genereer pseudo-willekeurige mask uit de seed
            rng_state = int.from_bytes(seed[:8], "big")
            sign = 1 if client_id < peer_id else -1
            for i in range(dimension):
                rng_state = (rng_state * 6364136223846793005 + 1) % (2**64)
                masks[i] += sign * (rng_state % (2**32))
        return masks

Bonawitz et al. beschreven een praktisch secure-aggregation-protocol in "Practical Secure Aggregation for Privacy-Preserving Machine Learning" (CCS 2017) dat client-uitval afhandelt en schaalt naar duizenden deelnemers. Google zette dit in productie in voor Gboard federated learning.

Methodologie voor red-team-beoordeling

Volg bij het red teamen van een federated-learning-deployment deze gestructureerde aanpak:

Fase 1: Reconnaissance

Bepaal welk FL-framework in gebruik is (TensorFlow Federated, PySyft, Flower, NVIDIA FLARE), het aggregatiealgoritme en de clientauthenticatiemechanismen. Controleer of secure aggregation of differential privacy is ingeschakeld.

Fase 2: Simulatie van clientcompromittering

Test wat er gebeurt wanneer één enkele client wordt gecompromitteerd:

# Voorbeeld: testen met het Flower-framework
# Kloon de doel-FL-setup en wijzig een client
pip install flwr torch torchvision
 
# Voer een vergiftigde client uit tegen de test-aggregatieserver
python poisoned_client.py \
  --server-address="127.0.0.1:8080" \
  --attack-type="backdoor" \
  --target-label=7 \
  --scaling-factor=10 \
  --poison-ratio=0.3

Fase 3: Aggregatierobuustheid

Test de bestendigheid van de aggregatieserver tegen Byzantijnse clients. Verhoog geleidelijk het aantal kwaadwillende clients en meet:

Degradatie van de nauwkeurigheid van het globale model
Slaagpercentage van de backdoor-aanval
Of norm-gebaseerde of afstand-gebaseerde anomaliedetectie wordt geactiveerd

Fase 4: Beoordeling van privacylek

Probeer gradiëntreconstructie-aanvallen tegen het communicatiekanaal. Meet de reconstructiekwaliteit bij verschillende batchgroottes en met verschillende niveaus van DP-ruis.

def assess_privacy_leakage(
    model: torch.nn.Module,
    client_gradients: List[torch.Tensor],
    original_data: torch.Tensor,
    original_labels: torch.Tensor,
) -> dict:
    """Assess how much private data can be reconstructed from gradients."""
    attack = GradientLeakageAttack(model)
    reconstructed_data, reconstructed_labels = attack.attack(
        observed_gradients=client_gradients,
        input_shape=original_data.shape[1:],
        num_classes=10,
        num_iterations=500,
    )
 
    # Meet de reconstructiekwaliteit
    mse = ((reconstructed_data - original_data) ** 2).mean().item()
    label_match = (reconstructed_labels == original_labels).float().mean().item()
 
    return {
        "reconstruction_mse": mse,
        "label_accuracy": label_match,
        "privacy_risk": "high" if mse < 0.1 else "medium" if mse < 0.5 else "low",
    }

Praktijkoverwegingen

Framework-specifieke kwetsbaarheden

Productie-FL-frameworks hebben hun eigen beveiligingsoverwegingen:

NVIDIA FLARE: Ondersteunt secure aggregation en clientauthenticatie via mutual TLS. De standaardconfiguratie schakelt echter mogelijk niet alle beveiligingsfuncties in. Red teams moeten verifiëren dat overseer en project.yml authenticatie afdwingen.
Flower (flwr): Flexibel, maar de beveiliging is grotendeels de verantwoordelijkheid van degene die het inzet. Het gRPC-kanaal moet versleuteld zijn, en aangepaste aggregatiestrategieën moeten worden geaudit op robuustheid.
PySyft: Biedt differential privacy en secure-computation-primitieven, maar vereist zorgvuldige configuratie om betekenisvolle privacygaranties te bereiken.

Implicaties voor regelgeving

Federated learning wordt vaak specifiek ingezet om te voldoen aan dataprotectieregelgeving (AVG, HIPAA). Als gradiëntlek-aanvallen beschermde data kunnen reconstrueren, voldoet de FL-deployment mogelijk niet daadwerkelijk aan de regelgeving, ondanks de architectonische scheiding van data. Dit is een kritieke bevinding voor red-team-rapporten — de organisatie kan een vals gevoel van privacybescherming hebben.

Verdedigingsaanbevelingen

Schakel secure aggregation in om gradiënt-gebaseerde datareconstructie te voorkomen
Pas differential privacy toe met een formeel geverifieerd privacybudget
Gebruik robuuste aggregatie (bijv. trimmed mean of Krum) in plaats van standaard FedAvg
Implementeer clientauthenticatie met hardware-attestatie waar mogelijk
Monitor updateverdelingen op afwijkende bijdragen die op vergiftiging kunnen wijzen
Beperk de invloed van clients door de norm van individuele updates te begrenzen
Audit FL-frameworks op standaardconfiguraties die mogelijk geen beveiligingsfuncties inschakelen

Referenties

McMahan et al. — "Communication-Efficient Learning of Deep Networks from Decentralized Data" (AISTATS 2017) — foundational FedAvg algorithm
Bagdasaryan et al. — "How To Back Door Federated Learning" (AISTATS 2020) — targeted backdoor attacks via model replacement
Zhu et al. — "Deep Leakage from Gradients" (NeurIPS 2019) — gradient-based data reconstruction
Fang et al. — "Local Model Poisoning Attacks to Byzantine-Robust Federated Learning" (USENIX Security 2020) — adaptive attacks against robust aggregation
Bonawitz et al. — "Practical Secure Aggregation for Privacy-Preserving Machine Learning" (CCS 2017) — secure aggregation protocol
Geiping et al. — "Inverting Gradients - How Easy Is It to Break Privacy in Federated Learning?" (NeurIPS 2020) — improved gradient inversion
MITRE ATLAS — AML.T0020 (Poisoning Training Data), AML.T0010 (ML Supply Chain Compromise)
NVIDIA FLARE — https://nvidia.github.io/NVFlare/
Flower Framework — https://flower.ai/

Federated Learning Security

Gevorderd18 min lezenBijgewerkt op 2026-03-20

Beveiligingsaanvallen op federated-learning-systemen, waaronder modelvergiftiging, data-inferentie en exploitatie van Byzantijnse fouten.

infrastructure federated-learning model-poisoning privacy

Overzicht

Architectuur van Federated Learning

Hoe Federated Averaging werkt

De server stuurt het huidige globale model naar een subset van clients
Elke client traint het model gedurende meerdere epochs op lokale data
Clients sturen bijgewerkte modelparameters (of gradiënten) terug naar de server
De server aggregeert de updates, doorgaans via gewogen middeling
Het proces herhaalt zich tot convergentie

import torch
import torch.nn as nn
from typing import Dict, List, Tuple
import copy
 
class FederatedServer:
    """Simplified federated averaging server."""
 
    def __init__(self, global_model: nn.Module):
        self.global_model = global_model
        self.round_number = 0
 
    def aggregate(
        self, client_updates: List[Dict[str, torch.Tensor]], weights: List[float]
    ) -> None:
        """Aggregate client updates using weighted averaging."""
        total_weight = sum(weights)
        global_state = self.global_model.state_dict()
 
        # Initialiseer geaggregeerde state met nullen
        aggregated = {
            key: torch.zeros_like(param) for key, param in global_state.items()
        }
 
        # Gewogen gemiddelde van alle client-updates
        for update, weight in zip(client_updates, weights):
            for key in aggregated:
                aggregated[key] += (weight / total_weight) * update[key]
 
        self.global_model.load_state_dict(aggregated)
        self.round_number += 1
 
    def distribute_model(self) -> Dict[str, torch.Tensor]:
        """Send current global model to clients."""
        return copy.deepcopy(self.global_model.state_dict())

Kaart van het aanvalsoppervlak

De federated-learning-architectuur stelt aanvalsoppervlakken bloot bij elk component:

Component	Aanvalsoppervlak	Voorbeeldaanvallen
Client-apparaten	Lokale modeltraining	Datavergiftiging, modelvervanging
Communicatiekanaal	Transmissie van gradiënten/gewichten	Gradiëntonderschepping, man-in-the-middle
Aggregatieserver	Combinatielogica van updates	Byzantijnse aanvallen, aggregatiemanipulatie
Globaal model	Gedeelde parameters	Backdoor-injectie, modelinversie
Clientselectie	Deelnameprotocol	Sybil-aanvallen, free-riding

Modelvergiftigingsaanvallen

Ongerichte vergiftiging

import torch
import torch.nn as nn
from typing import Dict
 
class UntargetedPoisoner:
    """Untargeted model poisoning — degrades global model accuracy."""
 
    def __init__(self, model: nn.Module, amplification_factor: float = 10.0):
        self.model = model
        self.amplification_factor = amplification_factor
 
    def generate_poisoned_update(
        self, honest_update: Dict[str, torch.Tensor]
    ) -> Dict[str, torch.Tensor]:
        """Generate a poisoned update by negating and amplifying honest gradients."""
        poisoned = {}
        for key, param in honest_update.items():
            # Keer de gradiëntrichting om en versterk
            poisoned[key] = -self.amplification_factor * param
        return poisoned
 
    def generate_noise_update(
        self, reference_update: Dict[str, torch.Tensor]
    ) -> Dict[str, torch.Tensor]:
        """Generate random noise update scaled to match legitimate update norms."""
        poisoned = {}
        for key, param in reference_update.items():
            noise = torch.randn_like(param)
            # Schaal de ruis om overeen te komen met de norm van de legitieme update
            scale = param.norm() / (noise.norm() + 1e-8)
            poisoned[key] = noise * scale
        return poisoned

Gerichte backdoor-aanvallen

De aanval werkt door een lokaal model te trainen op vergiftigde data die een triggerpatroon bevat, en de resulterende update vervolgens op te schalen om middeling met eerlijke clients te overwinnen:

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, Dataset
from typing import Dict, Optional, Tuple
import copy
 
class BackdoorPoisoner:
    """Targeted backdoor attack against federated learning."""
 
    def __init__(
        self,
        model: nn.Module,
        target_label: int,
        trigger_pattern: torch.Tensor,
        trigger_position: Tuple[int, int] = (0, 0),
        scaling_factor: float = 1.0,
    ):
        self.model = model
        self.target_label = target_label
        self.trigger_pattern = trigger_pattern
        self.trigger_position = trigger_position
        self.scaling_factor = scaling_factor
 
    def apply_trigger(self, images: torch.Tensor) -> torch.Tensor:
        """Stamp a trigger pattern onto images."""
        triggered = images.clone()
        h, w = self.trigger_pattern.shape[-2:]
        y, x = self.trigger_position
        triggered[:, :, y : y + h, x : x + w] = self.trigger_pattern
        return triggered
 
    def create_poisoned_batch(
        self, images: torch.Tensor, labels: torch.Tensor, poison_ratio: float = 0.5
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        """Create a mixed batch of clean and poisoned samples."""
        batch_size = images.size(0)
        num_poison = int(batch_size * poison_ratio)
 
        poisoned_images = images.clone()
        poisoned_labels = labels.clone()
 
        # Pas de trigger toe en wijzig labels voor de vergiftigde subset
        poisoned_images[:num_poison] = self.apply_trigger(images[:num_poison])
        poisoned_labels[:num_poison] = self.target_label
 
        return poisoned_images, poisoned_labels
 
    def train_and_scale(
        self,
        global_model_state: Dict[str, torch.Tensor],
        poisoned_loader: DataLoader,
        num_epochs: int = 5,
        lr: float = 0.01,
        num_total_clients: int = 10,
    ) -> Dict[str, torch.Tensor]:
        """Train on poisoned data and scale update to survive aggregation."""
        local_model = copy.deepcopy(self.model)
        local_model.load_state_dict(global_model_state)
 
        optimizer = torch.optim.SGD(local_model.parameters(), lr=lr)
        criterion = nn.CrossEntropyLoss()
 
        local_model.train()
        for epoch in range(num_epochs):
            for images, labels in poisoned_loader:
                poisoned_images, poisoned_labels = self.create_poisoned_batch(
                    images, labels
                )
                optimizer.zero_grad()
                outputs = local_model(poisoned_images)
                loss = criterion(outputs, poisoned_labels)
                loss.backward()
                optimizer.step()
 
        # Schaal de update zodat hij de aggregatie domineert
        # Na het middelen van n clients is de bijdrage van de vergiftigde update 1/n
        # Schalen met n zorgt ervoor dat de backdoor de middeling overleeft
        poisoned_state = local_model.state_dict()
        scaled_update = {}
        for key in global_model_state:
            delta = poisoned_state[key] - global_model_state[key]
            scaled_update[key] = (
                global_model_state[key]
                + delta * num_total_clients * self.scaling_factor
            )
 
        return scaled_update

Gradiënt-gebaseerde datareconstructie

Deep Leakage from Gradients

De aanval werkt door een dummy-input te optimaliseren zodat deze gradiënten produceert die overeenkomen met de waargenomen gradiënten:

import torch
import torch.nn as nn
from typing import List, Tuple, Optional
 
class GradientLeakageAttack:
    """Reconstruct private training data from observed gradients."""
 
    def __init__(self, model: nn.Module, device: str = "cpu"):
        self.model = model
        self.device = device
 
    def attack(
        self,
        observed_gradients: List[torch.Tensor],
        input_shape: Tuple[int, ...],
        num_classes: int,
        num_iterations: int = 300,
        lr: float = 1.0,
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        """
        Reconstruct training data from observed gradients.
 
        Args:
            observed_gradients: List of gradient tensors observed from the client
            input_shape: Shape of the input data (e.g., (3, 32, 32) for CIFAR-10)
            num_classes: Number of output classes
            num_iterations: Optimization iterations
            lr: Learning rate for reconstruction
 
        Returns:
            Tuple of (reconstructed_data, reconstructed_labels)
        """
        # Initialiseer willekeurige dummy-data en labels
        dummy_data = torch.randn(1, *input_shape, device=self.device, requires_grad=True)
        dummy_label = torch.randn(1, num_classes, device=self.device, requires_grad=True)
 
        optimizer = torch.optim.LBFGS([dummy_data, dummy_label], lr=lr)
        criterion = nn.CrossEntropyLoss()
 
        history = []
 
        for iteration in range(num_iterations):
            def closure():
                optimizer.zero_grad()
                self.model.zero_grad()
 
                # Forward pass met dummy-data
                pred = self.model(dummy_data)
                dummy_loss = criterion(pred, torch.softmax(dummy_label, dim=-1))
 
                # Bereken gradiënten van de dummy-data
                dummy_gradients = torch.autograd.grad(
                    dummy_loss, self.model.parameters(), create_graph=True
                )
 
                # Minimaliseer de afstand tussen dummy- en waargenomen gradiënten
                gradient_diff = sum(
                    ((dg - og) ** 2).sum()
                    for dg, og in zip(dummy_gradients, observed_gradients)
                )
 
                gradient_diff.backward()
                return gradient_diff
 
            loss = optimizer.step(closure)
            if iteration % 50 == 0:
                history.append(loss.item())
 
        reconstructed_label = torch.argmax(dummy_label, dim=-1)
        return dummy_data.detach(), reconstructed_label.detach()

Membership inference in federated-omgevingen

import torch
import torch.nn as nn
from typing import Dict, List
 
class FederatedMembershipInference:
    """Determine if a specific sample was in a client's training data."""
 
    def __init__(self, model: nn.Module, threshold: float = 0.5):
        self.model = model
        self.threshold = threshold
 
    def compute_loss_change(
        self,
        target_sample: torch.Tensor,
        target_label: torch.Tensor,
        model_before: Dict[str, torch.Tensor],
        model_after: Dict[str, torch.Tensor],
    ) -> float:
        """Measure how much the loss on a target sample changed after a client update."""
        criterion = nn.CrossEntropyLoss()
 
        # Verlies vóór de client-update
        self.model.load_state_dict(model_before)
        self.model.eval()
        with torch.no_grad():
            loss_before = criterion(
                self.model(target_sample.unsqueeze(0)), target_label.unsqueeze(0)
            ).item()
 
        # Verlies na de client-update
        self.model.load_state_dict(model_after)
        self.model.eval()
        with torch.no_grad():
            loss_after = criterion(
                self.model(target_sample.unsqueeze(0)), target_label.unsqueeze(0)
            ).item()
 
        return loss_before - loss_after
 
    def infer_membership(
        self,
        target_samples: List[torch.Tensor],
        target_labels: List[torch.Tensor],
        model_before: Dict[str, torch.Tensor],
        model_after: Dict[str, torch.Tensor],
    ) -> List[bool]:
        """Infer membership for multiple samples."""
        results = []
        for sample, label in zip(target_samples, target_labels):
            loss_change = self.compute_loss_change(
                sample, label, model_before, model_after
            )
            # Een grote positieve verliesverandering wijst erop dat het sample in de trainingsset zat
            results.append(loss_change > self.threshold)
        return results

Sybil- en free-rider-aanvallen

Sybil-aanvallen

import torch
import torch.nn as nn
from typing import Dict, List
import copy
 
class SybilAttacker:
    """Launch a Sybil attack by controlling multiple federated clients."""
 
    def __init__(
        self,
        num_sybil_clients: int,
        poisoned_update_fn,  # callable die een vergiftigde update genereert
    ):
        self.num_sybil_clients = num_sybil_clients
        self.poisoned_update_fn = poisoned_update_fn
 
    def generate_sybil_updates(
        self,
        global_model_state: Dict[str, torch.Tensor],
        add_noise: bool = True,
        noise_scale: float = 0.01,
    ) -> List[Dict[str, torch.Tensor]]:
        """Generate multiple slightly varied poisoned updates to avoid detection."""
        base_update = self.poisoned_update_fn(global_model_state)
        updates = []
 
        for i in range(self.num_sybil_clients):
            if add_noise and i > 0:
                # Voeg kleine willekeurige verstoringen toe zodat updates niet identiek zijn
                noisy_update = {}
                for key, param in base_update.items():
                    noise = torch.randn_like(param) * noise_scale * param.abs().mean()
                    noisy_update[key] = param + noise
                updates.append(noisy_update)
            else:
                updates.append(copy.deepcopy(base_update))
 
        return updates

Free-rider-aanvallen

Byzantijns-bestendige aggregatie

Robuuste aggregatieregels

import torch
from typing import Dict, List
 
class RobustAggregator:
    """Byzantine-resilient aggregation methods for federated learning."""
 
    @staticmethod
    def coordinate_wise_median(
        updates: List[Dict[str, torch.Tensor]],
    ) -> Dict[str, torch.Tensor]:
        """Coordinate-wise median aggregation (Yin et al., 2018)."""
        keys = updates[0].keys()
        aggregated = {}
 
        for key in keys:
            stacked = torch.stack([u[key] for u in updates])
            aggregated[key] = torch.median(stacked, dim=0).values
 
        return aggregated
 
    @staticmethod
    def trimmed_mean(
        updates: List[Dict[str, torch.Tensor]], trim_ratio: float = 0.1
    ) -> Dict[str, torch.Tensor]:
        """Trimmed mean — remove extreme values before averaging."""
        keys = updates[0].keys()
        aggregated = {}
        n = len(updates)
        trim_count = int(n * trim_ratio)
 
        for key in keys:
            stacked = torch.stack([u[key] for u in updates])
            sorted_updates, _ = torch.sort(stacked, dim=0)
            # Verwijder de bovenste en onderste trim_count waarden
            trimmed = sorted_updates[trim_count : n - trim_count]
            aggregated[key] = trimmed.mean(dim=0)
 
        return aggregated
 
    @staticmethod
    def krum(
        updates: List[Dict[str, torch.Tensor]], num_byzantine: int
    ) -> Dict[str, torch.Tensor]:
        """
        Multi-Krum aggregation (Blanchard et al., 2017).
        Selects the update closest to its neighbors.
        """
        n = len(updates)
        num_select = n - num_byzantine - 2
 
        # Plat de updates uit voor afstandsberekening
        flat_updates = []
        for update in updates:
            flat = torch.cat([param.flatten() for param in update.values()])
            flat_updates.append(flat)
 
        # Bereken paarsgewijze afstanden
        distances = torch.zeros(n, n)
        for i in range(n):
            for j in range(i + 1, n):
                dist = (flat_updates[i] - flat_updates[j]).norm()
                distances[i][j] = dist
                distances[j][i] = dist
 
        # Voor elke update: som de afstanden tot de dichtstbijzijnde buren
        scores = []
        for i in range(n):
            sorted_dists, _ = torch.sort(distances[i])
            # Som van afstanden tot de num_select dichtstbijzijnde buren (zonder zichzelf)
            score = sorted_dists[1 : num_select + 1].sum()
            scores.append(score)
 
        # Selecteer de update met de laagste score
        best_idx = int(torch.argmin(torch.tensor(scores)))
        return updates[best_idx]

Differential privacy in federated learning

import torch
from typing import Dict
 
class DifferentiallyPrivateAggregator:
    """Federated aggregation with differential privacy guarantees."""
 
    def __init__(
        self,
        clip_norm: float = 1.0,
        noise_multiplier: float = 1.0,
        num_clients_per_round: int = 10,
    ):
        self.clip_norm = clip_norm
        self.noise_multiplier = noise_multiplier
        self.num_clients_per_round = num_clients_per_round
 
    def clip_update(
        self, update: Dict[str, torch.Tensor]
    ) -> Dict[str, torch.Tensor]:
        """Clip update to bounded L2 norm."""
        flat = torch.cat([param.flatten() for param in update.values()])
        total_norm = flat.norm()
 
        clip_factor = min(1.0, self.clip_norm / (total_norm + 1e-8))
        return {key: param * clip_factor for key, param in update.items()}
 
    def aggregate_with_dp(
        self, updates: list[Dict[str, torch.Tensor]]
    ) -> Dict[str, torch.Tensor]:
        """Aggregate with clipping and noise addition."""
        clipped_updates = [self.clip_update(u) for u in updates]
 
        # Middel de geclipte updates
        keys = clipped_updates[0].keys()
        aggregated = {}
        n = len(clipped_updates)
 
        for key in keys:
            summed = sum(u[key] for u in clipped_updates)
            averaged = summed / n
 
            # Voeg gekalibreerde Gaussiaanse ruis toe
            noise_std = (
                self.clip_norm * self.noise_multiplier / n
            )
            noise = torch.randn_like(averaged) * noise_std
            aggregated[key] = averaged + noise
 
        return aggregated

Secure aggregation-protocollen

import secrets
import hashlib
from typing import List, Tuple
 
class SimpleSecretSharing:
    """
    Simplified additive secret sharing for secure aggregation.
    In production, use established libraries like TF Federated or PySyft.
    """
 
    @staticmethod
    def create_shares(
        value: int, num_shares: int, modulus: int = 2**32
    ) -> List[int]:
        """Split a value into n additive shares."""
        shares = [secrets.randbelow(modulus) for _ in range(num_shares - 1)]
        # De laatste share zorgt ervoor dat alle shares optellen tot de oorspronkelijke waarde
        last_share = (value - sum(shares)) % modulus
        shares.append(last_share)
        return shares
 
    @staticmethod
    def reconstruct(shares: List[int], modulus: int = 2**32) -> int:
        """Reconstruct value from all shares."""
        return sum(shares) % modulus
 
    @staticmethod
    def generate_pairwise_masks(
        client_id: int, peer_ids: List[int], seed_base: bytes, dimension: int
    ) -> List[int]:
        """Generate pairwise canceling masks between clients."""
        masks = [0] * dimension
        for peer_id in peer_ids:
            # Deterministische seed uit het geordende paar
            pair = (min(client_id, peer_id), max(client_id, peer_id))
            seed = hashlib.sha256(seed_base + str(pair).encode()).digest()
            # Genereer pseudo-willekeurige mask uit de seed
            rng_state = int.from_bytes(seed[:8], "big")
            sign = 1 if client_id < peer_id else -1
            for i in range(dimension):
                rng_state = (rng_state * 6364136223846793005 + 1) % (2**64)
                masks[i] += sign * (rng_state % (2**32))
        return masks

Methodologie voor red-team-beoordeling

Volg bij het red teamen van een federated-learning-deployment deze gestructureerde aanpak:

Fase 1: Reconnaissance

Fase 2: Simulatie van clientcompromittering

Test wat er gebeurt wanneer één enkele client wordt gecompromitteerd:

# Voorbeeld: testen met het Flower-framework
# Kloon de doel-FL-setup en wijzig een client
pip install flwr torch torchvision
 
# Voer een vergiftigde client uit tegen de test-aggregatieserver
python poisoned_client.py \
  --server-address="127.0.0.1:8080" \
  --attack-type="backdoor" \
  --target-label=7 \
  --scaling-factor=10 \
  --poison-ratio=0.3

Fase 3: Aggregatierobuustheid

Test de bestendigheid van de aggregatieserver tegen Byzantijnse clients. Verhoog geleidelijk het aantal kwaadwillende clients en meet:

Degradatie van de nauwkeurigheid van het globale model
Slaagpercentage van de backdoor-aanval
Of norm-gebaseerde of afstand-gebaseerde anomaliedetectie wordt geactiveerd

Fase 4: Beoordeling van privacylek

Probeer gradiëntreconstructie-aanvallen tegen het communicatiekanaal. Meet de reconstructiekwaliteit bij verschillende batchgroottes en met verschillende niveaus van DP-ruis.

def assess_privacy_leakage(
    model: torch.nn.Module,
    client_gradients: List[torch.Tensor],
    original_data: torch.Tensor,
    original_labels: torch.Tensor,
) -> dict:
    """Assess how much private data can be reconstructed from gradients."""
    attack = GradientLeakageAttack(model)
    reconstructed_data, reconstructed_labels = attack.attack(
        observed_gradients=client_gradients,
        input_shape=original_data.shape[1:],
        num_classes=10,
        num_iterations=500,
    )
 
    # Meet de reconstructiekwaliteit
    mse = ((reconstructed_data - original_data) ** 2).mean().item()
    label_match = (reconstructed_labels == original_labels).float().mean().item()
 
    return {
        "reconstruction_mse": mse,
        "label_accuracy": label_match,
        "privacy_risk": "high" if mse < 0.1 else "medium" if mse < 0.5 else "low",
    }

Praktijkoverwegingen

Framework-specifieke kwetsbaarheden

Productie-FL-frameworks hebben hun eigen beveiligingsoverwegingen:

NVIDIA FLARE: Ondersteunt secure aggregation en clientauthenticatie via mutual TLS. De standaardconfiguratie schakelt echter mogelijk niet alle beveiligingsfuncties in. Red teams moeten verifiëren dat overseer en project.yml authenticatie afdwingen.
Flower (flwr): Flexibel, maar de beveiliging is grotendeels de verantwoordelijkheid van degene die het inzet. Het gRPC-kanaal moet versleuteld zijn, en aangepaste aggregatiestrategieën moeten worden geaudit op robuustheid.
PySyft: Biedt differential privacy en secure-computation-primitieven, maar vereist zorgvuldige configuratie om betekenisvolle privacygaranties te bereiken.

Implicaties voor regelgeving

Verdedigingsaanbevelingen

Schakel secure aggregation in om gradiënt-gebaseerde datareconstructie te voorkomen
Pas differential privacy toe met een formeel geverifieerd privacybudget
Gebruik robuuste aggregatie (bijv. trimmed mean of Krum) in plaats van standaard FedAvg
Implementeer clientauthenticatie met hardware-attestatie waar mogelijk
Monitor updateverdelingen op afwijkende bijdragen die op vergiftiging kunnen wijzen
Beperk de invloed van clients door de norm van individuele updates te begrenzen
Audit FL-frameworks op standaardconfiguraties die mogelijk geen beveiligingsfuncties inschakelen

Referenties

McMahan et al. — "Communication-Efficient Learning of Deep Networks from Decentralized Data" (AISTATS 2017) — foundational FedAvg algorithm
Bagdasaryan et al. — "How To Back Door Federated Learning" (AISTATS 2020) — targeted backdoor attacks via model replacement
Zhu et al. — "Deep Leakage from Gradients" (NeurIPS 2019) — gradient-based data reconstruction
Fang et al. — "Local Model Poisoning Attacks to Byzantine-Robust Federated Learning" (USENIX Security 2020) — adaptive attacks against robust aggregation
Bonawitz et al. — "Practical Secure Aggregation for Privacy-Preserving Machine Learning" (CCS 2017) — secure aggregation protocol
Geiping et al. — "Inverting Gradients - How Easy Is It to Break Privacy in Federated Learning?" (NeurIPS 2020) — improved gradient inversion
MITRE ATLAS — AML.T0020 (Poisoning Training Data), AML.T0010 (ML Supply Chain Compromise)
NVIDIA FLARE — https://nvidia.github.io/NVFlare/
Flower Framework — https://flower.ai/

Federated Learning Security

Gerelateerde artikelen

Federated Learning Security

Gerelateerde artikelen