From 0f5ae0c3d7170985d1988a210ccebc7d6493fe42 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Mon, 28 Sep 2026 11:47:31 +0000 Subject: [PATCH 1/3] sdk: an egress floor, stated at join and held on every call join({ egressRequireLabels }) / join(egress_require_labels=) is sam-node's egress.require_labels for an SDK member: every provider the session calls must attest all of its pairs, on top of a call's required labels, MCP and HTTP alike, however the peer was named. A call cannot waive or widen it; an unmet floor is a LabelsNotSatisfiedError. The HTTP path (request, fetch, MeshTransport) now verifies the provider through the mutual /sam/auth handshake before sending anything, floor or not, as the MCP path and sam-node's egress proxy already did; a positive verdict is kept per peer for five minutes, sam-node's labelGateTTL. The integration runners hold a floor the mesh satisfies, and one extra runner per SDK, whose floor nobody attests, is refused by the node and by the other members before anything is sent. --- sdk/README.md | 26 +++-- sdk/js/README.md | 4 +- sdk/js/src/conformance-join.ts | 24 ++-- sdk/js/src/index.ts | 2 +- sdk/js/src/mcp.test.ts | 24 +++- sdk/js/src/mcp.ts | 38 ++++++- sdk/js/src/session.test.ts | 86 +++++++++++++- sdk/js/src/session.ts | 37 +++++- sdk/python/README.md | 4 +- sdk/python/src/agent_mesh/__init__.py | 3 +- sdk/python/src/agent_mesh/conformance_join.py | 17 ++- sdk/python/src/agent_mesh/httpx_transport.py | 2 +- sdk/python/src/agent_mesh/mcp_client.py | 26 ++++- sdk/python/src/agent_mesh/session.py | 32 +++++- sdk/python/tests/test_libp2p_http.py | 4 +- sdk/python/tests/test_mcp.py | 36 +++++- sdk/python/tests/test_session.py | 106 +++++++++++++++++- site/content/docs/guides/native-sdks.md | 8 +- tests/integration/sdk_mesh_test.go | 48 +++++++- 19 files changed, 467 insertions(+), 60 deletions(-) diff --git a/sdk/README.md b/sdk/README.md index 89deccd2..cbca7984 100644 --- a/sdk/README.md +++ b/sdk/README.md @@ -460,13 +460,25 @@ holds against the control plane's records. - `/sam/mcp/1.0.0` client: `session.openMCP(peer, "mcp://")` sends the `AuthFrame` naming the service, verifies the provider's credential and the caller's required labels (`checkPeerLabels`: several pairs are - met by any one of them, as `api.LabelCheck` joins them with `or`; the - conjunction is the operator's egress floor, which only `sam-node` has), - then runs the - official MCP client over the varint-framed stream. JS: a `Transport` for - `@modelcontextprotocol/sdk`; Python: a pair of memory streams pumped to - and from the libp2p stream for `mcp.ClientSession`. `""` as the target is - the provider's own catalog (`list_local_services`, `get_mesh_info`). + met by any one of them, as `api.LabelCheck` joins them with `or`), then + runs the official MCP client over the varint-framed stream. JS: a + `Transport` for `@modelcontextprotocol/sdk`; Python: a pair of memory + streams pumped to and from the libp2p stream for `mcp.ClientSession`. + `""` as the target is the provider's own catalog (`list_local_services`, + `get_mesh_info`). +- Egress floor: `join({ egressRequireLabels })` (`join(egress_require_labels=)`) + is `sam-node`'s `egress.require_labels` for an SDK member. The floor is + the conjunction (`api.LabelFloorCheck` joins the pairs with `,`): every + provider the session calls must attest all of them, on top of a call's + required labels, on every outbound call however the peer was named, MCP + and HTTP alike. Stated once at join and held for the session; a call + cannot waive or widen it. The three implementations agree on it, as they + do on the caller's requirement. The HTTP path (`request`, `fetch`, + `MeshTransport`) verifies the provider with or without a floor, through + the mutual `/sam/auth/1.0.0` handshake, as `sam-node`'s `VerifyPeerLabels` + does before its egress proxy sends anything; a positive verdict is kept + per peer for five minutes (`labelGateTTL`), a miss never. Refusals are + `LabelsNotSatisfiedError`, which names the floor. - `session.listTools(peer, service)` and `session.callTool(peer, service, tool, args)` on top of that. - Tests. Unit: each SDK calls a tool on an in-process provider that serves diff --git a/sdk/js/README.md b/sdk/js/README.md index 1624f4d2..4cccc11d 100644 --- a/sdk/js/README.md +++ b/sdk/js/README.md @@ -168,7 +168,9 @@ process.on("SIGTERM", stop); `enroll` reuses the identity and credential saved in `stateDir` when they are still valid for that control plane, and needs exactly one of `bootstrapTokenPath`, `bootstrapToken` or `jwt` otherwise. Read tokens from -a file or the environment; do not put them on a command line. +a file or the environment; do not put them on a command line. `labels` are +attested at enrollment; `join({ egressRequireLabels })` is the floor every +peer the session calls must attest, all of it, held for the session. A plaintext `http://` control plane is accepted only on loopback. Pass `allowInsecure: true` for a network you trust. diff --git a/sdk/js/src/conformance-join.ts b/sdk/js/src/conformance-join.ts index e00d4f0f..9fae3f01 100644 --- a/sdk/js/src/conformance-join.ts +++ b/sdk/js/src/conformance-join.ts @@ -78,6 +78,16 @@ function emit(obj: unknown): void { process.stdout.write(JSON.stringify(obj) + "\n"); } +/** Labels from an environment variable written "k=v,k2=v2". */ +function labelsFromEnv(name: string): Record { + return Object.fromEntries( + (process.env[name] ?? "") + .split(",") + .filter((pair) => pair.includes("=")) + .map((pair) => pair.split("=", 2) as [string, string]), + ); +} + function failure(cmd: string | undefined, err: unknown): unknown { return { cmd, ok: false, error: err instanceof Error ? `${err.name}: ${err.message}` : String(err) }; } @@ -172,14 +182,11 @@ async function main(): Promise { const stateDir = requireEnv("SAM_SDK_STATE_DIR"); const allowInsecure = process.env.SAM_INSECURE_CONTROL_PLANE === "1"; const listenAddrs = (process.env.SAM_SDK_LISTEN_ADDRS ?? "").split(",").filter((a) => a !== ""); - // Labels this member declares at enrollment, "k=v,k2=v2"; the policy's - // allowed_labels decide whether the control plane attests them. - const labels = Object.fromEntries( - (process.env.SAM_SDK_LABELS ?? "") - .split(",") - .filter((pair) => pair.includes("=")) - .map((pair) => pair.split("=", 2) as [string, string]), - ); + // Labels this member declares at enrollment; the policy's allowed_labels + // decide whether the control plane attests them. SAM_SDK_EGRESS_REQUIRE_LABELS + // is the floor every provider this member calls must attest. + const labels = labelsFromEnv("SAM_SDK_LABELS"); + const egressRequireLabels = labelsFromEnv("SAM_SDK_EGRESS_REQUIRE_LABELS"); const mesh = await AgentMesh.enroll({ controlPlaneUrl, @@ -197,6 +204,7 @@ async function main(): Promise { const session = await mesh.join({ listenAddrs, ...(routerAddresses !== undefined ? { routerAddresses } : {}), + ...(Object.keys(egressRequireLabels).length > 0 ? { egressRequireLabels } : {}), signal: AbortSignal.timeout(20_000), controlPlaneSyncIntervalMs: 0, controlPlaneSyncJitterMs: 0, diff --git a/sdk/js/src/index.ts b/sdk/js/src/index.ts index 47d2b496..45da9677 100644 --- a/sdk/js/src/index.ts +++ b/sdk/js/src/index.ts @@ -43,7 +43,7 @@ export { BiscuitVerificationError, ROLE_ROUTER, requireRole, verifyPeerBiscuit, export { AUTH_HANDLER_OPTIONS, AUTH_PROTOCOL, MCP_PROTOCOL, AuthRejectedError, authenticateWithPeer, authStreamHandler } from "./auth.ts"; export { createMeshHost, type MeshHost, type MeshHostOptions } from "./host.ts"; export { DHT_PROTOCOL, isServiceType, parseServiceTarget, serviceCID, type ServiceType } from "./discovery.ts"; -export { LabelsNotSatisfiedError, StreamTransport, openMCPSession, requireLabels, type MCPSession, type MCPSessionOptions } from "./mcp.ts"; +export { LabelsNotSatisfiedError, StreamTransport, openMCPSession, requireEgressLabels, requireLabels, type MCPSession, type MCPSessionOptions } from "./mcp.ts"; export { AuthorizationError, authorizeCaller, type AuthorizeRequest, type ProviderAuthorizerOptions } from "./authorizer.ts"; export { DEFAULT_A2A_NAME, diff --git a/sdk/js/src/mcp.test.ts b/sdk/js/src/mcp.test.ts index 7be179f7..aec677fa 100644 --- a/sdk/js/src/mcp.test.ts +++ b/sdk/js/src/mcp.test.ts @@ -36,7 +36,7 @@ import { ROLE_NODE } from "./controlplane.ts"; import { parseServiceTarget, serviceCID } from "./discovery.ts"; import { AuthFrameSchema, AuthResponseSchema } from "./gen/sam_pb.ts"; import { Identity } from "./identity.ts"; -import { LabelsNotSatisfiedError, StreamTransport, openMCPSession, requireLabels } from "./mcp.ts"; +import { LabelsNotSatisfiedError, StreamTransport, openMCPSession, requireEgressLabels, requireLabels } from "./mcp.ts"; type Wasm = Awaited>; @@ -186,6 +186,28 @@ test("a requirement of several labels is met by any one of them, as sam-node's c requireLabels(attesting({}), {}); }); +test("the egress floor is met only by every one of its pairs, as sam-node's api.LabelFloorCheck", () => { + const attesting = (labels: Record) => ({ peerId: "p", expiration: new Date(), verifyingKey: cpKey, roles: [], labels }); + requireEgressLabels(attesting({ region: "eu", team: "platform" }), { region: "eu" }); + requireEgressLabels(attesting({ region: "eu", team: "platform" }), { region: "eu", team: "platform" }); + // one pair short is a refusal that names the whole floor + assert.throws(() => requireEgressLabels(attesting({ region: "eu" }), { region: "eu", team: "platform" }), /peer p does not attest the egress floor: region=eu, team=platform/); + assert.throws(() => requireEgressLabels(attesting({ region: "us" }), { region: "eu" }), LabelsNotSatisfiedError); + // no floor is no floor + requireEgressLabels(attesting({}), {}); + requireEgressLabels(attesting({}), undefined); +}); + +test("the egress floor is held on the MCP path beside the caller's requirement", async () => { + const conn = await caller.dial(provider.getMultiaddrs()[0] as Parameters[0]); + const ok = await openMCPSession(conn, frame("mcp://calc"), [cpKey], {}, { region: "eu" }); + await ok.close(); + await assert.rejects(openMCPSession(conn, frame("mcp://calc"), [cpKey], {}, { region: "eu", team: "platform" }), LabelsNotSatisfiedError); + // Both apply when both are set: neither one's pairs stand in for the other's. + await assert.rejects(openMCPSession(conn, frame("mcp://calc"), [cpKey], { requiredLabels: { region: "eu" } }, { team: "platform" }), LabelsNotSatisfiedError); + await assert.rejects(openMCPSession(conn, frame("mcp://calc"), [cpKey], { requiredLabels: { team: "platform" } }, { region: "eu" }), LabelsNotSatisfiedError); +}); + test("a caller the provider cannot verify gets no session", async () => { const conn = await caller.dial(provider.getMultiaddrs()[0] as Parameters[0]); const forged = new wasm.KeyPair(wasm.SignatureAlgorithm.Ed25519); diff --git a/sdk/js/src/mcp.ts b/sdk/js/src/mcp.ts index ae67c268..7d75768f 100644 --- a/sdk/js/src/mcp.ts +++ b/sdk/js/src/mcp.ts @@ -102,10 +102,13 @@ export interface MCPSession { close(): Promise; } -/** The provider's credential carries none of the labels the caller requires (checkPeerLabels). */ +/** + * The provider's credential lacks what the caller requires (any one pair) or + * what the session's egress floor requires (every pair), as checkPeerLabels refuses. + */ export class LabelsNotSatisfiedError extends Error { - constructor(peerId: string, required: string[]) { - super(`peer ${peerId} carries none of the required labels: ${required.join(", ")}`); + constructor(peerId: string, required: string[], what = "carries none of the required labels") { + super(`peer ${peerId} ${what}: ${required.join(", ")}`); this.name = "LabelsNotSatisfiedError"; } } @@ -113,8 +116,8 @@ export class LabelsNotSatisfiedError extends Error { /** * A caller's requirement is satisfied by any one pair, as sam-node's * api.LabelCheck (`check if label(k1, v1) or label(k2, v2)`): several pairs - * mean "any of these will do". The operator's egress floor is the - * conjunction, and sam-node's alone. + * mean "any of these will do". The egress floor (requireEgressLabels) is the + * conjunction. */ export function requireLabels(provider: VerifiedBiscuit, required: Record | undefined): void { if (!required) { @@ -130,16 +133,38 @@ export function requireLabels(provider: VerifiedBiscuit, required: Record | undefined): void { + if (!required) { + return; + } + const pairs = Object.entries(required); + if (pairs.every(([k, v]) => provider.labels[k] === v)) { + return; + } + throw new LabelsNotSatisfiedError( + provider.peerId, + pairs.map(([k, v]) => `${k}=${v}`), + "does not attest the egress floor", + ); +} + /** * Opens /sam/mcp/1.0.0 to a connected provider for targetService ("" is the * provider's own catalog), verifies the provider, and returns a connected - * MCP client. frame is this member's AuthFrame for that service. + * MCP client. frame is this member's AuthFrame for that service; egressRequireLabels + * is the session's, not the caller's (requireEgressLabels). */ export async function openMCPSession( conn: Connection, frame: Uint8Array, trustedKeys: Uint8Array[], options: MCPSessionOptions = {}, + egressRequireLabels?: Record, ): Promise { const signal = options.signal ?? AbortSignal.timeout(AUTH_HANDSHAKE_TIMEOUT_MS); const stream = await conn.newStream(MCP_PROTOCOL, { signal, runOnLimitedConnection: true }); @@ -154,6 +179,7 @@ export async function openMCPSession( } provider = await verifyPeerBiscuit(resp.biscuit, conn.remotePeer.toString(), trustedKeys); requireLabels(provider, options.requiredLabels); + requireEgressLabels(provider, egressRequireLabels); } catch (err) { await stream.close().catch(() => stream.abort(err instanceof Error ? err : new Error(String(err)))); if (err instanceof BiscuitVerificationError) { diff --git a/sdk/js/src/session.test.ts b/sdk/js/src/session.test.ts index 6829ac42..1d04bb10 100644 --- a/sdk/js/src/session.test.ts +++ b/sdk/js/src/session.test.ts @@ -30,12 +30,15 @@ import { multiaddr } from "@multiformats/multiaddr"; import { createLibp2p } from "libp2p"; import assert from "node:assert/strict"; import { after, before, test } from "node:test"; -import { AUTH_PROTOCOL, AuthRejectedError, authenticateWithPeer, authStreamHandler } from "./auth.ts"; -import { ROLE_ROUTER, loadBiscuit, verifyPeerBiscuit } from "./biscuit.ts"; +import { AUTH_HANDLER_OPTIONS, AUTH_PROTOCOL, AuthRejectedError, authenticateWithPeer, authStreamHandler } from "./auth.ts"; +import { BiscuitVerificationError, ROLE_ROUTER, loadBiscuit, verifyPeerBiscuit } from "./biscuit.ts"; import { ROLE_NODE } from "./controlplane.ts"; import { BootstrapEnrollRequestSchema, BootstrapEnrollResponseSchema, EnrollmentStatus, KeysResponseSchema, AuthFrameSchema } from "./gen/sam_pb.ts"; import { Identity } from "./identity.ts"; +import { HTTP_HANDLER_OPTIONS, HTTP_PROTOCOL, a2aEndpoint, httpIngressHandler } from "./libp2p-http.ts"; +import { LabelsNotSatisfiedError } from "./mcp.ts"; import { AgentMesh } from "./mesh.ts"; +import { MeshSession, type JoinOptions } from "./session.ts"; type Wasm = Awaited>; @@ -46,14 +49,26 @@ let router: Libp2p; let routerAddr: string; let routerBiscuit: Uint8Array; -function mint(peerId: string, role: string, expiration = "2035-01-01T00:00:00Z"): Uint8Array { +function mint(peerId: string, role: string, expiration = "2035-01-01T00:00:00Z", labels: Record = {}): Uint8Array { const b = wasm.Biscuit.builder(); b.addFact(wasm.Fact.fromString(`node(${JSON.stringify(peerId)})`)); + b.addFact(wasm.Fact.fromString(`client_peer_id(${JSON.stringify(peerId)})`)); b.addFact(wasm.Fact.fromString(`expiration(${expiration})`)); b.addFact(wasm.Fact.fromString(`role(${JSON.stringify(role)})`)); + for (const [k, v] of Object.entries(labels)) { + b.addFact(wasm.Fact.fromString(`label(${JSON.stringify(k)}, ${JSON.stringify(v)})`)); + } return b.build(cpKeyPair.getPrivateKey()).toBytes(); } +// What the control plane renders for a policy granting the node role every +// A2A service on any target, and nothing else. +const POLICY_RULES = [ + `granted_service_all("a2a") <- role("sam:role:node")`, + `granted_service_all("sam:system") <- role("sam:role:node")`, + `target_unrestricted(true) <- role("sam:role:node")`, +]; + function proto(bytes: Uint8Array): Response { return new Response(Buffer.from(bytes), { status: 200, headers: { "Content-Type": "application/x-protobuf" } }); } @@ -97,6 +112,8 @@ before(async () => { connectionEncrypters: [tls()], streamMuxers: [yamux()], services: { identify: identify(), relay: circuitRelayServer() }, + // Every member here dials from loopback; the per-host inbound rate limit would refuse the later ones. + connectionManager: { inboundConnectionThreshold: Infinity }, }); routerBiscuit = mint(router.peerId.toString(), ROLE_ROUTER); await router.handle( @@ -342,3 +359,66 @@ test("an explicit rejection is reported with its reason", async () => { // The verifier does not depend on libp2p; a token for another peer is refused before any network I/O. await assert.rejects(verifyPeerBiscuit(routerBiscuit, "12D3KooWA4Xop1JaT3MHxwYMkCepYsv4iPVopMXwCz5iHYdBfeSB", [cpKey]), /not bound to peer/); }); + +test("an egress floor stated at join is held on the HTTP path, and the provider is verified with or without one", async () => { + // A provider answering /sam/auth and /libp2p-http as a member does, with + // whatever credential mintFor gives it; the handshakes it answers are counted. + const handshakes: string[] = []; + const serve = async (mintFor: (peerId: string) => Uint8Array): Promise => { + const identity = Identity.generate(); + const biscuit = mintFor(identity.peerId); + const host = await createLibp2p({ + privateKey: privateKeyFromProtobuf(identity.toLibp2pPrivateKey()), + addresses: { listen: ["/ip4/127.0.0.1/tcp/0"] }, + transports: [tcp()], + connectionEncrypters: [tls()], + streamMuxers: [yamux()], + services: { identify: identify() }, + }); + await host.handle(AUTH_PROTOCOL, authStreamHandler({ ownBiscuit: () => biscuit, trustedKeys: () => [cpKey], onAuthenticated: (peerId) => handshakes.push(peerId) }), AUTH_HANDLER_OPTIONS); + await host.handle(HTTP_PROTOCOL, httpIngressHandler(a2aEndpoint({ handler: () => Response.json({ ok: true }) }), { ownBiscuit: () => biscuit, trustedKeys: () => [cpKey], policyRules: () => POLICY_RULES }), HTTP_HANDLER_OPTIONS); + return host; + }; + const provider = await serve((peerId) => mint(peerId, ROLE_NODE, undefined, { region: "eu" })); + // Enrolled nowhere: a credential no trusted key signed. + const forged = new wasm.KeyPair(wasm.SignatureAlgorithm.Ed25519); + const impostor = await serve((peerId) => { + const b = wasm.Biscuit.builder(); + b.addFact(wasm.Fact.fromString(`node(${JSON.stringify(peerId)})`)); + b.addFact(wasm.Fact.fromString("expiration(2035-01-01T00:00:00Z)")); + return b.build(forged.getPrivateKey()).toBytes(); + }); + const addrOf = (host: Libp2p) => (host.getMultiaddrs()[0] as ReturnType).toString(); + const sessions: MeshSession[] = []; + const join = async (options: JoinOptions): Promise => { + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: fakeControlPlane([routerAddr]) }); + const session = await mesh.join({ refreshLeadMs: 0, reserveRelay: false, ...options }); + sessions.push(session); + return session; + }; + try { + const held = await join({ egressRequireLabels: { region: "eu" } }); + const missed = await join({ egressRequireLabels: { region: "eu", team: "platform" } }); + const plain = await join({}); + const card = MeshSession.meshURL(provider.peerId.toString(), "a2a://agent", "/card"); + // Met: one handshake verifies the provider; the verdict is kept for the next calls. + assert.equal((await held.request(addrOf(provider), "a2a://agent", "/card")).status, 200); + assert.equal((await held.request(addrOf(provider), "a2a://agent", "/card")).status, 200); + assert.equal((await held.fetch()(card)).status, 200); + assert.deepEqual(handshakes, [held.peerId]); + + // Missed: every path refuses, and a refusal is not kept: each call asks again. + await assert.rejects(missed.request(addrOf(provider), "a2a://agent", "/card"), LabelsNotSatisfiedError); + await assert.rejects(missed.fetch()(card), LabelsNotSatisfiedError); + assert.deepEqual(handshakes, [held.peerId, missed.peerId, missed.peerId]); + + // No floor: no gate, but the provider is verified all the same. + assert.equal((await plain.request(addrOf(provider), "a2a://agent", "/card")).status, 200); + await assert.rejects(plain.request(addrOf(impostor), "a2a://agent", "/card"), BiscuitVerificationError); + // A banned provider is refused before any handshake. + plain.banned.add(provider.peerId.toString(), Date.now()); + await assert.rejects(plain.request(addrOf(provider), "a2a://agent", "/card"), /banned/); + } finally { + await Promise.all([...sessions.map((s) => s.close()), provider.stop(), impostor.stop()]); + } +}); diff --git a/sdk/js/src/session.ts b/sdk/js/src/session.ts index 8973bd14..5c30f4ec 100644 --- a/sdk/js/src/session.ts +++ b/sdk/js/src/session.ts @@ -22,7 +22,7 @@ import { ROLE_ROUTER, requireRole, type VerifiedBiscuit } from "./biscuit.ts"; import { canonicalPeerId } from "./identity.ts"; import { isServiceType, parseServiceTarget, serviceCID } from "./discovery.ts"; import { createMeshHost, listenThroughRelay, type MeshHost, type MeshHostOptions, type RelayListener } from "./host.ts"; -import { openMCPSession, type MCPSession, type MCPSessionOptions } from "./mcp.ts"; +import { openMCPSession, requireEgressLabels, type MCPSession, type MCPSessionOptions } from "./mcp.ts"; import type { AgentMesh, ControlPlaneSync } from "./mesh.ts"; import { HTTP_HANDLER_OPTIONS, @@ -75,6 +75,12 @@ export interface JoinOptions extends MeshHostOptions { * how long the member is unreachable at most before it reserves again. */ relayCheckIntervalMs?: number; + /** + * sam-node's egress.require_labels for an SDK member: every provider this + * session calls must attest all of these pairs, on top of a call's + * requiredLabels. Held on every call, MCP and HTTP alike; no call waives it. + */ + egressRequireLabels?: Record; /** Bounds the whole join. */ signal?: AbortSignal; } @@ -132,6 +138,8 @@ const DEFAULT_CONTROL_PLANE_SYNC_MS = 15 * 60 * 1000; const FIRST_CONTROL_PLANE_SYNC_MS = 2_000; const DEFAULT_CONTROL_PLANE_SYNC_JITTER_MS = 2_000; const DEFAULT_RELAY_CHECK_MS = 30 * 1000; +/** How long a provider's positive egress verdict is kept; sam-node's labelGateTTL. */ +const EGRESS_VERDICT_TTL_MS = 5 * 60 * 1000; /** * A member that is on the mesh: a libp2p host authenticated with at least @@ -161,6 +169,9 @@ export class MeshSession { readonly #syncIntervalMs: number; readonly #syncJitterMs: number; #policyRules: string[] | undefined; + readonly #egressRequireLabels: Record | undefined; + /** Peers verified as enrolled and holding the floor, until when; misses are never kept. */ + readonly #egressVerdicts = new Map(); #closed = false; constructor(mesh: AgentMesh, node: MeshHost, routers: AdmittedRouter[], authenticatedPeers: Map, banned: BanSet, options: JoinOptions, relayListener?: RelayListener) { @@ -175,6 +186,7 @@ export class MeshSession { this.#policySyncMs = options.policySyncIntervalMs ?? DEFAULT_POLICY_SYNC_MS; this.#syncIntervalMs = options.controlPlaneSyncIntervalMs ?? DEFAULT_CONTROL_PLANE_SYNC_MS; this.#syncJitterMs = options.controlPlaneSyncJitterMs ?? DEFAULT_CONTROL_PLANE_SYNC_JITTER_MS; + this.#egressRequireLabels = options.egressRequireLabels; this.#scheduleRefresh(); this.#listenForEvents(); this.#keepRouterAdmissions(); @@ -458,7 +470,7 @@ export class MeshSession { */ async openMCP(peer: Peer, targetService: string, options: MCPSessionOptions = {}): Promise { const conn = await this.connect(peer, options.signal); - return openMCPSession(conn, this.mesh.authFrame(targetService, options.agent ?? ""), this.mesh.credential.controlPlaneKeys, options); + return openMCPSession(conn, this.mesh.authFrame(targetService, options.agent ?? ""), this.mesh.credential.controlPlaneKeys, options, this.#egressRequireLabels); } /** Lists the tools a provider serves for a service. */ @@ -619,10 +631,27 @@ export class MeshSession { * way sam-node's egress proxy does for /sam////. */ async request(peer: Peer, targetService: string, path: string, options: HTTPRequestOptions = {}): Promise { - const conn = await this.connect(peer, options.signal); + const conn = await this.#egressConnection(peer, options.signal); return httpRequestOverStream(conn, this.mesh.credential.biscuit, targetService, path, options); } + /** + * The connection an HTTP call goes out on, its peer verified as an enrolled + * member holding the floor before anything is sent (sam-node's VerifyPeerLabels). + */ + async #egressConnection(peer: Peer, signal?: AbortSignal): Promise { + const conn = await this.connect(peer, signal); + const peerId = conn.remotePeer.toString(); + const until = this.#egressVerdicts.get(peerId); + if (until !== undefined && until.getTime() > Date.now()) { + return conn; + } + const provider = await authenticateWithPeer(conn, this.mesh.authFrame(), this.mesh.credential.controlPlaneKeys); + requireEgressLabels(provider, this.#egressRequireLabels); + this.#egressVerdicts.set(peerId, new Date(Date.now() + EGRESS_VERDICT_TTL_MS)); + return conn; + } + /** * A `fetch` bound to the mesh, for clients built on fetch such as the A2A * SDK's (`fetchImpl`): a request to http://mesh/sam//// @@ -633,7 +662,7 @@ export class MeshSession { return async (input, init) => { const request = new Request(input, init); const { peerId } = splitMeshURL(new URL(request.url)); - const conn = await this.connect(peerId, request.signal); + const conn = await this.#egressConnection(peerId, request.signal); const streamOptions: { agent?: string; signal?: AbortSignal } = {}; if (options.agent !== undefined) { streamOptions.agent = options.agent; diff --git a/sdk/python/README.md b/sdk/python/README.md index ee8165e5..99b17648 100644 --- a/sdk/python/README.md +++ b/sdk/python/README.md @@ -176,7 +176,9 @@ trio.run(main) `enroll` reuses the identity and credential saved in `state_dir` when they are still valid for that control plane, and needs exactly one of `bootstrap_token_path`, `bootstrap_token` or `jwt` otherwise. Read tokens -from a file or the environment; do not put them on a command line. +from a file or the environment; do not put them on a command line. `labels` +are attested at enrollment; `join(egress_require_labels=)` is the floor +every peer the session calls must attest, all of it, held for the session. A plaintext `http://` control plane is accepted only on loopback. Pass `allow_insecure=True` for a network you trust. diff --git a/sdk/python/src/agent_mesh/__init__.py b/sdk/python/src/agent_mesh/__init__.py index 08979a2a..e510d3fd 100644 --- a/sdk/python/src/agent_mesh/__init__.py +++ b/sdk/python/src/agent_mesh/__init__.py @@ -47,7 +47,7 @@ mesh_http_target, open_http_request, ) -from .mcp_client import LabelsNotSatisfiedError, ToolCallResult, ToolInfo, open_mcp_session, require_labels +from .mcp_client import LabelsNotSatisfiedError, ToolCallResult, ToolInfo, open_mcp_session, require_egress_labels, require_labels from .mesh import AgentMesh, ControlPlaneSync from .relay import dial_through_relay, reserve_relay from .session import AdmittedRouter, MeshSession, Peer @@ -117,6 +117,7 @@ "peer_id_from_public_key", "refresh_challenge", "register_challenge", + "require_egress_labels", "require_labels", "require_role", "reserve_relay", diff --git a/sdk/python/src/agent_mesh/conformance_join.py b/sdk/python/src/agent_mesh/conformance_join.py index d97604dc..3543e0e3 100644 --- a/sdk/python/src/agent_mesh/conformance_join.py +++ b/sdk/python/src/agent_mesh/conformance_join.py @@ -69,6 +69,11 @@ def _emit(obj: dict) -> None: print(json.dumps(obj), flush=True) +def _labels_from_env(name: str) -> dict[str, str]: + """Labels from an environment variable written "k=v,k2=v2".""" + return dict(pair.split("=", 1) for pair in os.environ.get(name, "").split(",") if "=" in pair) + + def _root_cause(err: BaseException) -> BaseException: # trio wraps a failure in one ExceptionGroup per nursery it crossed. while isinstance(err, BaseExceptionGroup) and len(err.exceptions) == 1: @@ -160,9 +165,11 @@ async def main() -> None: state_dir = _require_env("SAM_SDK_STATE_DIR") allow_insecure = os.environ.get("SAM_INSECURE_CONTROL_PLANE") == "1" listen = [a for a in os.environ.get("SAM_SDK_LISTEN_ADDRS", "").split(",") if a] - # Labels this member declares at enrollment, "k=v,k2=v2"; the policy's - # allowed_labels decide whether the control plane attests them. - labels = dict(pair.split("=", 1) for pair in os.environ.get("SAM_SDK_LABELS", "").split(",") if "=" in pair) + # Labels this member declares at enrollment; the policy's allowed_labels + # decide whether the control plane attests them. SAM_SDK_EGRESS_REQUIRE_LABELS + # is the floor every provider this member calls must attest. + labels = _labels_from_env("SAM_SDK_LABELS") + egress_require_labels = _labels_from_env("SAM_SDK_EGRESS_REQUIRE_LABELS") mesh = AgentMesh.enroll( control_plane_url, @@ -177,7 +184,9 @@ async def main() -> None: # can put two members on different routers. only = {p for p in os.environ.get("SAM_SDK_ROUTERS", "").split(",") if p} router_addresses = [a for a in mesh.credential.router_addresses if any(a.endswith(f"/p2p/{p}") for p in only)] if only else None - async with mesh.join(listen_addrs=listen, router_addresses=router_addresses, control_plane_sync_interval=0, control_plane_sync_jitter=0) as session: + async with mesh.join( + listen_addrs=listen, router_addresses=router_addresses, control_plane_sync_interval=0, control_plane_sync_jitter=0, egress_require_labels=egress_require_labels or None + ) as session: _emit( { "sdk": "python", diff --git a/sdk/python/src/agent_mesh/httpx_transport.py b/sdk/python/src/agent_mesh/httpx_transport.py index d945e3fd..888b433b 100644 --- a/sdk/python/src/agent_mesh/httpx_transport.py +++ b/sdk/python/src/agent_mesh/httpx_transport.py @@ -77,7 +77,7 @@ def __init__(self, session: "MeshSession", *, agent: str = "", timeout: float = async def handle_async_request(self, request: httpx.Request) -> httpx.Response: peer_text, target = split_mesh_url(request.url) - peer_id = await self._session.connect(peer_text) + peer_id = await self._session._egress_peer(peer_text) # noqa: SLF001 - the session's verified egress path, not a caller option body = await request.aread() headers = {k.decode("latin-1"): v.decode("latin-1") for k, v in request.headers.raw} response = await open_http_request( diff --git a/sdk/python/src/agent_mesh/mcp_client.py b/sdk/python/src/agent_mesh/mcp_client.py index 005e07c5..72ee3cc2 100644 --- a/sdk/python/src/agent_mesh/mcp_client.py +++ b/sdk/python/src/agent_mesh/mcp_client.py @@ -46,17 +46,19 @@ class LabelsNotSatisfiedError(Exception): - """The provider's credential carries none of the labels the caller requires (checkPeerLabels).""" + """The provider's credential lacks what the caller requires (any one pair) + or what the session's egress floor requires (every pair), as + checkPeerLabels refuses.""" - def __init__(self, peer_id: str, required: Sequence[str]): - super().__init__(f"peer {peer_id} carries none of the required labels: {', '.join(required)}") + def __init__(self, peer_id: str, required: Sequence[str], what: str = "carries none of the required labels"): + super().__init__(f"peer {peer_id} {what}: {', '.join(required)}") def require_labels(provider: VerifiedBiscuit, required: Optional[Mapping[str, str]]) -> None: """A caller's requirement is satisfied by any one pair, as sam-node's api.LabelCheck (`check if label(k1, v1) or label(k2, v2)`): several pairs - mean "any of these will do". The operator's egress floor is the - conjunction, and sam-node's alone.""" + mean "any of these will do". The egress floor (require_egress_labels) is the + conjunction.""" if not required: return if any(provider.labels.get(k) == v for k, v in required.items()): @@ -64,6 +66,15 @@ def require_labels(provider: VerifiedBiscuit, required: Optional[Mapping[str, st raise LabelsNotSatisfiedError(provider.peer_id, [f"{k}={v}" for k, v in required.items()]) +def require_egress_labels(provider: VerifiedBiscuit, required: Optional[Mapping[str, str]]) -> None: + """The egress floor is met only by every one of its pairs, as sam-node's + api.LabelFloorCheck (`check if label(k1, v1), label(k2, v2)`) for + egress.require_labels: a floor takes no alternatives. Empty is no floor.""" + if not required or all(provider.labels.get(k) == v for k, v in required.items()): + return + raise LabelsNotSatisfiedError(provider.peer_id, [f"{k}={v}" for k, v in required.items()], "does not attest the egress floor") + + @dataclass class ToolInfo: """One entry of a provider's tool list.""" @@ -97,10 +108,12 @@ async def open_mcp_session( trusted_keys: Sequence[bytes], *, required_labels: Optional[Mapping[str, str]] = None, + egress_require_labels: Optional[Mapping[str, str]] = None, ) -> AsyncIterator[tuple[ClientSession, VerifiedBiscuit]]: """Opens /sam/mcp/1.0.0 to a connected provider with `frame`, this member's AuthFrame naming the service, verifies the provider and yields an - initialized MCP ClientSession with the provider's credential.""" + initialized MCP ClientSession with the provider's credential; egress_require_labels + is the session's, not the caller's (require_egress_labels).""" stream = await open_stream(host, peer_id, MCP_PROTOCOL, AUTH_HANDSHAKE_TIMEOUT) try: with trio.fail_after(AUTH_HANDSHAKE_TIMEOUT): @@ -116,6 +129,7 @@ async def open_mcp_session( except BiscuitVerificationError as err: raise AuthRejectedError(str(peer_id), f"provider credential rejected: {err}") from err require_labels(provider, required_labels) + require_egress_labels(provider, egress_require_labels) except trio.TooSlowError as err: await stream.close() raise AuthRejectedError(str(peer_id), "handshake timed out") from err diff --git a/sdk/python/src/agent_mesh/session.py b/sdk/python/src/agent_mesh/session.py index e9cf93e5..648a4b8f 100644 --- a/sdk/python/src/agent_mesh/session.py +++ b/sdk/python/src/agent_mesh/session.py @@ -55,7 +55,7 @@ http_request_over_stream, mesh_http_target, ) -from .mcp_client import ToolCallResult, ToolInfo, open_mcp_session, tool_call_result +from .mcp_client import ToolCallResult, ToolInfo, open_mcp_session, require_egress_labels, tool_call_result from .relay import STOP_PROTOCOL, dial_through_relay, reserve_relay, split_circuit_address, stop_stream_handler from .sync import GOSSIP_EVENTS_TOPIC, BanSet, verify_mesh_event @@ -81,6 +81,8 @@ DEFAULT_CONTROL_PLANE_SYNC = 15 * 60.0 FIRST_CONTROL_PLANE_SYNC = 2.0 DEFAULT_CONTROL_PLANE_SYNC_JITTER = 2.0 +# How long a provider's positive egress verdict is kept; sam-node's labelGateTTL. +EGRESS_VERDICT_TTL = 5 * 60.0 # How a caller names the peer it wants to reach: a provider `discover` returned, # a peer id, or a multiaddr. For a provider or a peer id the SDK dials the @@ -143,6 +145,10 @@ class MeshSession: banned: BanSet = field(default_factory=BanSet) # This member's agent, once accept_a2a was called. endpoint: Optional[A2AEndpoint] = None + # sam-node's egress.require_labels for an SDK member: every provider this + # session calls must attest all of these pairs, on top of a call's + # required_labels. Held on every call, MCP and HTTP alike; no call waives it. + egress_require_labels: Optional[Mapping[str, str]] = None policy_sync_interval: float = DEFAULT_POLICY_SYNC control_plane_sync_interval: float = DEFAULT_CONTROL_PLANE_SYNC control_plane_sync_jitter: float = DEFAULT_CONTROL_PLANE_SYNC_JITTER @@ -153,6 +159,8 @@ class MeshSession: _policy_rules: Optional[list[str]] = field(default=None, repr=False) _sync_lock: trio.Lock = field(default_factory=trio.Lock, repr=False) _sync_trigger: trio.Event = field(default_factory=trio.Event, repr=False) + # Peers verified as enrolled and holding the floor, until when; misses are never kept. + _egress_verdicts: dict[str, float] = field(default_factory=dict, repr=False) @property def peer_id(self) -> str: @@ -352,7 +360,9 @@ def open_mcp( async def opened() -> AsyncIterator[tuple[ClientSession, VerifiedBiscuit]]: peer_id = await self.connect(peer) frame = self.mesh.auth_frame(target_service, agent) - async with open_mcp_session(self.host, peer_id, frame, self.mesh.credential.control_plane_keys, required_labels=required_labels) as opened_session: + async with open_mcp_session( + self.host, peer_id, frame, self.mesh.credential.control_plane_keys, required_labels=required_labels, egress_require_labels=self.egress_require_labels + ) as opened_session: yield opened_session return opened() @@ -506,11 +516,22 @@ async def request( ) -> HTTPResponse: """Calls an inference or A2A service on a provider over /libp2p-http, the way sam-node's egress proxy does for /sam////.""" - peer_id = await self.connect(peer) + peer_id = await self._egress_peer(peer) return await http_request_over_stream( self.host, peer_id, self.mesh.credential.biscuit, target_service, path, method=method, headers=headers, body=body, agent=agent ) + async def _egress_peer(self, peer: Peer) -> ID: + """The peer an HTTP call goes out to, verified as an enrolled member + holding the floor before anything is sent (sam-node's VerifyPeerLabels).""" + peer_id = await self.connect(peer) + if self._egress_verdicts.get(str(peer_id), 0.0) > time.monotonic(): + return peer_id + provider = await authenticate_with_peer(self.host, peer_id, self.mesh.auth_frame(), self.mesh.credential.control_plane_keys) + require_egress_labels(provider, self.egress_require_labels) + self._egress_verdicts[str(peer_id)] = time.monotonic() + EGRESS_VERDICT_TTL + return peer_id + async def accept_a2a(self, target: Union[str, HTTPHandler], *, name: str = DEFAULT_A2A_NAME) -> str: """Makes this member's agent reachable: other members call it as `a2a://` by peer ID, through a router, and the SDK answers @@ -584,11 +605,13 @@ async def join_mesh( policy_sync_interval: float = DEFAULT_POLICY_SYNC, control_plane_sync_interval: float = DEFAULT_CONTROL_PLANE_SYNC, control_plane_sync_jitter: float = DEFAULT_CONTROL_PLANE_SYNC_JITTER, + egress_require_labels: Optional[Mapping[str, str]] = None, ) -> AsyncIterator[MeshSession]: """Implements AgentMesh.join(); lives here to keep mesh.py free of libp2p. router_addresses names the routers to join through instead of the ones the credential lists; a peer behind another router is still reached, see - MeshSession.connect.""" + MeshSession.connect. egress_require_labels is the floor every provider + this member calls must attest, see MeshSession.""" router_addrs = [multiaddr.Multiaddr(a) for a in (mesh.credential.router_addresses if router_addresses is None else router_addresses)] if not router_addrs: raise RuntimeError("credential lists no router addresses; the control plane had no active router at enrollment" if router_addresses is None else "router_addresses names no router") @@ -624,6 +647,7 @@ async def join_mesh( routers=admitted, authenticated_peers=authenticated, banned=banned, + egress_require_labels=egress_require_labels, policy_sync_interval=policy_sync_interval, control_plane_sync_interval=control_plane_sync_interval, control_plane_sync_jitter=control_plane_sync_jitter, diff --git a/sdk/python/tests/test_libp2p_http.py b/sdk/python/tests/test_libp2p_http.py index df842940..b231ab3e 100644 --- a/sdk/python/tests/test_libp2p_http.py +++ b/sdk/python/tests/test_libp2p_http.py @@ -141,13 +141,13 @@ class _Mesh: class FakeSession: """What MeshTransport needs of a session: the host, the credential and - connect(); the peer is already connected here.""" + _egress_peer(); the peer is already connected and verified here.""" def __init__(self, host, biscuit: bytes) -> None: self.host = host self.mesh = _Mesh(_Credential(biscuit)) - async def connect(self, peer) -> ID: + async def _egress_peer(self, peer) -> ID: return ID.from_base58(str(peer)) diff --git a/sdk/python/tests/test_mcp.py b/sdk/python/tests/test_mcp.py index 9c9e04dd..6d049b68 100644 --- a/sdk/python/tests/test_mcp.py +++ b/sdk/python/tests/test_mcp.py @@ -37,7 +37,7 @@ from agent_mesh.controlplane import ROLE_NODE from agent_mesh.discovery import parse_service_target, service_key from agent_mesh.identity import Identity -from agent_mesh.mcp_client import LabelsNotSatisfiedError, open_mcp_session, require_labels, tool_call_result +from agent_mesh.mcp_client import LabelsNotSatisfiedError, open_mcp_session, require_egress_labels, require_labels, tool_call_result from .test_session import CP, CP_KEY, libp2p_host @@ -197,6 +197,20 @@ async def main(): async with open_mcp_session(caller, pid, frame(caller_biscuit, "mcp://calc"), [CP_KEY], required_labels={"region": "eu", "team": "platform"}): pass + # The egress floor is met only by every one of its pairs, beside the caller's requirement. + async with open_mcp_session(caller, pid, frame(caller_biscuit, "mcp://calc"), [CP_KEY], egress_require_labels={"region": "eu"}): + pass + with pytest.raises(LabelsNotSatisfiedError, match="does not attest the egress floor: region=eu, team=platform"): + async with open_mcp_session(caller, pid, frame(caller_biscuit, "mcp://calc"), [CP_KEY], egress_require_labels={"region": "eu", "team": "platform"}): + pass + # Both apply when both are set: neither one's pairs stand in for the other's. + with pytest.raises(LabelsNotSatisfiedError): + async with open_mcp_session(caller, pid, frame(caller_biscuit, "mcp://calc"), [CP_KEY], required_labels={"region": "eu"}, egress_require_labels={"team": "platform"}): + pass + with pytest.raises(LabelsNotSatisfiedError): + async with open_mcp_session(caller, pid, frame(caller_biscuit, "mcp://calc"), [CP_KEY], required_labels={"team": "platform"}, egress_require_labels={"region": "eu"}): + pass + # A provider whose credential the caller does not trust is rejected. with pytest.raises(AuthRejectedError): async with open_mcp_session(caller, pid, frame(caller_biscuit, "mcp://calc"), [ba.KeyPair().public_key.to_bytes()]): @@ -247,3 +261,23 @@ def attesting(labels: dict) -> VerifiedBiscuit: # an empty requirement is no requirement require_labels(attesting({}), {}) require_labels(attesting({}), None) + + +def test_the_egress_floor_is_met_only_by_every_one_of_its_pairs(): + """sam-node's api.LabelFloorCheck for egress.require_labels, run through + the SDK's predicate: a floor takes no alternatives.""" + from datetime import datetime, timezone + + def attesting(labels: dict) -> VerifiedBiscuit: + return VerifiedBiscuit(peer_id="p", expiration=datetime.now(timezone.utc), verifying_key=CP_KEY, roles=[], labels=labels) + + require_egress_labels(attesting({"region": "eu", "team": "platform"}), {"region": "eu"}) + require_egress_labels(attesting({"region": "eu", "team": "platform"}), {"region": "eu", "team": "platform"}) + # one pair short is a refusal that names the whole floor + with pytest.raises(LabelsNotSatisfiedError, match="peer p does not attest the egress floor: region=eu, team=platform"): + require_egress_labels(attesting({"region": "eu"}), {"region": "eu", "team": "platform"}) + with pytest.raises(LabelsNotSatisfiedError): + require_egress_labels(attesting({"region": "us"}), {"region": "eu"}) + # no floor is no floor + require_egress_labels(attesting({}), {}) + require_egress_labels(attesting({}), None) diff --git a/sdk/python/tests/test_session.py b/sdk/python/tests/test_session.py index 3ab1229c..a26afacc 100644 --- a/sdk/python/tests/test_session.py +++ b/sdk/python/tests/test_session.py @@ -21,6 +21,7 @@ import urllib.parse import biscuit_auth as ba +import httpx import multiaddr import pytest import trio @@ -38,11 +39,16 @@ from agent_mesh._proto import circuit_pb2 as circuit from agent_mesh._proto import sam_pb2 as pb from agent_mesh.auth import AUTH_PROTOCOL, auth_stream_handler, authenticate_with_peer -from agent_mesh.biscuit import ROLE_ROUTER +from agent_mesh.authorizer import ProviderAuthorizerOptions +from agent_mesh.biscuit import ROLE_ROUTER, BiscuitVerificationError from agent_mesh.controlplane import ROLE_NODE +from agent_mesh.httpx_transport import MeshTransport from agent_mesh.identity import Identity +from agent_mesh.libp2p_http import HTTP_PROTOCOL, A2AEndpoint, HTTPResponse, ProviderOptions, http_ingress_handler +from agent_mesh.mcp_client import LabelsNotSatisfiedError from agent_mesh.mesh import AgentMesh from agent_mesh.relay import HOP_PROTOCOL as RELAY_HOP_PROTOCOL +from agent_mesh.session import MeshSession from google.protobuf.timestamp_pb2 import Timestamp @@ -61,10 +67,23 @@ def _ts_s(seconds: int) -> Timestamp: CP_KEY = CP.public_key.to_bytes() -def mint(peer_id: str, role: str, expiration: str = "2035-01-01T00:00:00Z") -> bytes: - return ba.BiscuitBuilder( - "node({p}); expiration(" + expiration + "); role({r});", {"p": peer_id, "r": role} - ).build(CP.private_key).to_bytes() +def mint(peer_id: str, role: str, expiration: str = "2035-01-01T00:00:00Z", labels: dict[str, str] | None = None) -> bytes: + code = "node({p}); client_peer_id({p}); expiration(" + expiration + "); role({r});" + params = {"p": peer_id, "r": role} + for i, (k, v) in enumerate((labels or {}).items()): + code += f" label({{k{i}}}, {{v{i}}});" + params[f"k{i}"] = k + params[f"v{i}"] = v + return ba.BiscuitBuilder(code, params).build(CP.private_key).to_bytes() + + +# What the control plane renders for a policy granting the node role every +# A2A service on any target, and nothing else. +POLICY_RULES = [ + 'granted_service_all("a2a") <- role("sam:role:node")', + 'granted_service_all("sam:system") <- role("sam:role:node")', + 'target_unrestricted(true) <- role("sam:role:node")', +] def fake_control_plane(router_addresses): @@ -151,6 +170,34 @@ async def run(): return router, addr_box[0] +async def start_provider(nursery, biscuit_for, handshakes: list[str]): + """A provider answering /sam/auth and /libp2p-http as a member does, with + whatever credential biscuit_for gives it; the handshakes it answers are + recorded in handshakes.""" + identity = Identity.generate() + host = libp2p_host(identity) + biscuit = biscuit_for(identity.peer_id) + host.set_stream_handler(AUTH_PROTOCOL, auth_stream_handler(lambda: biscuit, lambda: [CP_KEY], on_authenticated=lambda peer, _v: handshakes.append(peer))) + + async def card(_request, _caller) -> HTTPResponse: + return HTTPResponse(status=200, body=b'{"ok": true}') + + options = ProviderOptions(authorizer=ProviderAuthorizerOptions(trusted_keys=lambda: [CP_KEY], own_biscuit=lambda: biscuit, policy_rules=lambda: POLICY_RULES)) + host.set_stream_handler(HTTP_PROTOCOL, http_ingress_handler(A2AEndpoint(target=card), options)) + started = trio.Event() + addr_box = [] + + async def run(): + async with host.run(listen_addrs=[multiaddr.Multiaddr("/ip4/127.0.0.1/tcp/0")]): + addr_box.append(f"{host.get_addrs()[0]}") + started.set() + await trio.sleep_forever() + + nursery.start_soon(run) + await started.wait() + return host, addr_box[0] + + async def with_timeout(seconds, fn): with trio.fail_after(seconds): return await fn() @@ -389,3 +436,52 @@ async def main(): nursery.cancel_scope.cancel() trio.run(with_timeout, 60, main) + + +def test_an_egress_floor_stated_at_join_is_held_on_the_http_path(): + """The floor is held on request() and on MeshTransport, and the provider + is verified as an enrolled member with or without one.""" + + async def main(): + async with trio.open_nursery() as nursery: + _, router_addr = await start_router(nursery) + handshakes: list[str] = [] + provider, provider_addr = await start_provider(nursery, lambda p: mint(p, ROLE_NODE, labels={"region": "eu"}), handshakes) + # Enrolled nowhere: a credential no trusted key signed. + forged = ba.KeyPair() + _, impostor_addr = await start_provider( + nursery, lambda p: ba.BiscuitBuilder("node({p}); expiration(2035-01-01T00:00:00Z);", {"p": p}).build(forged.private_key).to_bytes(), handshakes + ) + + def join(**options): + mesh = AgentMesh.enroll("http://127.0.0.1:1", bootstrap_token="sbt", transport=fake_control_plane([router_addr])) + return mesh.join(reserve=False, refresh_lead=0, **options) + + async with join(egress_require_labels={"region": "eu"}) as held, join(egress_require_labels={"region": "eu", "team": "platform"}) as missed, join() as plain: + card = MeshSession.mesh_url(str(provider.get_id()), "a2a://agent", "/card") + # Met: one handshake verifies the provider; the verdict is kept for the next calls. + assert (await held.request(provider_addr, "a2a://agent", "/card")).status == 200 + assert (await held.request(provider_addr, "a2a://agent", "/card")).status == 200 + async with httpx.AsyncClient(transport=MeshTransport(held)) as client: + assert (await client.get(card)).status_code == 200 + assert handshakes == [held.peer_id] + + # Missed: every path refuses, and a refusal is not kept: each call asks again. + with pytest.raises(LabelsNotSatisfiedError): + await missed.request(provider_addr, "a2a://agent", "/card") + async with httpx.AsyncClient(transport=MeshTransport(missed)) as client: + with pytest.raises(LabelsNotSatisfiedError): + await client.get(card) + assert handshakes == [held.peer_id, missed.peer_id, missed.peer_id] + + # No floor: no gate, but the provider is verified all the same. + assert (await plain.request(provider_addr, "a2a://agent", "/card")).status == 200 + with pytest.raises(BiscuitVerificationError): + await plain.request(impostor_addr, "a2a://agent", "/card") + # A banned provider is refused before any handshake. + plain.banned.add(str(provider.get_id()), int(time.time() * 1000)) + with pytest.raises(PermissionError, match="banned"): + await plain.request(provider_addr, "a2a://agent", "/card") + nursery.cancel_scope.cancel() + + trio.run(with_timeout, 60, main) diff --git a/site/content/docs/guides/native-sdks.md b/site/content/docs/guides/native-sdks.md index 9f5fcecf..665b7307 100644 --- a/site/content/docs/guides/native-sdks.md +++ b/site/content/docs/guides/native-sdks.md @@ -885,7 +885,13 @@ circuit because it admitted the agent too. Either way the SDK verifies the peer's credential before sending anything, and `requiredLabels` (`required_labels` in Python) refuses a peer whose control-plane-attested labels carry none of the pairs you ask for; one matching pair is enough, -as with `X-Sam-Required-Labels` on a `sam-node`. +as with `X-Sam-Required-Labels` on a `sam-node`. A floor is the other way +round: `join({ egressRequireLabels })` (`join(egress_require_labels=)`) +names labels every peer the session calls must attest, all of them, as +`egress.require_labels` does for a `sam-node`. It is stated once at `join` +and held for the session, on every call and however the peer was named; +the agent's calls cannot waive or widen it. It is the program author's +floor, not the operator's: nothing outside the process sets it. `acceptA2A` (`accept_a2a`) fetched the mesh policy and started answering. Every caller must present a credential signed by a trusted control plane diff --git a/tests/integration/sdk_mesh_test.go b/tests/integration/sdk_mesh_test.go index e64adca0..61508bf1 100644 --- a/tests/integration/sdk_mesh_test.go +++ b/tests/integration/sdk_mesh_test.go @@ -496,6 +496,38 @@ func TestNativeSDKsMesh(t *testing.T) { } }) + // A member whose egress floor nobody attests is refused before anything is + // sent, MCP and HTTP alike; the members above carry a floor the mesh does + // satisfy (launchSDKMember). MCP to an SDK member fails before the floor: no /sam/mcp. + t.Run("egress-floor", func(t *testing.T) { + for _, launcher := range sdkMemberLaunchers { + cmd, skip := launcher.cmd(root) + if skip != "" { + continue + } + floored := launchSDKMember(t, launcher.name+"-floored", cmd, root, baseURL, adminToken, "SAM_SDK_EGRESS_REQUIRE_LABELS=team=nobody") + byFloor := func(what, target, err string) { + if !strings.Contains(err, "LabelsNotSatisfied") { + t.Errorf("%s %s to %s was not refused by the floor: %q", floored.name, what, target, err) + } + } + nodeRelayAddr := samNode.peerID.String() + byFloor("tools", "the node", floored.toolsRequiring(t, nodeRelayAddr, "mcp://"+serviceName, nil).Error) + byFloor("call", "the node", floored.callRaw(t, nodeRelayAddr, "mcp://"+serviceName, "add", nil).Error) + byFloor("http", "the node", floored.httpRaw(t, nodeRelayAddr, "egress://"+sdkMeshEgressHost, "/").Error) + for _, m := range members { + byFloor("http", m.name, floored.httpRaw(t, m.report.PeerID, "a2a://agent", "/card").Error) + if res := floored.toolsRequiring(t, m.report.PeerID, "mcp://"+serviceName, nil); res.OK { + t.Errorf("%s listed tools of %s, an SDK member: %v", floored.name, m.name, res.Tools) + } + if res := floored.callRaw(t, m.report.PeerID, "mcp://"+serviceName, "add", nil); res.OK { + t.Errorf("%s called a tool of %s, an SDK member: %+v", floored.name, m.name, res) + } + } + floored.quit(t) + } + }) + // Every SDK member is an agent: it accepts A2A requests for a2a://agent, // answered in the runner's process, reachable by peer ID through the // router. It publishes nothing; the policy rules it evaluates are the @@ -848,7 +880,8 @@ func startSDKMember(t *testing.T, name string, cmd *exec.Cmd, root, baseURL, adm } // launchSDKMember starts a conformance-join runner with a fresh bootstrap -// token and the labels the matrix uses, plus env, and reads its join report. +// token, the labels the matrix uses and an egress floor every provider in the +// mesh satisfies, plus env (a later entry wins), and reads its join report. func launchSDKMember(t *testing.T, name string, cmd *exec.Cmd, root, baseURL, adminToken string, env ...string) *sdkMember { t.Helper() tokenPath := filepath.Join(t.TempDir(), "bootstrap.token") @@ -860,6 +893,7 @@ func launchSDKMember(t *testing.T, name string, cmd *exec.Cmd, root, baseURL, ad "SAM_BOOTSTRAP_TOKEN_PATH="+tokenPath, "SAM_SDK_STATE_DIR="+filepath.Join(t.TempDir(), "state"), "SAM_SDK_LABELS="+labelsEnv(sdkMeshLabels), + "SAM_SDK_EGRESS_REQUIRE_LABELS=region="+sdkMeshLabels["region"], ), env...) cmd.Dir = root m := &sdkMember{name: name, cmd: cmd, stderr: &bytes.Buffer{}} @@ -1144,13 +1178,21 @@ type sdkHTTPResult struct { Body string `json:"body"` } -// http asks the member to call an inference or A2A service over /libp2p-http. -func (m *sdkMember) http(t *testing.T, addr, service, path string) sdkHTTPResult { +// httpRaw asks the member to call an inference or A2A service over +// /libp2p-http, and reports a refusal instead of failing. +func (m *sdkMember) httpRaw(t *testing.T, addr, service, path string) sdkHTTPResult { t.Helper() var res sdkHTTPResult if line := m.send(t, map[string]string{"cmd": "http", "addr": addr, "service": service, "path": path}); json.Unmarshal(line, &res) != nil { t.Fatalf("%s member: http answered %q", m.name, line) } + return res +} + +// http is httpRaw that must be answered. +func (m *sdkMember) http(t *testing.T, addr, service, path string) sdkHTTPResult { + t.Helper() + res := m.httpRaw(t, addr, service, path) if !res.OK { t.Fatalf("%s member could not call %s%s at %s: %s", m.name, service, path, addr, res.Error) } From caf520f5f3407f3b77f7ad6b6a6132b6d92fa1c7 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Mon, 28 Sep 2026 13:27:06 +0000 Subject: [PATCH 2/3] sdk: only a node passes the egress gate, and a malformed floor fails the runner sam-node's checkPeerLabels requires the node role on the provider before the requirement and the floor are evaluated; the HTTP-path gate in both SDKs now does the same, so a router's or an admin's credential attesting the floor is still not a provider. The runners' label parser refused nothing: a pair without "=" was dropped and a floor spelled wrong became no floor. It now fails on such a pair, as sam-node's parseRequiredLabels does. Two doc sentences reworded. --- sdk/README.md | 4 ++-- sdk/js/src/conformance-join.ts | 23 +++++++++++++------ sdk/js/src/session.test.ts | 6 ++++- sdk/js/src/session.ts | 5 +++- sdk/python/src/agent_mesh/conformance_join.py | 13 +++++++++-- sdk/python/src/agent_mesh/session.py | 5 +++- sdk/python/tests/test_session.py | 5 ++++ site/content/docs/guides/native-sdks.md | 4 ++-- 8 files changed, 49 insertions(+), 16 deletions(-) diff --git a/sdk/README.md b/sdk/README.md index cbca7984..fba48d7d 100644 --- a/sdk/README.md +++ b/sdk/README.md @@ -477,8 +477,8 @@ holds against the control plane's records. `MeshTransport`) verifies the provider with or without a floor, through the mutual `/sam/auth/1.0.0` handshake, as `sam-node`'s `VerifyPeerLabels` does before its egress proxy sends anything; a positive verdict is kept - per peer for five minutes (`labelGateTTL`), a miss never. Refusals are - `LabelsNotSatisfiedError`, which names the floor. + per peer for five minutes (`labelGateTTL`); a refusal is not kept. An + unmet floor is a `LabelsNotSatisfiedError` naming the floor. - `session.listTools(peer, service)` and `session.callTool(peer, service, tool, args)` on top of that. - Tests. Unit: each SDK calls a tool on an in-process provider that serves diff --git a/sdk/js/src/conformance-join.ts b/sdk/js/src/conformance-join.ts index 9fae3f01..54a12f6f 100644 --- a/sdk/js/src/conformance-join.ts +++ b/sdk/js/src/conformance-join.ts @@ -78,14 +78,23 @@ function emit(obj: unknown): void { process.stdout.write(JSON.stringify(obj) + "\n"); } -/** Labels from an environment variable written "k=v,k2=v2". */ +/** + * Labels from an environment variable written "k=v,k2=v2"; a pair without + * "=" is refused (parseRequiredLabels), so a typo never switches a floor off. + */ function labelsFromEnv(name: string): Record { - return Object.fromEntries( - (process.env[name] ?? "") - .split(",") - .filter((pair) => pair.includes("=")) - .map((pair) => pair.split("=", 2) as [string, string]), - ); + const out: Record = {}; + for (const pair of (process.env[name] ?? "").split(",")) { + if (pair.trim() === "") { + continue; + } + const eq = pair.indexOf("="); + if (eq === -1) { + throw new Error(`${name}: invalid label ${JSON.stringify(pair.trim())}: expected key=value`); + } + out[pair.slice(0, eq).trim()] = pair.slice(eq + 1).trim(); + } + return out; } function failure(cmd: string | undefined, err: unknown): unknown { diff --git a/sdk/js/src/session.test.ts b/sdk/js/src/session.test.ts index 1d04bb10..1b000131 100644 --- a/sdk/js/src/session.test.ts +++ b/sdk/js/src/session.test.ts @@ -388,6 +388,8 @@ test("an egress floor stated at join is held on the HTTP path, and the provider b.addFact(wasm.Fact.fromString("expiration(2035-01-01T00:00:00Z)")); return b.build(forged.getPrivateKey()).toBytes(); }); + // Enrolled and attesting the floor, but not a node: a router hosts no service. + const notANode = await serve((peerId) => mint(peerId, ROLE_ROUTER, undefined, { region: "eu" })); const addrOf = (host: Libp2p) => (host.getMultiaddrs()[0] as ReturnType).toString(); const sessions: MeshSession[] = []; const join = async (options: JoinOptions): Promise => { @@ -418,7 +420,9 @@ test("an egress floor stated at join is held on the HTTP path, and the provider // A banned provider is refused before any handshake. plain.banned.add(provider.peerId.toString(), Date.now()); await assert.rejects(plain.request(addrOf(provider), "a2a://agent", "/card"), /banned/); + // Only nodes host services, as sam-node's checkPeerLabels requires. + await assert.rejects(held.request(addrOf(notANode), "a2a://agent", "/card"), /lacks expected role "sam:role:node"/); } finally { - await Promise.all([...sessions.map((s) => s.close()), provider.stop(), impostor.stop()]); + await Promise.all([...sessions.map((s) => s.close()), provider.stop(), impostor.stop(), notANode.stop()]); } }); diff --git a/sdk/js/src/session.ts b/sdk/js/src/session.ts index 5c30f4ec..5274fbb1 100644 --- a/sdk/js/src/session.ts +++ b/sdk/js/src/session.ts @@ -19,6 +19,7 @@ import { peerIdFromString } from "@libp2p/peer-id"; import { isMultiaddr, multiaddr, type Multiaddr } from "@multiformats/multiaddr"; import { AUTH_HANDLER_OPTIONS, AUTH_PROTOCOL, authenticateWithPeer, authStreamHandler } from "./auth.ts"; import { ROLE_ROUTER, requireRole, type VerifiedBiscuit } from "./biscuit.ts"; +import { ROLE_NODE } from "./controlplane.ts"; import { canonicalPeerId } from "./identity.ts"; import { isServiceType, parseServiceTarget, serviceCID } from "./discovery.ts"; import { createMeshHost, listenThroughRelay, type MeshHost, type MeshHostOptions, type RelayListener } from "./host.ts"; @@ -637,7 +638,7 @@ export class MeshSession { /** * The connection an HTTP call goes out on, its peer verified as an enrolled - * member holding the floor before anything is sent (sam-node's VerifyPeerLabels). + * node holding the floor before anything is sent (sam-node's VerifyPeerLabels). */ async #egressConnection(peer: Peer, signal?: AbortSignal): Promise { const conn = await this.connect(peer, signal); @@ -647,6 +648,8 @@ export class MeshSession { return conn; } const provider = await authenticateWithPeer(conn, this.mesh.authFrame(), this.mesh.credential.controlPlaneKeys); + // Only nodes host services; a router's or an admin's credential is a member, not a provider. + requireRole(provider, ROLE_NODE); requireEgressLabels(provider, this.#egressRequireLabels); this.#egressVerdicts.set(peerId, new Date(Date.now() + EGRESS_VERDICT_TTL_MS)); return conn; diff --git a/sdk/python/src/agent_mesh/conformance_join.py b/sdk/python/src/agent_mesh/conformance_join.py index 3543e0e3..7e52dad1 100644 --- a/sdk/python/src/agent_mesh/conformance_join.py +++ b/sdk/python/src/agent_mesh/conformance_join.py @@ -70,8 +70,17 @@ def _emit(obj: dict) -> None: def _labels_from_env(name: str) -> dict[str, str]: - """Labels from an environment variable written "k=v,k2=v2".""" - return dict(pair.split("=", 1) for pair in os.environ.get(name, "").split(",") if "=" in pair) + """Labels from an environment variable written "k=v,k2=v2"; a pair without + "=" is refused (parseRequiredLabels), so a typo never switches a floor off.""" + out: dict[str, str] = {} + for pair in os.environ.get(name, "").split(","): + if not pair.strip(): + continue + key, sep, value = pair.partition("=") + if not sep: + raise SystemExit(f"{name}: invalid label {pair.strip()!r}: expected key=value") + out[key.strip()] = value.strip() + return out def _root_cause(err: BaseException) -> BaseException: diff --git a/sdk/python/src/agent_mesh/session.py b/sdk/python/src/agent_mesh/session.py index 648a4b8f..6d1ef571 100644 --- a/sdk/python/src/agent_mesh/session.py +++ b/sdk/python/src/agent_mesh/session.py @@ -40,6 +40,7 @@ from .auth import AUTH_PROTOCOL, auth_stream_handler, authenticate_with_peer from .authorizer import ProviderAuthorizerOptions from .biscuit import ROLE_ROUTER, VerifiedBiscuit, require_role +from .controlplane import ROLE_NODE from .discovery import DiscoveredProvider, find_peer, find_providers, parse_service_target, service_key from .host import create_mesh_host, dial, dial_addrs, peer_info from .httpx_transport import MESH_PATH_PREFIX @@ -522,12 +523,14 @@ async def request( ) async def _egress_peer(self, peer: Peer) -> ID: - """The peer an HTTP call goes out to, verified as an enrolled member + """The peer an HTTP call goes out to, verified as an enrolled node holding the floor before anything is sent (sam-node's VerifyPeerLabels).""" peer_id = await self.connect(peer) if self._egress_verdicts.get(str(peer_id), 0.0) > time.monotonic(): return peer_id provider = await authenticate_with_peer(self.host, peer_id, self.mesh.auth_frame(), self.mesh.credential.control_plane_keys) + # Only nodes host services; a router's or an admin's credential is a member, not a provider. + require_role(provider, ROLE_NODE) require_egress_labels(provider, self.egress_require_labels) self._egress_verdicts[str(peer_id)] = time.monotonic() + EGRESS_VERDICT_TTL return peer_id diff --git a/sdk/python/tests/test_session.py b/sdk/python/tests/test_session.py index a26afacc..fdbe5e8e 100644 --- a/sdk/python/tests/test_session.py +++ b/sdk/python/tests/test_session.py @@ -452,6 +452,8 @@ async def main(): _, impostor_addr = await start_provider( nursery, lambda p: ba.BiscuitBuilder("node({p}); expiration(2035-01-01T00:00:00Z);", {"p": p}).build(forged.private_key).to_bytes(), handshakes ) + # Enrolled and attesting the floor, but not a node: a router hosts no service. + _, not_a_node_addr = await start_provider(nursery, lambda p: mint(p, ROLE_ROUTER, labels={"region": "eu"}), handshakes) def join(**options): mesh = AgentMesh.enroll("http://127.0.0.1:1", bootstrap_token="sbt", transport=fake_control_plane([router_addr])) @@ -482,6 +484,9 @@ def join(**options): plain.banned.add(str(provider.get_id()), int(time.time() * 1000)) with pytest.raises(PermissionError, match="banned"): await plain.request(provider_addr, "a2a://agent", "/card") + # Only nodes host services, as sam-node's checkPeerLabels requires. + with pytest.raises(BiscuitVerificationError, match="lacks expected role 'sam:role:node'"): + await held.request(not_a_node_addr, "a2a://agent", "/card") nursery.cancel_scope.cancel() trio.run(with_timeout, 60, main) diff --git a/site/content/docs/guides/native-sdks.md b/site/content/docs/guides/native-sdks.md index 665b7307..12b310b6 100644 --- a/site/content/docs/guides/native-sdks.md +++ b/site/content/docs/guides/native-sdks.md @@ -890,8 +890,8 @@ round: `join({ egressRequireLabels })` (`join(egress_require_labels=)`) names labels every peer the session calls must attest, all of them, as `egress.require_labels` does for a `sam-node`. It is stated once at `join` and held for the session, on every call and however the peer was named; -the agent's calls cannot waive or widen it. It is the program author's -floor, not the operator's: nothing outside the process sets it. +the agent's calls cannot waive or widen it. The floor belongs to the +program that calls `join`; no configuration outside the process sets it. `acceptA2A` (`accept_a2a`) fetched the mesh policy and started answering. Every caller must present a credential signed by a trusted control plane From 12135afdb46afc353bd426fe5db53107c382230c Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Mon, 28 Sep 2026 13:57:28 +0000 Subject: [PATCH 3/3] sdk: only a node passes the MCP gate too The same node-role check as on the HTTP path, in openMCPSession and open_mcp_session right after the provider's credential verifies, as sam-node's checkPeerLabels runs it for every service call. --- sdk/js/src/mcp.test.ts | 14 +++++++++++++- sdk/js/src/mcp.ts | 5 ++++- sdk/python/src/agent_mesh/mcp_client.py | 5 ++++- sdk/python/tests/test_mcp.py | 13 ++++++++++--- 4 files changed, 31 insertions(+), 6 deletions(-) diff --git a/sdk/js/src/mcp.test.ts b/sdk/js/src/mcp.test.ts index aec677fa..b0fca7bd 100644 --- a/sdk/js/src/mcp.test.ts +++ b/sdk/js/src/mcp.test.ts @@ -31,7 +31,7 @@ import { readFileSync } from "node:fs"; import { after, before, test } from "node:test"; import { z } from "zod"; import { AuthRejectedError, MAX_AUTH_FRAME_BYTES, MCP_PROTOCOL } from "./auth.ts"; -import { loadBiscuit, verifyPeerBiscuit } from "./biscuit.ts"; +import { ROLE_ROUTER, loadBiscuit, verifyPeerBiscuit } from "./biscuit.ts"; import { ROLE_NODE } from "./controlplane.ts"; import { parseServiceTarget, serviceCID } from "./discovery.ts"; import { AuthFrameSchema, AuthResponseSchema } from "./gen/sam_pb.ts"; @@ -228,3 +228,15 @@ test("a service the provider does not have ends the session before MCP starts", const conn = await caller.dial(provider.getMultiaddrs()[0] as Parameters[0]); await assert.rejects(openMCPSession(conn, frame("mcp://no-such-service"), [cpKey], { signal: AbortSignal.timeout(3000) })); }); + +test("only a node is a provider, as sam-node's checkPeerLabels requires", async () => { + // The provider answers with the credential it holds at the time; a router's, attesting the floor, is not a provider's. + const nodeBiscuit = providerBiscuit; + providerBiscuit = mint(provider.peerId.toString(), ROLE_ROUTER, { region: "eu" }); + try { + const conn = await caller.dial(provider.getMultiaddrs()[0] as Parameters[0]); + await assert.rejects(openMCPSession(conn, frame("mcp://calc"), [cpKey], {}, { region: "eu" }), /lacks expected role "sam:role:node"/); + } finally { + providerBiscuit = nodeBiscuit; + } +}); diff --git a/sdk/js/src/mcp.ts b/sdk/js/src/mcp.ts index 7d75768f..fb92d6b1 100644 --- a/sdk/js/src/mcp.ts +++ b/sdk/js/src/mcp.ts @@ -22,7 +22,8 @@ import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; import { JSONRPCMessageSchema, type JSONRPCMessage } from "@modelcontextprotocol/sdk/types.js"; import { AUTH_HANDSHAKE_TIMEOUT_MS, AuthRejectedError, MAX_AUTH_FRAME_BYTES, MCP_PROTOCOL } from "./auth.ts"; -import { BiscuitVerificationError, verifyPeerBiscuit, type VerifiedBiscuit } from "./biscuit.ts"; +import { BiscuitVerificationError, requireRole, verifyPeerBiscuit, type VerifiedBiscuit } from "./biscuit.ts"; +import { ROLE_NODE } from "./controlplane.ts"; import { decodeAuthResponse } from "./credential.ts"; /** go-msgio's default message cap, which sam-node's StreamTransport uses. */ @@ -178,6 +179,8 @@ export async function openMCPSession( throw new AuthRejectedError(conn.remotePeer.toString(), resp.error || "no reason given"); } provider = await verifyPeerBiscuit(resp.biscuit, conn.remotePeer.toString(), trustedKeys); + // Only nodes host services; a router's or an admin's credential is a member, not a provider. + requireRole(provider, ROLE_NODE); requireLabels(provider, options.requiredLabels); requireEgressLabels(provider, egressRequireLabels); } catch (err) { diff --git a/sdk/python/src/agent_mesh/mcp_client.py b/sdk/python/src/agent_mesh/mcp_client.py index 72ee3cc2..61e8574b 100644 --- a/sdk/python/src/agent_mesh/mcp_client.py +++ b/sdk/python/src/agent_mesh/mcp_client.py @@ -34,7 +34,8 @@ from ._proto import sam_pb2 as pb from .auth import AUTH_HANDSHAKE_TIMEOUT, MAX_AUTH_FRAME_BYTES, MCP_PROTOCOL, AuthRejectedError -from .biscuit import BiscuitVerificationError, VerifiedBiscuit, verify_peer_biscuit +from .biscuit import BiscuitVerificationError, VerifiedBiscuit, require_role, verify_peer_biscuit +from .controlplane import ROLE_NODE from .host import open_stream logger = logging.getLogger("agent_mesh") @@ -126,6 +127,8 @@ async def open_mcp_session( raise AuthRejectedError(str(peer_id), resp.error or "no reason given") try: provider = verify_peer_biscuit(resp.biscuit, str(peer_id), trusted_keys) + # Only nodes host services; a router's or an admin's credential is a member, not a provider. + require_role(provider, ROLE_NODE) except BiscuitVerificationError as err: raise AuthRejectedError(str(peer_id), f"provider credential rejected: {err}") from err require_labels(provider, required_labels) diff --git a/sdk/python/tests/test_mcp.py b/sdk/python/tests/test_mcp.py index 6d049b68..d5f4002f 100644 --- a/sdk/python/tests/test_mcp.py +++ b/sdk/python/tests/test_mcp.py @@ -33,7 +33,7 @@ from agent_mesh._proto import sam_pb2 as pb from agent_mesh.auth import MCP_PROTOCOL, AuthRejectedError -from agent_mesh.biscuit import VerifiedBiscuit, verify_peer_biscuit +from agent_mesh.biscuit import ROLE_ROUTER, VerifiedBiscuit, verify_peer_biscuit from agent_mesh.controlplane import ROLE_NODE from agent_mesh.discovery import parse_service_target, service_key from agent_mesh.identity import Identity @@ -129,10 +129,10 @@ async def pump_out(): return handle -async def start_provider(nursery, labels=None): +async def start_provider(nursery, labels=None, role=ROLE_NODE): identity = Identity.generate() host = libp2p_host(identity) - biscuit = mint(identity.peer_id, ROLE_NODE, labels) + biscuit = mint(identity.peer_id, role, labels) served: list[str] = [] host.set_stream_handler(MCP_PROTOCOL, mcp_stream_handler(biscuit, served)) started = trio.Event() @@ -227,6 +227,13 @@ async def main(): with trio.fail_after(5): async with open_mcp_session(caller, pid, frame(caller_biscuit, "mcp://no-such-service"), [CP_KEY]): pass + + # Only a node is a provider, as sam-node's checkPeerLabels requires: a router attesting the floor is not. + router, router_addr, _ = await start_provider(nursery, labels={"region": "eu"}, role=ROLE_ROUTER) + await caller.connect(info_from_p2p_addr(router_addr)) + with pytest.raises(AuthRejectedError, match="lacks expected role 'sam:role:node'"): + async with open_mcp_session(caller, router.get_id(), frame(caller_biscuit, "mcp://calc"), [CP_KEY], egress_require_labels={"region": "eu"}): + pass nursery.cancel_scope.cancel() async def with_timeout():