Secagg Manager

@abstractmethod
def add(self, secagg_id: str, parties: List[str], context: Dict[str, int], experiment_id: Union[str, None]):
    """Add a new data entry in component secagg element database"""

@abstractmethod
def get(self, secagg_id: str, experiment_id: Union[str, None]):
    """Search for a data entry in component secagg element database"""

@abstractmethod
def remove(self, secagg_id: str, experiment_id: Union[str, None]) -> bool:
    """Remove a data entry from component secagg element database"""

def add(
        self,
        secagg_id: str,
        parties: List[str],
        context: Dict[str, int],
        experiment_id: None = None
) -> None:
    """Add a new data entry for a context element in the secagg table

    Check that no entry exists yet for this `secagg_id` in the table.

    Args:
        secagg_id: secure aggregation ID key of the entry
        parties: list of parties participating in this secagg context element
        context: the (full) biprime number shared with other parties
        experiment_id: unused argument
    """
    # Trust argument type and value check from calling class (`SecaggSetup`, `Node`)
    self._add_generic(
        SecaggElementTypes.BIPRIME,
        secagg_id,
        parties,
        {
            'context': context,
            'type': BiprimeType.DYNAMIC.value
        }
    )

def get(self, secagg_id: str, experiment_id: None = None) -> Union[dict, None]:
    """Search for data entry with given `secagg_id` in the secagg table

    Check that there is at most one entry with this unique secagg ID.

    Args:
        secagg_id: secure aggregation ID key to search
        experiment_id: unused argument.

    Returns:
        A dict containing all values for the secagg element for this `secagg_id` if it exists,
            or None if no element exists for this `secagg_id`
    """
    # Trust argument type and value check from calling class (`SecaggSetup`, `Node`)
    element = self._get_generic(secagg_id)
    self._raise_error_incompatible_requested_entry(element,
                                                   SecaggElementTypes.BIPRIME,
                                                   secagg_id,
                                                   None,  # a biprime is not associated to a specific experiment
                                                   'getting')

    # type is internal to this class, need not transmit to caller
    if isinstance(element, dict) and 'type' in element:
        # `deepcopy` avoids  any risk of error related to database implementation
        element = copy.deepcopy(element)
        del element['type']

    return element

def is_default_biprime(self, secagg_id: str) -> bool:
    """Search for default (non dynamic) data entry with given `secagg_id` in the biprime table.

    Args:
        secagg_id: secure aggregation ID key to search

    Returns:
        True if a default biprime entry exists for this `secagg_id`, False if not
    """
    # Trust argument type and value check from calling class (`SecaggSetup`, `Node`)
    element = self._get_generic(secagg_id)
    return isinstance(element, dict) and 'type' in element and element['type'] == BiprimeType.DEFAULT.value

def remove(self, secagg_id: str, experiment_id: None = None) -> bool:
    """Remove data entry for this `secagg_id` from the secagg table

    Args:
        secagg_id: secure aggregation ID key of the entry
        experiment_id: unused argument

    Returns:
        True if an entry existed (and was removed) for this `secagg_id`,
            False if no entry existed for this `secagg_id`

    Raises:
        FedbiomedSecaggError: trying to remove a non-dynamic biprime

    """
    # Trust argument type and value check from calling class (`SecaggSetup`, `Node`)

    # Can only remove dynamic biprimes
    element = self._get_generic(secagg_id)
    if isinstance(element, dict) and ('type' not in element or element['type'] != BiprimeType.DYNAMIC.value):
        errmess = f'{ErrorNumbers.FB623.value}: not authorized to remove non-dynamic biprime "{secagg_id}"'
        logger.error(errmess)
        raise FedbiomedSecaggError(errmess)

    return self._remove_generic(secagg_id, SecaggElementTypes.BIPRIME)

def update_default_biprimes(self, allow_default_biprimes: bool, default_biprimes_dir: str) -> None:
    """Update the default entries in the biprime table.

    If `allow_default_biprimes` is True, then add or update the default biprimes from the *.json
        files in `default_biprimes_dir` directory.

    In all cases, remove the other default biprimes existing in the biprime table.

    Args:
        allow_default_biprimes: if True, then accept default biprimes from files
        default_biprimes_dir: directory containing the default biprimes files

    Raises:
        FedbiomedSecaggError: cannot update default biprimes
    """
    # Read and check the new proposed default biprime values from files
    if allow_default_biprimes:
        default_biprimes_new = self._read_default_biprimes(default_biprimes_dir)
    else:
        default_biprimes_new = []

    # Read the existing default biprimes in DB
    try:
        default_biprimes_current = self._table.search(
            self._query.type.exists() &
            (self._query.type == BiprimeType.DEFAULT.value)
        )
    except Exception as e:
        errmess = f'{ErrorNumbers.FB623.value}: database search operation failed for default biprimes: {e}'
        logger.error(errmess)
        raise FedbiomedSecaggError(errmess)

    # Remove existing default biprimes not in the new proposed values
    bp_new_ids = set(bp['secagg_id'] for bp in default_biprimes_new)
    bp_current_ids = set(bp['secagg_id'] for bp in default_biprimes_current)
    bp_remove_ids = list(bp_current_ids - bp_new_ids)

    try:
        self._table.remove(self._query.secagg_id.one_of(bp_remove_ids))
    except Exception as e:
        errmess = f'{ErrorNumbers.FB623.value}: database remove operation failed for ' \
                  f'obsolete default biprimes {bp_remove_ids}: {e}'
        logger.error(errmess)
        raise FedbiomedSecaggError(errmess)

    # Save or update the new default biprimes
    for bp in default_biprimes_new:
        try:
            self._table.upsert(
                {
                    'secagg_version': bp['secagg_version'],
                    'secagg_id': bp['secagg_id'],
                    'parties': None,
                    'secagg_elem': SecaggElementTypes.BIPRIME.value,
                    'type': BiprimeType.DEFAULT.value,
                    'context': {
                        'biprime': bp['biprime'],
                        'max_keysize': bp['max_keysize']
                    },
                },
                self._query.secagg_id == bp['secagg_id']
            )
        except Exception as e:
            errmess = f'{ErrorNumbers.FB623.value}: database upsert operation failed for ' \
                      f'default biprime {bp["secagg_id"]}: {e}'
            logger.error(errmess)
            raise FedbiomedSecaggError(errmess)

def add(self, secagg_id: str, parties: List[str], context: Dict[str, bytes], experiment_id: str):
    """Add a new data entry for a context element in the secagg table

    Check that no entry exists yet for this `secagg_id` in the table.

    Args:
        secagg_id: secure aggregation ID key of the entry
        parties: list of parties participating in this secagg context element
        experiment_id: ID of the experiment to which this secagg context element is attached
        context: server key part held by this party
    """
    # Save key pairs as `str`` since it is the format support by JSON. Need to convert to `base64` first
    context_json = {node_id: str(base64.b64encode(key), 'utf-8') for node_id, key in context.items()}

    self._add_generic(
        SecaggElementTypes.DIFFIE_HELLMAN,
        secagg_id,
        parties,
        {'experiment_id': experiment_id, 'context': context_json}
    )

def get(self, secagg_id: str, experiment_id: str) -> Union[dict, None]:
    """Search for data entry with given `secagg_id`

    Check that there is at most one entry with this unique secagg ID.

    If there is an entry for this `secagg_id`, check it is associated with experiment `experiment_id`

    Args:
        secagg_id: secure aggregation ID key to search
        experiment_id: the experiment ID associated with the secagg entry

    Returns:
        A dict containing all values for the secagg element for this `secagg_id` if it exists,
            or None if no element exists for this `secagg_id`
    """
    # Trust argument type and value check from calling class (`SecaggSetup`, `Node`)
    element = self._get_generic(secagg_id)
    self._raise_error_incompatible_requested_entry(element,
                                                   SecaggElementTypes.DIFFIE_HELLMAN,
                                                   secagg_id,
                                                   experiment_id,
                                                   'getting')

    if element:
        # Need to convert to keys as bytes
        context_bytes = {
            node_id: bytes(base64.b64decode(key)) \
                for node_id, key in element['context'].items()}
        element['context'] = context_bytes

    return element

def remove(self, secagg_id: str, experiment_id: str) -> bool:
    """Remove data entry for this `secagg_id` from the secagg table

    Check that the experiment ID for the table entry and the current experiment match

    Args:
        secagg_id: secure aggregation ID key of the entry
        experiment_id: experiment ID of the current experiment

    Returns:
        True if an entry existed (and was removed) for this `secagg_id`,
            False if no entry existed for this `secagg_id`

    Raises:
        FedbiomedSecaggError: database entry does not belong to `experiment_id`
    """
    # Trust argument type and value check from calling class for `secagg_id` (`SecaggSetup`, but not `Node`)
    # Don't trust `Node` for `experiment_id` type (may give `None`) but this is not an issue
    element = self._get_generic(secagg_id)
    self._raise_error_incompatible_requested_entry(element,
                                                   SecaggElementTypes.DIFFIE_HELLMAN,
                                                   secagg_id,
                                                   experiment_id,
                                                   'removing')
    return self._remove_generic(secagg_id, SecaggElementTypes.DIFFIE_HELLMAN)

def add(self, secagg_id: str, parties: List[str], context: Dict[str, int], experiment_id: str):
    """Add a new data entry for a context element in the secagg table

    Check that no entry exists yet for this `secagg_id` in the table.

    Args:
        secagg_id: secure aggregation ID key of the entry
        parties: list of parties participating in this secagg context element
        context: server key part held by this party
        experiment_id: ID of the experiment to which this secagg context element is attached
    """

    # Trust argument type and value check from calling class (`SecaggSetup`, but not `Node`)
    self._add_generic(
        SecaggElementTypes.SERVER_KEY,
        secagg_id,
        parties,
        {'experiment_id': experiment_id, 'context': context}
    )

def get(self, secagg_id: str, experiment_id: str) -> Union[dict, None]:
    """Search for data entry with given `secagg_id`

    Check that there is at most one entry with this unique secagg ID.

    If there is an entry for this `secagg_id`, check it is associated with experiment `experiment_id`

    Args:
        secagg_id: secure aggregation ID key to search
        experiment_id: the experiment ID associated with the secagg entry

    Returns:
        A dict containing all values for the secagg element for this `secagg_id` if it exists,
            or None if no element exists for this `secagg_id`
    """

    # Trust argument type and value check from calling class (`SecaggSetup`, `Node`)
    element = self._get_generic(secagg_id)
    self._raise_error_incompatible_requested_entry(element,
                                                   SecaggElementTypes.SERVER_KEY,
                                                   secagg_id,
                                                   experiment_id,
                                                   'getting')

    return element

def remove(self, secagg_id: str, experiment_id: str) -> bool:
    """Remove data entry for this `secagg_id` from the secagg table

    Check that the experiment ID for the table entry and the current experiment match

    Args:
        secagg_id: secure aggregation ID key of the entry
        experiment_id: experiment ID of the current experiment

    Returns:
        True if an entry existed (and was removed) for this `secagg_id`,
            False if no entry existed for this `secagg_id`
    """

    # Trust argument type and value check from calling class for `secagg_id` (`SecaggSetup`, but not `Node`)
    # Don't trust `Node` for `experiment_id` type (may give `None`) but this is not an issue
    element = self._get_generic(secagg_id)
    self._raise_error_incompatible_requested_entry(element,
                                                   SecaggElementTypes.SERVER_KEY,
                                                   secagg_id,
                                                   experiment_id,
                                                   'removing')
    return self._remove_generic(secagg_id, SecaggElementTypes.SERVER_KEY)