Merge pull request #529 from FedML-AI/test/v0.7.0

Test/v0.7.0

python/examples/cross_silo/mqtt_s3_fedavg_defense_mnist_lr_example/config/crfl/fedml_config.yaml

@@ -0,0 +1,66 @@
+common_args:
+  training_type: "cross_silo"
+  scenario: "horizontal"
+  using_mlops: false
+  random_seed: 0
+  config_version: release
+
+environment_args:
+  bootstrap: config/bootstrap.sh
+
+data_args:
+  dataset: "mnist"
+  data_cache_dir: ~/fedml_data
+  partition_method: "hetero"
+  partition_alpha: 0.5
+
+model_args:
+  model: "lr"
+  model_file_cache_folder: "./model_file_cache" # will be filled by the server automatically
+  global_model_file_path: "./model_file_cache/global_model.pt"
+
+train_args:
+  federated_optimizer: "FedAvg"
+  # for CLI running, this can be None; in MLOps deployment, `client_id_list` will be replaced with real-time selected devices
+  client_id_list:
+  # for FoolsGold Defense, if use_memory is true, then client_num_in_total should be equal to client_number_per_round
+  client_num_in_total: 1000
+  client_num_per_round: 4
+  comm_round: 10
+  epochs: 1
+  batch_size: 10
+  client_optimizer: sgd
+  learning_rate: 0.03
+  weight_decay: 0.001
+
+validation_args:
+  frequency_of_the_test: 1
+
+device_args:
+  worker_num: 4
+  using_gpu: false
+  gpu_mapping_file: config/gpu_mapping.yaml
+  gpu_mapping_key: mapping_default
+
+comm_args:
+  backend: "MQTT_S3"
+  mqtt_config_path:
+  s3_config_path:
+  grpc_ipconfig_path: ./config/grpc_ipconfig.csv
+
+tracking_args:
+  # the default log path is at ~/fedml-client/fedml/logs/ and ~/fedml-server/fedml/logs/
+  enable_wandb: false
+  wandb_key: ee0b5f53d949c84cee7decbe7a629e63fb2f8408
+  wandb_project: fedml
+  wandb_name: fedml_torch_fedavg_mnist_lr
+
+attack_args:
+  enable_attack: false
+  attack_type: None
+
+defense_args:
+  enable_defense: true
+  defense_type: crfl
+  federated_optimizer: FedAvg
+  sigma: 0.02

python/fedml/core/alg_frame/server_aggregator.py

@@ -57,6 +57,8 @@ def on_after_aggregation(self, aggregated_model_or_grad: Dict) -> Dict:
         if FedMLDifferentialPrivacy.get_instance().is_cdp_enabled():
             logging.info("-----add central DP noise ----")
             aggregated_model_or_grad = FedMLDifferentialPrivacy.get_instance().add_noise(aggregated_model_or_grad)
+        if FedMLDefender.get_instance().is_defense_enabled():
+            aggregated_model_or_grad = FedMLDefender.get_instance().defend_after_aggregation(aggregated_model_or_grad)
         return aggregated_model_or_grad
 
     @abstractmethod

python/fedml/core/dp/fed_privacy_mechanism.py

@@ -1,6 +1,5 @@
 import logging
 from fedml.core.dp.budget_accountant import BudgetAccountant
-from fedml.core.dp.common.utils import check_params
 from fedml.core.dp.mechanisms import Laplace, Gaussian
 
 
@@ -14,9 +13,11 @@ def get_instance():
         return FedMLDifferentialPrivacy._dp_instance
 
     def __init__(self):
+        self.epsilon = None
         self.is_dp_enabled = False
         self.dp_type = None
         self.dp = None
+        self.enable_accountant = False
 
     def init(self, args):
         if hasattr(args, "enable_dp") and args.enable_dp:
@@ -26,9 +27,10 @@ def init(self, args):
             self.is_dp_enabled = True
             self.mechanism_type = args.mechanism_type.lower()
             self.dp_type = args.dp_type.lower().strip()
-            check_params(args.epsilon, args.delta, args.sensitivity)
-            self.epsilon = args.epsilon
-            self.delta = args.delta
+
+            if hasattr(args, "accountant_type") and args.accountant_type in ["adding"]:
+                self.enable_accountant = True
+                self.epsilon = args.epsilon
             if self.dp_type not in ["cdp", "ldp"]:
                 raise ValueError(
                     "DP type can only be cdp (for central DP) and ldp (for local DP)! "
@@ -38,9 +40,7 @@ def init(self, args):
                     epsilon=args.epsilon, delta=args.delta, sensitivity=args.sensitivity
                 )
             elif self.mechanism_type == "gaussian":
-                self.dp = Gaussian(
-                    epsilon=args.epsilon, delta=args.delta, sensitivity=args.sensitivity
-                )
+                self.dp = Gaussian(args)
             else:
                 raise NotImplementedError("DP mechanism not implemented!")
             self.accountant = BudgetAccountant()
@@ -61,7 +61,8 @@ def add_noise(self, grad):
         new_grad = dict()
         for k in grad.keys():
             new_grad[k] = self._compute_new_grad(grad[k])
-        self.accountant.spend(epsilon=self.epsilon, delta=0)
+        if self.enable_accountant:
+            self.accountant.spend(epsilon=self.epsilon, delta=0)
         return new_grad
 
     def _compute_new_grad(self, grad):

python/fedml/core/dp/mechanisms/gaussian.py

@@ -1,21 +1,34 @@
 import numpy as np
 import torch
 from .base_dp_mechanism import BaseDPMechanism
+from ..common.utils import check_params
 
 
 class Gaussian(BaseDPMechanism):
-    def __init__(self, *, epsilon, delta, sensitivity):
-        if epsilon == 0 or delta == 0:
-            raise ValueError("Neither Epsilon nor Delta can be zero")
-        if epsilon > 1.0:
-            raise ValueError(
-                "Epsilon cannot be greater than 1. "
+    def __init__(self, args):
+        if hasattr(args, "sigma") and isinstance(args.sigma, float):
+            self._scale = args.sigma
+        elif hasattr(args, "epsilon") and hasattr(args, "delta") and hasattr(args, "sensitivity"):
+            check_params(args.epsilon, args.delta, args.sensitivity)
+            if args.epsilon == 0 or args.delta == 0:
+                raise ValueError("Neither Epsilon nor Delta can be zero")
+            if args.epsilon > 1.0:
+                raise ValueError(
+                    "Epsilon cannot be greater than 1. "
+                )
+            self._scale = (
+                np.sqrt(2 * np.log(1.25 / float(args.delta)))
+                * float(args.sensitivity)
+                / float(args.epsilon)
             )
-        self._scale = (
-            np.sqrt(2 * np.log(1.25 / float(delta)))
-            * float(sensitivity)
-            / float(epsilon)
-        )
+        else:
+            raise ValueError("Missing necessary parameters for Gaussian Mechanism")
+
+    @classmethod
+    def add_noise_using_sigma(cls, sigma, size):
+        if not isinstance(sigma, float):
+            raise ValueError("sigma should be a float")
+        return torch.normal(mean=0, std=sigma, size=size)
 
     def compute_noise(self, size):
         return torch.normal(mean=0, std=self._scale, size=size)

python/fedml/core/dp/mechanisms/laplace.py

@@ -1,6 +1,7 @@
 import numpy as np
 import torch
 from .base_dp_mechanism import BaseDPMechanism
+from ..common.utils import check_params
 
 
 class Laplace(BaseDPMechanism):
@@ -9,6 +10,7 @@ class Laplace(BaseDPMechanism):
     """
 
     def __init__(self, *, epsilon, delta=0.0, sensitivity):
+        check_params(epsilon, delta, sensitivity)
         self.scale = float(sensitivity) / (float(epsilon) - np.log(1 - float(delta)))
 
     def compute_noise(self, size):

python/fedml/core/security/common/utils.py

@@ -22,16 +22,13 @@ def is_weight_param(k):
 def compute_euclidean_distance(v1, v2):
     return (v1 - v2).norm()
 
+def compute_model_norm(model):
+    return vectorize_weight(model).norm()
 
 def compute_middle_point(alphas, model_list):
     """
-
-    Args:
         alphas: weights of model_dict
         model_dict: a model submitted by a user
-
-    Returns:
-
     """
     sum_batch = torch.zeros(model_list[0].shape)
     for a, a_batch_w in zip(alphas, model_list):

python/fedml/core/security/constants.py

@@ -8,6 +8,7 @@
 DEFENSE_DP = "dp"
 DEFENSE_RFA = "rfa"
 DEFENSE_FOOLSGOLD = "foolsgold"
+DEFENSE_CRFL = "crfl"
 
 ATTACK_METHOD_BYZANTINE_ATTACK = "byzantine"
 ATTACK_METHOD_DLG = "dlg"

python/fedml/core/security/defense/cclip_defense.py

@@ -1,8 +1,5 @@
-import random
 from typing import Callable, List, Tuple, Dict, Any
-
 import numpy as np
-
 from .defense_base import BaseDefenseMethod
 from ..common import utils
 from ..common.bucket import Bucket

python/fedml/core/security/defense/crfl_defense.py

@@ -0,0 +1,87 @@
+from .defense_base import BaseDefenseMethod
+from typing import Callable, List, Tuple, Dict, Any
+from ..common import utils
+from ...dp.mechanisms import Gaussian
+
+"""
+CRFL: Certifiably Robust Federated Learning against Backdoor Attacks (ICML 2021)
+http://proceedings.mlr.press/v139/xie21a/xie21a.pdf
+"""
+
+
+class CRFLDefense(BaseDefenseMethod):
+    def __init__(self, config):
+        self.config = config
+        self.epoch = 1
+        if hasattr(config, "clip_threshold"):
+            self.clip_threshold = config.clip_threshold
+        else:
+            self.clip_threshold = None
+        if hasattr(config, "sigma") and isinstance(config.sigma, float):
+            self.sigma = config.sigma
+        else:
+            self.sigma = 0.01  # in the code of CRFL, the author set sigma to 0.01
+
+    def run(
+            self,
+            raw_client_grad_list: List[Tuple[float, Dict]],
+            base_aggregation_func: Callable = None,
+            extra_auxiliary_info: Any = None,
+    ):
+        new_grad_list = self.defend_before_aggregation(
+            raw_client_grad_list, extra_auxiliary_info
+        )
+        avg_params = self.defend_on_aggregation(new_grad_list, base_aggregation_func)
+        return self.defend_after_aggregation(avg_params)
+
+    def defend_before_aggregation(
+            self,
+            raw_client_grad_list: List[Tuple[float, Dict]],
+            extra_auxiliary_info: Any = None,
+    ):
+        return raw_client_grad_list
+
+    def defend_on_aggregation(
+            self,
+            raw_client_grad_list: List[Tuple[float, Dict]],
+            base_aggregation_func: Callable = None,
+            extra_auxiliary_info: Any = None,
+    ):
+        avg_params = base_aggregation_func(args=self.config, raw_grad_list=raw_client_grad_list)
+        """
+        clip the global model; dynamic threshold is adjusted according to the dataset;
+        in the experiment, the authors set the dynamic threshold as follows:
+            dataset == MNIST: dynamic_thres = epoch * 0.1 + 2
+            dataseet == LOAN: dynamic_thres = epoch * 0.025 + 2
+            datset == EMNIST: dynamic_thres = epoch * 0.25 + 4
+        """
+        print(f"avg params = {avg_params}")
+        dynamic_threshold = self.epoch * 0.1 + 2
+        if self.clip_threshold is None or self.clip_threshold > dynamic_threshold:
+            self.clip_threshold = dynamic_threshold
+        self.epoch += 1
+
+        print(f"self.clip_threshold={self.clip_threshold}")
+        new_model = self.clip_weight_norm(avg_params, self.clip_threshold)
+        # the output model is new model; later the algo adds dp noise to the global model
+        return new_model
+
+    def defend_after_aggregation(self, global_model):
+        # todo: to discuss with chaoyang: the output is the clipped model (real model);
+        # add dp noise to the real model and sent the permuted model to clients; how to get the last iteration?
+        new_global_model = dict()
+        for k in global_model.keys():
+            new_global_model[k] = global_model[k] + Gaussian.add_noise_using_sigma(self.sigma, global_model[k].shape)
+        return new_global_model
+
+    @staticmethod
+    def clip_weight_norm(model, clip_threshold):
+        total_norm = utils.compute_model_norm(model)
+        print(f"total_norm = {total_norm}")
+        if total_norm > clip_threshold:
+            clip_coef = clip_threshold / (total_norm + 1e-6)
+            new_model = dict()
+            for k in model.keys():
+                new_model[k] = model[k] * clip_coef
+            return new_model
+        return model