Laboratório 10 - Explorando Ataques e Defesas em FL

Envenenamento de Dados
ataque de envenenamento de dados ocorre quando um ou mais clientes inserem exemplos maliciosos em seus conjuntos locais de treinamento para manipular o modelo global. Diferente de um ataque direto ao servidor, ele explora o fato de que os dados ficam distribuídos e privados, tornando difícil verificar sua integridade.
Um exemplo simples seria um usuário que, em um sistema de reconhecimento de escrita, altera propositalmente etiquetas de imagens do número “7” para “1”. Assim, ao participar do treinamento federado, suas atualizações contaminam o modelo e podem levar a erros de classificação sistemáticos. Outro cenário é o ataque backdoor: um cliente insere dados com um padrão visual específico (como um ponto vermelho em uma imagem) e força o modelo a associá-lo a uma classe incorreta, mesmo que o desempenho geral do modelo pareça normal.
Detectar esse tipo de ataque é desafiador porque o servidor não tem acesso direto aos dados originais, apenas às atualizações de modelo, que podem parecer estatisticamente plausíveis. Além disso, distinguir contribuições legítimas de usuários com dados naturalmente enviesados de atualizações maliciosas é complexo. Métodos de defesa baseados em estatística, robustez agregada ou análise de gradientes ajudam, mas precisam equilibrar segurança com a preservação da privacidade e da eficiência do treinamento.
Alterando o código do Cliente para Realizar o Ataque
def flip_labels(self, num_classes=10):
self.y_train = (self.y_train + 1) % num_classes
self.y_test = (self.y_test + 1) % num_classes
Dessa forma, podemos realizar a troca de todos os labels de um cliente. Porém, apenas um rótulo poderia ser flipado para introduzir um comportamento específico no meolo. Mas nesse caso, só iremos alterar todos os rótulos para degradar a eficiência do modelo.
Em seguida, devemos incluir o ataque no código do cliente do Laboratório 9. Para isso, vamos incluir um atributo no construtor para definir o tipo de ataque que queremos realizar, em seguida, devemos aplicar uma condição informando quais clientes irão realizar o ataque. Nessa caso, vamos considerar clientes com id par cid % 2 == 0
class Cliente(fl.client.NumPyClient):
def __init__(self, cid, niid, num_clients, dirichlet_alpha, attack_type=None, filename=None):
...
self.attack_type = attack_type
...
if self.attack_type == 'label_flipping' and self.cid % 2 == 0:
self.flip_labels()
Preparando a simulação
import tensorflow as tf
import flwr as fl
from client import Cliente
from server import Servidor
import warnings
warnings.filterwarnings("ignore")
def create_client(cid):
client = Cliente(cid, NIID, NCLIENTS, DIRICHLET_ALPHA, ATTACK_TYPE, FILENAME)
return client.to_client()
class Simulation():
def __init__(self):
self.server = Servidor(num_clients=NCLIENTS, dirichlet_alpha=DIRICHLET_ALPHA, fraction_fit=FRACTION_FIT)
def run_simulation(self):
fl.simulation.start_simulation(
client_fn = create_client,
num_clients = NCLIENTS,
config = fl.server.ServerConfig(num_rounds=NROUNDS),
strategy = self.server,
client_resources = {"num_cpus": 6, "num_gpus": 0.0},)
Executando Simulação - Data poisoning
Após definir a classe de simulação e definir os parâmetros de configuração desejados, podemos inicializar a simulação do ambiente federado com o ataque. Ao final da simulação os resultados estarão salvos nos arquivos {filename}_train.csv
para os dados de treino e {filename}_evaluate.csv
para os dados de teste.
NCLIENTS = 3
NROUNDS = 5
NIID = False
DIRICHLET_ALPHA = 0.1
FRACTION_FIT = 0.5
ATTACK_TYPE = 'label_flipping' # 'label-flip',
FILENAME = ATTACK_TYPE
Simulation().run_simulation()
Visualizando Performance dos clientes
utilizando o código de plotagem do Laboratório 9, criamos um arquivo plot_utils.py
definindo as funções de plotagem plot_loss_acc(FILENAME)
e também a plot_performance_distribution(FILENAME, NCLIENTS)
para que possamos utilizar nesse laboratório
import plot_utils
plot_utils.plot_loss_accuracy(FILENAME)

Como podemos observar o envenenamento dos dados causou um degradação no desempenho do modelo. Portanto, é essencial desenvolver métodos de detecção de ataques para garantir a integridade e robustez do modelo. Esses métodos devem detectar os ataques apenas analisando os pesos do modelo, tornando a detecção mais complexa.
Model Poisoning Attack
o ataque de envenenamento de modelo (model poisoning) ocorre quando um participante malicioso não apenas manipula seus dados locais, mas modifica intencionalmente as atualizações (gradientes ou pesos) que envia ao agregador para corromper o modelo global, por exemplo, escalando e direcionando gradientes para forçar uma queda de desempenho em certas classes ou implantar um backdoor que ativa um comportamento errôneo na presença de um gatilho específico; ao contrário do data poisoning, aqui o atacante trabalha diretamente na manipulação das atualizações para contornar agregadores robustos e pode coordenar múltiplos clientes para amplificar o efeito. Detectar esse ataque é difícil porque as atualizações maliciosas podem parecer estatisticamente plausíveis, podem ser camufladas por ruído natural e aproveitam a falta de acesso a dados locais; técnicas de defesa incluem detecção de outliers nas atualizações, agregação robusta (p. ex. medianas, Krum), monitoramento de perda/validação e verificação por múltiplas rodadas, mas todas precisam balancear eficácia, custo computacional e privacidade.
Para realizar o ataque de model poisoning, vamos alterar o código do cliente adicionando a função model_poisoning()
a qual vai alterar o pesos dos modelos de uma ou todas as camadas específicas de acordo como shape
da camada. Nesse exemplo, colocamos um ruindo normal * 10
mas outras alterações podem ser incluídas, como por exemplo, reversão dos pesos, zerar alguma camada, alterar apenas alguns pesos, etc.
NCLIENTS = 3
NROUNDS = 5
NIID = False
DIRICHLET_ALPHA = 0.1
FRACTION_FIT = 0.5
ATTACK_TYPE = 'model_poisoning'
FILENAME = ATTACK_TYPE
def model_poisoning(self, layer):
weights = self.model.get_weights()
if str(layer).lower() != 'all':
shape = weights[layer].shape
weights[layer] = weights[layer] + np.random.normal(0, 1, shape) * 10
else:
for layer in range(len(weights)):
shape = weights[layer].shape
weights[layer] = weights[layer] + np.random.normal(0, 1, shape) * 10
self.model.set_weights(weights)
Note que o envenenamento de modelo deve ser realizado após o treinamento do cliente, assim devemos adicionar uma condição de verificação no método fit()
, além disso, também devemos definir quais clientes irão realizar o ataque. Nesse exemplo, vamos utilizar a mesma condição do ataque de envenenamento de dados.
Após definir os parâmetros para o ataque, podemos executar a simulação executando o método run_simulation()
Simulation().run_simulation()
Assim que a simulação terminar, os podemos realizar as plotagens para analizar a performance do modelo da mesma forma que foi utilizada no método de label_fliping
, entretanto, devemos alterar o nome do arquivo que iremos consultar os dados.
import plot_utils
plot_utils.plot_performance_distribution(FILENAME, NCLIENTS)

Os resultados mostraram uma degradação mais severa nos desepenho do modelo, uma vez que o modelo todo foi comprometido. Do mesmo modo que o ataque anterior, devemos desenvolver soluções para detecção desse tipo de ataque apenas olhando para os modelos compartilhados pelos clientes para garantir a integridade e robustez do modelo.
Inversão de Gradientes
A inversão de gradientes é um ataque onde o cliente malicioso ou servidor consegue recuperar os dados utilizados no treinamento pelos clientes apenas utilizando os modelos compartilhados pelos clientes. Uma técnica muito utilizada para esse ataque é a DLG Deep Leakage from Gradients, e vamos analisa-la detalhadamente a seguir. Para executar os e defesas contra inversão de gradientes, vamos precisar de funções auxiliares implementadas nesse módulo utils. Dessa forma, devemos fazer os seguintes importes
import utils
import torch
import torch.nn.functional as F
from tqdm import tqdm
from torchvision import transforms
import numpy as np
Assim, podemos carregar um dataset que será utilizado como exemplo para realizar o ataque de inversão, os datasets disponíveis incluem:
- MNIST
- CIFAR10
- CIFAR100
- FashionMNIST
Entratanto, outros datasets disponíveis no huggingface também pode ser incluídos.
Para carregar o dataset, podemos utilizar o seguinte código
dataset, transfor = utils.cria_dataset('cifar10')
modelo = utils.cria_modelo('cifar10')
Em seguida, devemos selecionar uma imagem que será utilizada no nosso exemplo para realizar a inversão e também gerar um ruído aleatório que será utilizado para se aproximar da imagem alvo
imagem, label = utils.get_imagem_e_label(21910, dataset, transfor, num_classes=10)
imagem_fake, label_fake = utils.cria_ruido(imagem.size(), label.size())
utils.compara_imagens(imagem, imagem_fake)
A imagem a seguir apresenta um exemplo:

A imagem a seguir apresenta o ataque DLG, onde a ideia é baseado nos gradientes reais compartilhados pelo cliente utilizar o output dos gradientes do ruído para minimizar a distância ente eles. Entetanto, ao invés de atulizar os pesos do modelo para reduzir tal distância, os pixels da imagem são atualizados.

Executando o Ataque de Inversão
Primeiramente, devemos simular o comportamento de um cliente honesto, onde ele irá passar sua imagem real no modelo e compartilhar os gradientes do modelo, para isso vamos utilizar o seguinte código:
output_modelo = modelo(imagem)
criterion = utils.cross_entropy_for_onehot
logits = criterion(output_modelo, label)
gradientes = torch.autograd.grad(logits, modelo.parameters())
gradientes_compartilhados = list((_.detach().clone() for _ in gradientes))
Em seguida, baseado nos gradientes compartilhados, vamos simular a captura desses gradientes pelo servidor para tentar realizar a inversão. Para isso, vamos precisar gerar o ruído e aplicar o ataque de DLG nos gradientes, utilizando o seguinte código
imagem_fake, label_fake = utils.cria_ruido(imagem.size(), label.size())
optimizer_dlg = torch.optim.LBFGS([imagem_fake, label_fake], )
loss_dlg = torch.nn.CrossEntropyLoss(reduction='none')
historico = run_ataque(modelo, imagem_fake, label_fake, gradientes_compartilhados,
optimizer_dlg, loss_dlg, ataque='dlg')
utils.plot_historico_ataque(historico, label_fake)
Como podemos obervar, após 140 iterações do ataque podemos reconstruir a imagem utilizada no treinamento pelo cliente.

Portanto, podemos concluir que mesmo sem o compartilhamento dos dados, os modelos compartilhados são capazes de revelar os dados utilizados no treinamento. Assim, técnicas para prevensão desse tipo de ataque devem ser utilizada como prunning e privacidade diferencial
Dificultando a Inversão de Gradientes via Prunning
O Prunning é um método de esparsificação de modelos, no qual uma máscara de 0s
e 1s
é definida para zerar os pesos do modelo, consequentemente gerando um modelo esparso.

Essa poda pode ser Estruturada (zerando neurônios do modelo), quando Não Estruturada (zerando conexões do modelo). Assim, reduzindo a quantidade de informação compartilhada, onde apenas os pesos que estejam com máscara igual a 1 serão compartilhados, fica mais defícil realizar a reconstrução.
Nesse exemplo, vamos utilizar uma poda de 40%
do modelo para analisar o impacto na reconstrução da imagem. Assim, podemos aplicar a poda e gerar os gradientes simulando o cliente alvo da seguinte forma
modelo = utils.cria_modelo('cifar10')
utils.poda_modelo(modelo, 0.4)
output_modelo = modelo(imagem)
criterion = utils.cross_entropy_for_onehot
logits = criterion(output_modelo, label)
gradientes = torch.autograd.grad(logits, modelo.parameters())
gradientes_compartilhados = list((_.detach().clone() for _ in gradientes))

Com os resultados, podemos observar que a poda deixou a reconstrução mais dificil, mesmo assim, ainda foi possível reconhecer a imagem alvo. Nesse cenário, uma poda maior poderia ser utilizada, porém isso pode degradar o desempenho do modelo drasticamente. Para exemplificar, vamos apresentar a reconstrução com 80%
de poda.

Nesse caso, não conseguimos recuperar a imagem alvo, o que é resultado da alta esparsidade do modelo. Entretanto, uma análise adicional para verficar o impacto na performance do modelo deve ser realizada.
Melhorando a Segurança e Evitando Ataques de Inversão
Para melhorar a segurança de soluções de FL sem degradar a performance, métodos de Criptografia Homomórfica devem ser incorporados, entretanto, métodos para redução de overhead de comunicação devem ser propostos para reduzir o alto overhead gerado pela criptografia