From a152a10f4a3391be87cb24fda661d0335ff2c5be Mon Sep 17 00:00:00 2001 From: agent-4b796a86eacc591f Date: Thu, 16 Apr 2026 22:51:09 +0200 Subject: [PATCH] build(agent): molt-az#4b796a iteration --- README.md | 6 ++++++ .../__init__.py | 3 ++- .../attestation.py | 10 ++++++++++ .../bridge.py | 12 ++++++++++++ .../governance.py | 10 ++++++++-- .../privacy.py | 17 +++++++++++++++++ .../transport.py | 11 +++++++++++ 7 files changed, 66 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 0ef8c81..8300073 100644 --- a/README.md +++ b/README.md @@ -36,3 +36,9 @@ How to run tests: bash test.sh This README also serves as a marketing and onboarding document to explain the architecture and how to contribute. + +Notes on MVP extensions (EnergiBridge focus): +- EnergiBridge provides a canonical interoperability layer mapping GridGuard primitives to a CatOpt-like representation with Objects, Morphisms, and PlanDelta. +- Attestation and DID-based identities are wired into AttestedAgent and governance flows; key rotation and hardware-backed attestations are supported in MVP form. +- Verifiable optimization is stubbed with a ZK-proof placeholder; privacy budgets and secure aggregation hooks are available for exploration. +- Governance ledger supports append-only entries with signatures and public anchoring options. diff --git a/src/gridguard_secure_attested_cross_domain_e/__init__.py b/src/gridguard_secure_attested_cross_domain_e/__init__.py index ea3a777..9ae3032 100644 --- a/src/gridguard_secure_attested_cross_domain_e/__init__.py +++ b/src/gridguard_secure_attested_cross_domain_e/__init__.py @@ -11,7 +11,7 @@ from .transport import TransportLayer from .governance import GovernanceLedger, DeltaSync from .marketplace import AdaptersMarketplace from .simulation import SimulationHarness -from .privacy import SecureAggregator +from .privacy import SecureAggregator, PrivacyBudget from .dsl import LocalProblem, SharedVariables, PlanDelta, PolicyBlock, AttestationHint from .bridge import to_canonical, from_canonical @@ -25,6 +25,7 @@ __all__ = [ "AdaptersMarketplace", "SimulationHarness", "SecureAggregator", + "PrivacyBudget", "LocalProblem", "SharedVariables", "PlanDelta", diff --git a/src/gridguard_secure_attested_cross_domain_e/attestation.py b/src/gridguard_secure_attested_cross_domain_e/attestation.py index 9acedba..9ae8740 100644 --- a/src/gridguard_secure_attested_cross_domain_e/attestation.py +++ b/src/gridguard_secure_attested_cross_domain_e/attestation.py @@ -21,3 +21,13 @@ class AttestedAgent: @property def credential(self) -> Optional[str]: return self._credential + + def verify_credential(self, credential: str) -> bool: + """Lightweight credential verifier. + + In a real deployment this would cryptographically verify the + remote attestation report. Here we perform a deterministic check + against the produced credential for testability and auditing. + """ + expected = f"attest-{self.agent_id}-{self.hardware}-v1" + return credential == expected diff --git a/src/gridguard_secure_attested_cross_domain_e/bridge.py b/src/gridguard_secure_attested_cross_domain_e/bridge.py index eb52487..2b0479f 100644 --- a/src/gridguard_secure_attested_cross_domain_e/bridge.py +++ b/src/gridguard_secure_attested_cross_domain_e/bridge.py @@ -25,3 +25,15 @@ def from_canonical(canonical: Dict[str, Any]) -> Dict[str, Any]: """Inverse of to_canonical for MVP playgrounds.""" objs = canonical.get("Objects", {}) return {"LocalProblems": objs.get("LocalProblems", [])} + + +class EnergiBridge: + """EnergiBridge façade: canonical <-> local representation bridge for MVP.""" + + @staticmethod + def to_canonical(local_problem: Dict[str, Any]) -> Dict[str, Any]: + return to_canonical(local_problem) + + @staticmethod + def from_canonical(canonical: Dict[str, Any]) -> Dict[str, Any]: + return from_canonical(canonical) diff --git a/src/gridguard_secure_attested_cross_domain_e/governance.py b/src/gridguard_secure_attested_cross_domain_e/governance.py index f127c72..4766257 100644 --- a/src/gridguard_secure_attested_cross_domain_e/governance.py +++ b/src/gridguard_secure_attested_cross_domain_e/governance.py @@ -36,7 +36,7 @@ class DeltaSync: return merged - def anchor_to_public(self, public_anchor_url: str) -> str: + def anchor_to_public(self, public_anchor_url: str, events: List[Dict[str, Any]]) -> str: """Create a simple anchor string for cross-organization auditability. This simulates anchoring the current ledger state to an external, public @@ -44,6 +44,12 @@ class DeltaSync: combining the provided base with a hash of the ledger contents. """ # Simple hash of all events to serve as a tamper-evident root - ledger_bytes = str(self.get_events()).encode() + ledger_bytes = str(events).encode() root_hash = hashlib.sha256(ledger_bytes).hexdigest() return f"{public_anchor_url}#root={root_hash}" + + # Lightweight verifier for external proofs (optional enhancement) + @staticmethod + def verify_proof(proof: Dict[str, Any]) -> bool: + # In MVP, a proof is valid if it contains a truthy 'valid' flag set to True + return bool(proof) and bool(proof.get("valid")) diff --git a/src/gridguard_secure_attested_cross_domain_e/privacy.py b/src/gridguard_secure_attested_cross_domain_e/privacy.py index 3d040bc..00ecaf9 100644 --- a/src/gridguard_secure_attested_cross_domain_e/privacy.py +++ b/src/gridguard_secure_attested_cross_domain_e/privacy.py @@ -27,3 +27,20 @@ class SecureAggregator: 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 diff --git a/src/gridguard_secure_attested_cross_domain_e/transport.py b/src/gridguard_secure_attested_cross_domain_e/transport.py index 1df1192..57adc6c 100644 --- a/src/gridguard_secure_attested_cross_domain_e/transport.py +++ b/src/gridguard_secure_attested_cross_domain_e/transport.py @@ -20,3 +20,14 @@ class TransportLayer: def rotate_keys(self) -> None: self._current_key = f"KEY_{hash(self._current_key) & 0xFFFFFFFF}" + + # Lightweight helpers for MVP: establish a mutual-authenticated channel + def establish_secure_channel(self, peer: str) -> dict: + """Establish a secured channel with a peer, performing a mock mutual-auth check.""" + if peer not in self._peer_keys: + # If no prior key exists, simulate a handshake producing a fresh key + self.create_channel(peer) + return self.create_channel(peer) + + def get_peer_key(self, peer: str) -> str: + return self._peer_keys.get(peer, "")