联邦学习后门攻击代码阅读——backdoors101

代码地址

• https://github.com/MrWater98/backdoors101

论文地址

• https://arxiv.org/abs/1807.00459
• https://arxiv.org/abs/2005.03823

代码阅读目标

• 清楚概括代码运行过程。
• 能将代码和论文相对应。
• 收集目前还不了解的内容。
• 只阅读和联邦学习相关的内容

代码结构

│ attack.py
│ helper.py
│ requirements.txt
│ training.py
├─configs
　　│ cifar10_params.yaml
　　│ cifar_fed.yaml
　　│ imagenet_params.yaml
　　│ mnist_params.yaml
　　│ pipa_params.yaml
├─dataset
　　│ celeba.py
　　│ multi_mnist_loader.py
　　│ pipa.py
　　│ vggface.py
├─losses
　　│ loss_functions.py
├─metrics
　　│ accuracy_metric.py
　　│ metric.py
　　│ test_loss_metric.py
├─models
　　│ face_ident.py
　　│ model.py
　　│ resnet.py
　　│ simple.py
　　│ vgg.py
　　│ word_model.py
　　│ init.py
├─src
　　│ attack_vectors.png
　　│ calculator.png
　　│ complex.png
　　│ pipa.png
　　│ pixel_vs_semantic.png
├─synthesizers
　　│ complex_synthesizer.py
　　│ pattern_synthesizer.py
　　│ physical_synthesizer.py
　　│ singlepixel_synthesizer.py
　　│ synthesizer.py
├─tasks
　　│ │ batch.py
　　│ │ celeba_helper.py
　　│ │ cifar10_task.py
　　│ │ imagenet_task.py
　　│ │ imdb_helper.py
　　│ │ mnist_task.py
　　│ │ multimnist_helper.py
　　│ │ pipa_task.py
　　│ │ task.py
　　│ │ vggface_helper.py
　　│ │
　　│ └─fl
　　　　│ cifarfed_task.py
　　　　│ fl_emnist_task.py
　　　　│ fl_reddit_task.py
　　　　│ fl_task.py
　　　　│ fl_user.py
　　└─utils
　　│ │ index.html
　　│ │ min_norm_solvers.py
　　│ │ parameters.py
　　│ │ utils.py
　　│ │ init.py

运行方法

1. 安装所有依赖pip install -r requirements.txt
2. 创建两个目录runs和saved_models
3. 通过tensorboard --logdir=runs/创建tensorboard
4. 通过python training.py --name mnist --params configs/mnist_params.yaml --commit none来运行其中的一个训练集
   1. 这里运行的是mnist_params.yaml
   2. 如果需要运行的是联邦学习的，则需要调整name后面的参数和params后面的参数。

梳理流程

trainning.py

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Backdoors')
    parser.add_argument('--params', dest='params', default='utils/params.yaml')
    parser.add_argument('--name', dest='name', required=True)
    parser.add_argument('--commit', dest='commit',
                        default=get_current_git_hash())

    args = parser.parse_args()
	# 第二个参数设定的.yaml最重要，name只是确认你创建的文件的名字
    with open(args.params) as f:
        params = yaml.load(f, Loader=yaml.FullLoader)
	
    params['current_time'] = datetime.now().strftime('%b.%d_%H.%M.%S')
    params['commit'] = args.commit
    params['name'] = args.name
    
    helper = Helper(params) 
    logger.warning(create_table(params))
	
    try:
        # 参数fl来自于cifar_fed.yaml
        if helper.params.fl:
            fl_run(helper)
        else:
            run(helper)

• 前面基本都是好理解的，我们现在进入到Helper的构造函数

Helper.py

params: Params = None
    # https://docs.python.org/3/library/typing.html#typing.Union
    # 这应该是意为这要不是Task，要不是FederatedLearningTask
    task: Union[Task, FederatedLearningTask] = None
    # 来源于 https://github.com/MrWater98/backdoors101 
    # 将未backdoored的输入转化为backdoored的输入
    synthesizer: Synthesizer = None
    # 包括了多个人物的同步器和损失率的计算
    attack: Attack = None
    # tensorboard的结果
    tb_writer: SummaryWriter = None

    def __init__(self, params):
        self.params = Params(**params)

        self.times = {'backward': list(), 'forward': list(), 'step': list(),
                      'scales': list(), 'total': list(), 'poison': list()}
        if self.params.random_seed is not None:
            self.fix_random(self.params.random_seed)
        # 创建结果的文件夹
        self.make_folders()
        # 找到训练用的xx_task文件，用默认构造函数获取构造后的结果
        self.make_task()
        # 基本同上
        self.make_synthesizer()
        # 拿到Attakc对象
        self.attack = Attack(self.params, self.synthesizer)
        # neural cleanse 识别和减轻神经网络中的后门攻击的手段
        self.nc = True if 'neural_cleanse' in self.params.loss_tasks else False
        
        self.best_loss = float('inf')

• 然后我们会进入make_task(), make_synthesizer()和Attack()
• 他们的基本目标都是找到参数中对应的类并获得他们类中存储的对象和方法。

fl_task.py

• 本函数从make_task()中进入

class FederatedLearningTask(Task):
    fl_train_loaders: List[Any] = None
    ignored_weights = ['num_batches_tracked']#['tracked', 'running']
    adversaries: List[int] = None

    def init_task(self):
        # load_data对应的是cifarfed_task.py
        self.load_data()
        # build_model就只是创建一个18层的残差网络
        self.model = self.build_model()
        # resume_model是当有训练过的模型后，则会使用之前的模型
        self.resume_model()
        # 使用cpu或者gpu训练，这里主要是使用cpu
        self.model = self.model.to(self.params.device)

        self.local_model = self.build_model().to(self.params.device)
        # 直接用entrophy了
        self.criterion = self.make_criterion()
        # 这个是选择攻击的样本
        self.adversaries = self.sample_adversaries()
		# 评价矩阵
        self.metrics = [AccuracyMetric(), TestLossMetric(self.criterion)]
        
        self.set_input_shape()
        return

load_data()

• 基本的意思是从cifar10_task.py调用load_cifar_data，然后分给每个用户500个图像，模拟他们是联邦学习的客户端。

def load_data(self) -> None:
        self.load_cifar_data()
        if self.params.fl_sample_dirichlet:
            # sample indices for participants using Dirichlet distribution
            indices_per_participant = self.sample_dirichlet_train_data(
                self.params.fl_total_participants,
                alpha=self.params.fl_dirichlet_alpha)
            train_loaders = [(pos, self.get_train(indices)) for pos, indices in
                             indices_per_participant.items()]
        else:
            # sample indices for participants that are equally
            # split to 500 images per participant
            all_range = list(range(len(self.train_dataset)))
            random.shuffle(all_range)
            train_loaders = [self.get_train_old(all_range, pos)
                             for pos in
                             range(self.params.fl_total_participants)]
        self.fl_train_loaders = train_loaders
        return

fl_run和run_fl_round

• 我们现在已经获得好了所有需要的材料，进入跑模型的阶段了。

def fl_run(hlpr: Helper):
    for epoch in range(hlpr.params.start_epoch,
                       hlpr.params.epochs + 1):
        run_fl_round(hlpr, epoch)
        metric = test(hlpr, epoch, backdoor=False)
        test(hlpr, epoch, backdoor=True)

        hlpr.save_model(hlpr.task.model, epoch, metric)


def run_fl_round(hlpr, epoch):
    # 获得global的模型
    global_model = hlpr.task.model
    # 获得local的模型
    local_model = hlpr.task.local_model

    round_participants = hlpr.task.sample_users_for_round(epoch)
    weight_accumulator = hlpr.task.get_empty_accumulator()
    # tqdm是python的进度条库，基本是基于对象迭代
    for user in tqdm(round_participants):
        # 将参数从global_model复制到local_model
        hlpr.task.copy_params(global_model, local_model)
        # 一个对象，会保存当前状态，并根据梯度更新参数
        optimizer = hlpr.task.make_optimizer(local_model)
        for local_epoch in range(hlpr.params.fl_local_epochs):
            # 如果是恶意的用户，则执行进攻的训练
            if user.compromised:
                train(hlpr, local_epoch, local_model, optimizer,
                      user.train_loader, attack=True)
            # 如果是非恶意的用户，则执行非进攻的训练
            else:
                train(hlpr, local_epoch, local_model, optimizer,
                      user.train_loader, attack=False)
        # 然后来更新global的模型
        local_update = hlpr.task.get_fl_update(local_model, global_model)
        # 如果用户是恶意用户，还会更新梯度
        if user.compromised:
            hlpr.attack.fl_scale_update(local_update)
        # 存疑，感觉是积累当前的权重变化
        hlpr.task.accumulate_weights(weight_accumulator, local_update)
    # 所有用户完成之后，更新全局的模型
    hlpr.task.update_global_model(weight_accumulator, global_model)

• 这里fl_run都比较好理解，但是run_fl_round相对来说比较难理解。我把注释写在了代码上。

trainning

• 现在进入了训练的阶段，可以观察代码是如何训练和攻击当前模型的了。