feat(e2e): EncryptedTrustBridge — handshake-gated encrypted channels (microsoft#1238)

imran-siddique · Copilot · web-flow · commit e77f0faea336 · 2026-04-20T12:36:38.000-07:00
* docs: refresh proposals index — close stale, add fresh submissions - Replaced 14 closed pre-release proposals with 6 fresh submissions (v3.2.0) - New active: CoSAI WS4 microsoft#86, OWASP ASI #12, CrewAI #5562, AutoGen #7613, Google ADK #5418, MCP Servers #3988 - Removed stale awesome-list PRs and misc integration issues - Cleaned up summary table Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * feat(e2e): EncryptedTrustBridge — handshake-gated encrypted channels (microsoft#1226) Integrates TrustHandshake with SecureChannel so agents must pass trust verification before an encrypted channel is established. New files: - agentmesh/encryption/bridge.py — EncryptedTrustBridge - open_secure_channel(): verify peer then X3DH + Double Ratchet - accept_secure_channel(): responder side - publish_prekey_bundle(): distribute pre-keys - Session management: get/close/close_all - Agent DIDs bound as associated data - tests/test_encrypted_bridge.py — 10 tests covering: - Full open/accept/exchange flow - 5-turn bidirectional conversation - Session close with key cleanup - Multi-session management (close all) - Pre-key bundle publishing with/without OTK - Associated data binding Closes microsoft#1226 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
diff --git a/packages/agent-mesh/src/agentmesh/encryption/bridge.py b/packages/agent-mesh/src/agentmesh/encryption/bridge.py
@@ -0,0 +1,234 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+"""Encrypted trust bridge — TrustHandshake + SecureChannel integration.
+
+Extends TrustBridge.verify_peer() to optionally establish an E2E
+encrypted SecureChannel after successful authentication. The trust
+handshake authenticates the peer (Ed25519 challenge-response + trust
+score check), then X3DH + Double Ratchet provide forward-secret
+encrypted messaging.
+
+Usage:
+    bridge = EncryptedTrustBridge(
+        agent_did="did:mesh:alice",
+        key_manager=alice_key_manager,
+    )
+    channel = await bridge.open_secure_channel("did:mesh:bob", bob_bundle)
+    ciphertext = channel.send(b"governed action")
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import Optional
+
+from agentmesh.encryption.channel import ChannelEstablishment, SecureChannel
+from agentmesh.encryption.x3dh import InMemoryPreKeyStore, PreKeyBundle, PreKeyStore, X3DHKeyManager
+from agentmesh.trust.bridge import TrustBridge
+from agentmesh.trust.handshake import HandshakeResult
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class EncryptedPeerSession:
+    """An active encrypted session with a verified peer."""
+
+    peer_did: str
+    channel: SecureChannel
+    handshake_result: HandshakeResult
+    establishment: ChannelEstablishment | None = None
+
+
+class EncryptedTrustBridge:
+    """Trust bridge that gates encrypted channels on successful handshake.
+
+    Agents must pass the trust handshake (identity verification + trust
+    score threshold) before an encrypted channel is established. Peers
+    that fail the handshake never reach the key exchange step.
+    """
+
+    def __init__(
+        self,
+        agent_did: str,
+        key_manager: X3DHKeyManager,
+        trust_bridge: TrustBridge | None = None,
+        prekey_store: PreKeyStore | None = None,
+        min_trust_score: int = 700,
+    ) -> None:
+        """Initialize the encrypted trust bridge.
+
+        Args:
+            agent_did: This agent's DID.
+            key_manager: X3DH key manager with this agent's identity keys.
+            trust_bridge: Optional existing TrustBridge instance. If None,
+                a new one is created.
+            prekey_store: Pre-key storage backend. Defaults to in-memory.
+            min_trust_score: Minimum trust score required to open a channel.
+        """
+        self._agent_did = agent_did
+        self._key_manager = key_manager
+        self._bridge = trust_bridge or TrustBridge(agent_did=agent_did)
+        self._prekey_store = prekey_store or InMemoryPreKeyStore()
+        self._min_trust_score = min_trust_score
+        self._sessions: dict[str, EncryptedPeerSession] = {}
+
+        # Ensure we have a signed pre-key for receiving channels
+        if key_manager.signed_pre_key is None:
+            key_manager.generate_signed_pre_key()
+
+    @property
+    def agent_did(self) -> str:
+        return self._agent_did
+
+    @property
+    def active_sessions(self) -> dict[str, EncryptedPeerSession]:
+        """Return all active encrypted sessions, keyed by peer DID."""
+        return dict(self._sessions)
+
+    def publish_prekey_bundle(self, include_otk: bool = True) -> PreKeyBundle:
+        """Generate and publish this agent's pre-key bundle.
+
+        Args:
+            include_otk: Whether to include a one-time pre-key.
+
+        Returns:
+            The public PreKeyBundle for distribution to peers.
+        """
+        otk_id = None
+        if include_otk:
+            otks = self._key_manager.generate_one_time_pre_keys(1)
+            otk_id = otks[0].key_id
+
+        bundle = self._key_manager.get_public_bundle(otk_id=otk_id)
+        self._prekey_store.store_bundle(self._agent_did, bundle)
+        return bundle
+
+    async def open_secure_channel(
+        self,
+        peer_did: str,
+        peer_bundle: PreKeyBundle,
+        required_trust_score: int | None = None,
+        skip_handshake: bool = False,
+    ) -> SecureChannel:
+        """Open an E2E encrypted channel to a peer after trust verification.
+
+        The flow is:
+        1. Run TrustHandshake (Ed25519 challenge-response + trust check)
+        2. If verified, run X3DH key agreement
+        3. Initialize Double Ratchet from shared secret
+        4. Return SecureChannel ready for send/receive
+
+        Args:
+            peer_did: The peer agent's DID.
+            peer_bundle: The peer's published pre-key bundle.
+            required_trust_score: Override the default trust threshold.
+            skip_handshake: If True, skip trust verification (for testing
+                or pre-verified peers only).
+
+        Returns:
+            A SecureChannel for encrypted communication.
+
+        Raises:
+            PermissionError: If the peer fails trust verification.
+        """
+        threshold = required_trust_score or self._min_trust_score
+
+        # Step 1: Trust verification
+        if not skip_handshake:
+            result = await self._bridge.verify_peer(
+                peer_did=peer_did,
+                required_trust_score=threshold,
+            )
+            if not result.verified:
+                raise PermissionError(
+                    f"Peer {peer_did} failed trust verification: {result.rejection_reason}"
+                )
+            handshake_result = result
+        else:
+            handshake_result = HandshakeResult(
+                verified=True,
+                peer_did=peer_did,
+                trust_score=threshold,
+                trust_level="trusted",
+            )
+
+        # Step 2: X3DH + Double Ratchet via SecureChannel
+        ad = f"{self._agent_did}|{peer_did}".encode()
+        channel, establishment = SecureChannel.create_sender(
+            self._key_manager, peer_bundle, associated_data=ad
+        )
+
+        session = EncryptedPeerSession(
+            peer_did=peer_did,
+            channel=channel,
+            handshake_result=handshake_result,
+            establishment=establishment,
+        )
+        self._sessions[peer_did] = session
+
+        logger.info(
+            "Opened encrypted channel to %s (trust=%d, level=%s)",
+            peer_did,
+            handshake_result.trust_score,
+            handshake_result.trust_level,
+        )
+        return channel
+
+    def accept_secure_channel(
+        self,
+        peer_did: str,
+        establishment: ChannelEstablishment,
+        handshake_result: HandshakeResult | None = None,
+    ) -> SecureChannel:
+        """Accept an incoming encrypted channel from a verified peer.
+
+        Args:
+            peer_did: The initiating peer's DID.
+            establishment: The ChannelEstablishment from the initiator.
+            handshake_result: Optional handshake result if already verified.
+
+        Returns:
+            A SecureChannel for encrypted communication.
+        """
+        ad = f"{peer_did}|{self._agent_did}".encode()
+        channel = SecureChannel.create_receiver(
+            self._key_manager, establishment, associated_data=ad
+        )
+
+        session = EncryptedPeerSession(
+            peer_did=peer_did,
+            channel=channel,
+            handshake_result=handshake_result or HandshakeResult(
+                verified=True, peer_did=peer_did,
+            ),
+        )
+        self._sessions[peer_did] = session
+
+        logger.info("Accepted encrypted channel from %s", peer_did)
+        return channel
+
+    def get_session(self, peer_did: str) -> EncryptedPeerSession | None:
+        """Get an active session with a peer."""
+        return self._sessions.get(peer_did)
+
+    def close_session(self, peer_did: str) -> bool:
+        """Close an encrypted session and clear key material.
+
+        Returns:
+            True if a session existed and was closed.
+        """
+        session = self._sessions.pop(peer_did, None)
+        if session is None:
+            return False
+        session.channel.close()
+        logger.info("Closed encrypted session with %s", peer_did)
+        return True
+
+    def close_all_sessions(self) -> int:
+        """Close all active sessions. Returns the number closed."""
+        count = len(self._sessions)
+        for peer_did in list(self._sessions.keys()):
+            self.close_session(peer_did)
+        return count
diff --git a/packages/agent-mesh/tests/test_encrypted_bridge.py b/packages/agent-mesh/tests/test_encrypted_bridge.py
@@ -0,0 +1,160 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+"""Tests for EncryptedTrustBridge — handshake + encrypted channel integration."""
+
+import pytest
+from nacl.signing import SigningKey
+
+from agentmesh.encryption.bridge import EncryptedPeerSession, EncryptedTrustBridge
+from agentmesh.encryption.channel import SecureChannel
+from agentmesh.encryption.x3dh import X3DHKeyManager
+
+
+def _make_manager() -> X3DHKeyManager:
+    sk = SigningKey.generate()
+    return X3DHKeyManager.from_ed25519_keys(
+        bytes(sk) + bytes(sk.verify_key), bytes(sk.verify_key)
+    )
+
+
+def _make_bridge(did: str) -> tuple[EncryptedTrustBridge, X3DHKeyManager]:
+    mgr = _make_manager()
+    bridge = EncryptedTrustBridge(agent_did=did, key_manager=mgr, min_trust_score=500)
+    return bridge, mgr
+
+
+class TestEncryptedTrustBridge:
+    @pytest.mark.asyncio
+    async def test_open_and_accept_channel(self):
+        """Full flow: Alice opens channel, Bob accepts, they exchange messages."""
+        alice_bridge, alice_mgr = _make_bridge("did:mesh:alice")
+        bob_bridge, bob_mgr = _make_bridge("did:mesh:bob")
+
+        bob_bundle = bob_bridge.publish_prekey_bundle()
+
+        alice_ch = await alice_bridge.open_secure_channel(
+            "did:mesh:bob", bob_bundle, skip_handshake=True
+        )
+        alice_session = alice_bridge.get_session("did:mesh:bob")
+        assert alice_session is not None
+
+        bob_ch = bob_bridge.accept_secure_channel(
+            "did:mesh:alice", alice_session.establishment
+        )
+
+        enc = alice_ch.send(b"hello bob")
+        assert bob_ch.receive(enc) == b"hello bob"
+
+        enc2 = bob_ch.send(b"hello alice")
+        assert alice_ch.receive(enc2) == b"hello alice"
+
+    @pytest.mark.asyncio
+    async def test_bidirectional_conversation(self):
+        """Multi-turn conversation through encrypted bridge."""
+        alice_bridge, _ = _make_bridge("did:mesh:alice")
+        bob_bridge, _ = _make_bridge("did:mesh:bob")
+
+        bob_bundle = bob_bridge.publish_prekey_bundle()
+        alice_ch = await alice_bridge.open_secure_channel(
+            "did:mesh:bob", bob_bundle, skip_handshake=True
+        )
+        bob_ch = bob_bridge.accept_secure_channel(
+            "did:mesh:alice", alice_bridge.get_session("did:mesh:bob").establishment
+        )
+
+        for i in range(5):
+            enc = alice_ch.send(f"alice-{i}".encode())
+            assert bob_ch.receive(enc) == f"alice-{i}".encode()
+
+            enc = bob_ch.send(f"bob-{i}".encode())
+            assert alice_ch.receive(enc) == f"bob-{i}".encode()
+
+    @pytest.mark.asyncio
+    async def test_close_session(self):
+        alice_bridge, _ = _make_bridge("did:mesh:alice")
+        bob_bridge, _ = _make_bridge("did:mesh:bob")
+
+        bob_bundle = bob_bridge.publish_prekey_bundle()
+        alice_ch = await alice_bridge.open_secure_channel(
+            "did:mesh:bob", bob_bundle, skip_handshake=True
+        )
+
+        assert alice_bridge.close_session("did:mesh:bob") is True
+        assert alice_bridge.get_session("did:mesh:bob") is None
+        assert alice_ch.is_closed is True
+
+    @pytest.mark.asyncio
+    async def test_close_nonexistent_session(self):
+        bridge, _ = _make_bridge("did:mesh:alice")
+        assert bridge.close_session("did:mesh:unknown") is False
+
+    @pytest.mark.asyncio
+    async def test_close_all_sessions(self):
+        alice_bridge, _ = _make_bridge("did:mesh:alice")
+        bob_bridge, _ = _make_bridge("did:mesh:bob")
+        carol_bridge, _ = _make_bridge("did:mesh:carol")
+
+        bob_bundle = bob_bridge.publish_prekey_bundle()
+        carol_bundle = carol_bridge.publish_prekey_bundle()
+
+        await alice_bridge.open_secure_channel(
+            "did:mesh:bob", bob_bundle, skip_handshake=True
+        )
+        await alice_bridge.open_secure_channel(
+            "did:mesh:carol", carol_bundle, skip_handshake=True
+        )
+
+        assert len(alice_bridge.active_sessions) == 2
+        closed = alice_bridge.close_all_sessions()
+        assert closed == 2
+        assert len(alice_bridge.active_sessions) == 0
+
+    @pytest.mark.asyncio
+    async def test_publish_prekey_bundle(self):
+        bridge, mgr = _make_bridge("did:mesh:alice")
+        bundle = bridge.publish_prekey_bundle(include_otk=True)
+        assert len(bundle.identity_key) == 32
+        assert len(bundle.signed_pre_key) == 32
+        assert bundle.one_time_pre_key is not None
+
+    @pytest.mark.asyncio
+    async def test_publish_prekey_bundle_no_otk(self):
+        bridge, _ = _make_bridge("did:mesh:alice")
+        bundle = bridge.publish_prekey_bundle(include_otk=False)
+        assert bundle.one_time_pre_key is None
+
+    @pytest.mark.asyncio
+    async def test_active_sessions_property(self):
+        bridge, _ = _make_bridge("did:mesh:alice")
+        assert len(bridge.active_sessions) == 0
+
+        bob_bridge, _ = _make_bridge("did:mesh:bob")
+        bob_bundle = bob_bridge.publish_prekey_bundle()
+        await bridge.open_secure_channel("did:mesh:bob", bob_bundle, skip_handshake=True)
+
+        sessions = bridge.active_sessions
+        assert "did:mesh:bob" in sessions
+        assert isinstance(sessions["did:mesh:bob"], EncryptedPeerSession)
+
+    @pytest.mark.asyncio
+    async def test_agent_did_property(self):
+        bridge, _ = _make_bridge("did:mesh:test-agent")
+        assert bridge.agent_did == "did:mesh:test-agent"
+
+    @pytest.mark.asyncio
+    async def test_channel_with_associated_data(self):
+        """Channels bind agent DIDs as associated data."""
+        alice_bridge, _ = _make_bridge("did:mesh:alice")
+        bob_bridge, _ = _make_bridge("did:mesh:bob")
+
+        bob_bundle = bob_bridge.publish_prekey_bundle()
+        alice_ch = await alice_bridge.open_secure_channel(
+            "did:mesh:bob", bob_bundle, skip_handshake=True
+        )
+        bob_ch = bob_bridge.accept_secure_channel(
+            "did:mesh:alice", alice_bridge.get_session("did:mesh:bob").establishment
+        )
+
+        # Messages work because AD matches
+        enc = alice_ch.send(b"test")
+        assert bob_ch.receive(enc) == b"test"
