multiple collaborators per certificate

Signed-off-by: hasan7n <[email protected]>

Author: hasan7n

openfl/component/aggregator/aggregator.py

@@ -40,6 +40,7 @@ def __init__(self,
                  aggregator_uuid,
                  federation_uuid,
                  authorized_cols,
+                 cn_mapping,
 
                  init_state_path,
                  best_state_path,
@@ -75,6 +76,7 @@ def __init__(self,
 
         # if the collaborator requests a delta, this value is set to true
         self.authorized_cols = authorized_cols
+        self.cn_mapping = cn_mapping
         self.uuid = aggregator_uuid
         self.federation_uuid = federation_uuid
         self.assigner = assigner
@@ -225,7 +227,7 @@ def valid_collaborator_cn_and_id(self, cert_common_name,
         # FIXME: '' instead of None is just for protobuf compatibility.
         #  Cleaner solution?
         if self.single_col_cert_common_name == '':
-            return (cert_common_name == collaborator_common_name
+            return (cert_common_name == self.cn_mapping[collaborator_common_name]
                     and collaborator_common_name in self.authorized_cols)
         # otherwise, common_name must be in whitelist and
         # collaborator_common_name must be in authorized_cols

openfl/federated/plan/plan.py

@@ -135,9 +135,15 @@ def parse(plan_config_path: Path, cols_config_path: Path = None,
                 gandlf_config['output_dir'] = gandlf_config.get('output_dir', '.')
                 plan.config['task_runner']['settings']['gandlf_config'] = gandlf_config
 
-            plan.authorized_cols = Plan.load(cols_config_path).get(
+            cols_info = Plan.load(cols_config_path).get(
                 'collaborators', []
             )
+            if isinstance(cols_info, list):
+                plan.cn_mapping = {col: col for col in cols_info}
+                plan.authorized_cols = cols_info
+            else:
+                plan.cn_mapping = cols_info
+                plan.authorized_cols = list(cols_info.keys())
 
             # TODO: Does this need to be a YAML file? Probably want to use key
             #  value as the plan hash
@@ -223,6 +229,7 @@ def __init__(self):
         """Initialize."""
         self.config = {}  # dictionary containing patched plan definition
         self.authorized_cols = []  # authorized collaborator list
+        self.cn_mapping = {}  # expected cert common name for each collaborator
         self.cols_data_paths = {}  # collaborator data paths dict
 
         self.collaborator_ = None  # collaborator object
@@ -338,6 +345,7 @@ def get_aggregator(self, tensor_dict=None):
         defaults[SETTINGS]['aggregator_uuid'] = self.aggregator_uuid
         defaults[SETTINGS]['federation_uuid'] = self.federation_uuid
         defaults[SETTINGS]['authorized_cols'] = self.authorized_cols
+        defaults[SETTINGS]['cn_mapping'] = self.cn_mapping
         defaults[SETTINGS]['assigner'] = self.get_assigner()
         defaults[SETTINGS]['compression_pipeline'] = self.get_tensor_pipe()
         defaults[SETTINGS]['straggler_handling_policy'] = self.get_straggler_handling_policy()

openfl/interface/collaborator.py

@@ -137,6 +137,7 @@ def register_data_path(collaborator_name, data_path=None, silent=False):
 def generate_cert_request_(collaborator_name,
                            silent, skip_package):
     """Generate certificate request for the collaborator."""
+    # TODO: this should take an extra argument: common_name
     generate_cert_request(collaborator_name, silent, skip_package)
 
 
@@ -304,7 +305,7 @@ def certify(collaborator_name, silent, request_pkg=None, import_=False):
     from openfl.utilities.utils import rmtree
 
     common_name = f'{collaborator_name}'.lower()
-
+    # TODO: read and parse CSR and use the actual CN
     if not import_:
         if request_pkg:
             Path(f'{CERT_DIR}/client').mkdir(parents=True, exist_ok=True)

openfl/interface/interactive_api/experiment.py

@@ -377,6 +377,7 @@ def _prepare_plan(self, model_provider, data_loader,
         self.plan.authorized_cols = [
             name for name, info in shard_registry.items() if info['is_online']
         ]
+        self.plan.cn_mapping = {name:name for name in self.plan.authorized_cols}
         # Network part of the plan
         # We keep in mind that an aggregator FQND will be the same as the directors FQDN
         # We just choose a port randomly from plan hash

openfl/transport/grpc/aggregator_server.py

@@ -84,7 +84,7 @@ def validate_collaborator(self, request, context):
                     common_name, collaborator_common_name):
                 # Random delay in authentication failures
                 sleep(5 * random())  # nosec
-                context.abort(
+                context.abort(  # TODO? add the expected CN in the error msg
                     StatusCode.UNAUTHENTICATED,
                     f'Invalid collaborator. CN: |{common_name}| '
                     f'collaborator_common_name: |{collaborator_common_name}|')