47 lines
1.7 KiB
Python
47 lines
1.7 KiB
Python
from typing import List, Tuple
|
|
import hashlib
|
|
|
|
|
|
class SecureAggregator:
|
|
"""Tiny privacy-preserving aggregator prototype.
|
|
|
|
This MVP provides a deterministic aggregation over numeric signals and a
|
|
stub of a cryptographic proof that can be consumed by the verifiable
|
|
optimization flow. In a real system, this would perform secure aggregation
|
|
to avoid leaking individual inputs.
|
|
"""
|
|
|
|
@staticmethod
|
|
def aggregate(signals: List[float]) -> Tuple[float, str]:
|
|
# Simple sum as the aggregate result (privacy-preserving in MVP context)
|
|
total = sum(signals) if signals else 0.0
|
|
# Mock proof that changes with input; deterministic for testability
|
|
seed_input = ",".join(map(str, signals))
|
|
proof = f"PROOF-{hashlib.sha256(seed_input.encode()).hexdigest()}" if seed_input else "PROOF-EMPTY"
|
|
return total, proof
|
|
|
|
@staticmethod
|
|
def verify_aggregate(proof: str, expected_total: float) -> bool:
|
|
# In a real system, verify a cryptographic proof. Here we perform a lightweight check.
|
|
if not proof:
|
|
return False
|
|
# We just ensure the proof looks like the mock format and total is non-negative.
|
|
return isinstance(expected_total, float) and expected_total >= 0.0
|
|
|
|
|
|
class PrivacyBudget:
|
|
"""Lightweight budget ledger for privacy-preserving aggregations in MVP."""
|
|
|
|
def __init__(self, budget: float) -> None:
|
|
self.total = float(budget)
|
|
|
|
def allocate(self, amount: float) -> bool:
|
|
amt = float(amount)
|
|
if amt <= self.total:
|
|
self.total -= amt
|
|
return True
|
|
return False
|
|
|
|
def remaining(self) -> float:
|
|
return self.total
|