聯邦學習投毒（訓練管線）

┌─────────────┐     模型更新           ┌──────────────────┐
│ 參與者 1    │ ───────────────────── │                  │
│ （誠實）    │                       │  聚合            │
├─────────────┤     模型更新           │  伺服器          │
│ 參與者 2    │ ───────────────────── │                  │
│ （誠實）    │                       │  FedAvg /        │
├─────────────┤     被投毒之更新       │  Robust Agg      │
│ 參與者 3    │ ───────────────────── │                  │
│ （惡意）    │                       │  ──── 全域 ──▶   │
├─────────────┤     模型更新           │       模型       │
│ 參與者 N    │ ───────────────────── │                  │
│ （誠實）    │                       └──────────────────┘
└─────────────┘

def federated_average(participant_updates: list[dict], weights: list[float]) -> dict:
    """FedAvg：參與者模型更新之加權平均。
    易受投毒，因它等級地信任所有更新。"""
    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,
):
    """打造將全域模型朝目標行為轉移之被投毒更新。"""
    # 步驟 1：計算朝目標行為之梯度
    malicious_model.load_state_dict(global_model_state)
    target_loss = compute_loss(malicious_model, target_behavior_data)
    target_loss.backward()
 
    # 步驟 2：計算更新 delta
    poisoned_update = {}
    for name, param in malicious_model.named_parameters():
        clean_delta = param.data - global_model_state[name]
        target_delta = -param.grad  # 朝目標之梯度下降
 
        # 將乾淨行為與被投毒目標混合
        poisoned_update[name] = clean_delta + scaling_factor * target_delta
 
    # 步驟 3：縮放以主宰聚合
    # 具 N 個參與者時，以 N 縮放確保被投毒更新
    # 超過所有誠實參與者之總和
    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,
):
    """計算經 FedAvg 後將全域模型
    替換為攻擊者目標模型之更新。"""
    replacement_update = {}
    for name in target_model_state:
        # 經 FedAvg 後：new_global = global + lr * avg(updates)
        # 我們要：new_global = target_model
        # 故：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
):
    """受約束以通過以 norm 為本偵測之後門更新。"""
    update = {}
    for name in backdoored_model.state_dict():
        update[name] = backdoored_model.state_dict()[name] - global_model.state_dict()[name]
 
    # 計算更新 norm
    total_norm = torch.sqrt(sum(
        (update[name] ** 2).sum() for name in update
    ))
 
    # 必要時縮小以保持於偵測門檻內
    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：選擇最接近多數之更新。
    對 < 50% 惡意參與者為 Byzantine 穩健。"""
    n = len(updates)
    scores = []
 
    for i in range(n):
        # 計算至所有其他更新之距離
        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())
 
        # 分數 = n - num_byzantine - 2 個最近距離之總和
        distances.sort()
        score = sum(distances[:n - num_byzantine - 2])
        scores.append(score)
 
    # 選擇最低分數（最接近多數）之更新
    best_idx = scores.index(min(scores))
    return updates[best_idx]