feat(runtime-host): discover peer mesh routes

native/runtime-host-peer/src/bindings.rs

@@ -31,6 +31,7 @@ use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot, watch};
 use crate::engine::{self, EngineCommand, PeerError, StreamCommand};
 
 type IncomingStreamReceiver = mpsc::Receiver<std::result::Result<Vec<u8>, PeerError>>;
+const IDENTITY_PAYLOAD_MAX_BYTES: usize = 8 * 1024;
 
 #[napi(object)]
 pub struct StartPeerEndpointOptions {
@@ -49,6 +50,12 @@ pub struct ConnectPeerOptions {
     pub direct_deadline_ms: u32,
 }
 
+#[napi(object)]
+pub struct PeerIdentitySignature {
+    pub public_key: Buffer,
+    pub signature: Buffer,
+}
+
 #[napi]
 pub struct PeerEndpoint {
     peer_id: String,
@@ -301,6 +308,38 @@ pub async fn ensure_peer_identity(key_path: String) -> Result<String> {
         .map_err(peer_error)
 }
 
+#[napi]
+pub async fn sign_peer_identity(
+    key_path: String,
+    expected_peer_id: String,
+    payload: Buffer,
+) -> Result<PeerIdentitySignature> {
+    validate_identity_payload(&payload)?;
+    let signed = engine::sign_identity(
+        PathBuf::from(key_path),
+        parse_peer_id(&expected_peer_id)?,
+        &payload,
+    )
+    .await
+    .map_err(peer_error)?;
+    Ok(PeerIdentitySignature {
+        public_key: signed.public_key.into(),
+        signature: signed.signature.into(),
+    })
+}
+
+#[napi]
+pub fn verify_peer_identity(
+    peer_id: String,
+    public_key: Buffer,
+    payload: Buffer,
+    signature: Buffer,
+) -> Result<bool> {
+    validate_identity_payload(&payload)?;
+    engine::verify_identity(parse_peer_id(&peer_id)?, &public_key, &payload, &signature)
+        .map_err(peer_error)
+}
+
 fn wrap_stream(stream: engine::PeerStream) -> Result<PeerStream> {
     Ok(PeerStream {
         peer_id: stream.peer_id.to_string(),
@@ -327,6 +366,16 @@ fn parse_addresses(values: Vec<String>, label: &str) -> Result<Vec<Multiaddr>> {
         .collect()
 }
 
+fn validate_identity_payload(payload: &[u8]) -> Result<()> {
+    if payload.is_empty() || payload.len() > IDENTITY_PAYLOAD_MAX_BYTES {
+        return Err(Error::new(
+            Status::InvalidArg,
+            "identity payload must be between 1 and 8192 bytes",
+        ));
+    }
+    Ok(())
+}
+
 fn peer_error(error: PeerError) -> Error {
     Error::new(
         Status::GenericFailure,

native/runtime-host-peer/src/engine.rs

@@ -84,6 +84,11 @@ pub struct StartedEndpoint {
     pub thread: thread::JoinHandle<()>,
 }
 
+pub struct IdentitySignature {
+    pub public_key: Vec<u8>,
+    pub signature: Vec<u8>,
+}
+
 pub struct ConnectOptions {
     pub request_id: u32,
     pub peer_id: PeerId,
@@ -249,6 +254,37 @@ pub async fn ensure_identity(key_path: PathBuf) -> Result<PeerId, PeerError> {
         .to_peer_id())
 }
 
+pub async fn sign_identity(
+    key_path: PathBuf,
+    expected_peer_id: PeerId,
+    payload: &[u8],
+) -> Result<IdentitySignature, PeerError> {
+    let key = identity_store::load_key(&key_path).await?;
+    if PeerId::from(key.public()) != expected_peer_id {
+        return Err(PeerError::new(
+            "peer_identity_mismatch",
+            "the persisted peer identity does not match the expected PeerId",
+        ));
+    }
+    Ok(IdentitySignature {
+        public_key: key.public().encode_protobuf(),
+        signature: key
+            .sign(payload)
+            .map_err(|error| PeerError::new("peer_native_failed", error.to_string()))?,
+    })
+}
+
+pub fn verify_identity(
+    peer_id: PeerId,
+    public_key: &[u8],
+    payload: &[u8],
+    signature: &[u8],
+) -> Result<bool, PeerError> {
+    let public_key = identity::PublicKey::try_decode_protobuf(public_key)
+        .map_err(|error| PeerError::new("peer_native_failed", error.to_string()))?;
+    Ok(PeerId::from(&public_key) == peer_id && public_key.verify(payload, signature))
+}
+
 pub fn start(options: StartOptions) -> Result<StartedEndpoint, PeerError> {
     let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel(1);
     let (command_tx, command_rx) = mpsc::channel(COMMAND_CAPACITY);
@@ -1228,6 +1264,38 @@ fn native_error(error: impl std::fmt::Display) -> PeerError {
 mod tests {
     use super::*;
 
+    #[tokio::test]
+    async fn identity_signature_is_bound_to_peer_and_payload() {
+        let root = std::env::temp_dir().join(format!("maka-peer-signature-{}", PeerId::random()));
+        std::fs::create_dir_all(&root).expect("create test root");
+        let key_path = root.join("peer.key");
+        let peer_id = ensure_identity(key_path.clone())
+            .await
+            .expect("create identity");
+        let proof = sign_identity(key_path, peer_id, b"route")
+            .await
+            .expect("sign payload");
+
+        assert!(
+            verify_identity(peer_id, &proof.public_key, b"route", &proof.signature)
+                .expect("verify signature")
+        );
+        assert!(
+            !verify_identity(peer_id, &proof.public_key, b"other", &proof.signature)
+                .expect("reject changed payload")
+        );
+        assert!(
+            !verify_identity(
+                PeerId::random(),
+                &proof.public_key,
+                b"route",
+                &proof.signature,
+            )
+            .expect("reject changed peer")
+        );
+        std::fs::remove_dir_all(root).expect("remove test root");
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn mesh_control_survives_repeated_application_streams_on_one_endpoint() {
         let root = std::env::temp_dir().join(format!("maka-peer-test-{}", PeerId::random()));

native/runtime-host-peer/src/lib.rs

@@ -21,6 +21,6 @@ mod bindings;
 mod engine;
 
 pub use bindings::{
-    ConnectPeerOptions, PeerEndpoint, PeerStream, StartPeerEndpointOptions, ensure_peer_identity,
-    start_peer_endpoint,
+    ConnectPeerOptions, PeerEndpoint, PeerIdentitySignature, PeerStream, StartPeerEndpointOptions,
+    ensure_peer_identity, sign_peer_identity, start_peer_endpoint, verify_peer_identity,
 };

packages/runtime-host/src/__tests__/peer-listener.test.ts

@@ -21,14 +21,12 @@ import assert from 'node:assert/strict';
 import { setImmediate as waitForImmediate } from 'node:timers/promises';
 import { test } from 'node:test';
 import { createRuntimeHostPeerListener } from '../server/peer-listener.js';
-import type {
-  RuntimeHostPeerNativeEndpoint,
-  RuntimeHostPeerNativeStream,
-} from '../transport/peer-native.js';
+import type { RuntimeHostPeerClient } from '../client/peer-client.js';
+import type { RuntimeHostPeerNativeStream } from '../transport/peer-native.js';
 
 test('bounds and aborts pending peer authentication', async () => {
   const streams = Array.from({ length: 17 }, (_, index) => pendingStream(`remote-peer-${index}`));
-  const listener = createRuntimeHostPeerListener(endpointWith([...streams]), {} as never, () => {});
+  const listener = createRuntimeHostPeerListener(peerWith([...streams]), {} as never, () => {});
   await waitForImmediate();
 
   assert.equal(streams.filter((stream) => stream.aborted).length, 1);
@@ -43,7 +41,7 @@ test('bounds and aborts pending peer authentication', async () => {
 test('expires a peer that does not send its credential', async (context) => {
   context.mock.timers.enable({ apis: ['setTimeout'] });
   const stream = pendingStream();
-  const listener = createRuntimeHostPeerListener(endpointWith([stream]), {} as never, () => {});
+  const listener = createRuntimeHostPeerListener(peerWith([stream]), {} as never, () => {});
   await waitForImmediate();
 
   context.mock.timers.tick(5_000);
@@ -55,7 +53,7 @@ test('expires a peer that does not send its credential', async (context) => {
 test('reports an explicit authentication rejection before closing the stream', async () => {
   const stream = recordingStream(Buffer.from('{"v":1,"credential":"rejected"}\n'));
   const listener = createRuntimeHostPeerListener(
-    endpointWith([stream]),
+    peerWith([stream]),
     { authenticate: () => null } as never,
     () => {},
   );
@@ -86,7 +84,7 @@ test('rechecks peer authority at admission after the authentication response is
   let authentications = 0;
   let accepted = false;
   const listener = createRuntimeHostPeerListener(
-    endpointWith([stream]),
+    peerWith([stream]),
     {
       authenticate: () => (authentications++ === 0 ? { operationGrants: 'all' } : null),
     } as never,
@@ -110,7 +108,7 @@ test('bounds active application streams from one authenticated peer', async () =
   const streams = Array.from({ length: 5 }, () => authenticatedPendingStream('remote-peer'));
   let accepted = 0;
   const listener = createRuntimeHostPeerListener(
-    endpointWith([...streams]),
+    peerWith([...streams]),
     { authenticate: () => ({ operationGrants: 'all' }) } as never,
     () => {
       accepted += 1;
@@ -124,19 +122,28 @@ test('bounds active application streams from one authenticated peer', async () =
   await listener.cleanup();
 });
 
-function endpointWith(streams: RuntimeHostPeerNativeStream[]): RuntimeHostPeerNativeEndpoint {
+function peerWith(streams: RuntimeHostPeerNativeStream[]): RuntimeHostPeerClient {
   return {
-    peerId: 'peer',
-    listenAddresses: [],
+    identity: () => ({ peerId: 'peer', listenAddresses: [], coordinationRelays: [] }),
+    signIdentity: async () => {
+      throw new Error('not used');
+    },
+    verifyIdentity: () => false,
     connect: async () => {
       throw new Error('not used');
     },
     connectMeshControl: async () => {
       throw new Error('not used');
     },
-    cancelConnect: async () => false,
-    accept: async () => streams.shift() ?? null,
-    acceptMeshControl: async () => null,
+    serveApplication: async (onStream, signal) => {
+      for (const stream of streams) onStream(stream);
+      await new Promise<void>((resolve) =>
+        signal.addEventListener('abort', () => resolve(), { once: true }),
+      );
+    },
+    serveMeshControl: async () => {
+      throw new Error('not used');
+    },
     close: async () => undefined,
   };
 }