diff --git a/AGENTS.md b/AGENTS.md index 854e4fff..9cf4156f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -13,6 +13,7 @@ You are an expert software engineering assistant helping to develop, maintain, a * **Two-Token Model & Complementary Gateway/PEP Integration:** Inside the mesh, credentials are Biscuits; at both borders, credentials are standard JWTs (`site/content/docs/contributing/security-architecture.md`). Inbound platform credentials (OIDC, K8s SA, SPIFFE JWT-SVID) exchange into delegated Biscuits (`POST /token/exchange`); outbound verified Biscuits mint short-lived ES256 JWTs at the control plane (`POST /sts/token`) that the egress node exchanges at cloud STS endpoints (`CloudTokenExchanger`). `agentmesh-node` integrates with existing gateways (`agentgateway`, Istio, Envoy) via Envoy `ext_authz` / `ext_proc` and RFC 8693 `/oauth/token`, while native SDKs (`sdk/js`, `sdk/python`) attenuate and seal Biscuits in memory. * **Policy on Names:** egress policy, secret injection and routing decisions are made on the destination *name*, never on an IP. Deny by default. * **Zero Trust:** Enforce a Zero Trust architecture. Assume no implicit trust between nodes, control planes, routers, or external actors. All data passing through the API must be authenticated, authorized, and validated. +* **Nothing interacts with the mesh unauthenticated:** a router or a member serves a mesh protocol (the DHT, gossip, the relay, the data plane) only to a peer that completed the mesh handshake on that connection and whose credential has not lapsed; a stream from any other peer is reset and counted. The only protocols open to a peer that has not authenticated yet are the ones the handshake itself needs: the transport security handshake, identify, and `/mesh/auth`. A libp2p library that registers its own stream handlers (kad-dht, gossipsub) is built on a host wrapper that installs this check in front of every handler, and its routing table admits only authenticated peers. A connection that has not authenticated within the deadline is closed. This holds for reads as much as for writes: a directory lookup or a gossip subscription is mesh data. * **Zero config, safe defaults:** the goal is zero trust with zero configuration. A component started with no flags must be secure; every default is the safe value (deny, verify, short-lived, loopback-only, bounded), and an operator opts into anything looser with an explicit flag whose help text names the risk. Before adding a flag, option or knob, ask whether a safe default removes the need for it. * **Auditable and traceable behavior:** when a member is admitted, refused, disconnected, shed, skipped or retried, the process says so in a log line at the right level or in a counter, and the states a component is in (attached routers, authenticated peers, listed peer routers, leased addresses) are gauges. A silent `continue`, a swallowed error or a condition that short-circuits a decision without a trace is a defect even when the behavior is correct. When a failure in a test or on a testnet cannot be explained from the logs and metrics the binaries already emit, the fix includes the missing instrumentation, so that the next occurrence can be read from the data instead of reproduced. * **Simple UX:** Maintain a very simple User Experience. Configuration, CLI usage, and error messages must be intuitive, minimal, and explicitly clear. diff --git a/internal/node/auth_gate.go b/internal/node/auth_gate.go new file mode 100644 index 00000000..baa410d7 --- /dev/null +++ b/internal/node/auth_gate.go @@ -0,0 +1,137 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package node + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/network" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/libp2p/go-libp2p/core/protocol" +) + +// streamScopeUnauthenticated labels a stream the gate refused: its peer +// never authenticated while the stream waited. +const streamScopeUnauthenticated = "unauthenticated" + +// authWait bounds how long a stream waits for its peer's handshake: a +// router's auth deadline for the connection, so a peer that never +// authenticates is cut off with it. +const authWait = time.Minute + +const authPollInterval = 100 * time.Millisecond + +var errNotAuthenticated = errors.New("peer did not authenticate") + +// meshHost is the host the node's DHT and gossipsub are built on. A stream +// a library opens to a router runs the handshake with that router first, +// so the router serves it at once instead of holding it; a stream to another +// member waits for that member to authenticate, as it does when it calls. +// A stream a library serves is served only to a router this node +// authenticated with or a peer that authenticated with this node; any other +// waits for that, bounded, and is then refused and counted. +type meshHost struct { + host.Host + n *AgentMeshNode +} + +func (h *meshHost) SetStreamHandler(pid protocol.ID, handler network.StreamHandler) { + h.Host.SetStreamHandler(pid, h.n.requireAuthenticated(handler)) +} + +func (h *meshHost) SetStreamHandlerMatch(pid protocol.ID, match func(protocol.ID) bool, handler network.StreamHandler) { + h.Host.SetStreamHandlerMatch(pid, match, h.n.requireAuthenticated(handler)) +} + +func (h *meshHost) NewStream(ctx context.Context, p peer.ID, pids ...protocol.ID) (network.Stream, error) { + if err := h.n.ensurePeerAuthenticated(ctx, p); err != nil { + return nil, err + } + return h.Host.NewStream(ctx, p, pids...) +} + +// isPeerAuthenticated reports whether p is a router this node authenticated +// with on an open session, or a peer whose handshake with this node holds. +func (n *AgentMeshNode) isPeerAuthenticated(p peer.ID) bool { + return n.isAuthenticatedAndConnected(p) || n.isAdmitted(p) +} + +// ensurePeerAuthenticated returns once p is authenticated: for a router the +// control plane lists, by running the handshake (one at a time per router, +// shared with the one the connection notifee started). Any other peer +// authenticates with this node when it calls it; until it has, nothing is +// sent to it, and a lookup is not held waiting for a peer it cannot vouch +// for. +func (n *AgentMeshNode) ensurePeerAuthenticated(ctx context.Context, p peer.ID) error { + if n.isPeerAuthenticated(p) { + return nil + } + if !n.isCatalogRouter(p) { + return fmt.Errorf("%s: %w", p, errNotAuthenticated) + } + biscuitBytes, err := n.loadIdentityForAuth() + if err != nil { + return err + } + return n.authRouter(ctx, peer.AddrInfo{ID: p}, biscuitBytes) +} + +// requireAuthenticated wraps a stream handler so it runs only once the +// stream's peer is authenticated. +func (n *AgentMeshNode) requireAuthenticated(next network.StreamHandler) network.StreamHandler { + return func(s network.Stream) { + p := s.Conn().RemotePeer() + if err := n.awaitPeerAuthenticated(context.Background(), p); err != nil { + streamsRefusedTotal.WithLabelValues(streamScopeUnauthenticated, string(s.Protocol())).Inc() + logger.Debugf("[AuthZ] Refusing %s stream from %s: %v", s.Protocol(), p, err) + _ = s.Reset() + return + } + next(s) + } +} + +// awaitPeerAuthenticated returns once p is authenticated, or with an error +// when ctx ends, p's connections are gone, or authWait passes. +func (n *AgentMeshNode) awaitPeerAuthenticated(ctx context.Context, p peer.ID) error { + if n.isPeerAuthenticated(p) { + return nil + } + parent := ctx + ctx, cancel := context.WithTimeout(ctx, authWait) + defer cancel() + tick := time.NewTicker(authPollInterval) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + if parent.Err() != nil { + return parent.Err() + } + return errNotAuthenticated + case <-tick.C: + } + if n.isPeerAuthenticated(p) { + return nil + } + if n.Host.Network().Connectedness(p) != network.Connected { + return errNotAuthenticated + } + } +} diff --git a/internal/node/auth_gate_test.go b/internal/node/auth_gate_test.go new file mode 100644 index 00000000..332fe540 --- /dev/null +++ b/internal/node/auth_gate_test.go @@ -0,0 +1,153 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package node + +import ( + "context" + "errors" + "io" + "testing" + "time" + + "github.com/libp2p/go-libp2p" + "github.com/libp2p/go-libp2p/core/network" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/libp2p/go-libp2p/core/protocol" + "github.com/libp2p/go-msgio" + "google.golang.org/protobuf/proto" + + "github.com/google/agentmesh/api" +) + +const echoProtocol protocol.ID = "/test/echo/1.0.0" + +func echo(s network.Stream) { + defer func() { _ = s.Close() }() + _, _ = io.Copy(s, s) +} + +// A stream the node opens to a listed router it has not authenticated with +// runs the handshake first, once, shared with the one the connection +// notifee starts; a stream to a peer the node cannot vouch for is refused +// at once, so a lookup is never held on it. +func TestNodeStreamsToARouterFollowItsHandshake(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + + r1 := newFakeRouter(t, "/ip4/127.0.0.1/tcp/0") + r2 := newFakeRouterWithKey(t, r1.cpPub, r1.cpPriv, "/ip4/127.0.0.1/tcp/0") + node := r1.startNode(t, ctx, append(r1.routerAddrs, r2.routerAddrs...), func(o *Options) { o.Routers = 1 }) + other := r2 + if node.attachedRouters()[0] == r2.h.ID() { + other = r1 + } + other.h.SetStreamHandler(echoProtocol, echo) + gated := &meshHost{Host: node.Host, n: node} + + s, err := gated.NewStream(ctx, other.h.ID(), echoProtocol) + if err != nil { + t.Fatalf("NewStream to a listed router: %v", err) + } + _ = s.Close() + if !node.isAuthenticatedAndConnected(other.h.ID()) { + t.Fatal("the stream opened before the handshake with the router") + } + if got := other.handshakes.Load(); got != 1 { + t.Errorf("the router saw %d handshakes, want 1", got) + } + + stranger, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = stranger.Close() }) + stranger.SetStreamHandler(echoProtocol, echo) + if err := node.Host.Connect(ctx, peer.AddrInfo{ID: stranger.ID(), Addrs: stranger.Addrs()}); err != nil { + t.Fatal(err) + } + start := time.Now() + if _, err := gated.NewStream(ctx, stranger.ID(), echoProtocol); !errors.Is(err, errNotAuthenticated) { + t.Fatalf("NewStream to a peer that never authenticated: err = %v, want %v", err, errNotAuthenticated) + } + if took := time.Since(start); took > time.Second { + t.Fatalf("the refusal took %v; a lookup must not wait on a peer the node cannot vouch for", took) + } +} + +// A handler the node registers through its gated host serves a peer only +// once that peer authenticated with the node; a stream that arrives first +// waits for the handshake and is then served. +func TestNodeServesOnlyAuthenticatedPeers(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + + r := newFakeRouter(t, "/ip4/127.0.0.1/tcp/0") + node := r.startNode(t, ctx, r.routerAddrs) + gated := &meshHost{Host: node.Host, n: node} + gated.SetStreamHandler(echoProtocol, echo) + + caller, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = caller.Close() }) + if err := caller.Connect(ctx, peer.AddrInfo{ID: node.Host.ID(), Addrs: node.Host.Addrs()}); err != nil { + t.Fatal(err) + } + + s, err := caller.NewStream(ctx, node.Host.ID(), echoProtocol) + if err != nil { + t.Fatal(err) + } + defer func() { _ = s.Close() }() + if _, err := s.Write([]byte("hello")); err != nil { + t.Fatal(err) + } + buf := make([]byte, 5) + _ = s.SetReadDeadline(time.Now().Add(300 * time.Millisecond)) + if n, err := s.Read(buf); err == nil { + t.Fatalf("read %q before the handshake; the stream must wait for it", buf[:n]) + } + + // The caller authenticates with the node, as it does before a call. + as, err := caller.NewStream(ctx, node.Host.ID(), api.AuthProtocolID) + if err != nil { + t.Fatal(err) + } + frame, err := proto.Marshal(&api.AuthFrame{Biscuit: r.mint(caller.ID().String(), api.RoleNode)}) + if err != nil { + t.Fatal(err) + } + if err := msgio.NewVarintWriter(as).WriteMsg(frame); err != nil { + t.Fatal(err) + } + reply, err := msgio.NewVarintReaderSize(as, 1024*64).ReadMsg() + if err != nil { + t.Fatalf("handshake reply: %v", err) + } + var resp api.AuthResponse + if err := proto.Unmarshal(reply, &resp); err != nil || !resp.Success { + t.Fatalf("handshake refused: %v %s", err, resp.GetError()) + } + _ = as.Close() + + _ = s.SetReadDeadline(time.Now().Add(5 * time.Second)) + if _, err := io.ReadFull(s, buf); err != nil { + t.Fatalf("read after the handshake: %v", err) + } + if string(buf) != "hello" { + t.Fatalf("got %q, want %q", buf, "hello") + } +} diff --git a/internal/node/gossip_hardening_test.go b/internal/node/gossip_hardening_test.go index b419ba32..9d4a12c1 100644 --- a/internal/node/gossip_hardening_test.go +++ b/internal/node/gossip_hardening_test.go @@ -178,11 +178,14 @@ func TestGossipJunkFloodIsRejectedNotForwarded(t *testing.T) { victim.keysMu.Unlock() // attacker -> victim -> observer. The observer only ever hears what the - // victim chooses to forward. + // victim chooses to forward. Both authenticated with the victim, as a + // caller does: the victim gossips with no one who did not. attacker := newGossipPeer(t, ctx) observer := newGossipPeer(t, ctx) - attacker.connect(t, ctx, victim.Host) - observer.connect(t, ctx, victim.Host) + for _, g := range []*gossipPeer{attacker, observer} { + victim.authPeers.Store(g.host.ID(), time.Now().Add(time.Hour)) + g.connect(t, ctx, victim.Host) + } var mu sync.Mutex var forwardedJunk []string diff --git a/internal/node/node.go b/internal/node/node.go index f8c3c651..5125d159 100644 --- a/internal/node/node.go +++ b/internal/node/node.go @@ -688,8 +688,11 @@ func (n *AgentMeshNode) Start(ctx context.Context) error { } } + // The DHT and gossip are served by, and serve, authenticated peers only. + gated := &meshHost{Host: h, n: n} + // Initialize Rendezvous (DHT Client) - kdht, err := dht.New(h, dhtOptions(n.directorySettings())...) + kdht, err := dht.New(gated, dhtOptions(n.directorySettings())...) if err != nil { return err } @@ -715,7 +718,7 @@ func (n *AgentMeshNode) Start(ctx context.Context) error { // library default; it is pinned because the event validator and the // per-author rate limit key on msg.GetFrom(), which only the signature // makes trustworthy. - ps, err := pubsub.NewGossipSub(ctx, h, pubsub.WithMessageSignaturePolicy(pubsub.StrictSign)) + ps, err := pubsub.NewGossipSub(ctx, gated, pubsub.WithMessageSignaturePolicy(pubsub.StrictSign)) if err != nil { return err } diff --git a/internal/router/auth_gate.go b/internal/router/auth_gate.go new file mode 100644 index 00000000..22e1d06e --- /dev/null +++ b/internal/router/auth_gate.go @@ -0,0 +1,128 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package router + +import ( + "context" + "errors" + "time" + + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/network" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/libp2p/go-libp2p/core/protocol" +) + +// streamScopeUnauthenticated labels a stream the gate refused: its peer's +// connection ended before the peer authenticated. +const streamScopeUnauthenticated = "unauthenticated" + +// authPollInterval is how often a stream waiting on its peer's handshake +// checks for it. +const authPollInterval = 100 * time.Millisecond + +// defaultAuthWait bounds the wait for a peer's handshake when the router +// runs without an auth deadline, which would otherwise close the connection. +const defaultAuthWait = time.Minute + +var errNotAuthenticated = errors.New("peer did not authenticate") + +// authGate is the host the DHT and gossipsub are built on. A library +// registers its stream handlers through it, and a handler so registered +// serves a peer only once that peer completed the mesh handshake on this +// router; a stream the library opens waits for the same. The handshake +// itself, identify and the relay keep the plain host: the first two are +// what the handshake needs, the relay checks admission in its own ACL. +// +// A stream that arrives before its peer's handshake is held, not reset: a +// member's DHT and gossip streams open on connect, a moment before its +// handshake completes, and a reset would make each library drop the peer +// and retry on its own schedule. An unauthenticated peer's stream is held +// unserved until its connection closes at the auth deadline, and counted. +type authGate struct { + host.Host + r *Router +} + +func (g *authGate) SetStreamHandler(pid protocol.ID, h network.StreamHandler) { + g.Host.SetStreamHandler(pid, g.r.requireAuthenticated(h)) +} + +func (g *authGate) SetStreamHandlerMatch(pid protocol.ID, match func(protocol.ID) bool, h network.StreamHandler) { + g.Host.SetStreamHandlerMatch(pid, match, g.r.requireAuthenticated(h)) +} + +func (g *authGate) NewStream(ctx context.Context, p peer.ID, pids ...protocol.ID) (network.Stream, error) { + if err := g.r.awaitAuthenticated(ctx, p); err != nil { + return nil, err + } + return g.Host.NewStream(ctx, p, pids...) +} + +// requireAuthenticated wraps a stream handler so it runs only once the +// stream's peer has authenticated. +func (r *Router) requireAuthenticated(next network.StreamHandler) network.StreamHandler { + return func(s network.Stream) { + p := s.Conn().RemotePeer() + if err := r.awaitAuthenticated(r.lifetime(), p); err != nil { + streamsRefusedTotal.WithLabelValues(streamScopeUnauthenticated, string(s.Protocol())).Inc() + logger.Debugf("[AuthZ] Refusing %s stream from %s: %v", s.Protocol(), p, err) + _ = s.Reset() + return + } + next(s) + } +} + +// awaitAuthenticated returns once p has authenticated, or with an error +// when ctx ends, p's connections are gone, or the auth deadline passes. +func (r *Router) awaitAuthenticated(ctx context.Context, p peer.ID) error { + if r.isPeerAuthenticated(p) { + return nil + } + wait := r.config.AuthDeadline + if wait <= 0 { + wait = defaultAuthWait + } + parent := ctx + ctx, cancel := context.WithTimeout(ctx, wait) + defer cancel() + tick := time.NewTicker(authPollInterval) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + if parent.Err() != nil { + return parent.Err() + } + return errNotAuthenticated + case <-tick.C: + } + if r.isPeerAuthenticated(p) { + return nil + } + if r.Host.Network().Connectedness(p) != network.Connected { + return errNotAuthenticated + } + } +} + +// lifetime is the router's own context, for work not tied to a request. +func (r *Router) lifetime() context.Context { + if r.ctx != nil { + return r.ctx + } + return context.Background() +} diff --git a/internal/router/auth_gate_test.go b/internal/router/auth_gate_test.go new file mode 100644 index 00000000..990d8045 --- /dev/null +++ b/internal/router/auth_gate_test.go @@ -0,0 +1,247 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package router + +import ( + "context" + "errors" + "io" + "testing" + "time" + + "github.com/libp2p/go-libp2p" + dht "github.com/libp2p/go-libp2p-kad-dht" + "github.com/libp2p/go-libp2p/core/event" + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/network" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/libp2p/go-libp2p/core/protocol" + dto "github.com/prometheus/client_model/go" + + "github.com/google/agentmesh/api" +) + +const echoProtocol protocol.ID = "/test/echo/1.0.0" + +// newGatedRouter is a router whose DHT and gossip host is the gate, with a +// short auth deadline so an unauthenticated peer is cut off quickly. +func newGatedRouter(t *testing.T) (*Router, *authGate) { + t.Helper() + h, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = h.Close() }) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + r := &Router{Host: h, ctx: ctx, directory: api.DefaultDirectorySettings(), config: Options{AuthDeadline: 2 * time.Second}} + return r, &authGate{Host: h, r: r} +} + +func newClient(t *testing.T, r *Router) host.Host { + t.Helper() + c, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close() }) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := c.Connect(ctx, peer.AddrInfo{ID: r.Host.ID(), Addrs: r.Host.Addrs()}); err != nil { + t.Fatal(err) + } + return c +} + +// waitForInboundStream returns once the router holds a negotiated stream +// on proto from p, so the gate's handler has been dispatched for it. +func waitForInboundStream(t *testing.T, r *Router, p peer.ID, proto protocol.ID) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for { + for _, conn := range r.Host.Network().ConnsToPeer(p) { + for _, st := range conn.GetStreams() { + if st.Protocol() == proto { + return + } + } + } + if time.Now().After(deadline) { + t.Fatalf("the router never saw a %s stream from %s", proto, p) + } + time.Sleep(10 * time.Millisecond) + } +} + +func unauthenticatedRefusals(t *testing.T, proto protocol.ID) float64 { + t.Helper() + var m dto.Metric + if err := streamsRefusedTotal.WithLabelValues(streamScopeUnauthenticated, string(proto)).Write(&m); err != nil { + t.Fatal(err) + } + return m.GetCounter().GetValue() +} + +// A handler registered through the gate serves a peer only once it +// authenticated: a stream that arrives first waits for the handshake and is +// then served; a peer that never authenticates gets nothing, and its stream +// is counted as refused when its connection closes at the auth deadline. +func TestGateServesOnlyAuthenticatedPeers(t *testing.T) { + r, gated := newGatedRouter(t) + gated.SetStreamHandler(echoProtocol, func(s network.Stream) { + defer func() { _ = s.Close() }() + _, _ = io.Copy(s, s) + }) + + t.Run("a stream opened before the handshake is served after it", func(t *testing.T) { + c := newClient(t, r) + s, err := c.NewStream(context.Background(), r.Host.ID(), echoProtocol) + if err != nil { + t.Fatal(err) + } + defer func() { _ = s.Close() }() + if _, err := s.Write([]byte("hello")); err != nil { + t.Fatal(err) + } + buf := make([]byte, 5) + _ = s.SetReadDeadline(time.Now().Add(300 * time.Millisecond)) + if n, err := s.Read(buf); err == nil { + t.Fatalf("read %q before the handshake; the stream must wait for it", buf[:n]) + } + + r.authenticatedPeers.Store(c.ID(), time.Now().Add(time.Hour)) + _ = s.SetReadDeadline(time.Now().Add(5 * time.Second)) + if _, err := io.ReadFull(s, buf); err != nil { + t.Fatalf("read after the handshake: %v", err) + } + if string(buf) != "hello" { + t.Fatalf("got %q, want %q", buf, "hello") + } + }) + + t.Run("a peer that never authenticates is refused and counted", func(t *testing.T) { + before := unauthenticatedRefusals(t, echoProtocol) + c := newClient(t, r) + s, err := c.NewStream(context.Background(), r.Host.ID(), echoProtocol) + if err != nil { + t.Fatal(err) + } + if _, err := s.Write([]byte("hello")); err != nil { + t.Fatal(err) + } + waitForInboundStream(t, r, c.ID(), echoProtocol) + // The connection goes, as it would at the auth deadline. + if err := c.Network().ClosePeer(r.Host.ID()); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(5 * time.Second) + for unauthenticatedRefusals(t, echoProtocol) != before+1 { + if time.Now().After(deadline) { + t.Fatalf("refusals = %v, want %v: a stream from a peer that never authenticated must be counted", unauthenticatedRefusals(t, echoProtocol), before+1) + } + time.Sleep(20 * time.Millisecond) + } + }) + + t.Run("a stream the router opens waits for the handshake", func(t *testing.T) { + c := newClient(t, r) + c.SetStreamHandler(echoProtocol, func(s network.Stream) { _ = s.Close() }) + ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) + defer cancel() + if _, err := gated.NewStream(ctx, c.ID(), echoProtocol); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("NewStream to an unauthenticated peer: err = %v, want the context's deadline", err) + } + r.authenticatedPeers.Store(c.ID(), time.Now().Add(time.Hour)) + s, err := gated.NewStream(context.Background(), c.ID(), echoProtocol) + if err != nil { + t.Fatalf("NewStream to an authenticated peer: %v", err) + } + _ = s.Close() + }) +} + +// The router's DHT, built on the gate, answers a member's query only once +// the member authenticated, and its table admits authenticated peer routers +// and nothing else. +func TestRouterDHTServesAuthenticatedMembersAndTablesPeerRouters(t *testing.T) { + r, gated := newGatedRouter(t) + kdht, err := dht.New(gated, r.dhtOptions()...) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = kdht.Close() }) + r.DHT = kdht + + // A member: a DHT client that asks the router for the peers closest to a key. + member := newClient(t, r) + memberDHT, err := dht.New(member, dht.Mode(dht.ModeClient), dht.ProtocolPrefix(api.DHTProtocolPrefix), dht.BucketSize(3), dht.DisableAutoRefresh()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = memberDHT.Close() }) + if _, err := memberDHT.RoutingTable().TryAddPeer(r.Host.ID(), true, false); err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + if _, err := memberDHT.GetClosestPeers(ctx, "a key"); err == nil { + t.Fatal("an unauthenticated member's query was answered") + } + cancel() + + r.authenticatedPeers.Store(member.ID(), time.Now().Add(time.Hour)) + ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if _, err := memberDHT.GetClosestPeers(ctx, "a key"); err != nil { + t.Fatalf("an authenticated member's query failed: %v", err) + } + if kdht.RoutingTable().Find(member.ID()) != "" { + t.Fatal("a member entered the router's routing table") + } + + // A peer router in server mode: in the table once it authenticates, not + // before. The DHT hears of a peer when identify completes. + identified, err := r.Host.EventBus().Subscribe(new(event.EvtPeerIdentificationCompleted)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = identified.Close() }) + other := newClient(t, r) + otherDHT, err := dht.New(other, dht.Mode(dht.ModeServer), dht.ProtocolPrefix(api.DHTProtocolPrefix)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = otherDHT.Close() }) + for seen := false; !seen; { + select { + case e := <-identified.Out(): + seen = e.(event.EvtPeerIdentificationCompleted).Peer == other.ID() + case <-time.After(5 * time.Second): + t.Fatal("the router never identified the peer") + } + } + if kdht.RoutingTable().Find(other.ID()) != "" { + t.Fatal("a server-mode peer that never authenticated entered the routing table") + } + r.authenticatedPeers.Store(other.ID(), time.Now().Add(time.Hour)) + r.admitPeerRouter(other.ID()) + if kdht.RoutingTable().Find(other.ID()) == "" { + t.Fatal("an authenticated peer router is not in the routing table") + } + r.forgetPeer(other.ID()) + if kdht.RoutingTable().Find(other.ID()) != "" { + t.Fatal("a peer router whose authentication lapsed is still in the routing table") + } +} diff --git a/internal/router/dht.go b/internal/router/dht.go index 083ef6c8..44a5a4a5 100644 --- a/internal/router/dht.go +++ b/internal/router/dht.go @@ -20,6 +20,7 @@ import ( dht "github.com/libp2p/go-libp2p-kad-dht" "github.com/libp2p/go-libp2p-kad-dht/records" + "github.com/libp2p/go-libp2p/core/peer" "github.com/google/agentmesh/api" ) @@ -40,12 +41,46 @@ const dhtTableBucketSize = 64 // node that is gone drops out of discovery within it and two missed // announcements do not unlist one that is not. The library default (48h) // would keep every past identity listed. -func dhtOptions(directory api.DirectorySettings) []dht.Option { +// +// Its table holds, and its lookups ask, authenticated peer routers only: a +// peer in server mode that never authenticated would otherwise be handed +// queries and records. A router enters the table when it authenticates +// (admitPeerRouter), not when the DHT first hears of it. +func (r *Router) dhtOptions() []dht.Option { return []dht.Option{ dht.Mode(dht.ModeServer), dht.ProtocolPrefix(api.DHTProtocolPrefix), dht.BucketSize(dhtTableBucketSize), - dht.ProviderManagerOpts(records.ProviderAddrTTL(directory.RecordTTL), records.ProvideValidity(directory.RecordTTL)), + dht.ProviderManagerOpts(records.ProviderAddrTTL(r.directory.RecordTTL), records.ProvideValidity(r.directory.RecordTTL)), + dht.RoutingTableFilter(func(_ interface{}, p peer.ID) bool { return r.isAuthenticatedPeerRouter(p) }), + dht.QueryFilter(func(_ interface{}, ai peer.AddrInfo) bool { return r.isAuthenticatedPeerRouter(ai.ID) }), + } +} + +// isAuthenticatedPeerRouter reports whether p is a peer router that holds a +// current authentication with this router. +func (r *Router) isAuthenticatedPeerRouter(p peer.ID) bool { + if _, isRouter := r.peerRouters.Load(p); !isRouter { + return false + } + return r.isPeerAuthenticated(p) +} + +// admitPeerRouter records p as a peer router that authenticated and puts it +// in the DHT's table: the table's filter kept it out until now, and the +// DHT itself only adds a peer when it first hears of it. +func (r *Router) admitPeerRouter(p peer.ID) { + if _, known := r.peerRouters.LoadOrStore(p, true); !known { + logger.Infof("[Federation] Peer router %s authenticated with us", p) + r.Host.ConnManager().Protect(p, federationTag) + } + if r.DHT == nil { + return + } + if added, err := r.DHT.RoutingTable().TryAddPeer(p, true, false); err != nil { + logger.Warnf("[DHT] Peer router %s not added to the routing table: %v", p, err) + } else if added { + logger.Infof("[DHT] Peer router %s added to the routing table (%d routers)", p, r.DHT.RoutingTable().Size()) } } diff --git a/internal/router/dht_test.go b/internal/router/dht_test.go index 4f51cbd1..9025539c 100644 --- a/internal/router/dht_test.go +++ b/internal/router/dht_test.go @@ -32,7 +32,8 @@ func TestRouterDHTTableHoldsEveryRouter(t *testing.T) { t.Fatalf("libp2p.New: %v", err) } t.Cleanup(func() { _ = h.Close() }) - kdht, err := dht.New(h, dhtOptions(api.DefaultDirectorySettings())...) + r := &Router{Host: h, directory: api.DefaultDirectorySettings()} + kdht, err := dht.New(h, r.dhtOptions()...) if err != nil { t.Fatalf("dht.New: %v", err) } @@ -43,13 +44,19 @@ func TestRouterDHTTableHoldsEveryRouter(t *testing.T) { rt := kdht.RoutingTable() const routers = 40 - for i := 0; i < routers; i++ { + for added := 0; added < routers; { p, err := rt.GenRandPeerID(0) if err != nil { t.Fatalf("GenRandPeerID: %v", err) } + // GenRandPeerID draws from a 16-bit prefix table, so it repeats a + // peer now and then; a repeat is not a refusal. + if rt.Find(p) != "" { + continue + } if ok, err := rt.TryAddPeer(p, true, false); !ok { - t.Fatalf("router #%d refused from the table: %v; a bucket must hold at least %d", i+1, err, routers) + t.Fatalf("router #%d refused from the table: %v; a bucket must hold at least %d", added+1, err, routers) } + added++ } } diff --git a/internal/router/router.go b/internal/router/router.go index 6b778146..42face05 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -252,6 +252,9 @@ func (r *Router) forgetPeer(p peer.ID) { r.Host.ConnManager().UntagPeer(p, memberTag) r.Host.Peerstore().UpdateAddrs(p, r.reservationTTL(), 0) } + if r.DHT != nil { + r.DHT.RoutingTable().RemovePeer(p) + } } // AdmittedUntil reports when this router's admission of p lapses: the @@ -502,9 +505,11 @@ func (r *Router) Start() (retErr error) { return err } r.Host = hostNode + // The DHT and gossip serve, and are served by, authenticated peers only. + gated := &authGate{Host: hostNode, r: r} // Setup DHT - kadDHT, err := dht.New(hostNode, dhtOptions(r.directory)...) + kadDHT, err := dht.New(gated, r.dhtOptions()...) if err != nil { return err } @@ -526,7 +531,7 @@ func (r *Router) Start() (retErr error) { // Setup PubSub. StrictSign is the default; pinned because the event // validator trusts msg.GetFrom(). - ps, err := pubsub.NewGossipSub(r.ctx, hostNode, pubsub.WithMessageSignaturePolicy(pubsub.StrictSign)) + ps, err := pubsub.NewGossipSub(r.ctx, gated, pubsub.WithMessageSignaturePolicy(pubsub.StrictSign)) if err != nil { return err } @@ -1509,10 +1514,7 @@ func (r *Router) HandleAuthHandshake(s network.Stream) { // A router that proves the router role is a peer router from here on, // not from the next federation pass: it is never a member to shed. if identity.RequireRole(b, verifyingKey, api.RoleRouter, r.config.BiscuitTimeout) == nil { - if _, known := r.peerRouters.LoadOrStore(remotePeer, true); !known { - logger.Infof("[Federation] Peer router %s authenticated with us", remotePeer) - r.Host.ConnManager().Protect(remotePeer, federationTag) - } + r.admitPeerRouter(remotePeer) } // Send mutual response (our biscuit) @@ -1589,6 +1591,7 @@ func (r *Router) performMutualAuth(s network.Stream) error { } r.admitPeer(remotePeer, expiry) + r.admitPeerRouter(remotePeer) return nil } diff --git a/sdk/js/src/session.test.ts b/sdk/js/src/session.test.ts index ac3b2aaa..227d9314 100644 --- a/sdk/js/src/session.test.ts +++ b/sdk/js/src/session.test.ts @@ -405,6 +405,48 @@ test("a member on one router reaches a member on another", async () => { } }); +test("a connection to a listed router is authenticated, and released after a lookup", async () => { + // The DHT client connects to whichever listed router kademlia puts closest + // to a key. A router serves its DHT only to a member that authenticated + // on the connection, so the handshake runs on connection:open for any + // listed router, held or not; a router met this way is not held, and the + // connection is closed once the lookup ends. + const routerB = await createLibp2p({ + addresses: { listen: ["/ip4/127.0.0.1/tcp/0"] }, + transports: [tcp()], + connectionEncrypters: [tls()], + streamMuxers: [yamux()], + services: { identify: identify(), relay: circuitRelayServer() }, + }); + try { + const handshakesB: string[] = []; + await routerB.handle(AUTH_PROTOCOL, authStreamHandler({ ownBiscuit: () => mint(routerB.peerId.toString(), ROLE_ROUTER), trustedKeys: () => [cpKey], onAuthenticated: (p) => handshakesB.push(p) })); + const routerBAddr = (routerB.getMultiaddrs()[0] as ReturnType).toString(); + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: fakeControlPlane([routerAddr, routerBAddr]) }); + const session = await mesh.join({ refreshLeadMs: 0, routerAddresses: [routerAddr], reserveRelay: false }); + try { + assert.deepEqual(session.routers.map((r) => r.peerId), [router.peerId.toString()]); + + await session.node.dial(multiaddr(routerBAddr)); + const deadline = Date.now() + 10_000; + while (handshakesB.length === 0 && Date.now() < deadline) { + await new Promise((r) => setTimeout(r, 20)); + } + assert.deepEqual(handshakesB, [mesh.peerId], "the member authenticated on connecting to a listed router"); + assert.deepEqual(session.routers.map((r) => r.peerId), [router.peerId.toString()], "a router met for a lookup is not held"); + + await session.discover("mcp", undefined, { timeoutMs: 500 }).catch(() => {}); + const connected = session.node.getConnections().map((c) => c.remotePeer.toString()); + assert.ok(!connected.includes(routerB.peerId.toString()), "the lookup's connection to the router it does not hold is released"); + assert.ok(connected.includes(router.peerId.toString()), "the held router's connection stays"); + } finally { + await session.close(); + } + } finally { + await routerB.stop(); + } +}); + test("a resumed member adopts a rotated key before admitting routers", async () => { // The control plane rotated its signing key while this member was not // running: the routers already hold credentials under the new key, and diff --git a/sdk/js/src/session.ts b/sdk/js/src/session.ts index 91bb2b4b..63fc5852 100644 --- a/sdk/js/src/session.ts +++ b/sdk/js/src/session.ts @@ -499,18 +499,64 @@ export class MeshSession { * A router forgets the admission with the connection it came on, and * js-libp2p reconnects to a relay on its own, so every new connection to a * router runs the handshake again: the relay's own renewal of the - * reservation then still passes the router's check. + * reservation then still passes the router's check. A router serves its + * DHT and gossip only to a member that authenticated on the connection, + * so a connection the DHT client opens to any router the control plane + * lists is authenticated the same way; the router holds the streams the + * client opened meanwhile and serves them once the handshake is done. */ #keepRouterAdmissions(): void { this.node.addEventListener("connection:open", (evt) => { const conn = evt.detail; - if (this.#closed || !this.routers.some((r) => r.peerId === conn.remotePeer.toString())) { + if (this.#closed || !this.#isRouter(conn.remotePeer.toString())) { return; } - void authenticateWithPeer(conn, this.mesh.authFrame(), this.mesh.credential.controlPlaneKeys).catch(() => {}); + void this.#authenticateRouter(conn).catch(() => {}); }); } + /** Handshakes in flight by connection id: one per connection, whoever asks first. */ + readonly #routerHandshakes = new Map>(); + + /** + * Runs the handshake on a router connection once: the connection:open + * listener and an explicit admission both call this, and the second waits + * for the first instead of running a second handshake. + */ + #authenticateRouter(conn: Connection): Promise { + const pending = this.#routerHandshakes.get(conn.id); + if (pending !== undefined) { + return pending; + } + const run = authenticateWithPeer(conn, this.mesh.authFrame(), this.mesh.credential.controlPlaneKeys).finally(() => { + this.#routerHandshakes.delete(conn.id); + }); + this.#routerHandshakes.set(conn.id, run); + return run; + } + + /** Whether peerId is a router this member holds or one the mesh lists for it. */ + #isRouter(peerId: string): boolean { + return ( + this.routers.some((r) => r.peerId === peerId) || + candidatesFromRouterInfos(this.mesh.routers, this.mesh.credential.routerAddresses).some((c) => c.peerId === peerId) + ); + } + + /** + * Closes the connections a lookup opened to routers this member does not + * hold: the member keeps sessions with the routers it holds and with no + * other, and the next lookup connects and authenticates again. + */ + async #releaseLookupRouters(): Promise { + for (const conn of this.node.getConnections()) { + const peerId = conn.remotePeer.toString(); + if (this.#isRouter(peerId) && !this.routers.some((r) => r.peerId === peerId)) { + await this.node.hangUp(conn.remotePeer).catch(() => {}); + } + } + } + /** * Maintains held routers and relay reservations: when a held router's * connection or reservation drops, redials it first (2s, 4s, 8s by default) @@ -574,7 +620,7 @@ export class MeshSession { continue; } try { - const credential = await authenticateWithPeer(conn, this.mesh.authFrame(), this.mesh.credential.controlPlaneKeys); + const credential = await this.#authenticateRouter(conn); requireRole(credential, ROLE_ROUTER); const addr = connectedAddress(conn, r.peerId); if (this.#reserveRelay) { @@ -625,7 +671,7 @@ export class MeshSession { tried.add(cand.peerId); try { const conn = await this.node.dial(cand.addresses, signal !== undefined ? { signal } : {}); - const credential = await authenticateWithPeer(conn, this.mesh.authFrame(), this.mesh.credential.controlPlaneKeys); + const credential = await this.#authenticateRouter(conn); requireRole(credential, ROLE_ROUTER); const addr = connectedAddress(conn, cand.peerId); if (this.#reserveRelay) { @@ -724,6 +770,8 @@ export class MeshSession { found = await this.node.peerRouting.findPeer(peerIdFromString(peerId), { signal: lookup }); } catch { return []; + } finally { + await this.#releaseLookupRouters(); } const out: Multiaddr[] = []; for (const ma of found.multiaddrs) { @@ -753,7 +801,7 @@ export class MeshSession { /** Dials a router, runs the handshake and, its role verified, adds it to the admitted set. */ async #admitRouter(addr: Multiaddr, signal?: AbortSignal): Promise { const conn = await this.node.dial(addr, signal !== undefined ? { signal } : {}); - const credential = await authenticateWithPeer(conn, this.mesh.authFrame(), this.mesh.credential.controlPlaneKeys); + const credential = await this.#authenticateRouter(conn); requireRole(credential, ROLE_ROUTER); const peerId = conn.remotePeer.toString(); const router = { peerId, addr: connectedAddress(conn, peerId), credential }; @@ -837,6 +885,7 @@ export class MeshSession { } } finally { clearTimeout(settleTimer); + await this.#releaseLookupRouters(); } return [...found.values()]; } @@ -909,7 +958,7 @@ export class MeshSession { return; } try { - await authenticateWithPeer(conn, this.mesh.authFrame(), this.mesh.credential.controlPlaneKeys); + await this.#authenticateRouter(conn); } catch { // The router keeps the admission it has until the old credential lapses. } diff --git a/sdk/python/src/agent_mesh/discovery.py b/sdk/python/src/agent_mesh/discovery.py index 3d465189..272fa4ec 100644 --- a/sdk/python/src/agent_mesh/discovery.py +++ b/sdk/python/src/agent_mesh/discovery.py @@ -24,7 +24,7 @@ import hashlib import logging from dataclasses import dataclass, field -from typing import Callable, Iterable, Literal +from typing import Awaitable, Callable, Iterable, Literal import multiaddr import trio @@ -116,7 +116,18 @@ async def _query(host: IHost, peer_id: ID, req: kad.Message) -> kad.Message | No return resp if resp.type == req.type else None -async def find_peer(host: IHost, target: ID, seeds: Iterable[ID], replication: int = DEFAULT_DHT_REPLICATION) -> list[multiaddr.Multiaddr]: +Admit = Callable[[ID], Awaitable[None]] +"""Authenticates this member with a router before it is queried: a router +serves its DHT to a peer only once that peer passed the handshake.""" + + +async def find_peer( + host: IHost, + target: ID, + seeds: Iterable[ID], + replication: int = DEFAULT_DHT_REPLICATION, + admit: Admit | None = None, +) -> list[multiaddr.Multiaddr]: """The addresses the DHT knows for a peer, as go-libp2p-kad-dht's FindPeer learns them: a router answers FIND_NODE for a peer it is connected to with that peer and its addresses, its relayed one included, and names the @@ -130,11 +141,18 @@ def take(resp: kad.Message) -> bool: return True return False - await _walk(host, kad.Message(type=kad.Message.FIND_NODE, key=target.to_bytes()), seeds, take, skip={target}, replication=replication) + await _walk(host, kad.Message(type=kad.Message.FIND_NODE, key=target.to_bytes()), seeds, take, skip={target}, replication=replication, admit=admit) return found -async def find_providers(host: IHost, key: bytes, seeds: Iterable[ID], limit: int = 20, replication: int = DEFAULT_DHT_REPLICATION) -> list[DiscoveredProvider]: +async def find_providers( + host: IHost, + key: bytes, + seeds: Iterable[ID], + limit: int = 20, + replication: int = DEFAULT_DHT_REPLICATION, + admit: Admit | None = None, +) -> list[DiscoveredProvider]: """Asks the seed peers (the routers we are connected to) for providers of key and follows the closer peers they name for a few rounds.""" found: dict[str, DiscoveredProvider] = {} @@ -150,7 +168,7 @@ def take(resp: kad.Message) -> bool: entry.addrs.append(str(ma)) return bool(found) - await _walk(host, kad.Message(type=kad.Message.GET_PROVIDERS, key=key), seeds, take, replication=replication) + await _walk(host, kad.Message(type=kad.Message.GET_PROVIDERS, key=key), seeds, take, replication=replication, admit=admit) return list(found.values())[:limit] @@ -161,6 +179,7 @@ async def _walk( take: Callable[[kad.Message], bool], skip: frozenset[ID] | set[ID] = frozenset(), replication: int = DEFAULT_DHT_REPLICATION, + admit: Admit | None = None, ) -> None: """The Kademlia walk both lookups share. A round asks its peers at once and ends when an answer satisfies `take`; only otherwise are the closer @@ -170,8 +189,34 @@ async def _walk( may list a router this member cannot reach. A peer that did not answer its dial is not dialed again. A router names the routers closest to the key in order, and a record lives on the `replication` closest, so only - that many of them are followed from one round to the next.""" + that many of them are followed from one round to the next. + + A router serves a query only from a member that authenticated with it, + so `admit` runs before each query. A router the walk connected to for + the query is disconnected when the walk ends: the member keeps sessions + with the routers it holds and with no other.""" self_id = host.get_id() + held = set(host.get_connected_peers()) + try: + await _walk_rounds(host, req, seeds, take, skip, replication, admit, self_id) + finally: + for peer_id in set(host.get_connected_peers()) - held - {self_id}: + try: + await host.disconnect(peer_id) + except Exception as err: # noqa: BLE001 - already gone + logger.debug("dht: disconnect from %s after the walk: %s", peer_id, err) + + +async def _walk_rounds( + host: IHost, + req: kad.Message, + seeds: Iterable[ID], + take: Callable[[kad.Message], bool], + skip: frozenset[ID] | set[ID], + replication: int, + admit: Admit | None, + self_id: ID, +) -> None: asked: set[ID] = set() frontier: list[ID] = list(seeds) fanout = min(max(replication, 1), _MAX_PEERS_PER_ROUND) @@ -183,6 +228,12 @@ async def _walk( answers: list[kad.Message] = [] async def ask(peer_id: ID) -> None: + if admit is not None: + try: + await admit(peer_id) + except Exception as err: # noqa: BLE001 - a router that refuses us is skipped + logger.debug("dht: %s did not admit this member: %s", peer_id, err) + return resp = await _query(host, peer_id, req) if resp is not None: answers.append(resp) diff --git a/sdk/python/src/agent_mesh/session.py b/sdk/python/src/agent_mesh/session.py index 08f3734d..6dc35aeb 100644 --- a/sdk/python/src/agent_mesh/session.py +++ b/sdk/python/src/agent_mesh/session.py @@ -585,7 +585,7 @@ async def _routed_addresses(self, target: ID) -> tuple[list[multiaddr.Multiaddr] relays: list[multiaddr.Multiaddr] = [] seeds = [ID.from_base58(r.peer_id) for r in self.routers] suffix = f"/p2p/{target}" - for ma in await find_peer(self.host, target, seeds, self.mesh.dht_replication): + for ma in await find_peer(self.host, target, seeds, self.mesh.dht_replication, admit=self._admit_for_lookup): text = str(ma) if "/p2p-circuit" not in text: try: @@ -603,6 +603,15 @@ async def _routed_addresses(self, target: ID) -> tuple[list[multiaddr.Multiaddr] relays.append(relay_addr) return direct, relays + async def _admit_for_lookup(self, peer_id: ID) -> None: + """Runs the handshake with a router a lookup is about to query. A + router this member holds is authenticated already; any other serves + the query only once this member authenticated on the connection the + walk just opened, and the walk closes that connection when it ends.""" + if any(r.peer_id == str(peer_id) for r in self.routers): + return + await _authenticate_router(self.host, self.mesh, peer_id) + async def _admit_router(self, addr: multiaddr.Multiaddr) -> AdmittedRouter: """Dials a router, runs the handshake and, its role verified, adds it to the admitted set.""" @@ -649,7 +658,7 @@ async def discover(self, service: str, name: str | None = None, limit: int = 20) if service_type not in ("mcp", "inference", "a2a", "egress"): raise ValueError(f"service type must be mcp, inference, a2a or egress, got {service_type!r}") seeds = [ID.from_base58(r.peer_id) for r in self.routers] - return await find_providers(self.host, service_key(service_type, service_name), seeds, limit, self.mesh.dht_replication) + return await find_providers(self.host, service_key(service_type, service_name), seeds, limit, self.mesh.dht_replication, admit=self._admit_for_lookup) def open_mcp( self, diff --git a/sdk/python/tests/test_discovery.py b/sdk/python/tests/test_discovery.py index dd849719..9b7a27af 100644 --- a/sdk/python/tests/test_discovery.py +++ b/sdk/python/tests/test_discovery.py @@ -280,3 +280,51 @@ async def main(replication, want): trio.run(main, 3, [GONE_POD, GONE_POD_2, DARK_ROUTER], clock=trio.testing.MockClock(autojump_threshold=0)) trio.run(main, 1, [GONE_POD], clock=trio.testing.MockClock(autojump_threshold=0)) + + +def test_the_walk_authenticates_before_each_query_and_releases_what_it_connected(): + """A router serves a query only from a member that authenticated with it, + so `admit` runs for a router before it is queried: the held router and + the closer one the walk connects to for the record. That connection is + the walk's alone and is closed when the walk ends; the held router's + stays.""" + + class ReachableHost(_Host): + async def connect(self, info): + self.dials.append(info.peer_id) + self.connected.add(info.peer_id) + self._answers[info.peer_id] = lambda req: kad.Message(type=req.type, providerPeers=[_peer(PROVIDER, "/ip4/10.0.0.9/tcp/4001")]) + + def names_router_b(req): + return kad.Message(type=req.type, closerPeers=[_peer(ROUTER_B, "/ip4/10.0.0.2/tcp/4501")]) + + host = ReachableHost({ROUTER: names_router_b}) + admitted = [] + + async def admit(peer_id): + assert peer_id in host.connected, "admit runs on an open connection" + admitted.append(peer_id) + + async def main(): + found = await find_providers(host, b"key", [ROUTER], admit=admit) + assert [p.peer_id for p in found] == [str(PROVIDER)] + assert admitted == [ROUTER, ROUTER_B], "every router is admitted before it is queried, in walk order" + assert host.connected == {ROUTER}, "the router the walk connected to is released; the held one is kept" + + trio.run(main, clock=trio.testing.MockClock(autojump_threshold=0)) + + +def test_a_router_that_refuses_the_handshake_is_skipped(): + """A router that does not admit this member answers nothing; the walk + takes the other answers and moves on.""" + host = _Host({ROUTER: _after_a_rollout, ROUTER_B: _after_a_rollout}) + + async def admit(peer_id): + if peer_id == ROUTER_B: + raise RuntimeError("not admitted") + + async def main(): + found = await find_providers(host, b"key", [ROUTER, ROUTER_B], admit=admit) + assert [p.peer_id for p in found] == [str(PROVIDER)] + + trio.run(main, clock=trio.testing.MockClock(autojump_threshold=0)) diff --git a/site/content/docs/contributing/security-architecture.md b/site/content/docs/contributing/security-architecture.md index 29f818d7..244e50d1 100644 --- a/site/content/docs/contributing/security-architecture.md +++ b/site/content/docs/contributing/security-architecture.md @@ -594,3 +594,14 @@ response directly). members on charset grounds and quotes wildcard string literals with `strconv.Quote` before rendering `.starts_with(...)` or `.ends_with(...)` Datalog rules. +7. Nothing interacts with the mesh unauthenticated. A router or a member + serves a mesh protocol (the DHT, gossip, the relay, the data plane) only to + a peer that completed the credential handshake on that connection and whose + credential has not lapsed. The protocols open to a peer before that are the + ones the handshake needs: the transport security handshake, identify and + `/mesh/auth`. The DHT and gossipsub are built on a host wrapper that holds a + stream until its peer is authenticated and resets it when the connection + closes first, counted in `*_streams_refused_total{scope="unauthenticated"}`; + a router's routing table admits authenticated peer routers only. A + directory lookup and a gossip subscription are mesh data, so this covers + reads as much as writes. diff --git a/site/content/docs/reference/router.md b/site/content/docs/reference/router.md index aaa5342b..fadaed75 100644 --- a/site/content/docs/reference/router.md +++ b/site/content/docs/reference/router.md @@ -64,6 +64,13 @@ replica count. router on `/info` while the lease is live. 4. Runs the mutual credential handshake on every inbound connection and refuses peers whose credential does not verify and peers on the ban list. + Until a peer passes it, the router serves it nothing but the handshake: + a DHT or gossip stream it opens waits, unread, and is served once the + peer is authenticated, or reset when the connection closes at + `--auth-deadline` first. A member's streams open on connect, a moment + before its handshake completes, which is why they wait instead of being + refused outright. The routing table admits authenticated peer routers and + nothing else, and the router queries no one else either. 5. Refreshes its own credential before it expires, like a node. 6. Every 5 seconds, compares the members it holds, the peers that passed the handshake, with `--high-watermark`. At the mark it picks members at @@ -120,7 +127,7 @@ The metrics address exposes, next to the Go runtime metrics: | `agentmesh_router_goaway_sent_total{reason,result}` | Members sent away, by reason (`DRAINING`, `OVERLOADED`) and outcome: `ok` when the member took the message, `unsupported` when it does not speak the protocol and was closed instead, `failed` when the stream broke. A rising `OVERLOADED` count on a router that is not draining is a fleet that has outgrown its routers. | | `agentmesh_router_unauthenticated_connections_closed_total` | Connections closed at `--auth-deadline` without a handshake. Members handshake with every router they connect to, so a steady rate is members running a node release that predates that, and a burst is a scanner. | | `agentmesh_router_inbound_connections_refused_total` | Inbound connections refused by the per-address cap. | -| `agentmesh_router_streams_refused_total{scope,protocol}` | Streams the resource manager refused, by scope and protocol or service. A count on `/mesh/kad/1.0.0` or a `/meshsub/*` protocol is members the router keeps out of discovery; see the scalability guide. | +| `agentmesh_router_streams_refused_total{scope,protocol}` | Streams refused, by scope and protocol or service. `scope="unauthenticated"` is a stream from a peer that never passed the handshake, reset when its connection closed; a steady rate is a scanner, or a member whose handshake fails. The other scopes are the resource manager's: a count on `/mesh/kad/1.0.0` or a `/meshsub/*` protocol there is members the router keeps out of discovery; see the scalability guide. | | `agentmesh_router_busiest_connection_streams{transport}` | The most streams any one connection carries. The connection to a service the router relays for holds one stream per relayed caller, and QUIC admits 256 per connection; see the scalability guide. | | `libp2p_relaysvc_*` | The relay service's own counters: `reservations_total{type}` and `connections_total{type}` (`opened`, `closed`, `renewed`), requests by response status, rejections and bytes relayed. The difference between `opened` and `closed` is the number in use; read it against `agentmesh_router_relay_limit` to see how close the relay is to its budget. |