Privacy & Zero-Knowledge Proofs
Panda supports privacy-preserving smart contracts through encrypted state, zero-knowledge proofs, and secure computation patterns. The @private decorator and panda.crypto module enable contracts where sensitive data never appears on-chain in plaintext.
Private Contracts
Mark a contract as private to encrypt its state. The PrivateModel contract demonstrates XOR cipher encryption for model privacy, where weights are encrypted on-chain and only the owner (who holds the decryption key) can see the actual model parameters.
from panda import contract, constructor, call, query, event, private, proof
import hashlib
import json
@private
@contract
class PrivateModel:
"""Privacy-enabled linear regression model with encrypted state."""
class State:
encrypted_weights: str = ""
encrypted_bias: str = ""
encrypted_metadata: str = ""
owner: str = ""
key_hash: str = ""
is_initialized: bool = False
training_count: int = 0
input_dim: int = 0
@constructor
def deploy(self, ctx, input_dim=1):
"""Initialize the private model.
Args:
input_dim: number of input features
"""
self.state.owner = ctx.sender
self.state.input_dim = input_dim
# Derive an encryption key from owner + contract address
key_material = ctx.sender + ":" + ctx.contract_address
key_hash = hashlib.sha256(key_material.encode("utf-8")).hexdigest()
self.state.key_hash = key_hash
# Initialize weights to zeros, encrypt, and store
weights = [0.0] * input_dim
bias = 0.0
metadata = {
"input_dim": input_dim,
"learning_rate": 0.01,
"samples_trained": 0,
}
self.state.encrypted_weights = _encrypt_data(json.dumps(weights), key_hash)
self.state.encrypted_bias = _encrypt_data(json.dumps(bias), key_hash)
self.state.encrypted_metadata = _encrypt_data(json.dumps(metadata), key_hash)
self.state.is_initialized = True
self.emit(
event.PrivateModelCreated(
owner=ctx.sender,
input_dim=input_dim,
)
)
print(f"PrivateModel deployed: owner={ctx.sender}, input_dim={input_dim}")
With @private:
- Contract state is encrypted at rest using key material derived from owner + contract address
- Only the contract owner can read raw state
- Query results are computed inside a trusted execution environment
- Predictions use
@proof(type="validity")so correctness is verifiable via ZK proofs
The contract also includes encrypted training and inference methods. Here is the predict_private method, which decrypts input, runs inference on encrypted weights, and returns encrypted output:
@query
@proof(type="validity")
def predict_private(self, encrypted_input) -> str:
"""Run inference on encrypted input and return encrypted output."""
if not self.state.is_initialized:
raise RuntimeError("Model not initialized")
key_hash = self.state.key_hash
# Decrypt input
input_json = _decrypt_data(encrypted_input, key_hash)
input_data = json.loads(input_json)
x_data = input_data["x"]
input_dim = self.state.input_dim
# Normalize 1D input
if isinstance(x_data[0], (int, float)):
x_data = [[float(v)] for v in x_data]
# Decrypt model weights
weights = json.loads(_decrypt_data(self.state.encrypted_weights, key_hash))
bias = json.loads(_decrypt_data(self.state.encrypted_bias, key_hash))
# Run inference
predictions = []
for sample in x_data:
pred = bias
for i in range(input_dim):
pred += sample[i] * weights[i]
predictions.append(round(pred, 6))
# Encrypt output
output = {
"predictions": predictions,
"model_version": self.state.training_count,
}
encrypted_output = _encrypt_data(json.dumps(output), key_hash)
return encrypted_output
Use case: A company deploys a proprietary fraud detection model. Users submit encrypted transactions for scoring, but the model's learned weights (trade secrets) stay encrypted. The @proof(type="validity") decorator ensures predictions are verifiable without re-executing.
Zero-Knowledge Membership Proof
Prove you belong to a set without revealing which member you are. The ZKPMembership contract uses Merkle proofs for set membership verification and nullifiers to prevent double-use of authorizations.
from panda import contract, constructor, call, query, event, proof
from panda.crypto import merkle_root, merkle_verify, sha256 # noqa: F401
@contract
class ZKPMembership:
"""Anonymous authorization contract using Merkle proofs and nullifiers."""
class State:
root: str = ""
members_count: int = 0
owner: str = ""
authorized_actions: int = 0
nullifiers: dict = {}
all_members: list = []
@constructor
def deploy(self, ctx, members: list):
"""Deploy with an initial set of members.
Computes the Merkle root of all member IDs for future proof
verification.
Args:
members: list of member ID strings.
"""
if not isinstance(members, list) or len(members) == 0:
raise ValueError("must provide at least one member")
self.state.owner = ctx.sender
self.state.all_members = members
self.state.members_count = len(members)
self.state.root = merkle_root(members)
self.emit(
event.MembersAdded(
count=len(members),
root=self.state.root,
)
)
print(
f"ZKPMembership deployed: owner={ctx.sender}, members={len(members)}, root={self.state.root[:16]}..."
)
@call
@proof(type="validity")
def authorize(self, ctx, member_id: str, proof_data: list, nullifier: str):
"""Verify membership and record an anonymous authorization.
The caller proves they are a member of the whitelist by providing
a valid Merkle proof. The nullifier prevents the same member from
authorizing twice (it is sha256(member_id + secret) computed
off-chain).
Args:
member_id: the member's ID (leaf in the Merkle tree).
proof_data: Merkle proof -- list of {"hash": hex, "position": str}.
nullifier: sha256(member_id + secret) -- unique, unlinkable token.
"""
if not nullifier or len(nullifier) == 0:
raise ValueError("nullifier is required")
# Check nullifier has not been used
if nullifier in self.state.nullifiers:
raise ValueError("nullifier already used -- possible double authorization")
# Verify Merkle proof of membership
if not merkle_verify(member_id, proof_data, self.state.root):
raise ValueError("invalid membership proof")
# Record the nullifier
nullifiers = dict(self.state.nullifiers)
nullifiers[nullifier] = ctx.block_height
self.state.nullifiers = nullifiers
self.state.authorized_actions = self.state.authorized_actions + 1
self.emit(
event.MemberAuthorized(
nullifier=nullifier,
block=ctx.block_height,
total_authorizations=self.state.authorized_actions,
)
)
print(
f"Member authorized (nullifier: {nullifier[:16]}...), total={self.state.authorized_actions}"
)
@query
def verify_membership(self, member_id: str, proof_data: list) -> bool:
"""Stateless check: verify a Merkle proof against the current root."""
return merkle_verify(member_id, proof_data, self.state.root)
@query
def is_nullifier_used(self, nullifier: str) -> bool:
"""Check whether a nullifier has already been consumed."""
return nullifier in self.state.nullifiers
@query
def get_info(self) -> dict:
"""Return public contract information."""
return {
"root": self.state.root,
"members_count": self.state.members_count,
"authorized_actions": self.state.authorized_actions,
"nullifiers_used": len(self.state.nullifiers),
}
How ZK Proofs Work on Panda
- Off-chain: The member computes a nullifier (
sha256(member_id + secret)) and a Merkle proof of their membership - On-chain: The
authorizemethod (decorated with@proof(type="validity")) verifies the Merkle proof and checks the nullifier has not been used - Result: The contract records that a valid member authorized an action, without learning which member it was
Secure Aggregation
Multiple parties compute a sum of their private values without revealing individual contributions. The SecureAggregation contract implements the Bonawitz et al. (2017) pairwise masking protocol used in production federated learning systems.
from panda import contract, constructor, call, query, event
from panda.crypto import derive_key
@contract
class SecureAggregation:
"""Multi-party secure sum via pairwise masking."""
class State:
participants: list = []
round: int = 0
submissions: dict = {}
phase: str = "registration"
result: float = 0.0
owner: str = ""
min_participants: int = 2
completed_rounds: int = 0
@constructor
def deploy(self, ctx, min_participants: int):
"""Initialize the secure aggregation contract.
Args:
min_participants: minimum number of parties required per round.
"""
if min_participants < 2:
raise ValueError("min_participants must be at least 2")
self.state.owner = ctx.sender
self.state.min_participants = min_participants
self.state.round = 1
print(
f"SecureAggregation deployed: owner={ctx.sender}, min_participants={min_participants}"
)
@call
def register(self, ctx):
"""Register for the current aggregation round."""
if self.state.phase != "registration":
raise ValueError("registration is not open")
if ctx.sender in self.state.participants:
raise ValueError("already registered")
participants = list(self.state.participants)
participants.append(ctx.sender)
self.state.participants = participants
self.emit(
event.ParticipantRegistered(
participant=ctx.sender,
round=self.state.round,
total_registered=len(participants),
)
)
@call
def submit_masked(self, ctx, masked_value: float):
"""Submit a masked value for the current round.
The caller must compute their masked value OFF-CHAIN as:
masked_value = private_value + sum_of_pairwise_masks
For each other participant j, the pairwise mask is computed as:
mask = int(derive_key(min(my_addr, j_addr) + max(my_addr, j_addr),
str(round))[:8], 16) / 2**32
If my_addr < j_addr: add the mask
If my_addr > j_addr: subtract the mask
"""
if self.state.phase != "submission":
raise ValueError("submission is not open")
if ctx.sender not in self.state.participants:
raise ValueError("not registered for this round")
if ctx.sender in self.state.submissions:
raise ValueError("already submitted")
submissions = dict(self.state.submissions)
submissions[ctx.sender] = masked_value
self.state.submissions = submissions
self.emit(
event.ValueSubmitted(
participant=ctx.sender,
round=self.state.round,
submissions_count=len(submissions),
)
)
@call
def finalize(self, ctx):
"""Compute the aggregate result and reset for the next round.
All registered participants must have submitted their masked values.
The sum of masked values equals the true sum because pairwise masks
cancel out.
"""
if self.state.phase != "submission":
raise ValueError("not in submission phase")
if len(self.state.submissions) != len(self.state.participants):
raise ValueError(
f"not all participants submitted: "
f"{len(self.state.submissions)}/{len(self.state.participants)}"
)
# Sum all masked values -- masks cancel, yielding true sum
total = 0.0
for addr in self.state.participants:
total += self.state.submissions[addr]
self.state.result = total
completed = self.state.completed_rounds + 1
self.state.completed_rounds = completed
current_round = self.state.round
self.emit(
event.RoundFinalized(
round=current_round,
result=round(total, 6),
num_participants=len(self.state.participants),
)
)
# Reset for next round
self.state.round = current_round + 1
self.state.participants = []
self.state.submissions = {}
self.state.phase = "registration"
The protocol works as follows:
- Registration: Parties register for the current round
- Submission: Each party computes pairwise masks OFF-CHAIN using
derive_key. For each pair (i, j), both parties derive the same mask. The party with the smaller address ADDS the mask; the party with the larger address SUBTRACTS it. Each party submitsvalue + sum_of_pairwise_masks - Finalization: The contract sums all masked submissions. Because every mask appears once as +mask and once as -mask, they cancel out, yielding the true sum of all private values
Use case: Salary surveys where participants prove their contribution without revealing individual salaries.
Federated Learning
Coordinate distributed model training across multiple data holders. The FederatedTrainer contract implements Federated Stochastic Gradient Descent (FedSGD) for a multi-output linear model.
from panda import contract, constructor, call, query, event, proof
@contract
class FederatedTrainer:
"""On-chain federated linear model with validity proofs on gradient updates."""
class State:
weights: list = []
bias: float = 0.0
round: int = 0
contributions: int = 0
contributors: list = []
owner: str = ""
input_dim: int = 0
output_dim: int = 1
total_samples_seen: int = 0
# Multi-output support: weights is flat [input_dim * output_dim], row-major
biases: list = []
@constructor
def deploy(self, ctx, input_dim=1, output_dim=1):
"""Deploy the federated model with specified dimensions."""
if input_dim < 1 or input_dim > 10000:
raise ValueError(f"input_dim must be between 1 and 10000, got {input_dim}")
if output_dim < 1 or output_dim > 10000:
raise ValueError(f"output_dim must be between 1 and 10000, got {output_dim}")
if input_dim * output_dim > 100_000_000:
raise ValueError(
f"input_dim * output_dim must be <= 100,000,000, got {input_dim * output_dim}"
)
self.state.owner = ctx.sender
self.state.input_dim = input_dim
self.state.output_dim = output_dim
self.state.weights = [0.0] * (input_dim * output_dim)
self.state.bias = 0.0
self.state.biases = [0.0] * output_dim
self.state.contributors = []
self.state.contributions = 0
self.state.round = 0
self.state.total_samples_seen = 0
self.emit(
event.ModelInitialized(
input_dim=input_dim,
output_dim=output_dim,
owner=ctx.sender,
)
)
@call
@proof(type="validity")
def contribute_gradient(self, ctx, x, y, learning_rate=0.01):
"""Submit a gradient update computed on private local data.
The contract computes the gradient of MSE loss for the provided
data batch and applies a single SGD step.
Args:
x: list of input samples (shape: [n_samples] or [n_samples, input_dim])
y: list of target values (shape: [n_samples] or [n_samples, output_dim])
learning_rate: SGD learning rate (default 0.01)
"""
if not isinstance(x, list) or len(x) == 0:
raise ValueError("x must be a non-empty list")
if not isinstance(y, list) or len(y) == 0:
raise ValueError("y must be a non-empty list")
if len(x) != len(y):
raise ValueError(f"x and y must have same length: {len(x)} != {len(y)}")
input_dim = self.state.input_dim
output_dim = self.state.output_dim
n_samples = len(x)
# Normalize input format: ensure x is always 2D [n_samples, input_dim]
if isinstance(x[0], (int, float)):
if input_dim == 1:
x_2d = [[float(xi)] for xi in x]
else:
raise ValueError(f"Expected {input_dim}-D input vectors, got scalar values")
else:
x_2d = [list(row) for row in x]
# ... gradient computation and SGD update ...
# (full implementation computes predictions, MSE loss gradients,
# and applies weight updates in-place)
self.state.contributions = self.state.contributions + 1
self.state.round = self.state.round + 1
self.state.total_samples_seen = self.state.total_samples_seen + n_samples
self.emit(
event.GradientApplied(
contributor=ctx.sender,
round=self.state.round,
samples=n_samples,
)
)
@query
def predict(self, x) -> list:
"""Run inference on the current global model."""
input_dim = self.state.input_dim
output_dim = self.state.output_dim
if not isinstance(x, list) or len(x) == 0:
return []
# Normalize to 2D
if isinstance(x[0], (int, float)):
if input_dim == 1:
x_2d = [[float(xi)] for xi in x]
else:
raise ValueError(f"Expected {input_dim}-D input vectors, got scalar values")
else:
x_2d = [list(row) for row in x]
biases = list(self.state.biases) if self.state.biases else [self.state.bias] * output_dim
weights = self.state.weights
results = []
for s in range(len(x_2d)):
preds = []
for j in range(output_dim):
val = biases[j]
for i in range(input_dim):
val += x_2d[s][i] * weights[i * output_dim + j]
preds.append(val)
if output_dim == 1:
results.append(preds[0])
else:
results.append(preds)
return results
@query
def model_info(self) -> dict:
"""Return current model state and training statistics."""
return {
"weights": self.state.weights,
"bias": self.state.bias,
"biases": self.state.biases,
"round": self.state.round,
"contributions": self.state.contributions,
"contributors": self.state.contributors,
"input_dim": self.state.input_dim,
"output_dim": self.state.output_dim,
"total_samples_seen": self.state.total_samples_seen,
}
Each participant trains locally on their private data, submits gradient updates via contribute_gradient (which is decorated with @proof(type="validity") for verifiable execution), and the contract applies SGD updates to the global model. Raw data never leaves the participant.
Differential Privacy
The DifferentialPrivacy contract maintains aggregate statistics and returns differentially-private answers using the Laplace mechanism. Noise is calibrated to the query sensitivity and drawn from a deterministic pseudo-random source (SHA-256 hash of query context) for on-chain reproducibility.
import math
from panda import contract, constructor, call, query, event
from panda.crypto import sha256
@contract
class DifferentialPrivacy:
"""On-chain DP statistics engine with Laplace noise and epsilon tracking."""
class State:
data_count: int = 0
data_sum: float = 0.0
data_sum_squares: float = 0.0
epsilon_budget: float = 0.0
epsilon_spent: float = 0.0
owner: str = ""
sensitivity: float = 1.0
contributions: int = 0
query_count: int = 0
@constructor
def deploy(self, ctx, epsilon_budget: float, sensitivity: float):
"""Initialize the DP engine with a total privacy budget and sensitivity."""
if epsilon_budget <= 0:
raise ValueError("epsilon_budget must be positive")
if sensitivity <= 0:
raise ValueError("sensitivity must be positive")
self.state.owner = ctx.sender
self.state.epsilon_budget = epsilon_budget
self.state.sensitivity = sensitivity
@call
def contribute(self, ctx, values: list):
"""Add data points to the aggregate statistics.
Only running aggregates are stored -- individual values are never
persisted on-chain.
"""
if not isinstance(values, list) or len(values) == 0:
raise ValueError("values must be a non-empty list")
for v in values:
if not isinstance(v, (int, float)):
raise ValueError("all values must be numeric")
n = len(values)
self.state.data_count = self.state.data_count + n
self.state.data_sum = self.state.data_sum + sum(float(v) for v in values)
self.state.data_sum_squares = self.state.data_sum_squares + sum(
float(v) ** 2 for v in values
)
self.state.contributions = self.state.contributions + 1
self.emit(
event.DataContributed(
contributor=ctx.sender,
num_values=n,
total_count=self.state.data_count,
)
)
def _laplace_noise(self, seed_str, sensitivity, epsilon):
"""Generate deterministic Laplace noise from a seed string."""
h = sha256(seed_str)
# Map first 16 hex chars (8 bytes) to float in [0, 1)
u = int(h[:16], 16) / (2**64)
# Clamp u away from exact 0.0 and 0.5 to avoid log(0)
if u == 0.0:
u = 1e-15
if u == 0.5:
u = 0.5 + 1e-15
b = sensitivity / epsilon
sign = 1.0 if u >= 0.5 else -1.0
noise = -b * sign * math.log(1.0 - 2.0 * abs(u - 0.5))
return noise
def _spend_budget(self, epsilon_per_query):
"""Deduct epsilon from the budget. Raises if exhausted."""
remaining = self.state.epsilon_budget - self.state.epsilon_spent
if epsilon_per_query > remaining:
raise ValueError(
f"privacy budget exhausted: need {epsilon_per_query}, remaining {remaining}"
)
self.state.epsilon_spent = self.state.epsilon_spent + epsilon_per_query
self.state.query_count = self.state.query_count + 1
@call
def private_mean(self, ctx) -> dict:
"""Return the mean with calibrated Laplace noise."""
if self.state.data_count == 0:
raise ValueError("no data contributed yet")
epsilon_per_query = 0.1
self._spend_budget(epsilon_per_query)
true_mean = self.state.data_sum / self.state.data_count
# Sensitivity of mean = sensitivity / n
mean_sensitivity = self.state.sensitivity / self.state.data_count
seed = "mean" + str(ctx.block_height) + ctx.sender + str(self.state.query_count)
noise = self._laplace_noise(seed, mean_sensitivity, epsilon_per_query)
noised_mean = true_mean + noise
return {
"value": round(noised_mean, 6),
"epsilon_spent": epsilon_per_query,
"remaining_budget": round(self.state.epsilon_budget - self.state.epsilon_spent, 6),
}
@call
def private_variance(self, ctx) -> dict:
"""Return the variance with calibrated Laplace noise."""
if self.state.data_count < 2:
raise ValueError("need at least 2 data points for variance")
epsilon_per_query = 0.1
self._spend_budget(epsilon_per_query)
n = self.state.data_count
mean = self.state.data_sum / n
# Variance = E[X^2] - (E[X])^2
variance = (self.state.data_sum_squares / n) - (mean**2)
# Sensitivity of variance = sensitivity^2 / n
var_sensitivity = (self.state.sensitivity**2) / n
seed = "variance" + str(ctx.block_height) + ctx.sender + str(self.state.query_count)
noise = self._laplace_noise(seed, var_sensitivity, epsilon_per_query)
noised_variance = variance + noise
return {
"value": round(noised_variance, 6),
"epsilon_spent": epsilon_per_query,
"remaining_budget": round(self.state.epsilon_budget - self.state.epsilon_spent, 6),
}
@query
def budget_status(self) -> dict:
"""Return the current privacy budget status."""
return {
"epsilon_budget": self.state.epsilon_budget,
"epsilon_spent": round(self.state.epsilon_spent, 6),
"remaining": round(self.state.epsilon_budget - self.state.epsilon_spent, 6),
"queries_executed": self.state.query_count,
"data_count": self.state.data_count,
"contributions": self.state.contributions,
}
The key design choices:
- Deterministic noise: The Laplace noise is derived from
sha256(query_context)using the inverse CDF method, so identical queries at the same block height produce identical noise -- critical for on-chain determinism - Budget tracking: The contract tracks cumulative epsilon and refuses queries once the privacy budget is exhausted
- Aggregate-only storage: Individual data values are never stored; only running aggregates (count, sum, sum of squares) are persisted
Differential Privacy in Federated Learning
The DPFederatedLearner contract combines federated gradient descent with formal differential privacy (DP-SGD). Each contributor clips their gradient to a fixed L2 norm, the contract adds calibrated Gaussian noise, and a privacy accountant tracks cumulative epsilon spend against a finite budget. This provides (epsilon, delta)-differential privacy guarantees as described by Abadi et al. ("Deep Learning with Differential Privacy", CCS 2016).
import math
from panda import contract, constructor, call, query, event, proof
from panda.crypto import sha256
@contract
class DPFederatedLearner:
"""Federated linear model with DP-SGD gradient protection."""
class State:
weights: list = []
bias: float = 0.0
input_dim: int = 0
round: int = 0
clip_norm: float = 1.0
noise_scale: float = 0.0
epsilon_per_round: float = 0.0
delta: float = 0.00001
total_epsilon_spent: float = 0.0
epsilon_budget: float = 0.0
contributors: list = []
contributions_this_round: list = []
required_contributions: int = 1
learning_rate: float = 0.01
owner: str = ""
total_samples: int = 0
@constructor
def deploy(
self,
ctx,
input_dim: int,
clip_norm: float,
epsilon_budget: float,
delta: float,
contributions_per_round: int,
):
"""Initialize the DP federated model.
Args:
input_dim: number of input features for the linear model.
clip_norm: maximum L2 norm for gradient clipping (C).
epsilon_budget: total differential privacy budget.
delta: probability of privacy failure (typically 1e-5).
contributions_per_round: number of gradient submissions needed
before the owner can trigger aggregation.
"""
if input_dim <= 0:
raise ValueError("input_dim must be positive")
if input_dim > 10000:
raise ValueError(f"input_dim must be <= 10000, got {input_dim}")
if clip_norm <= 0:
raise ValueError("clip_norm must be positive")
if epsilon_budget <= 0:
raise ValueError("epsilon_budget must be positive")
if delta <= 0 or delta >= 1:
raise ValueError("delta must be in (0, 1)")
if contributions_per_round < 1:
raise ValueError("contributions_per_round must be >= 1")
self.state.owner = ctx.sender
self.state.input_dim = input_dim
self.state.weights = [0.0] * input_dim
self.state.bias = 0.0
self.state.clip_norm = clip_norm
self.state.epsilon_budget = epsilon_budget
self.state.delta = delta
self.state.required_contributions = contributions_per_round
self.state.learning_rate = 0.01
self.state.round = 0
self.state.total_epsilon_spent = 0.0
self.state.contributors = []
self.state.contributions_this_round = []
self.state.total_samples = 0
# Compute noise scale: sigma = sqrt(2 * ln(1.25 / delta)) * C / epsilon
# We use epsilon_per_round = epsilon_budget / sqrt(expected_rounds)
# For simplicity, set epsilon_per_round so noise_scale is well-defined.
# Start with epsilon_per_round = epsilon_budget / 10 (expect ~100 rounds
# via basic composition: total_eps = eps_per_round * sqrt(T)).
eps_per_round = epsilon_budget / 10.0
self.state.epsilon_per_round = eps_per_round
noise_scale = math.sqrt(2.0 * math.log(1.25 / delta)) * clip_norm / eps_per_round
self.state.noise_scale = noise_scale
self.emit(
event.ModelDeployed(
owner=ctx.sender,
input_dim=input_dim,
clip_norm=clip_norm,
epsilon_budget=epsilon_budget,
delta=delta,
noise_scale=round(noise_scale, 6),
)
)
The contribute_dp_gradient method is where the DP mechanism is enforced. Each contributor submits a pre-clipped gradient, and the contract verifies the norm bound and adds deterministic Gaussian noise:
def _deterministic_gaussian(self, seed):
"""Generate a deterministic Gaussian sample via Box-Muller transform.
Uses SHA-256 hashes of the seed to produce two uniform values u1, u2
in (0, 1), then applies the Box-Muller transform:
z = sqrt(-2 * ln(u1)) * cos(2 * pi * u2)
Args:
seed: a string that uniquely identifies this noise draw.
Returns:
A float drawn from a standard normal distribution (deterministically).
"""
h1 = sha256(seed + ":u1")
h2 = sha256(seed + ":u2")
u1 = max(1e-10, int(h1[:16], 16) / (2**64))
u2 = int(h2[:16], 16) / (2**64)
z = math.sqrt(-2.0 * math.log(u1)) * math.cos(2.0 * math.pi * u2)
return z
def _clip_gradient(self, gradient, max_norm):
"""Clip a gradient vector to a maximum L2 norm.
If ||gradient||_2 > max_norm, rescale so ||g_clipped||_2 = max_norm.
This bounds the sensitivity of any single contribution.
Args:
gradient: list of float gradient values.
max_norm: maximum allowed L2 norm.
Returns:
List of clipped gradient values.
"""
norm = math.sqrt(sum(g * g for g in gradient))
if norm > max_norm:
scale = max_norm / norm
return [g * scale for g in gradient]
return list(gradient)
@call
@proof(type="validity")
def contribute_dp_gradient(
self, ctx, clipped_gradient: list, clipped_bias_grad: float, num_samples: int
):
"""Submit a DP-protected gradient update for the current round.
The contributor is expected to clip their gradient locally before
submission. The contract verifies the norm bound, then adds
calibrated Gaussian noise to each dimension for differential privacy.
Args:
clipped_gradient: pre-clipped weight gradient (length = input_dim).
clipped_bias_grad: pre-clipped bias gradient (scalar).
num_samples: number of private samples used to compute this gradient.
"""
remaining = self.state.epsilon_budget - self.state.total_epsilon_spent
if remaining <= 0:
raise ValueError("privacy budget exhausted")
if len(clipped_gradient) != self.state.input_dim:
raise ValueError(
f"gradient length {len(clipped_gradient)} != input_dim {self.state.input_dim}"
)
if num_samples <= 0:
raise ValueError("num_samples must be positive")
# Verify the gradient norm is within the clip bound (with small tolerance)
grad_norm = math.sqrt(sum(g * g for g in clipped_gradient))
tolerance = 1e-6
if grad_norm > self.state.clip_norm + tolerance:
# Re-clip to enforce the bound on-chain
clipped_gradient = self._clip_gradient(clipped_gradient, self.state.clip_norm)
# Add deterministic Gaussian noise to each gradient dimension
noised_gradient = []
current_round = self.state.round
sender = ctx.sender
for dim_idx in range(self.state.input_dim):
seed = sender + str(current_round) + str(dim_idx)
noise = self._deterministic_gaussian(seed) * self.state.noise_scale
noised_gradient.append(clipped_gradient[dim_idx] + noise)
# Noise the bias gradient as well
bias_seed = sender + str(current_round) + "bias"
bias_noise = self._deterministic_gaussian(bias_seed) * self.state.noise_scale
noised_bias_grad = clipped_bias_grad + bias_noise
# Track contributor
if sender not in self.state.contributors:
contributors = list(self.state.contributors)
contributors.append(sender)
self.state.contributors = contributors
# Accumulate this contribution for the current round
contributions = list(self.state.contributions_this_round)
contributions.append(
{
"sender": sender,
"gradient": noised_gradient,
"bias_grad": noised_bias_grad,
"num_samples": num_samples,
}
)
self.state.contributions_this_round = contributions
self.state.total_samples = self.state.total_samples + num_samples
self.emit(
event.GradientContributed(
contributor=sender,
round=current_round,
num_samples=num_samples,
grad_norm=round(grad_norm, 6),
)
)
Once enough contributions are collected, the owner calls force_aggregate to average the noised gradients and apply an SGD step. Privacy accounting uses basic composition: total_epsilon = epsilon_per_round * sqrt(num_rounds).
@call
def force_aggregate(self, ctx):
"""Aggregate accumulated gradient contributions and apply SGD step.
Only the contract owner can trigger aggregation. All accumulated
noised gradients are averaged and applied to the global model weights.
The round counter increments and privacy budget is updated.
Privacy accounting uses basic composition:
total_epsilon = epsilon_per_round * sqrt(num_rounds)
"""
if ctx.sender != self.state.owner:
raise ValueError("only owner can force aggregation")
contributions = self.state.contributions_this_round
if len(contributions) == 0:
raise ValueError("no contributions to aggregate")
num_contribs = len(contributions)
input_dim = self.state.input_dim
# Average the noised gradients
avg_gradient = [0.0] * input_dim
avg_bias_grad = 0.0
for contrib in contributions:
for i in range(input_dim):
avg_gradient[i] += contrib["gradient"][i]
avg_bias_grad += contrib["bias_grad"]
for i in range(input_dim):
avg_gradient[i] /= num_contribs
avg_bias_grad /= num_contribs
# Apply SGD step: w = w - lr * avg_gradient
lr = self.state.learning_rate
new_weights = list(self.state.weights)
for i in range(input_dim):
new_weights[i] -= lr * avg_gradient[i]
new_bias = self.state.bias - lr * avg_bias_grad
self.state.weights = new_weights
self.state.bias = new_bias
# Advance round
new_round = self.state.round + 1
self.state.round = new_round
self.state.contributions_this_round = []
# Update privacy accounting: total_eps = eps_per_round * sqrt(rounds)
self.state.total_epsilon_spent = self.state.epsilon_per_round * math.sqrt(float(new_round))
# Check if budget is exhausted
if self.state.total_epsilon_spent >= self.state.epsilon_budget:
self.emit(
event.BudgetExhausted(
total_epsilon=round(self.state.total_epsilon_spent, 6),
budget=self.state.epsilon_budget,
rounds_completed=new_round,
)
)
self.emit(
event.RoundAggregated(
round=new_round,
num_contributions=num_contribs,
total_epsilon_spent=round(self.state.total_epsilon_spent, 6),
)
)
The privacy_report query returns a complete summary of the privacy accounting state:
@query
def privacy_report(self) -> dict:
"""Return detailed privacy accounting information.
Returns:
dict with epsilon spent, remaining budget, noise scale,
delta, and per-round epsilon.
"""
remaining = self.state.epsilon_budget - self.state.total_epsilon_spent
return {
"epsilon_budget": self.state.epsilon_budget,
"epsilon_spent": round(self.state.total_epsilon_spent, 6),
"epsilon_remaining": round(max(0.0, remaining), 6),
"epsilon_per_round": round(self.state.epsilon_per_round, 6),
"noise_scale": round(self.state.noise_scale, 6),
"delta": self.state.delta,
"rounds_completed": self.state.round,
"clip_norm": self.state.clip_norm,
}
How DP-SGD Works in This Contract
L2 gradient clipping bounds the influence of any single contributor. Before submission, each party clips their gradient vector so that its L2 norm does not exceed clip_norm (C). The contract verifies this bound on-chain and re-clips if necessary. This ensures the sensitivity of any single gradient contribution is at most C, regardless of the private data used to compute it.
Gaussian noise calibration provides the formal DP guarantee. The contract computes a noise scale using the analytic Gaussian mechanism: sigma = sqrt(2 * ln(1.25 / delta)) * C / epsilon_per_round. Each dimension of the gradient receives independent Gaussian noise at this scale. The noise is generated deterministically via the Box-Muller transform using SHA-256 seeds derived from the sender address, round number, and dimension index -- ensuring identical noise across all validators.
Epsilon budget tracking limits the total information leakage across all rounds. The contract uses basic composition: after T aggregation rounds, the total epsilon spent is epsilon_per_round * sqrt(T). Once total_epsilon_spent reaches the configured epsilon_budget, the contract rejects further gradient contributions. The privacy_report query lets anyone inspect how much budget remains.
The @proof(type="validity") decorator on contribute_dp_gradient ensures that the gradient clipping and noise addition are verifiable. A ZK validity proof attests that the contract correctly clipped the gradient and added the prescribed noise, without requiring re-execution.
Secure Federated Learning
The SecureFederatedLearner contract implements federated gradient descent with secure aggregation, following the Bonawitz et al. ("Practical Secure Aggregation for Privacy-Preserving Machine Learning", CCS 2017) pairwise masking protocol. The contract never sees any individual party's true gradient -- only the aggregate sum, which reveals no single party's contribution.
from panda import contract, constructor, call, query, event
@contract
class SecureFederatedLearner:
"""Federated linear model with secure aggregation via pairwise masking."""
class State:
weights: list = []
bias: float = 0.0
input_dim: int = 0
round: int = 0
registered_parties: list = []
masked_submissions: dict = {}
phase: str = "registration"
learning_rate: float = 0.01
min_parties: int = 2
owner: str = ""
contributions: int = 0
total_rounds: int = 0
@constructor
def deploy(self, ctx, input_dim: int, learning_rate: float, min_parties: int):
"""Initialize the secure federated model.
Args:
input_dim: number of input features for the linear model.
learning_rate: SGD learning rate applied during aggregation.
min_parties: minimum number of parties required per round.
"""
if input_dim <= 0:
raise ValueError("input_dim must be positive")
if input_dim > 10000:
raise ValueError(f"input_dim must be <= 10000, got {input_dim}")
if learning_rate <= 0:
raise ValueError("learning_rate must be positive")
if min_parties < 2:
raise ValueError("min_parties must be >= 2")
self.state.owner = ctx.sender
self.state.input_dim = input_dim
self.state.weights = [0.0] * input_dim
self.state.bias = 0.0
self.state.learning_rate = learning_rate
self.state.min_parties = min_parties
self.state.phase = "registration"
self.state.round = 0
self.state.total_rounds = 0
self.state.registered_parties = []
self.state.masked_submissions = {}
self.state.contributions = 0
self.emit(
event.RoundOpened(
round=0,
phase="registration",
min_parties=min_parties,
owner=ctx.sender,
)
)
Participants register for a round, then submit masked gradients once the owner opens the submission phase:
<!-- Source: contracts/privacy/federated/secure_federated_learner.py --> @call
def register_for_round(self, ctx):
"""Register to participate in the current training round.
Can only be called during the registration phase. Each address
can register at most once per round.
"""
if self.state.phase != "registration":
raise ValueError(f"registration closed: current phase is '{self.state.phase}'")
sender = ctx.sender
parties = list(self.state.registered_parties)
if sender in parties:
raise ValueError("already registered for this round")
parties.append(sender)
self.state.registered_parties = parties
self.emit(
event.PartyRegistered(
party=sender,
round=self.state.round,
total_registered=len(parties),
)
)
@call
def start_submission(self, ctx):
"""Transition from registration to submission phase.
Only the contract owner can trigger this, and only after at least
min_parties have registered. Once submission starts, no more
registrations are accepted.
"""
if ctx.sender != self.state.owner:
raise ValueError("only owner can start submission phase")
if self.state.phase != "registration":
raise ValueError("not in registration phase")
num_registered = len(self.state.registered_parties)
if num_registered < self.state.min_parties:
raise ValueError(
f"need at least {self.state.min_parties} parties, only {num_registered} registered"
)
self.state.phase = "submission"
self.state.masked_submissions = {}
@call
def submit_masked_gradient(self, ctx, masked_weights: list, masked_bias: float):
"""Submit a masked gradient for the current round.
The caller must compute pairwise masks OFF-CHAIN with every other
registered party and add/subtract them from their true gradient
before submission. The masking protocol:
For each pair (addr_i, addr_j) where addr_i < addr_j:
- shared_key = derive_key(min(addr_i,addr_j) + max(addr_i,addr_j),
str(round))
- mask = float values derived from shared_key
- addr_i ADDS mask to their gradient
- addr_j SUBTRACTS mask from their gradient
When all masked gradients are summed, the pairwise masks cancel.
Args:
masked_weights: gradient for weights with pairwise masks applied
(length = input_dim).
masked_bias: gradient for bias with pairwise masks applied.
"""
if self.state.phase != "submission":
raise ValueError(f"not in submission phase: current phase is '{self.state.phase}'")
sender = ctx.sender
parties = self.state.registered_parties
if sender not in parties:
raise ValueError("not registered for this round")
if len(masked_weights) != self.state.input_dim:
raise ValueError(
f"gradient length {len(masked_weights)} != input_dim {self.state.input_dim}"
)
submissions = dict(self.state.masked_submissions)
if sender in submissions:
raise ValueError("already submitted for this round")
submissions[sender] = {
"weights": list(masked_weights),
"bias": float(masked_bias),
}
self.state.masked_submissions = submissions
self.state.contributions = self.state.contributions + 1
self.emit(
event.GradientSubmitted(
party=sender,
round=self.state.round,
total_submitted=len(submissions),
total_registered=len(parties),
)
)
Once all registered parties have submitted, the owner triggers aggregation. The sum of all masked gradients cancels the pairwise masks, yielding the true aggregate gradient:
<!-- Source: contracts/privacy/federated/secure_federated_learner.py --> @call
def aggregate(self, ctx):
"""Aggregate all masked submissions and apply the SGD step.
Sums all masked gradients (pairwise masks cancel in the sum),
divides by the number of parties to get the average gradient,
and applies a single SGD step to the global model. Resets state
for the next round.
Only the contract owner can trigger aggregation, and only after
all registered parties have submitted.
"""
if ctx.sender != self.state.owner:
raise ValueError("only owner can aggregate")
if self.state.phase != "submission":
raise ValueError("not in submission phase")
submissions = self.state.masked_submissions
parties = self.state.registered_parties
num_parties = len(parties)
if len(submissions) < num_parties:
raise ValueError(f"waiting for submissions: {len(submissions)}/{num_parties}")
input_dim = self.state.input_dim
# Sum all masked gradients -- pairwise masks cancel in the sum
agg_weights = [0.0] * input_dim
agg_bias = 0.0
for party_addr in parties:
sub = submissions[party_addr]
for i in range(input_dim):
agg_weights[i] += sub["weights"][i]
agg_bias += sub["bias"]
# Average
for i in range(input_dim):
agg_weights[i] /= num_parties
agg_bias /= num_parties
# Apply SGD step: w = w - lr * avg_gradient
lr = self.state.learning_rate
new_weights = list(self.state.weights)
for i in range(input_dim):
new_weights[i] -= lr * agg_weights[i]
new_bias = self.state.bias - lr * agg_bias
self.state.weights = new_weights
self.state.bias = new_bias
# Advance to next round
new_round = self.state.round + 1
self.state.round = new_round
self.state.total_rounds = self.state.total_rounds + 1
self.state.phase = "registration"
self.state.registered_parties = []
self.state.masked_submissions = {}
self.emit(
event.RoundAggregated(
round=new_round,
num_parties=num_parties,
)
)
The round_status query lets participants inspect the current phase and registration count:
@query
def round_status(self) -> dict:
"""Return the status of the current training round.
Returns:
dict with current phase, registered party count, and
submission count.
"""
return {
"round": self.state.round,
"phase": self.state.phase,
"registered_parties": len(self.state.registered_parties),
"submissions_received": len(self.state.masked_submissions),
"min_parties": self.state.min_parties,
}
How Secure Aggregation Works in This Contract
The protocol progresses through three phases per round:
Registration phase. Parties call register_for_round to join the current training round. Each address can register at most once. The owner waits until at least min_parties have registered before opening submission.
Submission phase. The owner calls start_submission to transition the contract. Each registered party then computes their true gradient locally on private data. Before submitting, they generate pairwise masks OFF-CHAIN using derive_key() from the panda.crypto module. For each pair of parties (addr_i, addr_j) where addr_i < addr_j, both parties derive the same deterministic mask from derive_key(min(addr_i, addr_j) + max(addr_i, addr_j), str(round)). Party i adds the mask to their gradient; party j subtracts it. Each party then submits their masked gradient via submit_masked_gradient.
Aggregation phase. Once all registered parties have submitted, the owner calls aggregate. The contract sums all masked gradients. Because each pairwise mask appears exactly once with a positive sign and once with a negative sign, all masks cancel in the sum. The result is the true aggregate gradient, which is then averaged and applied as a single SGD step. The contract resets to the registration phase for the next round.
Security guarantee. The contract (and any on-chain observer) only ever sees masked individual gradients and the final aggregate. Recovering any single party's true gradient would require colluding with ALL other parties in that round to reconstruct every pairwise mask. As long as at least one other honest participant exists, individual gradients remain hidden.
Phase progression. Each training round follows the cycle: registration (parties join) -> submission (masked gradients submitted) -> aggregation (masks cancel, SGD step applied) -> registration (next round). The round_status query reports the current phase, number of registered parties, and number of submissions received, so participants can coordinate off-chain.
Privacy Patterns Summary
| Pattern | Privacy Guarantee | Use Case |
|---|---|---|
@private + XOR/AES encryption | Encrypted state | Proprietary models, trade secrets |
| ZK Membership (Merkle + nullifiers) | Anonymous authentication | Voting, whistleblowing |
| Secure Aggregation (pairwise masking) | Individual values hidden | Surveys, benchmarks |
| Federated Learning (gradient SGD) | Raw data stays local | Collaborative ML |
| Differential Privacy (Laplace noise) | Statistical privacy | Census data, salary surveys |
| DP Federated Learning (DP-SGD) | Gradient clipping + Gaussian noise + epsilon budget | Privacy-preserving collaborative ML |
| Secure Federated Learning (pairwise masking) | Individual gradients hidden via mask cancellation | Multi-party model training |
Try It
- Open the Playground and select the Private Model template
- See the Crypto & Security guide for commit-reveal and Merkle proofs
- See the Proofs guide for
@proofdecorator details and ZK backend options