sdk/js: reach peers through a router the control plane names by /dnsaddr

.github/k8s/sam-sdk-canary-template.yaml

@@ -233,21 +233,27 @@ data:
     #!/bin/sh
     # Sends the agent beside this container a message with the A2A SDK's
     # client every INTERVAL seconds, by the peer ID it printed to
-    # /var/run/sam/agent.log, and keeps
-    # /var/run/sam/reachable present while the last call succeeded; the
-    # readiness probe reads it. One JSON line per call, for a log-based
-    # metric. The first call enrolls with the projected token; the later ones
-    # resume from SAM_STATE_DIR.
+    # $SAM_CANARY_DIR/agent.log, and keeps $SAM_CANARY_DIR/reachable present
+    # while the last call succeeded; the readiness probe reads it. One JSON
+    # line per call, for a log-based metric. The first call enrolls with the
+    # projected token; the later ones resume from SAM_STATE_DIR.
+    # tests/integration runs this script against the example programs.
     set -u
     INTERVAL=${INTERVAL:-300}
-    LOG=/var/run/sam/agent.log
-    REACHABLE=/var/run/sam/reachable
+    LOG=${SAM_CANARY_DIR:-/var/run/sam}/agent.log
+    REACHABLE=${SAM_CANARY_DIR:-/var/run/sam}/reachable
 
     verdict() {
       printf '{"canary":"%s","ok":%s,"agent_peer":"%s","call_s":%d,"error":"%s"}\n' \
         "$CANARY" "$1" "${AGENT_PEER:-}" "$2" "$(printf '%s' "$3" | tr -d '"\n' | cut -c1-300)"
     }
 
+    # The line naming the error: Node prints it before the stack trace and
+    # Python after, so the end of the output is not the place to look.
+    reason() {
+      printf '%s\n' "$1" | grep -m1 -E '^[A-Za-z_.]*(Error|Exception)\b' || printf '%s' "$1" | tail -c 200
+    }
+
     while :; do
       AGENT_PEER=$(grep -o 'accepting a2a://agent as [^ ]*' "$LOG" 2>/dev/null | tail -1 | cut -d' ' -f4)
       if [ -z "$AGENT_PEER" ]; then
@@ -262,7 +268,7 @@ data:
         sleep "$INTERVAL"
       else
         rm -f "$REACHABLE"
-        verdict false $(( $(date +%s) - start )) "$(printf '%s' "$out" | tail -c 200)"
+        verdict false $(( $(date +%s) - start )) "$(reason "$out")"
         # A failed call is retried soon; the readiness probe already shows it.
         sleep 30
       fi

hack/lint.sh

@@ -21,17 +21,17 @@ set -o pipefail
 REPO_ROOT=$(dirname "${BASH_SOURCE[0]}")/..
 
 cd $REPO_ROOT
-docker run --rm -v $(pwd):/app -w /app golangci/golangci-lint:v2.11.4 golangci-lint run -v
+docker run --rm -v $(pwd):/app -w /app golangci/golangci-lint:v2.14.0 golangci-lint run -v
 
 # nano-init is a separate module, so ./... above does not reach it.
-docker run --rm -v $(pwd):/app -w /app/cmd/nano-init golangci/golangci-lint:v2.11.4 golangci-lint run -v
+docker run --rm -v $(pwd):/app -w /app/cmd/nano-init golangci/golangci-lint:v2.14.0 golangci-lint run -v
 
 # golangci-lint has no deadcode linter (removed upstream in v1.49) and its
 # replacement, "unused", ignores exported identifiers. This catches exported
 # code that is unreachable from every binary and test.
 # mobile/ is exported to Android over cgo/FFI and development/examples/ is sample code.
 DEADCODE_EXCLUDES='^(mobile/|development/examples/)'
-deadcode_report=$(go run golang.org/x/tools/cmd/deadcode@v0.40.0 -test ./... | grep -Ev "${DEADCODE_EXCLUDES}" || true)
+deadcode_report=$(go run golang.org/x/tools/cmd/deadcode@v0.50.0 -test ./... | grep -Ev "${DEADCODE_EXCLUDES}" || true)
 if [[ -n "${deadcode_report}" ]]; then
   echo "Dead code detected (unreachable from any binary or test):"
   echo "${deadcode_report}"

internal/node/inference_service.go

@@ -110,11 +110,10 @@ func (s *InferenceService) trackActive(next http.Handler) http.Handler {
 
 func (s *InferenceService) newInferenceProxy() http.Handler {
 	return &httputil.ReverseProxy{
-		Director: func(req *http.Request) {
-			if _, ok := req.Header["User-Agent"]; !ok {
-				req.Header.Set("User-Agent", "")
-			}
-		},
+		// The transport addresses the backend; the proxy itself strips the
+		// inbound Forwarded and X-Forwarded-* headers and blanks a missing
+		// User-Agent.
+		Rewrite: func(*httputil.ProxyRequest) {},
 		Transport: &inferenceTransport{
 			backend: s.backendURL,
 			auth:    s.backendAuth,
@@ -132,11 +131,6 @@ type inferenceTransport struct {
 func (t *inferenceTransport) RoundTrip(req *http.Request) (*http.Response, error) {
 	attemptReq := req.Clone(req.Context())
 	attemptReq.Header.Del("Accept-Encoding") // Prevent gzipped response from breaking token tracking
-	// A Director proxy does not manage X-Forwarded-For, and RemoteAddr here is
-	// a peer id, so an inbound value would reach the backend as-is.
-	attemptReq.Header.Del("X-Forwarded-For")
-	attemptReq.Header.Del("X-Forwarded-Host")
-	attemptReq.Header.Del("X-Forwarded-Proto")
 	t.auth.apply(attemptReq.Header)
 
 	attemptReq.URL.Scheme = t.backend.Scheme

internal/node/inference_test.go

@@ -41,6 +41,49 @@ func getCounterValue(peerID, model, tokenType string) float64 {
 	return m.GetCounter().GetValue()
 }
 
+// The backend sees no forwarding header a caller sent and no Go default
+// User-Agent: ReverseProxy strips Forwarded and X-Forwarded-* before its
+// Rewrite runs and blanks a missing User-Agent, the same in Rewrite mode
+// as the Director used to do by hand.
+func TestInferenceService_ForwardingHeadersDoNotReachTheBackend(t *testing.T) {
+	var seen http.Header
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		seen = r.Header.Clone()
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer server.Close()
+
+	svc := &InferenceService{
+		baseService: baseService{
+			info:    &api.ServiceInfo{Type: api.ServiceType_SERVICE_TYPE_INFERENCE, Name: "test-inference-headers"},
+			backend: &api.RegisterServiceRequest_TargetUrl{TargetUrl: server.URL},
+		},
+	}
+	if err := svc.Init(context.Background()); err != nil {
+		t.Fatalf("Init failed: %v", err)
+	}
+
+	req := httptest.NewRequest("POST", "http://localhost/v1/chat/completions", strings.NewReader(`{}`))
+	req.RemoteAddr = "12D3KooWpeer"
+	req.Header.Set("Forwarded", "for=spoofed;host=spoofed;proto=spoofed")
+	req.Header.Set("X-Forwarded-For", "spoofed")
+	req.Header.Set("X-Forwarded-Host", "spoofed")
+	req.Header.Set("X-Forwarded-Proto", "spoofed")
+	w := httptest.NewRecorder()
+	svc.Handler().ServeHTTP(w, req)
+	if w.Code != http.StatusNoContent {
+		t.Fatalf("status = %d, want %d", w.Code, http.StatusNoContent)
+	}
+	for _, name := range []string{"Forwarded", "X-Forwarded-For", "X-Forwarded-Host", "X-Forwarded-Proto"} {
+		if got, ok := seen[name]; ok {
+			t.Errorf("backend saw %s: %q", name, got)
+		}
+	}
+	if got := seen.Get("User-Agent"); got != "" {
+		t.Errorf("backend saw User-Agent %q, want none", got)
+	}
+}
+
 func TestInferenceService_TokenAccountability_JSON(t *testing.T) {
 	peerID := "test-peer-json"
 	modelName := "test-model-json"

internal/node/sidecar.go

@@ -690,16 +690,16 @@ func createEgressProxy(node *SamNode) http.Handler {
 	transport := libp2phttp.NewTransport(node.Host)
 
 	proxy := &httputil.ReverseProxy{
-		Director: func(req *http.Request) {
-			ctx := allowLimitedEgressConn(req.Context())
-			*req = *req.WithContext(ctx)
+		Rewrite: func(pr *httputil.ProxyRequest) {
+			req := pr.Out.WithContext(allowLimitedEgressConn(pr.Out.Context()))
+			pr.Out = req
 
 			parts := strings.SplitN(req.URL.Path, "/", 6)
 			if len(parts) < 5 {
 				return
 			}
 			peerID := parts[2]
-			node.prepareEgressPeer(ctx, peerID)
+			node.prepareEgressPeer(req.Context(), peerID)
 			serviceType := parts[3]
 			serviceName := parts[4]
 			upstreamPath := ""
@@ -763,8 +763,8 @@ func createEgressProxy(node *SamNode) http.Handler {
 			return
 		}
 		// The verdict is for the canonical peer, so the dial must name the
-		// same form: rewrite the segment the Director will re-parse rather
-		// than let a non-canonical spelling travel past the gate.
+		// same form: rewrite the segment the proxy's Rewrite will re-parse
+		// rather than let a non-canonical spelling travel past the gate.
 		if canonical := pid.String(); route.peerID != canonical {
 			parts := strings.SplitN(r.URL.Path, "/", 6)
 			if len(parts) >= 3 {

internal/standalone/standalone_test.go

@@ -202,4 +202,3 @@ func TestResponseRecorder(t *testing.T) {
 		t.Fatalf("rec.Header(Content-Length) = %q, want 12", got)
 	}
 }
-

internal/storage/round_trip_test.go

@@ -47,7 +47,7 @@ func requireAllFieldsSet(t *testing.T, v any, skip ...string) {
 	}
 
 	val := reflect.ValueOf(v)
-	if val.Kind() == reflect.Ptr {
+	if val.Kind() == reflect.Pointer {
 		val = val.Elem()
 	}
 	for i := 0; i < val.NumField(); i++ {
@@ -73,7 +73,7 @@ func requireFieldsRoundTrip(t *testing.T, want, got any, skip ...string) {
 
 	w := reflect.ValueOf(want)
 	g := reflect.ValueOf(got)
-	if w.Kind() == reflect.Ptr {
+	if w.Kind() == reflect.Pointer {
 		w, g = w.Elem(), g.Elem()
 	}
 	for i := 0; i < w.NumField(); i++ {

sdk/js/src/host.ts

@@ -28,7 +28,7 @@ import { ping } from "@libp2p/ping";
 import { tcp } from "@libp2p/tcp";
 import { tls } from "@libp2p/tls";
 import type { Multiaddr } from "@multiformats/multiaddr";
-import { createLibp2p } from "libp2p";
+import { createLibp2p, type Libp2pOptions } from "libp2p";
 import { DHT_PROTOCOL } from "./discovery.ts";
 import type { Identity } from "./identity.ts";
 
@@ -44,6 +44,8 @@ export interface MeshHostOptions {
    * it current from /info and the gossip events.
    */
   banned?: { has(peerId: string): boolean };
+  /** The resolver for `/dnsaddr` and `/dns*` addresses; the system's by default. */
+  dns?: Libp2pOptions["dns"];
 }
 
 /** The services a mesh host runs; `services.pubsub` carries the control plane's events. */
@@ -55,6 +57,7 @@ export async function createMeshHost(identity: Identity, options: MeshHostOption
   return createLibp2p({
     privateKey: privateKeyFromProtobuf(identity.toLibp2pPrivateKey()),
     addresses: { listen: options.listenAddrs ?? [] },
+    ...(options.dns !== undefined ? { dns: options.dns } : {}),
     transports: [tcp(), circuitRelayTransport()],
     connectionEncrypters: [tls()],
     streamMuxers: [yamux()],

sdk/js/src/session.test.ts

@@ -156,6 +156,39 @@ test("join authenticates with the router, reserves a relay slot and answers peer
   }
 });
 
+test("a router handed out as /dnsaddr still relays to peers", async () => {
+  // The testnets advertise their routers as /dnsaddr/<host>/p2p/<id>, one
+  // TXT lookup away from the addresses. A stub resolver answers with the
+  // real router's addresses, as the records would.
+  const dnsaddr = `/dnsaddr/router.test/p2p/${router.peerId.toString()}`;
+  const dns = {
+    query: async (domain: string) => {
+      assert.equal(domain, "_dnsaddr.router.test");
+      return { Answer: router.getMultiaddrs().map((ma) => ({ name: domain, type: 16, TTL: 60, data: `dnsaddr=${ma.toString()}` })) };
+    },
+  } as unknown as NonNullable<Parameters<typeof createLibp2p>[0]>["dns"];
+
+  const agent = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: fakeControlPlane([routerAddr]) });
+  const agentSession = await agent.join({ refreshLeadMs: 0 });
+  const caller = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: fakeControlPlane([dnsaddr]) });
+  const callerSession = await caller.join({ refreshLeadMs: 0, dns });
+  try {
+    // The router is known by the address the connection was made on: the
+    // relay address for a peer is dialable as it stands.
+    const routerAddrs = callerSession.routers.map((r) => r.addr.toString());
+    assert.deepEqual(routerAddrs, [routerAddr]);
+    assert.ok(callerSession.relayAddresses.some((ma) => ma.toString().startsWith(routerAddr)), `reserved on ${callerSession.relayAddresses.map(String).join(",")}`);
+    const targets = callerSession.dialTargets(agent.peerId).addrs.map(String);
+    assert.deepEqual(targets, [`${routerAddr}/p2p-circuit/p2p/${agent.peerId}`]);
+
+    const verified = await callerSession.authenticate(agent.peerId);
+    assert.equal(verified.peerId, agent.peerId);
+  } finally {
+    await callerSession.close();
+    await agentSession.close();
+  }
+});
+
 test("join fails closed when the router is not a router", async () => {
   // A relay whose credential lacks the router role must not admit us to the mesh.
   const impostor = await createLibp2p({

sdk/js/src/session.ts

@@ -68,6 +68,14 @@ export interface JoinOptions extends MeshHostOptions {
 
 export interface AdmittedRouter {
   peerId: string;
+  /**
+   * The address the connection to the router was made on, resolved: a
+   * `/ip4` or `/ip6` address ending in `/p2p/<router>`. Relay addresses
+   * are built on it. The control plane may hand a router out as
+   * `/dnsaddr/<host>/p2p/<router>`; js-libp2p resolves such an address to
+   * the TXT records themselves and drops what follows it, so a
+   * `/dnsaddr/.../p2p-circuit/p2p/<peer>` address reaches nobody.
+   */
   addr: Multiaddr;
   credential: VerifiedBiscuit;
 }
@@ -557,7 +565,7 @@ export async function joinMesh(mesh: AgentMesh, options: JoinOptions = {}): Prom
         // Enforced under the key that verified the token; a relay that is
         // not a router must not become our way onto the mesh.
         requireRole(credential, ROLE_ROUTER);
-        admitted.push({ peerId: routerPeer, addr, credential });
+        admitted.push({ peerId: routerPeer, addr: connectedAddress(conn, routerPeer), credential });
       } catch (err) {
         failures.push(`${addr.toString()}: ${err instanceof Error ? err.message : String(err)}`);
       }
@@ -583,6 +591,11 @@ function targetPeerOf(ma: Multiaddr): string | undefined {
   return last?.name === "p2p" && last.value !== undefined ? canonicalPeerId(last.value) : undefined;
 }
 
+/** The remote address of a connection, ending in `/p2p/<peerId>`. */
+function connectedAddress(conn: Connection, peerId: string): Multiaddr {
+  return targetPeerOf(conn.remoteAddr) === undefined ? conn.remoteAddr.encapsulate(`/p2p/${peerId}`) : conn.remoteAddr;
+}
+
 /** Canonicalizes a list from the control plane, dropping entries that are not peer IDs. */
 function canonicalPeerIds(ids: string[]): string[] {
   const out: string[] = [];