From c9a557d1c50b73c7bbb47f31f0c6069ead56015a Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 21:04:51 +0100 Subject: [PATCH 1/2] chore(deps): bump wire-mesh-core to 1.48.2 wire-mesh#170 fixed the session layer's relay-pairing tracking and added IncomingManageRequest.toDevice, read from each relay-data frame's own field; wire-mesh#175 (following, once implementation here surfaced the gap) made relay-hub actually forward that field onward to the receiving peer, which toDevice disambiguation depends on entirely. --- package.json | 2 +- pnpm-lock.yaml | 10 +++++----- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/package.json b/package.json index 71d7826f..45ea72bf 100644 --- a/package.json +++ b/package.json @@ -103,7 +103,7 @@ "cc-peer": "1.4.1", "preact": "10.29.7", "typebox": "1.3.6", - "wire-mesh-core": "1.30.1", + "wire-mesh-core": "1.48.2", "ws": "8.21.1", "zod": "4.4.3" }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 3b17eb32..f6f4aa96 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -198,8 +198,8 @@ importers: specifier: 1.3.6 version: 1.3.6 wire-mesh-core: - specifier: 1.30.1 - version: 1.30.1 + specifier: 1.48.2 + version: 1.48.2 ws: specifier: 8.21.1 version: 8.21.1 @@ -4471,8 +4471,8 @@ packages: engines: {node: '>=8'} hasBin: true - wire-mesh-core@1.30.1: - resolution: {integrity: sha512-MmTtt5MQXffDPOOndQ10UI3pz0EZ81xWtbpDCnAUwVcY4Nala16glE/li3c1VLiuErmWHzTjAljg7ULpX/8dZg==} + wire-mesh-core@1.48.2: + resolution: {integrity: sha512-s5oCOGG6LRs7PVZfPBvEaPL7aIGaFWC/NTg0tuSdm5txeSk949xPjlXmp4Wr/+Jnz5VTogaNv2f9Yfsyyt4CMw==} word-wrap@1.2.5: resolution: {integrity: sha512-BN22B5eaMMI9UMtjrGd5g5eCYPpCPDUy0FJXbYsaT5zYxjFOckS53SQDE3pWkVoWpHXVb3BrYcEN4Twa55B5cA==} @@ -9014,7 +9014,7 @@ snapshots: siginfo: 2.0.0 stackback: 0.0.2 - wire-mesh-core@1.30.1: + wire-mesh-core@1.48.2: dependencies: cbor2: 2.3.0 cddl.js: 1.0.1 From 0cedfb5d00ead1734f65bdfdfe105121b5d45330 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 21:05:07 +0100 Subject: [PATCH 2/2] feat(core): route hub-relayed requests to the correct local peer via toDevice Every inbound hub-relayed manage-request used to dispatch unconditionally against the gateway's own local state, even when it named a different device (toDevice) that this gateway also fronts as an ordinary local peer -- the known limitation hub-session.ts's own consume() doc comment flagged. It now forwards such a request on to that peer's own local mesh session (WireMeshTransport.sendToLocalPeer, extracted to hub-forwarding.ts) instead, falling back to the existing gateway-local dispatch when toDevice is absent or names the gateway itself, so a sender still on an older wire-mesh-core (never stamping toDevice) keeps working exactly as before. The forwarded command carries the request's already-verified true sender in a new "on-behalf-of" params field, and room-router.ts's own resolveHandle substitutes it for the request's dispatch identity once it reaches the local peer, so a capability token or a DM's own participant check still authenticates against the real remote origin rather than this gateway. resolveHandle only honours that field when the session's peer identity matches this side's own recorded coordinator device (WireMeshTransport.coordinatorDeviceHex) -- checked by identity rather than by which call established the session, since mesh formation's own reciprocal connectToPeer can reach a coordinator over a second, independently-accepted connection to that identical device. Also extracts readvertiseGossip into gossip-directory.ts, under this repo's max-lines cap once the above pushed wire-mesh-transport.ts over it. --- src/core/gossip-directory.ts | 42 +++++- src/core/hub-forwarding.ts | 14 ++ src/core/hub-session.ts | 38 ++++- src/core/room-router.ts | 40 ++++- src/core/wire-mesh-transport.ts | 52 +++---- .../hub-todevice-routing.integration.test.ts | 142 ++++++++++++++++++ 6 files changed, 291 insertions(+), 37 deletions(-) create mode 100644 src/test/hub-todevice-routing.integration.test.ts diff --git a/src/core/gossip-directory.ts b/src/core/gossip-directory.ts index 08f93f32..49b9de48 100644 --- a/src/core/gossip-directory.ts +++ b/src/core/gossip-directory.ts @@ -3,10 +3,20 @@ */ import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; -import type { DirectoryEntry } from "wire-mesh-core/domain/mesh-session"; +import type { + AcceptedMeshSession, + DirectoryEntry, +} from "wire-mesh-core/domain/mesh-session"; import type { PeerAdvert } from "wire-mesh-core/generated/protocol"; import { AgentStatus } from "./types.js"; -import { PRESENCE_GOSSIP_KEY } from "./wire-mesh-transport.js"; +import { + AGENT_SELF_GOSSIP_KEY, + HOSTED_ROOMS_GOSSIP_KEY, + PRESENCE_GOSSIP_KEY, + type AgentSelfAdvert, + type HostedRoomAdvert, +} from "./wire-mesh-transport.js"; +import type { HubSession } from "./hub-session.js"; /** Merges one session event's own directory into the mesh-wide knownDevices view (mutated in place), keeping the newer advert (by snapshot-seconds) whenever a device-id is already known from an earlier event or a different session. */ export function mergeKnownDevices( @@ -37,3 +47,31 @@ export function findPresenceAdvert( const status: unknown = entry.advert[PRESENCE_GOSSIP_KEY]; return AgentStatus.is(status) ? status : undefined; } + +/** Re-sends this side's own current presence status and currently-hosted rooms, together, onto every live session's gossip self-advert -- one gossip frame per tick carrying whichever of the two sources is wired in, rather than a separate frame per fact. Split out of wire-mesh-transport.ts's own WireMeshTransport class purely to keep that file under the repo's max-lines cap, the same reason mergeKnownDevices/findPresenceAdvert above already live here rather than there. A session that fails to send (mid-disconnect, most likely -- watchForDisconnect will independently notice and clean it up) is reported via onError and skipped, not allowed to stop the tick from reaching the rest of allSessions: a periodic broadcast to N peers is N independent operations, not one atomic unit. A no-op tick (neither source wired in, or no sessions exist yet) is expected and silent. The hub's own session (agent-comms#156) is gated separately from every ordinary local-peer session in allSessions: local mesh trust is a different layer (connect_request/introduce approval already gated it before it ever joined allSessions), but the hub session is a broadcast to every connected hub peer, trusted or not, and would otherwise leak this side's own presence/hosted-rooms/self-agent advert onto the hub regardless of GatewayTrust -- forwardAdvertsToHub/pushHubCatchUp's own hasAny gate exists to prevent exactly this for OTHER local peers' adverts, and this side's own self-advert deserves the identical gate, not a bypass. */ +export function readvertiseGossip( + allSessions: ReadonlySet, + hub: Readonly>, + hasAnyTrustedGateway: () => boolean, + onError: ((error: Error) => void) | undefined, + getCurrentPresence: (() => AgentStatus | undefined) | undefined, + getHostedRooms: (() => readonly HostedRoomAdvert[]) | undefined, + getSelfAgentAdvert: (() => AgentSelfAdvert | undefined) | undefined, +): void { + const extensions: Record = {}; + const status = getCurrentPresence?.(); + if (status !== undefined) extensions[PRESENCE_GOSSIP_KEY] = status; + const hostedRooms = getHostedRooms?.(); + if (hostedRooms !== undefined) + extensions[HOSTED_ROOMS_GOSSIP_KEY] = hostedRooms; + const selfAgentAdvert = getSelfAgentAdvert?.(); + if (selfAgentAdvert !== undefined) + extensions[AGENT_SELF_GOSSIP_KEY] = selfAgentAdvert; + if (Object.keys(extensions).length === 0) return; + for (const session of allSessions) { + if (hub.ownsSession(session) && !hasAnyTrustedGateway()) continue; + session.sendGossipUpdate(extensions).catch((error: unknown) => { + onError?.(error instanceof Error ? error : new Error(String(error))); + }); + } +} diff --git a/src/core/hub-forwarding.ts b/src/core/hub-forwarding.ts index 2aa8a04c..ae09a55d 100644 --- a/src/core/hub-forwarding.ts +++ b/src/core/hub-forwarding.ts @@ -9,6 +9,7 @@ import type { PeerAdvert, } from "wire-mesh-core/generated/protocol"; import type { + AcceptedMeshSession, DirectoryEntry, ManageOutcome, } from "wire-mesh-core/domain/mesh-session"; @@ -47,6 +48,19 @@ export function pushHubCatchUp( forwardAdvertsToHub(hub, catchUp, onError, hasAnyTrustedGateway); } +/** Sends a room-domain manage-request to a LOCAL peer session only (peerSessions), never falling back to hub routing -- HubSession's own toDevice-forwarding leg (agent-comms#184: a hub-relayed request explicitly addressed to a non-gateway local peer this gateway also fronts), wired in as WireMeshTransport's forwardToLocalPeer dependency. Returns undefined when no local session exists for that device-id, in which case HubSession falls back to dispatching the request against this gateway's own local state instead. */ +export function sendToLocalPeer( + peerSessions: ReadonlyMap, + memberId: string, + command: ManageCommand, + scope: Readonly, + token?: CapabilityToken, +): Promise | undefined { + const session = peerSessions.get(memberId); + if (session === undefined) return undefined; + return session.sendManageRequest(command, scope, undefined, token); +} + /** Routes a room-domain request through the hub's relay-connect/relay-data pairing when memberId isn't a local peer session -- WireMeshTransport.sendRoomRequest's own fallback, since the member may be a remote agent reachable only via this machine's gateway connection (agent-comms#155's local-to-remote leg). WireMeshTransport.sendRoomRequest itself gates memberId against the gateway trust boundary (agent-comms#156) before ever calling this, so by the time this runs memberId is already known-trusted -- this function stays focused on the hub-connectivity outcome alone. Resolves the same not_connected outcome sendRoomRequest already returned before the hub existed at all when this side isn't currently the gateway. */ export async function routeRoomRequestViaHub( hub: Readonly>, diff --git a/src/core/hub-session.ts b/src/core/hub-session.ts index b866dd39..6ea29a53 100644 --- a/src/core/hub-session.ts +++ b/src/core/hub-session.ts @@ -56,6 +56,13 @@ export interface HubSessionDeps { ) => Promise; /** The gateway trust boundary (agent-comms#156): whether the given device-id (hex) is currently trusted. Checked against every gossiped directory entry's own device and every relayed request's own fromDevice before this side merges or dispatches it -- see consume()/connect()'s own doc comments for exactly where and why. */ isTrusted: (deviceHex: string) => boolean; + /** Forwards a room-domain manage-request on to a specific LOCAL peer session (one this gateway is directly connected to over the ordinary local mesh, keyed by device-id hex) rather than dispatching it against this gateway's own local state -- consume()'s own toDevice disambiguation (agent-comms#184, wire-mesh-core 1.48.1's IncomingManageRequest.toDevice). Returns undefined when no local session exists for that device-id, in which case consume() falls back to handleRoomRequest exactly as it always has. */ + forwardToLocalPeer: ( + deviceHex: string, + command: ManageCommand, + scope: Readonly, + token?: CapabilityToken, + ) => Promise | undefined; } export class HubSession { @@ -132,14 +139,14 @@ export class HubSession { if (event.state.status === "closed") break; } })(); - this.consume(session); + this.consume(session, deviceIdToHex(identity.deviceId)); void (async () => { await this.watchDisconnect(session); })(); } - /** Dispatches inbound relayed manage-requests: each is handled with a handle keyed by the SENDING device (request.fromDevice names it on relay-routed requests), so every downstream consumer sees the true origin, never the hub -- the same discipline extends to a real room-domain verb (agent-comms#155's "remote to local" leg) as it already applied to the legacy opaque-frame path. Every request is first checked against the gateway trust boundary (agent-comms#156, deps.isTrusted): a request with no fromDevice at all (senderHex falls back to the literal string "hub-peer", never a real trusted device-id) or an unrecognised fromDevice is never dispatched to either path below -- a legacy FRAME_VERB message is silently dropped (matching isStateMutatingMessage's own swallow-and-ack style, so an untrusted sender learns nothing about why), and a room-domain request gets an explicit `unauthorized` error rather than being dispatched, so its caller fails fast instead of waiting out HUB_ROOM_REQUEST_TIMEOUT_MS's local-session-side counterpart. A legacy FRAME_VERB carrying state_sync/state_update is dropped before ever reaching onMessage/applyPatch even from an otherwise-trusted sender -- see isStateMutatingMessage's own doc for why: gateway trust says "this device's traffic is worth acting on," not "this device may directly overwrite this side's mesh state," which is a strictly stronger claim the trust boundary here was never meant to grant (a security review finding on agent-comms#169). A real room-domain verb (room.send, room.join, room.notify, ...) carries no equivalent risk -- it is independently gated by its own room:member capability token, verified regardless of which transport path it arrived over -- so a trusted sender's request is dispatched to the same roomVerbHandlers a local peer session's own drainSession uses, via handleRoomRequest, against THIS side's own local mesh state. Known limitation, inherited from wire-mesh-core's own session layer rather than something agent-comms can fix here: a session tracks at most one active relay pairing per remote device (mesh-session.ts's own single relayPeerDevice slot), so a request relayed here is dispatched as "addressed to this gateway's own agent" unconditionally -- there is no target-device disambiguation available to route it on to a DIFFERENT local peer this gateway also advertises. Forwarding this gateway's own agent's traffic is therefore correct; a remote request genuinely meant for another local peer behind this same gateway is not yet distinguishable from one meant for this gateway's own agent. */ - private consume(session: AcceptedMeshSession): void { + /** Dispatches inbound relayed manage-requests: each is handled with a handle keyed by the SENDING device (request.fromDevice names it on relay-routed requests), so every downstream consumer sees the true origin, never the hub -- the same discipline extends to a real room-domain verb (agent-comms#155's "remote to local" leg) as it already applied to the legacy opaque-frame path. Every request is first checked against the gateway trust boundary (agent-comms#156, deps.isTrusted): a request with no fromDevice at all (senderHex falls back to the literal string "hub-peer", never a real trusted device-id) or an unrecognised fromDevice is never dispatched to either path below -- a legacy FRAME_VERB message is silently dropped (matching isStateMutatingMessage's own swallow-and-ack style, so an untrusted sender learns nothing about why), and a room-domain request gets an explicit `unauthorized` error rather than being dispatched, so its caller fails fast instead of waiting out HUB_ROOM_REQUEST_TIMEOUT_MS's local-session-side counterpart. A legacy FRAME_VERB carrying state_sync/state_update is dropped before ever reaching onMessage/applyPatch even from an otherwise-trusted sender -- see isStateMutatingMessage's own doc for why: gateway trust says "this device's traffic is worth acting on," not "this device may directly overwrite this side's mesh state," which is a strictly stronger claim the trust boundary here was never meant to grant (a security review finding on agent-comms#169). A real room-domain verb (room.send, room.join, room.notify, ...) carries no equivalent risk -- it is independently gated by its own room:member capability token, verified regardless of which transport path it arrived over. Multi-device gateway routing (agent-comms#184, wire-mesh-core 1.48.1's own IncomingManageRequest.toDevice, read directly from each relay-data frame's own to-device field rather than guessed from pairing state): when the request carries a toDevice that names a different device than ownDeviceHex, the command is first re-stamped with an "on-behalf-of" params field naming senderHex (already verified trusted above) before deps.forwardToLocalPeer forwards it on to that device's own local mesh session (one this gateway is directly connected to, never merely gossiped-about) -- room-router.ts's own resolveHandle reads that field back out on the receiving end, so the forwarded request is attributed to the true remote sender there, not to this gateway, matching this same method's own "every downstream consumer sees the true origin, never the hub" discipline for the local hop too. Its outcome is relayed straight back. Only when forwardToLocalPeer finds no such local session (toDevice is absent, matches ownDeviceHex, or names a device this gateway doesn't actually front) does the request fall through to handleRoomRequest, dispatched to the same roomVerbHandlers a local peer session's own drainSession uses, against THIS side's own local mesh state -- correct for traffic genuinely addressed to this gateway's own agent, and the same fallback a sender still on a pre-#184 wire-mesh-core (never stamping toDevice at all) already relied on. */ + private consume(session: AcceptedMeshSession, ownDeviceHex: string): void { void (async () => { for await (const request of session.incomingManageRequests) { if (this.deps.isShuttingDown()) break; @@ -167,6 +174,31 @@ export class HubSession { await request.respond({ result: "ok" }).catch(() => undefined); continue; } + const toDeviceHex = + request.toDevice !== undefined + ? deviceIdToHex(request.toDevice) + : undefined; + if (toDeviceHex !== undefined && toDeviceHex !== ownDeviceHex) { + // Stamps the already-verified true sender (senderHex -- trusted above, never the "hub-peer" fallback, since an untrusted or fromDevice-less request already continued away) onto the forwarded command's own params, so the local peer's own resolveHandle (room-router.ts) can attribute the request to senderHex instead of this side's own device once it arrives over that peer's ordinary local-mesh session -- otherwise every downstream consumer at the local peer would see this gateway as the requester, never the real remote origin, defeating consume()'s own "true origin, never the hub" discipline for this forwarded leg specifically. + const forwardedCommand = { + ...request.command, + params: { + ...request.command.params, + "on-behalf-of": senderHex, + }, + }; + const forwarded = this.deps.forwardToLocalPeer( + toDeviceHex, + forwardedCommand, + request.scope, + request.token, + ); + if (forwarded !== undefined) { + const outcome = await forwarded; + await request.respond(outcome).catch(() => undefined); + continue; + } + } await this.deps.handleRoomRequest(request, handle); } })(); diff --git a/src/core/room-router.ts b/src/core/room-router.ts index c98f07bb..4c31d901 100644 --- a/src/core/room-router.ts +++ b/src/core/room-router.ts @@ -11,6 +11,7 @@ import type { IncomingManageRequest, ManageOutcome, } from "wire-mesh-core/domain/mesh-session"; +import { deviceIdFromHex } from "wire-mesh-core/domain/device-id"; import { isMeshMessage } from "./wire-protocol.js"; import type { MeshMessage } from "./wire-protocol.js"; import type { ConnectionHandle, TransportEvents } from "./transport.js"; @@ -80,6 +81,35 @@ function extractParamsVerb(params: unknown): string | undefined { return typeof params.verb === "string" ? params.verb : undefined; } +/** Reads a command's own "on-behalf-of" extension field (open params tail, the same convention room.send's "streaming-behavior" already uses) -- present only when a trusted gateway forwarded this request on behalf of a different device than the one it physically arrived over (agent-comms#184's own hub-session.ts toDevice-forwarding leg, which stamps the hub-relayed request's real fromDevice here before handing it to a local peer's own session). Returns undefined for anything that isn't a well-formed device-id-hex string, so a malformed or absent value silently falls back to the connection's own authenticated identity rather than ever producing a broken handle. */ +function onBehalfOfFromParams(params: unknown): string | undefined { + if (typeof params !== "object" || params === null) return undefined; + if (!("on-behalf-of" in params)) return undefined; + const claimed = params["on-behalf-of"]; + if (typeof claimed !== "string") return undefined; + try { + deviceIdFromHex(claimed); + } catch { + return undefined; + } + return claimed; +} + +/** Resolves the handle a request is actually dispatched with: the session's own authenticated connection identity, unless allowOnBehalfOf permits substituting a claimed "on-behalf-of" device from the request's own params. allowOnBehalfOf is true only when the session's peer identity matches this side's own recorded coordinator device (wire-mesh-transport.ts's consumeIncoming, keyed on identity rather than on which specific session connectToCoordinator itself dialled, since mesh formation's own reciprocal connectToPeer can reach this side over a second, independently-accepted connection to that identical device) -- the one relationship in this mesh where the far end already holds unconditional trust over this session (the pre-existing ungated introduce/coordinator-handoff path), the same trust boundary that lets it forward a hub-relayed request on to a different local peer at all. Safe for a capability-token-gated verb (room.send, ...) regardless: the token's own bearer is independently verified against the resolved handle, so a forged claim here would simply fail that check rather than succeed under a false identity -- this only changes outcomes for the ungated room.join first round, where the coordinator relationship is the sole trust boundary either way. */ +function resolveHandle( + request: IncomingManageRequest, + handle: Readonly, + allowOnBehalfOf: boolean, +): Readonly { + if (!allowOnBehalfOf) return handle; + const claimed = onBehalfOfFromParams(request.command.params); + if (claimed === undefined) return handle; + return { + id: claimed, + ...(handle.policy !== undefined ? { policy: handle.policy } : {}), + }; +} + /** Handles one already-verified-and-approved incoming request, given its authenticated connection handle. Registered per params.verb (room.send, room.join, ...), never per command.verb (the shared room:member capability every ordinary membership verb rides under) -- see this file's own header comment. */ export type RoomVerbHandler = ( request: IncomingManageRequest, @@ -99,10 +129,11 @@ export interface RoomRouter { request: IncomingManageRequest, handle: Readonly, ) => Promise; - /** Consumes one session's incomingManageRequests until it ends, dispatching each request via handleRequest. Becomes the single consumer of that iterable from this point on -- the caller must not also iterate the same session's incomingManageRequests itself once this is called. */ + /** Consumes one session's incomingManageRequests until it ends, dispatching each request via handleRequest. Becomes the single consumer of that iterable from this point on -- the caller must not also iterate the same session's incomingManageRequests itself once this is called. allowOnBehalfOf (default false) permits resolveHandle's own "on-behalf-of" substitution for every request on this session -- wire-mesh-transport.ts's consumeIncoming passes true only when this session's peer identity matches this side's own recorded coordinator device (see resolveHandle's own doc comment for why identity, not call site, is what's checked). */ drainSession: ( session: AcceptedMeshSession, handle: Readonly, + allowOnBehalfOf?: boolean, ) => void; } @@ -134,10 +165,13 @@ export function createRoomRouter(options: RoomRouterOptions): RoomRouter { return { handleRequest, - drainSession(session, handle) { + drainSession(session, handle, allowOnBehalfOf = false) { void (async () => { for await (const request of session.incomingManageRequests) { - await handleRequest(request, handle); + await handleRequest( + request, + resolveHandle(request, handle, allowOnBehalfOf), + ); } })(); }, diff --git a/src/core/wire-mesh-transport.ts b/src/core/wire-mesh-transport.ts index 824ad5ea..21675768 100644 --- a/src/core/wire-mesh-transport.ts +++ b/src/core/wire-mesh-transport.ts @@ -22,11 +22,16 @@ import { unregisterListener, type TrackedListener, } from "./listener-registry.js"; -import { findPresenceAdvert, mergeKnownDevices } from "./gossip-directory.js"; +import { + findPresenceAdvert, + mergeKnownDevices, + readvertiseGossip, +} from "./gossip-directory.js"; import { connectHubGateway, forwardAdvertsToHub, routeRoomRequestViaHub, + sendToLocalPeer, } from "./hub-forwarding.js"; import { acceptMeshSession, @@ -97,7 +102,7 @@ const PRESENCE_READVERTISE_INTERVAL_MS = export const PRESENCE_GOSSIP_KEY = "presence/status"; /** The domain-qualified gossip extension key this transport writes this side's own currently-hosted public/private rooms under -- the write half of P3.8's room-discovery replacement for createRoom's own broadcastPatch (agent-comms#48). Same namespacing convention as PRESENCE_GOSSIP_KEY. */ -const HOSTED_ROOMS_GOSSIP_KEY = "room/hosted"; +export const HOSTED_ROOMS_GOSSIP_KEY = "room/hosted"; /** The lightweight, gossip-safe shape a room advertises itself under: enough for a peer to display "this device hosts a discoverable room here" without exposing anything membership- or grant-related. Deliberately excludes secret rooms (never worth advertising at all) and every CRDT membership field a real Room carries -- a gossip-discovered entry is a hint pointing at a room to join, not a substitute for the real Room object join/admission still produces. */ export interface HostedRoomAdvert { @@ -191,8 +196,9 @@ export class WireMeshTransport implements MeshTransport { private readonly coordinatorListeners = new Map(); private defaultListenerId: string | undefined; - // -- The session dialled via connectToCoordinator, when this instance is not itself the coordinator -- + // -- The session dialled via connectToCoordinator, when this instance is not itself the coordinator -- and that same coordinator's own device-id hex, set alongside it and never cleared: consumeIncoming's own allowOnBehalfOf gate (see its doc comment) compares every session's authenticated peer identity against the hex, not merely "was this the specific session connectToCoordinator itself dialled", since mesh formation's own reciprocal connectToPeer can just as easily reach this transport over an accepted connection to the identical coordinator device. private coordinatorSession: AcceptedMeshSession | undefined; + private coordinatorDeviceHex: string | undefined; // -- The hub relay mode (agent-comms#151), owning its own file under the max-lines cap: see hub-session.ts for the full connection model. readonly hub: HubSession; @@ -293,6 +299,7 @@ export class WireMeshTransport implements MeshTransport { }, handleRoomRequest: this.roomRouter.handleRequest, isTrusted: (deviceHex) => this.gatewayTrust.isTrusted(deviceHex), + forwardToLocalPeer: sendToLocalPeer.bind(null, this.peerSessions), }); this.pendingConnectionTimeoutMs = pendingConnectionTimeoutMs; this.getCurrentPresence = getCurrentPresence; @@ -305,7 +312,15 @@ export class WireMeshTransport implements MeshTransport { getSelfAgentAdvert !== undefined ) { this.gossipInterval = setInterval(() => { - this.readvertiseGossip(); + readvertiseGossip( + this.allSessions, + this.hub, + () => this.gatewayTrust.hasAny(), + this.events.onError, + this.getCurrentPresence, + this.getHostedRooms, + this.getSelfAgentAdvert, + ); }, presenceReadvertiseIntervalMs); this.gossipInterval.unref(); } @@ -357,29 +372,6 @@ export class WireMeshTransport implements MeshTransport { } } - /** Re-sends this side's own current presence status and currently-hosted rooms, together, onto every live session's gossip self-advert -- one gossip frame per tick carrying whichever of the two sources is wired in, rather than a separate frame per fact. A session that fails to send (mid-disconnect, most likely -- watchForDisconnect will independently notice and clean it up) is reported via onError and skipped, not allowed to stop the tick from reaching the rest of allSessions: a periodic broadcast to N peers is N independent operations, not one atomic unit. A no-op tick (neither source wired in, or no sessions exist yet) is expected and silent. The hub's own session (agent-comms#156) is gated separately from every ordinary local-peer session in allSessions: local mesh trust is a different layer (connect_request/introduce approval already gated it before it ever joined allSessions), but the hub session is a broadcast to every connected hub peer, trusted or not, and would otherwise leak this side's own presence/hosted-rooms/self-agent advert onto the hub regardless of GatewayTrust -- forwardAdvertsToHub/pushHubCatchUp's own hasAny gate exists to prevent exactly this for OTHER local peers' adverts, and this side's own self-advert deserves the identical gate, not a bypass. */ - private readvertiseGossip(): void { - const extensions: Record = {}; - const status = this.getCurrentPresence?.(); - if (status !== undefined) extensions[PRESENCE_GOSSIP_KEY] = status; - const hostedRooms = this.getHostedRooms?.(); - if (hostedRooms !== undefined) - extensions[HOSTED_ROOMS_GOSSIP_KEY] = hostedRooms; - const selfAgentAdvert = this.getSelfAgentAdvert?.(); - if (selfAgentAdvert !== undefined) - extensions[AGENT_SELF_GOSSIP_KEY] = selfAgentAdvert; - if (Object.keys(extensions).length === 0) return; - for (const session of this.allSessions) { - if (this.hub.ownsSession(session) && !this.gatewayTrust.hasAny()) - continue; - session.sendGossipUpdate(extensions).catch((error: unknown) => { - this.events.onError?.( - error instanceof Error ? error : new Error(String(error)), - ); - }); - } - } - // -- Public getters -- get dataPort(): number { @@ -524,12 +516,13 @@ export class WireMeshTransport implements MeshTransport { })(); } - /** Delegates the whole post-approval drain loop to roomRouter -- this transport no longer decodes or routes a MeshMessage itself, only hands the session off. Also starts this session's own independent revocationAnnouncements drain, since a gossiped revocation is not a manage-request and has no verb for roomRouter to dispatch. */ + /** Delegates the whole post-approval drain loop to roomRouter -- this transport no longer decodes or routes a MeshMessage itself, only hands the session off. Also starts this session's own independent revocationAnnouncements drain, since a gossiped revocation is not a manage-request and has no verb for roomRouter to dispatch. Derives roomRouter's own allowOnBehalfOf gate itself, from whether this SPECIFIC handle's peer identity matches coordinatorDeviceHex -- never from which method established this particular session (mesh formation's reciprocal connectToPeer means the same coordinator device can just as easily reach this transport over an accepted connection as over the one connectToCoordinator itself dialled; see coordinatorDeviceHex's own field comment). A coordinator-less instance (coordinatorDeviceHex still undefined, e.g. this transport IS the coordinator) never allows it for anything. */ private consumeIncoming( session: AcceptedMeshSession, handle: Readonly, ): void { - this.roomRouter.drainSession(session, handle); + const allowOnBehalfOf = handle.id === this.coordinatorDeviceHex; + this.roomRouter.drainSession(session, handle, allowOnBehalfOf); this.drainRevocationAnnouncements(session); } @@ -639,6 +632,7 @@ export class WireMeshTransport implements MeshTransport { const handle: ConnectionHandle = { id: deviceIdToHex(coordinatorDeviceId), }; + this.coordinatorDeviceHex = handle.id; this.trackSession(handle.id, session); this.consumeIncoming(session, handle); this.watchForDisconnect(session, handle, handle.id); diff --git a/src/test/hub-todevice-routing.integration.test.ts b/src/test/hub-todevice-routing.integration.test.ts new file mode 100644 index 00000000..a5ecff6a --- /dev/null +++ b/src/test/hub-todevice-routing.integration.test.ts @@ -0,0 +1,142 @@ +/** + * Integration test for agent-comms#184: a hub-relayed manage-request explicitly addressed (relay-data's own `to-device` field, surfaced as `IncomingManageRequest.toDevice` since wire-mesh#170, wire-mesh-core 1.48.1) to a NON-gateway local peer must be routed on to that peer's own local mesh session, not dispatched against the gateway's own local state the way gateway-forwarding.integration.test.ts's "routes a DM to a remote gateway's own agent" case already covers for the gateway itself. Also proves the request's true remote origin (b1) survives the forwarding hop intact -- room-router.ts's own "on-behalf-of" mechanism, not just delivery reaching the right recipient. + * + * Same three-party shape as gateway-forwarding.integration.test.ts: machine A has a coordinator (a1, the gateway, dialled to the hub) and a second, ordinary local peer (a2) that is never itself connected to the hub. Machine B (b1) reaches a2 purely through the hub's relay pairing with a1 -- if toDevice routing didn't exist, b1's request would land at a1's own DM state instead of a2's, since a1's hub session has no way to tell the two apart. + */ + +import { afterEach, describe, expect, it } from "vitest"; +import { MeshStore } from "../core/mesh-store.js"; +import { dmRoomPath } from "../core/room-path.js"; +import { realHubOverWs } from "./hub-helpers.js"; +import { wireTestTransport } from "./test-transport.js"; + +let nextPort = 22_500; +function freshPort(): number { + nextPort += 1; + return nextPort; +} + +/** Short enough that a gossip re-advertisement (which carries a1's forwarded advert of a2) fires within a test's own poll budget, matching gateway-forwarding.integration.test.ts's own choice. */ +const FAST_GOSSIP_INTERVAL_MS = 50; + +const POLL_ATTEMPTS = 100; +const POLL_INTERVAL_MS = 50; +const sleep = async (ms: number): Promise => + new Promise((resolve) => { + setTimeout(resolve, ms); + }); + +/** Polls an async predicate until it holds -- gateway-forwarding.integration.test.ts's own established pattern for this, duplicated here rather than shared, since neither file exports it. */ +async function waitFor( + what: string, + check: () => Promise, +): Promise { + for (let i = 0; i < POLL_ATTEMPTS; i++) { + if (await check()) return; + await sleep(POLL_INTERVAL_MS); + } + expect(false, `timed out waiting for ${what}`).toBeTruthy(); +} + +const cleanups: (() => Promise)[] = []; + +afterEach(async () => { + for (const close of cleanups.splice(0)) { + await close(); + } +}); + +describe("hub toDevice routing", () => { + it("routes a DM addressed to a non-gateway local peer to that peer's own state, not the gateway's", async () => { + const hub = await realHubOverWs(); + cleanups.push(hub.close); + + const a1 = new MeshStore(freshPort(), hub.url); + await wireTestTransport(a1, undefined, undefined, FAST_GOSSIP_INTERVAL_MS); + await a1.init(); + cleanups.push(async () => a1.shutdown()); + await a1.registerAgent({ + name: "a1-gateway", + harness: "test", + cwd: "/test/a1", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + const a2 = new MeshStore(a1.coordinatorPort, hub.url); + await wireTestTransport(a2, undefined, undefined, FAST_GOSSIP_INTERVAL_MS); + await a2.init(); + cleanups.push(async () => a2.shutdown()); + await a2.registerAgent({ + name: "a2-local-peer", + harness: "test", + cwd: "/test/a2", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + const b1 = new MeshStore(freshPort(), hub.url); + await wireTestTransport(b1, undefined, undefined, FAST_GOSSIP_INTERVAL_MS); + await b1.init(); + cleanups.push(async () => b1.shutdown()); + await b1.registerAgent({ + name: "b1-remote", + harness: "test", + cwd: "/test/b1", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + // Deny-all by default (agent-comms#156): a1 needs a trusted remote gateway to forward a2's advert onto the hub at all (forwardAdvertsToHub's own hasAny gate), b1 needs a2's own device-id specifically trusted for sendRoomRequest's own outbound isTrusted gate to route to it via the hub, and a1 needs b1 trusted for consume()'s own inbound gate on the relayed request's fromDevice. + a1.addTrustedGateway(b1.peerId); + b1.addTrustedGateway(a2.peerId); + + // Waits for b1 to actually learn of a2 via the hub's own gossip/catch-up (mirroring gateway-forwarding.integration.test.ts's own first test) before addressing a request to it -- otherwise the relay hub has no adjacency entry for a2's device yet and silently drops the relay-connect (spec/relay-hub's own documented behaviour for an unknown target-device), which would time out for a reason unrelated to toDevice routing itself. + await waitFor("b1 to learn of a2 via the hub", async () => { + const agents = await b1.listAgents(b1.peerId); + return agents.some((agent) => agent.id === a2.peerId); + }); + + const dmPath = dmRoomPath(b1.peerId, a2.peerId); + const requestPromise = b1.requestDmAccess(a2.peerId); + await waitFor( + "a2 (not a1) to see b1's pending DM request, routed through a1's hub session via toDevice", + async () => { + await Promise.resolve(); + return a2.listPendingRoomJoins().length === 1; + }, + ); + const [pending] = a2.listPendingRoomJoins(); + expect(pending?.roomPath).toBe(dmPath); + expect(pending?.requesterId).toBe(b1.peerId); + // The gateway's own DM state must never have seen this request: proves toDevice routing actually redirected it to a2 rather than a1 additionally absorbing a copy. + expect(a1.listPendingRoomJoins()).toHaveLength(0); + + a2.acceptRoomJoin(dmPath, b1.peerId); + await requestPromise; + + const message = await b1.sendDm(b1.peerId, a2.peerId, "hello from b1"); + + await waitFor( + "a2 to receive the DM via the hub, routed through a1", + async () => { + const a2Dms = a2.serialise().dms[dmPath] ?? []; + return a2Dms.some((m) => m.id === message.id); + }, + ); + + const delivered = (a2.serialise().dms[dmPath] ?? []).find( + (m) => m.id === message.id, + ); + expect(delivered).toMatchObject({ + from: b1.peerId, + to: a2.peerId, + content: "hello from b1", + }); + // Again: the gateway's own DM state must stay empty throughout -- the message belongs to a2, never a1. + expect(a1.serialise().dms[dmPath] ?? []).toHaveLength(0); + }); +});