Federated Learning

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

# Gradient Inversion Attack: Deep Leakage from Gradients
# Zhu et al., NeurIPS 2019
 
import torch
import torch.nn.functional as F
 
def gradient_inversion_attack(
    model: torch.nn.Module,
    observed_gradients: List[torch.Tensor],
    image_shape: Tuple[int, int, int, int],  # (B, C, H, W)
    num_iterations: int = 300,
    learning_rate: float = 1.0,
    tv_weight: float = 0.01  # Total variation regularization
) -> torch.Tensor:
    """
    Reconstruct training data from observed gradients.
    
    The attack optimizes dummy inputs such that their gradients
    match the observed gradients. When successful, dummy inputs
    closely resemble the original training data.
    
    Mathematical formulation:
        x* = argmin_x || ∇L(model, x) - ∇_observed ||² + λ·TV(x)
    
    Where TV(x) is total variation regularization for smoothness.
    
    Args:
        model: The neural network model
        observed_gradients: Gradients shared by the victim client
        image_shape: Shape of images to reconstruct
        num_iterations: Number of optimization steps
        learning_rate: Learning rate for reconstruction optimizer
        tv_weight: Weight for total variation regularization
    
    Returns:
        Reconstructed images approximating training data
    """
    # Initialize random "dummy" data to optimize
    dummy_data = torch.randn(image_shape, requires_grad=True)
    dummy_labels = torch.randn((image_shape[0], num_classes), requires_grad=True)
    
    optimizer = torch.optim.LBFGS(
        [dummy_data, dummy_labels], 
        lr=learning_rate
    )
    
    for iteration in range(num_iterations):
        def closure():
            optimizer.zero_grad()
            
            # Compute gradients on dummy data
            dummy_outputs = model(dummy_data)
            dummy_loss = F.cross_entropy(dummy_outputs, F.softmax(dummy_labels, dim=-1))
            dummy_gradients = torch.autograd.grad(
                dummy_loss, 
                model.parameters(), 
                create_graph=True
            )
            
            # Minimize distance between dummy gradients and observed gradients
            gradient_distance = sum(
                ((dg - og) ** 2).sum()
                for dg, og in zip(dummy_gradients, observed_gradients)
            )
            
            # Total variation regularization (encourages smooth images)
            tv_loss = total_variation(dummy_data)
            
            total_loss = gradient_distance + tv_weight * tv_loss
            total_loss.backward()
            
            return total_loss
        
        optimizer.step(closure)
        
        # Clamp to valid image range
        with torch.no_grad():
            dummy_data.clamp_(0, 1)
    
    return dummy_data.detach()
 
 
def total_variation(images: torch.Tensor) -> torch.Tensor:
    """
    Total variation regularization.
    Penalizes large differences between adjacent pixels,
    encouraging smooth, natural-looking reconstructions.
    """
    diff_h = images[:, :, 1:, :] - images[:, :, :-1, :]
    diff_w = images[:, :, :, 1:] - images[:, :, :, :-1]
    return torch.mean(diff_h ** 2) + torch.mean(diff_w ** 2)
 
 
# Attack effectiveness demonstration
# With batch size 1, reconstruction achieves >90% PSNR
# Larger batches make individual sample recovery harder
# But metadata (class distribution, statistics) still leaks

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223

# Differential Privacy Mechanisms for Federated Learning
import numpy as np
from typing import Tuple, Callable
from scipy import stats
 
class DifferentialPrivacy:
    """
    Implementation of core differential privacy mechanisms
    used in federated learning.
    """
    
    @staticmethod
    def gaussian_mechanism(
        true_value: np.ndarray,
        sensitivity: float,
        epsilon: float,
        delta: float
    ) -> np.ndarray:
        """
        Gaussian mechanism for (ε, δ)-differential privacy.
        
        Adds Gaussian noise calibrated to the L2 sensitivity of the query.
        
        Noise scale: σ = Δ₂ · √(2 ln(1.25/δ)) / ε
        
        Where Δ₂ is the L2 sensitivity: max ||f(D) - f(D')||₂
        over all neighboring datasets D, D'.
        
        Args:
            true_value: The true query result to privatize
            sensitivity: L2 sensitivity of the query
            epsilon: Privacy parameter ε
            delta: Privacy parameter δ
        
        Returns:
            Noisy value satisfying (ε, δ)-DP
        """
        # Calculate noise scale using Gaussian mechanism formula
        # σ ≥ Δ₂ · √(2 ln(1.25/δ)) / ε
        sigma = sensitivity * np.sqrt(2 * np.log(1.25 / delta)) / epsilon
        
        # Sample Gaussian noise and add to true value
        noise = np.random.normal(0, sigma, size=true_value.shape)
        
        return true_value + noise
    
    @staticmethod
    def laplace_mechanism(
        true_value: np.ndarray,
        sensitivity: float,
        epsilon: float
    ) -> np.ndarray:
        """
        Laplace mechanism for ε-differential privacy (pure DP).
        
        Adds Laplace noise calibrated to the L1 sensitivity.
        
        Noise scale: b = Δ₁ / ε
        
        Where Δ₁ is the L1 sensitivity: max ||f(D) - f(D')||₁
        
        Provides pure DP (δ = 0) but requires more noise than Gaussian.
        
        Args:
            true_value: The true query result to privatize
            sensitivity: L1 sensitivity of the query
            epsilon: Privacy parameter ε
        
        Returns:
            Noisy value satisfying ε-DP
        """
        scale = sensitivity / epsilon
        noise = np.random.laplace(0, scale, size=true_value.shape)
        return true_value + noise
 
 
class DPFederatedLearning:
    """
    Differentially Private Federated Learning implementation
    following the DP-SGD approach (Abadi et al., 2016).
    """
    
    def __init__(
        self,
        target_epsilon: float,
        target_delta: float,
        clip_norm: float,
        noise_multiplier: float,
        num_rounds: int
    ):
        """
        Initialize DP-FL with privacy budget.
        
        Args:
            target_epsilon: Total privacy budget ε for all rounds
            target_delta: Target δ (typically 1/n for n users)
            clip_norm: Maximum L2 norm for client updates (sensitivity bound)
            noise_multiplier: σ/C ratio for Gaussian noise
            num_rounds: Total training rounds (affects privacy composition)
        """
        self.target_epsilon = target_epsilon
        self.target_delta = target_delta
        self.clip_norm = clip_norm
        self.noise_multiplier = noise_multiplier
        self.num_rounds = num_rounds
        
        # Track privacy spent so far
        self.rounds_completed = 0
    
    def clip_gradient(self, gradient: np.ndarray) -> np.ndarray:
        """
        Clip gradient to bound L2 sensitivity.
        
        Per-sample gradient clipping ensures that no single training
        example can influence the update by more than clip_norm.
        
        This is CRITICAL: without clipping, sensitivity is unbounded
        and no finite noise can achieve DP.
        """
        grad_norm = np.linalg.norm(gradient)
        
        if grad_norm > self.clip_norm:
            # Scale down to have exactly clip_norm magnitude
            gradient = gradient * (self.clip_norm / grad_norm)
        
        return gradient
    
    def add_noise_to_aggregate(
        self,
        aggregated_gradient: np.ndarray,
        num_clients: int
    ) -> np.ndarray:
        """
        Add calibrated Gaussian noise to aggregated update.
        
        The noise scale accounts for:
        1. The clip norm C (bounds sensitivity)
        2. The noise multiplier σ
        3. The number of clients (amplification via sampling)
        
        Returns:
            Noisy aggregate satisfying per-round DP guarantee
        """
        # Standard deviation of noise
        # σ_aggregate = noise_multiplier * C / num_clients
        sigma = self.noise_multiplier * self.clip_norm / num_clients
        
        noise = np.random.normal(0, sigma, size=aggregated_gradient.shape)
        noisy_aggregate = aggregated_gradient + noise
        
        self.rounds_completed += 1
        
        return noisy_aggregate
    
    def compute_privacy_spent(self) -> Tuple[float, float]:
        """
        Compute privacy budget spent using moments accountant.
        
        The Rényi Differential Privacy (RDP) framework provides tight
        privacy composition, essential for multi-round FL.
        
        Returns:
            (epsilon_spent, delta) tuple
        """
        # Simplified privacy accounting
        # In practice, use tensorflow-privacy or opacus for tight bounds
        from dp_accounting import compute_rdp, get_privacy_spent
        
        # Compute RDP at multiple orders
        orders = [1 + x / 10.0 for x in range(1, 100)]
        sampling_probability = 1.0  # If not subsampling clients
        
        rdp = compute_rdp(
            q=sampling_probability,
            noise_multiplier=self.noise_multiplier,
            steps=self.rounds_completed,
            orders=orders
        )
        
        epsilon_spent, _, _ = get_privacy_spent(
            orders, rdp, target_delta=self.target_delta
        )
        
        return epsilon_spent, self.target_delta
 
 
def dp_federated_averaging_round(
    global_model: np.ndarray,
    client_gradients: List[np.ndarray],
    dp_fl: DPFederatedLearning,
    client_weights: List[float]
) -> np.ndarray:
    """
    Execute one round of DP-FedAvg.
    
    Steps:
    1. Clip each client's gradient to bound sensitivity
    2. Compute weighted average of clipped gradients
    3. Add calibrated Gaussian noise to the aggregate
    4. Apply noisy update to global model
    """
    # Step 1: Clip each client's gradient
    clipped_gradients = [
        dp_fl.clip_gradient(grad) for grad in client_gradients
    ]
    
    # Step 2: Weighted average
    total_weight = sum(client_weights)
    aggregated = sum(
        (w / total_weight) * grad 
        for w, grad in zip(client_weights, clipped_gradients)
    )
    
    # Step 3: Add noise
    noisy_aggregate = dp_fl.add_noise_to_aggregate(
        aggregated, 
        num_clients=len(client_gradients)
    )
    
    # Step 4: Update model
    updated_model = global_model - noisy_aggregate  # Gradient descent
    
    return updated_model

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230

# Secure Aggregation Protocol (Simplified)
# Based on Bonawitz et al., CCS 2017
 
import secrets
from typing import Dict, List, Tuple
import numpy as np
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
 
class SecureAggregation:
    """
    Secure Aggregation Protocol Implementation.
    
    Key insight: Clients add pairwise masks that cancel in the sum.
    For clients i, j: mask_{i,j} = -mask_{j,i}
    When server sums all updates, masks cancel, revealing true sum.
    
    Security property: Server learns only Σᵢxᵢ, not individual xᵢ.
    
    Protocol phases:
    1. Key advertisement: Clients exchange Diffie-Hellman public keys
    2. Share keys: Clients secret-share their keys for dropout recovery
    3. Masked input: Clients submit masked updates
    4. Unmasking: Surviving clients help reconstruct dropped clients' masks
    """
    
    def __init__(self, num_clients: int, threshold: int):
        """
        Initialize SecAgg protocol.
        
        Args:
            num_clients: Total number of participating clients
            threshold: Minimum clients needed for aggregation
        """
        self.num_clients = num_clients
        self.threshold = threshold
        self.client_keys: Dict[int, bytes] = {}
        self.pairwise_masks: Dict[Tuple[int, int], np.ndarray] = {}
    
    def setup_pairwise_keys(self) -> Dict[int, Dict[int, bytes]]:
        """
        Phase 1: Clients exchange Diffie-Hellman keys to establish
        pairwise shared secrets.
        
        Each pair (i, j) derives a shared secret s_{i,j} = s_{j,i}
        using Diffie-Hellman key exchange.
        """
        # In practice, use proper DH key exchange
        # Here we simulate with random shared secrets
        shared_secrets = {}
        
        for i in range(self.num_clients):
            shared_secrets[i] = {}
            for j in range(self.num_clients):
                if i != j:
                    # Derive symmetric pairwise key
                    shared_secrets[i][j] = self._derive_pairwise_key(i, j)
        
        return shared_secrets
    
    def _derive_pairwise_key(self, i: int, j: int) -> bytes:
        """
        Derive pairwise key using HKDF.
        In practice, this uses Diffie-Hellman shared secret.
        """
        # Ensure consistent ordering for symmetric key
        pair = tuple(sorted([i, j]))
        seed = f"pair_{pair[0]}_{pair[1]}".encode()
        return HKDF(
            algorithm=hashes.SHA256(),
            length=32,
            salt=None,
            info=b'secagg-mask'
        ).derive(seed)
    
    def generate_pairwise_mask(
        self, 
        client_i: int, 
        client_j: int, 
        shape: Tuple[int, ...],
        shared_key: bytes
    ) -> np.ndarray:
        """
        Generate pairwise mask m_{i,j}.
        
        Critical property: m_{i,j} = -m_{j,i}
        This ensures masks cancel in the sum.
        
        Implementation: Use shared key as PRG seed, 
        negate if i > j.
        """
        # Use shared key to seed random number generator
        rng = np.random.default_rng(
            int.from_bytes(shared_key[:8], 'big')
        )
        
        # Generate mask
        mask = rng.standard_normal(shape).astype(np.float32)
        
        # Negate for one direction to ensure cancellation
        if client_i > client_j:
            mask = -mask
        
        return mask
    
    def mask_update(
        self,
        client_id: int,
        raw_update: np.ndarray,
        pairwise_keys: Dict[int, bytes]
    ) -> np.ndarray:
        """
        Mask a client's update for secure transmission.
        
        Masked update: ŷᵢ = xᵢ + Σⱼ m_{i,j} + rᵢ
        
        Where:
        - xᵢ is raw update
        - m_{i,j} are pairwise masks (cancel with j's contribution)
        - rᵢ is self-mask (for dropout recovery)
        """
        masked = raw_update.copy()
        
        # Add pairwise masks
        for j, key in pairwise_keys.items():
            mask = self.generate_pairwise_mask(
                client_id, j, raw_update.shape, key
            )
            masked += mask
        
        # Add self-mask (shared via secret sharing for dropout recovery)
        self_mask = self._generate_self_mask(client_id, raw_update.shape)
        masked += self_mask
        
        return masked
    
    def aggregate_masked_updates(
        self,
        masked_updates: Dict[int, np.ndarray],
        surviving_clients: List[int]
    ) -> np.ndarray:
        """
        Aggregate masked updates from surviving clients.
        
        For surviving clients, pairwise masks cancel:
        Σᵢ m_{i,j} + Σⱼ m_{j,i} = 0
        
        For dropped clients, surviving clients reconstruct
        the negative of dropped clients' self-masks.
        
        Result: Server learns only Σᵢ xᵢ
        """
        # Sum all masked updates
        aggregate = sum(masked_updates.values())
        
        # Handle dropped clients' self-masks
        dropped_clients = set(range(self.num_clients)) - set(surviving_clients)
        
        for dropped_id in dropped_clients:
            # Reconstruct dropped client's self-mask from secret shares
            # (held by surviving clients)
            reconstructed_self_mask = self._reconstruct_self_mask(
                dropped_id, surviving_clients
            )
            # Subtract to cancel the self-mask
            aggregate -= reconstructed_self_mask
        
        return aggregate
    
    def _generate_self_mask(
        self, 
        client_id: int, 
        shape: Tuple[int, ...]
    ) -> np.ndarray:
        """Generate client's self-mask for dropout recovery."""
        rng = np.random.default_rng(client_id * 1000)
        return rng.standard_normal(shape).astype(np.float32)
    
    def _reconstruct_self_mask(
        self, 
        dropped_id: int,
        surviving_clients: List[int]
    ) -> np.ndarray:
        """
        Reconstruct dropped client's self-mask from secret shares.
        Uses Shamir's secret sharing for threshold reconstruction.
        """
        # In practice: collect shares from t surviving clients
        # and use polynomial interpolation
        return self._generate_self_mask(dropped_id, self.update_shape)
 
 
class SecAggWithDropouts:
    """
    Production-ready SecAgg handling client dropouts.
    
    The protocol tolerates up to n-t dropouts while maintaining
    security and correctness, where t is the threshold.
    """
    
    def execute_protocol(
        self,
        client_updates: Dict[int, np.ndarray],
        dropout_probability: float = 0.1
    ) -> np.ndarray:
        """
        Execute full SecAgg protocol with dropout handling.
        
        Protocol rounds:
        1. Advertise keys
        2. Share keys  
        3. Submit masked inputs
        4. Unmask (handle dropouts)
        """
        # Simulate client dropouts
        active_clients = [
            cid for cid in client_updates.keys()
            if np.random.random() > dropout_probability
        ]
        
        if len(active_clients) < self.threshold:
            raise ProtocolFailedError(
                f"Only {len(active_clients)} clients survived, "
                f"need {self.threshold}"
            )
        
        # Execute aggregation with surviving clients
        return self._aggregate_with_recovery(
            {cid: client_updates[cid] for cid in active_clients}
        )

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184

# Privacy Accounting with Rényi Differential Privacy
from dataclasses import dataclass
from typing import List, Tuple
import numpy as np
from scipy import special
 
@dataclass
class PrivacyBudget:
    """Track privacy budget consumption."""
    target_epsilon: float
    target_delta: float
    consumed_epsilon: float = 0.0
    consumed_delta: float = 0.0
    
    def remaining(self) -> Tuple[float, float]:
        return (
            self.target_epsilon - self.consumed_epsilon,
            self.target_delta - self.consumed_delta
        )
    
    def is_exhausted(self) -> bool:
        return (
            self.consumed_epsilon >= self.target_epsilon or
            self.consumed_delta >= self.target_delta
        )
 
 
class RenyiDifferentialPrivacy:
    """
    Rényi Differential Privacy (RDP) accounting.
    
    Mironov, 2017: "Rényi Differential Privacy"
    
    Key advantages:
    1. Tight composition (RDP adds across mechanisms)
    2. Natural for subsampled Gaussian mechanism
    3. Converts to (ε, δ)-DP for final reporting
    """
    
    @staticmethod
    def compute_rdp_gaussian(
        sampling_rate: float,
        noise_multiplier: float,
        orders: List[float]
    ) -> List[float]:
        """
        Compute RDP for subsampled Gaussian mechanism.
        
        For Gaussian mechanism with subsampling probability q
        and noise multiplier σ (noise std = σ × sensitivity):
        
        RDP at order α ≈ α / (2σ²)  [simplified, full formula complex]
        
        Subsampling provides amplification: effective RDP is much smaller.
        
        Args:
            sampling_rate: Probability q of including each record
            noise_multiplier: σ = noise_std / sensitivity
            orders: Rényi orders α to compute RDP for
        
        Returns:
            RDP at each order
        """
        rdp = []
        for order in orders:
            if order == 1:
                # KL divergence (order 1 Rényi)
                rdp.append(0)  # Needs special handling
            elif sampling_rate == 1.0:
                # No subsampling: α / (2σ²)
                rdp.append(order / (2 * noise_multiplier ** 2))
            else:
                # With subsampling: use numerical integration
                # Approximation for small q:
                rdp.append(
                    np.log1p(
                        sampling_rate ** 2 * 
                        (np.exp(order / (2 * noise_multiplier ** 2)) - 1)
                    ) / (order - 1)
                )
        return rdp
    
    @staticmethod
    def rdp_to_epsilon(
        rdp_values: List[float],
        orders: List[float],
        target_delta: float
    ) -> float:
        """
        Convert RDP to (ε, δ)-DP.
        
        For each order α with RDP value ρ:
            ε ≤ ρ + log(1/δ) / (α - 1)
        
        Return minimum ε across all orders.
        """
        epsilons = []
        for rdp, order in zip(rdp_values, orders):
            if order == 1:
                epsilon = rdp  # KL divergence equals ε for order 1
            else:
                epsilon = rdp + np.log(1 / target_delta) / (order - 1)
            epsilons.append(epsilon)
        
        return min(epsilons)
    
    @staticmethod
    def compose_rdp(
        rdp_per_round: List[float],
        num_rounds: int
    ) -> List[float]:
        """
        Compose RDP across multiple rounds.
        
        RDP composes additively: total RDP = sum of per-round RDP.
        This is the beauty of RDP—simple addition gives tight bounds.
        """
        return [rdp * num_rounds for rdp in rdp_per_round]
 
 
def plan_private_training(
    target_epsilon: float,
    target_delta: float,
    num_rounds: int,
    samples_per_round: int,
    total_samples: int
) -> float:
    """
    Plan private training by computing required noise multiplier.
    
    Given privacy budget and training plan, compute the minimum
    noise_multiplier that stays within budget.
    
    Args:
        target_epsilon: Privacy budget ε
        target_delta: Privacy budget δ
        num_rounds: Number of training rounds
        samples_per_round: Samples used per round
        total_samples: Total samples in dataset
    
    Returns:
        Required noise_multiplier σ
    """
    sampling_rate = samples_per_round / total_samples
    orders = [1 + x / 10.0 for x in range(1, 100)]
    
    # Binary search for noise multiplier
    low, high = 0.1, 100.0
    
    while high - low > 0.01:
        mid = (low + high) / 2
        
        # Compute RDP for this noise multiplier
        rdp_per_round = RenyiDifferentialPrivacy.compute_rdp_gaussian(
            sampling_rate, mid, orders
        )
        
        # Compose across rounds
        total_rdp = RenyiDifferentialPrivacy.compose_rdp(
            rdp_per_round, num_rounds
        )
        
        # Convert to epsilon
        achieved_epsilon = RenyiDifferentialPrivacy.rdp_to_epsilon(
            total_rdp, orders, target_delta
        )
        
        if achieved_epsilon > target_epsilon:
            low = mid  # Need more noise
        else:
            high = mid  # Can use less noise
    
    return high
 
 
# Example usage
noise_multiplier = plan_private_training(
    target_epsilon=1.0,
    target_delta=1e-5,
    num_rounds=100,
    samples_per_round=256,
    total_samples=60000  # MNIST size
)
print(f"Required noise multiplier: {noise_multiplier:.2f}")