# Copyright 2025 Flower Labs GmbH. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""Message-based Central differential privacy with fixed clipping.
Papers: https://arxiv.org/abs/1712.07557, https://arxiv.org/abs/1710.06963
"""
from abc import ABC
from collections import OrderedDict
from collections.abc import Iterable
from logging import INFO, WARNING
from typing import Optional
from flwr.common import Array, ArrayRecord, ConfigRecord, Message, MetricRecord, log
from flwr.common.differential_privacy import (
add_gaussian_noise_inplace,
compute_clip_model_update,
compute_stdv,
)
from flwr.common.differential_privacy_constants import (
CLIENTS_DISCREPANCY_WARNING,
KEY_CLIPPING_NORM,
)
from flwr.server import Grid
from .strategy import Strategy
class DifferentialPrivacyFixedClippingBase(Strategy, ABC):
"""Base class for DP strategies with fixed clipping.
This class contains common functionality shared between server-side and
client-side fixed clipping implementations.
Parameters
----------
strategy : Strategy
The strategy to which DP functionalities will be added by this wrapper.
noise_multiplier : float
The noise multiplier for the Gaussian mechanism for model updates.
A value of 1.0 or higher is recommended for strong privacy.
clipping_norm : float
The value of the clipping norm.
num_sampled_clients : int
The number of clients that are sampled on each round.
"""
# pylint: disable=too-many-arguments,too-many-instance-attributes
def __init__(
self,
strategy: Strategy,
noise_multiplier: float,
clipping_norm: float,
num_sampled_clients: int,
) -> None:
super().__init__()
self.strategy = strategy
if noise_multiplier < 0:
raise ValueError("The noise multiplier should be a non-negative value.")
if clipping_norm <= 0:
raise ValueError("The clipping norm should be a positive value.")
if num_sampled_clients <= 0:
raise ValueError(
"The number of sampled clients should be a positive value."
)
self.noise_multiplier = noise_multiplier
self.clipping_norm = clipping_norm
self.num_sampled_clients = num_sampled_clients
def _add_noise_to_aggregated_arrays(
self, aggregated_arrays: ArrayRecord
) -> ArrayRecord:
"""Add Gaussian noise to aggregated arrays.
Parameters
----------
aggregated_arrays : ArrayRecord
The aggregated arrays to add noise to.
Returns
-------
ArrayRecord
The aggregated arrays with noise added.
"""
aggregated_ndarrays = aggregated_arrays.to_numpy_ndarrays()
stdv = compute_stdv(
self.noise_multiplier, self.clipping_norm, self.num_sampled_clients
)
add_gaussian_noise_inplace(aggregated_ndarrays, stdv)
log(
INFO,
"aggregate_fit: central DP noise with %.4f stdev added",
stdv,
)
return ArrayRecord(
OrderedDict(
{
k: Array(v)
for k, v in zip(aggregated_arrays.keys(), aggregated_ndarrays)
}
)
)
def configure_evaluate(
self, server_round: int, arrays: ArrayRecord, config: ConfigRecord, grid: Grid
) -> Iterable[Message]:
"""Configure the next round of federated evaluation."""
return self.strategy.configure_evaluate(server_round, arrays, config, grid)
def aggregate_evaluate(
self,
server_round: int,
replies: Iterable[Message],
) -> Optional[MetricRecord]:
"""Aggregate MetricRecords in the received Messages."""
return self.strategy.aggregate_evaluate(server_round, replies)
def summary(self) -> None:
"""Log summary configuration of the strategy."""
self.strategy.summary()
[docs]
class DifferentialPrivacyServerSideFixedClipping(DifferentialPrivacyFixedClippingBase):
"""Strategy wrapper for central DP with server-side fixed clipping.
Parameters
----------
strategy : Strategy
The strategy to which DP functionalities will be added by this wrapper.
noise_multiplier : float
The noise multiplier for the Gaussian mechanism for model updates.
A value of 1.0 or higher is recommended for strong privacy.
clipping_norm : float
The value of the clipping norm.
num_sampled_clients : int
The number of clients that are sampled on each round.
Examples
--------
Create a strategy::
strategy = fl.serverapp.FedAvg( ... )
Wrap the strategy with the `DifferentialPrivacyServerSideFixedClipping` wrapper::
dp_strategy = DifferentialPrivacyServerSideFixedClipping(
strategy, cfg.noise_multiplier, cfg.clipping_norm, cfg.num_sampled_clients
)
"""
def __init__(
self,
strategy: Strategy,
noise_multiplier: float,
clipping_norm: float,
num_sampled_clients: int,
) -> None:
super().__init__(strategy, noise_multiplier, clipping_norm, num_sampled_clients)
self.current_arrays: ArrayRecord = ArrayRecord()
def __repr__(self) -> str:
"""Compute a string representation of the strategy."""
return "Differential Privacy Strategy Wrapper (Server-Side Fixed Clipping)"
[docs]
def summary(self) -> None:
"""Log summary configuration of the strategy."""
log(INFO, "\t├──> DP settings:")
log(INFO, "\t│\t├── Noise multiplier: %s", self.noise_multiplier)
log(INFO, "\t│\t└── Clipping norm: %s", self.clipping_norm)
super().summary()
[docs]
def aggregate_train(
self,
server_round: int,
replies: Iterable[Message],
) -> tuple[Optional[ArrayRecord], Optional[MetricRecord]]:
"""Aggregate ArrayRecords and MetricRecords in the received Messages."""
if not validate_replies(replies, self.num_sampled_clients):
return None, None
# Clip arrays in replies
current_ndarrays = self.current_arrays.to_numpy_ndarrays()
for reply in replies:
for arr_name, record in reply.content.array_records.items():
# Clip
reply_ndarrays = record.to_numpy_ndarrays()
compute_clip_model_update(
param1=reply_ndarrays,
param2=current_ndarrays,
clipping_norm=self.clipping_norm,
)
# Replace content while preserving keys
reply.content[arr_name] = ArrayRecord(
OrderedDict(
{k: Array(v) for k, v in zip(record.keys(), reply_ndarrays)}
)
)
log(
INFO,
"aggregate_fit: parameters are clipped by value: %.4f.",
self.clipping_norm,
)
# Pass the new parameters for aggregation
aggregated_arrays, aggregated_metrics = self.strategy.aggregate_train(
server_round, replies
)
# Add Gaussian noise to the aggregated arrays
if aggregated_arrays:
aggregated_arrays = self._add_noise_to_aggregated_arrays(aggregated_arrays)
return aggregated_arrays, aggregated_metrics
[docs]
class DifferentialPrivacyClientSideFixedClipping(DifferentialPrivacyFixedClippingBase):
"""Strategy wrapper for central DP with client-side fixed clipping.
Use `fixedclipping_mod` modifier at the client side.
In comparison to `DifferentialPrivacyServerSideFixedClipping`,
which performs clipping on the server-side,
`DifferentialPrivacyClientSideFixedClipping` expects clipping to happen
on the client-side, usually by using the built-in `fixedclipping_mod`.
Parameters
----------
strategy : Strategy
The strategy to which DP functionalities will be added by this wrapper.
noise_multiplier : float
The noise multiplier for the Gaussian mechanism for model updates.
A value of 1.0 or higher is recommended for strong privacy.
clipping_norm : float
The value of the clipping norm.
num_sampled_clients : int
The number of clients that are sampled on each round.
Examples
--------
Create a strategy::
strategy = fl.serverapp.FedAvg(...)
Wrap the strategy with the `DifferentialPrivacyClientSideFixedClipping` wrapper::
dp_strategy = DifferentialPrivacyClientSideFixedClipping(
strategy, cfg.noise_multiplier, cfg.clipping_norm, cfg.num_sampled_clients
)
On the client, add the `fixedclipping_mod` to the client-side mods::
app = fl.client.ClientApp(mods=[fixedclipping_mod])
"""
def __repr__(self) -> str:
"""Compute a string representation of the strategy."""
return "Differential Privacy Strategy Wrapper (Client-Side Fixed Clipping)"
[docs]
def summary(self) -> None:
"""Log summary configuration of the strategy."""
log(INFO, "\t├──> DP settings:")
log(INFO, "\t│\t├── Noise multiplier: %s", self.noise_multiplier)
log(INFO, "\t│\t└── Clipping norm: %s", self.clipping_norm)
super().summary()
[docs]
def aggregate_train(
self,
server_round: int,
replies: Iterable[Message],
) -> tuple[Optional[ArrayRecord], Optional[MetricRecord]]:
"""Aggregate ArrayRecords and MetricRecords in the received Messages."""
if not validate_replies(replies, self.num_sampled_clients):
return None, None
# Aggregate
aggregated_arrays, aggregated_metrics = self.strategy.aggregate_train(
server_round, replies
)
# Add Gaussian noise to the aggregated arrays
if aggregated_arrays:
aggregated_arrays = self._add_noise_to_aggregated_arrays(aggregated_arrays)
return aggregated_arrays, aggregated_metrics
def validate_replies(replies: Iterable[Message], num_sampled_clients: int) -> bool:
"""Validate replies and log errors/warnings.
Arguments
----------
replies : Iterable[Message]
The replies to validate.
num_sampled_clients : int
The expected number of sampled clients.
Returns
-------
bool
True if replies are valid for aggregation, False otherwise.
"""
num_errors = 0
num_replies_with_content = 0
for msg in replies:
if msg.has_error():
log(
INFO,
"Received error in reply from node %d: %s",
msg.metadata.src_node_id,
msg.error,
)
num_errors += 1
else:
num_replies_with_content += 1
# Errors are not allowed
if num_errors:
log(
INFO,
"aggregate_train: Some clients reported errors. Skipping aggregation.",
)
return False
log(
INFO,
"aggregate_train: Received %s results and %s failures",
num_replies_with_content,
num_errors,
)
if num_replies_with_content != num_sampled_clients:
log(
WARNING,
CLIENTS_DISCREPANCY_WARNING,
num_replies_with_content,
num_sampled_clients,
)
return True