From a5bb58bb6531647450fbb1789e5bc0f1c26aceb5 Mon Sep 17 00:00:00 2001 From: HosniBelfeki Date: Fri, 4 Sep 2026 23:41:48 +0100 Subject: [PATCH 1/3] api,controlplane: publish the ban set in /info MeshEvent_BANNED is broadcast once, at the moment of the ban, and gossip has no replay. A node or router that restarted, or that was down when the event went out, therefore never learns the ban -- and nothing re-tells it, even though the control plane holds the ban durably in its own database. /info is already fetched by sam-node at startup and by sam-router every 30s, so it is where that catch-up belongs. This adds the whole ban set to ControlPlaneInfoResponse rather than an incremental feed, because the full set is also the only signal an *unban* has: there is no unban event, so a consumer that merged deltas could never let a peer back in. A failure to read the ban set fails the request rather than answering without the field. An empty list is a meaningful value -- it means nothing is banned -- so serving one on error would tell every consumer to clear its blocklist. api/sam.pb.go is regenerated with the pinned toolchain the committed file records (protoc v3.21.12, protoc-gen-go v1.36.12), verified beforehand to reproduce it byte for byte so this diff is only the new field. No consumers yet; they follow. --- api/sam.pb.go | 23 ++++++++-- api/sam.proto | 7 ++++ internal/controlplane/server.go | 14 +++++++ internal/controlplane/server_test.go | 63 ++++++++++++++++++++++++++++ internal/storage/sql_store.go | 20 +++++++++ internal/storage/storage.go | 6 +++ 6 files changed, 129 insertions(+), 4 deletions(-) diff --git a/api/sam.pb.go b/api/sam.pb.go index 4c1280ef..65101f11 100644 --- a/api/sam.pb.go +++ b/api/sam.pb.go @@ -1127,8 +1127,15 @@ type ControlPlaneInfoResponse struct { ClientId string `protobuf:"bytes,2,opt,name=client_id,json=clientId,proto3" json:"client_id,omitempty"` Audience string `protobuf:"bytes,3,opt,name=audience,proto3" json:"audience,omitempty"` RouterAddresses []string `protobuf:"bytes,4,rep,name=router_addresses,json=routerAddresses,proto3" json:"router_addresses,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // The complete set of currently banned peer IDs. Consumers reconcile their + // local blocklist against this list rather than merging into it, so that a + // peer the control plane no longer bans is unbanned everywhere without + // needing an event of its own. MeshEvent_BANNED stays the fast path for + // sub-second eviction; this is how a node that restarted or was offline + // when the event was published catches up. + BannedPeerIds []string `protobuf:"bytes,5,rep,name=banned_peer_ids,json=bannedPeerIds,proto3" json:"banned_peer_ids,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ControlPlaneInfoResponse) Reset() { @@ -1189,6 +1196,13 @@ func (x *ControlPlaneInfoResponse) GetRouterAddresses() []string { return nil } +func (x *ControlPlaneInfoResponse) GetBannedPeerIds() []string { + if x != nil { + return x.BannedPeerIds + } + return nil +} + type RouterLeaseRequest struct { state protoimpl.MessageState `protogen:"open.v1"` PeerId string `protobuf:"bytes,1,opt,name=peer_id,json=peerId,proto3" json:"peer_id,omitempty"` @@ -2977,13 +2991,14 @@ const file_api_sam_proto_rawDesc = "" + "\ttimestamp\x18\b \x01(\x03R\ttimestamp\x1a9\n" + "\vLabelsEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\x9f\x01\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xc7\x01\n" + "\x18ControlPlaneInfoResponse\x12\x1f\n" + "\voidc_issuer\x18\x01 \x01(\tR\n" + "oidcIssuer\x12\x1b\n" + "\tclient_id\x18\x02 \x01(\tR\bclientId\x12\x1a\n" + "\baudience\x18\x03 \x01(\tR\baudience\x12)\n" + - "\x10router_addresses\x18\x04 \x03(\tR\x0frouterAddresses\"\xa9\x01\n" + + "\x10router_addresses\x18\x04 \x03(\tR\x0frouterAddresses\x12&\n" + + "\x0fbanned_peer_ids\x18\x05 \x03(\tR\rbannedPeerIds\"\xa9\x01\n" + "\x12RouterLeaseRequest\x12\x17\n" + "\apeer_id\x18\x01 \x01(\tR\x06peerId\x12\x1c\n" + "\taddresses\x18\x02 \x03(\tR\taddresses\x12\x18\n" + diff --git a/api/sam.proto b/api/sam.proto index 9ae07c1f..b56826ff 100644 --- a/api/sam.proto +++ b/api/sam.proto @@ -171,6 +171,13 @@ message ControlPlaneInfoResponse { string client_id = 2; string audience = 3; repeated string router_addresses = 4; + // The complete set of currently banned peer IDs. Consumers reconcile their + // local blocklist against this list rather than merging into it, so that a + // peer the control plane no longer bans is unbanned everywhere without + // needing an event of its own. MeshEvent_BANNED stays the fast path for + // sub-second eviction; this is how a node that restarted or was offline + // when the event was published catches up. + repeated string banned_peer_ids = 5; } message RouterLeaseRequest { diff --git a/internal/controlplane/server.go b/internal/controlplane/server.go index 8a155cc9..5f398ba9 100644 --- a/internal/controlplane/server.go +++ b/internal/controlplane/server.go @@ -408,11 +408,25 @@ func (s *Server) HandleInfo(w http.ResponseWriter, r *http.Request) { routerAddrs = append(routerAddrs, r.Addresses...) } + // The ban set is published here, rather than only as a MeshEvent, because + // the event is broadcast once and gossip has no replay: a node or router + // that restarted or was offline at the time would otherwise never learn + // the ban. A failure to read it must not be served as an empty list -- + // that would read as "nothing is banned" and unban everyone -- so it is + // treated like any other failure to build this response. + bannedPeerIDs, err := s.store.ListBannedPeerIDs(r.Context()) + if err != nil { + logger.Errorf("Failed to retrieve banned peers: %v", err) + http.Error(w, "Internal server error", http.StatusInternalServerError) + return + } + resp := &api.ControlPlaneInfoResponse{ OidcIssuer: issuer, ClientId: clientID, Audience: aud, RouterAddresses: routerAddrs, // Reused this field for back-compatibility with bootstrap routers list + BannedPeerIds: bannedPeerIDs, } respData, err := proto.Marshal(resp) diff --git a/internal/controlplane/server_test.go b/internal/controlplane/server_test.go index 48d58a15..a58fdd63 100644 --- a/internal/controlplane/server_test.go +++ b/internal/controlplane/server_test.go @@ -2611,3 +2611,66 @@ func TestAdminBootstrapTokensList(t *testing.T) { t.Errorf("unauthenticated GET status = %s, want 401", resp.Status) } } + +// /info carries the whole ban set so a node or router that restarted, or was +// offline when MeshEvent_BANNED was published, can reconcile against it. A +// peer that is not banned must not appear, or consumers would blocklist it. +func TestHandleInfoPublishesBanSet(t *testing.T) { + issuer, _ := startCustomMockOIDC(t) + srv, st, _ := setupTestServer(t, issuer) + defer func() { + _ = srv.Close() + _ = st.Close() + }() + ctx := context.Background() + + const ( + bannedPeer = "12D3KooWBanned000000000000000000000000000000000001" + allowedPeer = "12D3KooWAllowed00000000000000000000000000000000002" + ) + for _, peerID := range []string{bannedPeer, allowedPeer} { + if err := st.EnrollNode(ctx, &storage.EnrolledNode{ + PeerID: peerID, + PublicKey: []byte("test-public-key"), + Biscuit: []byte("test-biscuit"), + Role: "agent", + ExpiresAt: time.Now().Add(time.Hour), + }); err != nil { + t.Fatalf("EnrollNode(%s): %v", peerID, err) + } + } + // The path /admin/revoke takes: enrollment always starts unbanned. + if err := st.SetNodeBanned(ctx, bannedPeer, true); err != nil { + t.Fatalf("SetNodeBanned: %v", err) + } + + rec := httptest.NewRecorder() + srv.HandleInfo(rec, httptest.NewRequest(http.MethodGet, "/info", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("unexpected /info status: %d", rec.Code) + } + var info api.ControlPlaneInfoResponse + if err := proto.Unmarshal(rec.Body.Bytes(), &info); err != nil { + t.Fatalf("unmarshal ControlPlaneInfoResponse: %v", err) + } + + got := info.GetBannedPeerIds() + if len(got) != 1 || got[0] != bannedPeer { + t.Errorf("banned_peer_ids = %v, want exactly [%s]", got, bannedPeer) + } + + // An unban must show up as absence, since that is the only signal + // consumers get: there is no unban event. + if err := st.SetNodeBanned(ctx, bannedPeer, false); err != nil { + t.Fatalf("SetNodeBanned(false): %v", err) + } + rec = httptest.NewRecorder() + srv.HandleInfo(rec, httptest.NewRequest(http.MethodGet, "/info", nil)) + var after api.ControlPlaneInfoResponse + if err := proto.Unmarshal(rec.Body.Bytes(), &after); err != nil { + t.Fatalf("unmarshal ControlPlaneInfoResponse: %v", err) + } + if len(after.GetBannedPeerIds()) != 0 { + t.Errorf("after unban banned_peer_ids = %v, want empty", after.GetBannedPeerIds()) + } +} diff --git a/internal/storage/sql_store.go b/internal/storage/sql_store.go index 86abd092..79291766 100644 --- a/internal/storage/sql_store.go +++ b/internal/storage/sql_store.go @@ -740,6 +740,26 @@ func (s *SQLStore) IsIdentityBanned(ctx context.Context, identity string) (bool, return true, nil } +// ListBannedPeerIDs implements Store. +func (s *SQLStore) ListBannedPeerIDs(ctx context.Context) ([]string, error) { + query := s.rebind(`SELECT peer_id FROM nodes WHERE banned = ?`) + rows, err := s.db.QueryContext(ctx, query, true) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + var peerIDs []string + for rows.Next() { + var peerID string + if err := rows.Scan(&peerID); err != nil { + return nil, err + } + peerIDs = append(peerIDs, peerID) + } + return peerIDs, rows.Err() +} + // UpsertRouterLease implements Store. func (s *SQLStore) UpsertRouterLease(ctx context.Context, lease *RouterLease) error { addrsBytes, err := json.Marshal(lease.Addresses) diff --git a/internal/storage/storage.go b/internal/storage/storage.go index bae522db..3158f4f6 100644 --- a/internal/storage/storage.go +++ b/internal/storage/storage.go @@ -177,6 +177,12 @@ type Store interface { // IsIdentityBanned checks if an identity is currently banned. IsIdentityBanned(ctx context.Context, identity string) (bool, error) + // ListBannedPeerIDs returns the peer IDs of every currently banned node. + // Callers publish this as the complete ban set (see + // ControlPlaneInfoResponse.banned_peer_ids), so it must be the whole list + // rather than a page of it. + ListBannedPeerIDs(ctx context.Context) ([]string, error) + // UpsertRouterLease updates or creates a lease for a sam-router. UpsertRouterLease(ctx context.Context, lease *RouterLease) error From a4ce1c313e4381e79737b66bfe4828072da285b0 Mon Sep 17 00:00:00 2001 From: HosniBelfeki Date: Fri, 4 Sep 2026 23:42:12 +0100 Subject: [PATCH 2/3] node,router: reconcile peer bans against the control plane's ban set Nodes and routers only learned bans from MeshEvent_BANNED, which is published once with no replay, so a restart or any downtime left them enforcing nothing. Both now reconcile against the ban set in /info, which they already fetch: sam-node at startup via SyncMeshConfig, sam-router in runFederationLoop. The event stays as the sub-second path for peers that are already up. Reconciliation replaces the local blocklist instead of merging into it. That is what makes an unban work -- the control plane can clear a ban and there is no event for it, so a peer that drops off the list has to drop out of the blocklist too. Replacing raises a race the router hits on startup, where the first /info is still in flight while gossip is already live: a stale answer would immediately undo a ban that had just arrived. The control plane writes the ban to its database before publishing the event, so an answer *issued after* a ban is guaranteed to carry it, while one issued before cannot speak to it at all. reconcileBannedPeers therefore takes the time the request was issued and leaves alone any ban recorded at or after it. bannedPeers stores that timestamp; every reader only tests presence. That boundary is inclusive, because the answer has to be strictly newer than the ban to say anything about it. A clock coarse enough to time the fetch and the ban in the same tick would otherwise read a simultaneous ban as the older of the two and drop it -- Windows does so readily, and under a strict comparison TestRouterGossipSubBannedEvent failed about one run in ten with the ban deleted at the instant it arrived. - node: SyncMeshConfig returns the ban set alongside the key and router addresses, and Options.BannedPeerIDs seeds the revocation cache in NewSamNode so the gater enforces existing bans from the first connection. A nil result means the control plane was not reached, which is not the same as "nothing is banned", so it clears nothing. - router: reconcileBannedPeers evicts any prior admission for a newly banned peer, and skips undecodable entries rather than discarding the valid ones. TestReconcileBannedPeers covers the unban case, which is the point of replacing rather than merging, the malformed-entry case, and a ban recorded in the same tick as the fetch. TestRouterGossipSubBannedEvent already covered the startup race described above; it fails without the fetch timestamp. --- cmd/sam-node/main.go | 8 +++- internal/node/controlplane.go | 21 ++++++--- internal/node/controlplane_test.go | 7 ++- internal/node/node.go | 13 ++++++ internal/node/options.go | 20 +++++--- internal/router/router.go | 74 +++++++++++++++++++++++++++++- internal/router/router_test.go | 66 ++++++++++++++++++++++++++ mobile/sam-node-ffi/ffi/ffi.go | 5 +- 8 files changed, 194 insertions(+), 20 deletions(-) diff --git a/cmd/sam-node/main.go b/cmd/sam-node/main.go index 8962adc0..111eae8d 100644 --- a/cmd/sam-node/main.go +++ b/cmd/sam-node/main.go @@ -350,7 +350,7 @@ func main() { var controlPlanePubKey ed25519.PublicKey var routerAddrs []multiaddr.Multiaddr - storedPubKey, syncedAddrs, err := node.SyncMeshConfig(context.Background(), store) + storedPubKey, syncedAddrs, bannedPeerIDs, err := node.SyncMeshConfig(context.Background(), store) if err != nil { logger.Warnf("Failed to sync mesh config: %v", err) } @@ -475,6 +475,7 @@ func main() { ControlPlanePubKey: controlPlanePubKey, RouterAddrs: routerAddrs, Store: store, + BannedPeerIDs: bannedPeerIDs, MeshID: meshFlag, DiscoveryInterval: discoveryIntervalFlag, ListenAddrs: listenAddrs, @@ -542,6 +543,7 @@ func main() { PrivKey: priv, RouterAddrs: initRouterAddrs, Store: store, + BannedPeerIDs: bannedPeerIDs, MeshID: meshFlag, DiscoveryInterval: discoveryIntervalFlag, ListenAddrs: listenAddrs, @@ -599,11 +601,12 @@ func main() { } enrollCancel() - storedPubKey, newRouterAddrs, err := node.SyncMeshConfig(context.Background(), store) + storedPubKey, newRouterAddrs, postEnrollBannedPeerIDs, err := node.SyncMeshConfig(context.Background(), store) if err != nil { logger.Warnf("Failed to sync mesh config post-enrollment: %v", err) } controlPlanePubKey = storedPubKey + bannedPeerIDs = postEnrollBannedPeerIDs logger.Debugf("listenAddrs: %v, allowLoopback: %v", listenAddrs, allowLoopbackFlag) meshNode, err = node.NewSamNode(node.Options{ @@ -611,6 +614,7 @@ func main() { ControlPlanePubKey: controlPlanePubKey, RouterAddrs: newRouterAddrs, Store: store, + BannedPeerIDs: bannedPeerIDs, MeshID: meshFlag, DiscoveryInterval: discoveryIntervalFlag, ListenAddrs: listenAddrs, diff --git a/internal/node/controlplane.go b/internal/node/controlplane.go index e6a3c37d..de40ed4a 100644 --- a/internal/node/controlplane.go +++ b/internal/node/controlplane.go @@ -134,17 +134,26 @@ func mergeTrustedKeys(existing []TrustedKey, fetched []ed25519.PublicKey, now ti // SyncMeshConfig loads the mesh configuration from the store, attempts to refresh it // via HTTP from the control plane, and updates the store if successful. -// It returns the control plane public key and the latest multiaddresses. -func SyncMeshConfig(ctx context.Context, s *Store) ([]byte, []multiaddr.Multiaddr, error) { +// It returns the control plane public key, the latest multiaddresses, and the +// control plane's current ban set. +// +// The ban set is deliberately not persisted. MeshEvent_BANNED is published once +// and gossip has no replay, so a node that restarted or was offline has to be +// told again; /info is that catch-up, and reading it fresh each start is also +// what makes an unban take effect. Nil means the control plane was not reached, +// which is not the same as "nothing is banned": callers must not treat it as an +// instruction to clear anything. +func SyncMeshConfig(ctx context.Context, s *Store) ([]byte, []multiaddr.Multiaddr, []string, error) { pubKey, storedAddrsStr, err := s.LoadMeshConfig() if err != nil { - return nil, nil, fmt.Errorf("failed to load mesh config from store: %w", err) + return nil, nil, nil, fmt.Errorf("failed to load mesh config from store: %w", err) } controlPlaneURL, err := s.LoadControlPlaneURL() if err != nil { - return nil, nil, fmt.Errorf("failed to load control plane URL from store: %w", err) + return nil, nil, nil, fmt.Errorf("failed to load control plane URL from store: %w", err) } + var bannedPeerIDs []string var routerAddrs []multiaddr.Multiaddr // Parse stored addresses @@ -163,7 +172,7 @@ func SyncMeshConfig(ctx context.Context, s *Store) ([]byte, []multiaddr.Multiadd info, err := FetchControlPlaneInfo(ctx, controlPlaneURL) if err != nil { logger.Warnf("Failed to fetch updated addresses via HTTP (using cached): %v", err) - } else if len(info.RouterAddresses) > 0 { + } else if bannedPeerIDs = info.GetBannedPeerIds(); len(info.RouterAddresses) > 0 { logger.Infof("Discovered latest router addresses: %v", info.RouterAddresses) var newRouterAddrs []multiaddr.Multiaddr for _, addrStr := range info.RouterAddresses { @@ -199,7 +208,7 @@ func SyncMeshConfig(ctx context.Context, s *Store) ([]byte, []multiaddr.Multiadd } } - return pubKey, routerAddrs, nil + return pubKey, routerAddrs, bannedPeerIDs, nil } // FetchMeshPolicy retrieves the latest mesh policy from the control plane's /policies endpoint using a biscuit token. diff --git a/internal/node/controlplane_test.go b/internal/node/controlplane_test.go index 04b6cc08..e47508e3 100644 --- a/internal/node/controlplane_test.go +++ b/internal/node/controlplane_test.go @@ -162,10 +162,13 @@ func TestSyncMeshConfig(t *testing.T) { defer store.Close() //nolint:errcheck // Initial store is empty, so SyncMeshConfig should just return empty - pubKey, addrs, err := SyncMeshConfig(context.Background(), store) + pubKey, addrs, bannedPeerIDs, err := SyncMeshConfig(context.Background(), store) if err != nil { t.Fatalf("SyncMeshConfig failed: %v", err) } + if len(bannedPeerIDs) != 0 { + t.Errorf("Expected no banned peers for empty store, got %v", bannedPeerIDs) + } if len(pubKey) != 0 || len(addrs) != 0 { t.Errorf("Expected empty result for empty store, got pubKey=%v, addrs=%v", pubKey, addrs) } @@ -180,7 +183,7 @@ func TestSyncMeshConfig(t *testing.T) { } // Call SyncMeshConfig, it should fetch new addrs from server - pubKey, addrs, err = SyncMeshConfig(context.Background(), store) + pubKey, addrs, _, err = SyncMeshConfig(context.Background(), store) if err != nil { t.Fatalf("SyncMeshConfig failed: %v", err) } diff --git a/internal/node/node.go b/internal/node/node.go index f6fc054f..dfd99550 100644 --- a/internal/node/node.go +++ b/internal/node/node.go @@ -299,6 +299,16 @@ func NewSamNode(cfg Options) (*SamNode, error) { if err != nil { return nil, fmt.Errorf("failed to create revocation cache: %w", err) } + // Seed from the control plane's ban set so the gater enforces existing + // bans from the first connection, instead of waiting for an event that + // was already published while this node was down. + for _, id := range cfg.BannedPeerIDs { + if _, err := peer.Decode(id); err != nil { + logger.Warnf("Ignoring undecodable banned peer ID %q from the control plane: %v", id, err) + continue + } + node.revokedPeers.Add(id, time.Now().Unix()) + } node.peerLabelGate, err = lru.New[string, time.Time](labelGateCacheSize) if err != nil { return nil, fmt.Errorf("failed to create label gate cache: %w", err) @@ -1276,6 +1286,9 @@ func (n *SamNode) handleBannedEvent(event *api.MeshEvent) { n.revokedPeers.Add(event.PeerId, event.Timestamp) } // Drop any prior admission, otherwise the relay ACL keeps honouring it. + // The cache entry above is not written to disk: a restarted node picks the + // ban back up from the control plane's ban set in /info (see + // SyncMeshConfig), which is also how an unban reaches it. if p, err := peer.Decode(event.PeerId); err == nil { n.authPeers.Delete(p) if n.Host != nil { diff --git a/internal/node/options.go b/internal/node/options.go index f70529a6..98f3b146 100644 --- a/internal/node/options.go +++ b/internal/node/options.go @@ -40,13 +40,19 @@ type Options struct { ControlPlanePubKey ed25519.PublicKey RouterAddrs []multiaddr.Multiaddr Store *Store - MeshID string - DiscoveryInterval string - ListenAddrs []string - EnableRelay bool - NodeConfig *NodeConfigComplete - KeyGracePeriod time.Duration - AllowLoopback bool + + // BannedPeerIDs seeds the revocation cache from the control plane's ban + // set (see SyncMeshConfig). Without it a restarted node would enforce no + // ban until the next MeshEvent_BANNED, which for an existing ban never + // comes. + BannedPeerIDs []string + MeshID string + DiscoveryInterval string + ListenAddrs []string + EnableRelay bool + NodeConfig *NodeConfigComplete + KeyGracePeriod time.Duration + AllowLoopback bool // AnnouncePrivateAddrs controls whether RFC1918/ULA addresses are published // to the mesh. Nil means true: private meshes reach each other over exactly // those addresses. Set false on nodes that are only reachable via routers or diff --git a/internal/router/router.go b/internal/router/router.go index 3f733d5b..bce3d493 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -721,7 +721,10 @@ func (r *Router) listenForControlPlaneEvents(ctx context.Context) { if bannedPeer, err := peer.Decode(event.PeerId); err == nil { logger.Infof("[Router Event] Received BANNED event for peer %s, evicting from authenticated peers and adding to blocklist", bannedPeer) r.authenticatedPeers.Delete(bannedPeer) - r.bannedPeers.Store(bannedPeer, true) + // The timestamp is what stops an /info answer that was + // already in flight from undoing this (see + // reconcileBannedPeers). + r.bannedPeers.Store(bannedPeer, time.Now()) if r.Host != nil { _ = r.Host.Network().ClosePeer(bannedPeer) } @@ -853,6 +856,67 @@ func (r *Router) renewLease() { } } +// reconcileBannedPeers replaces the local blocklist with the control plane's +// ban set. It is a replacement rather than a merge: the control plane can +// unban a peer, and there is no event for that, so a peer that has dropped off +// the list has to drop out of the blocklist too. +// +// fetchedAt is when the request that produced peerIDs was issued, and it is +// what makes the replacement safe. A ban is written to the control plane's +// database before MeshEvent_BANNED is published, so a response *issued after* +// the ban is guaranteed to carry it, and its absence really does mean +// unbanned. A response issued before the ban cannot say anything about it, so +// bans recorded at or after fetchedAt are left alone -- otherwise an /info +// still in flight when an event arrived would immediately undo it, which is a +// race the router hits on startup, where the first poll overlaps live gossip. +// +// The boundary is inclusive because the answer has to be strictly newer than +// the ban to speak to it. A clock coarse enough to time the fetch and the ban +// in the same tick -- Windows readily does -- otherwise reads a simultaneous +// ban as the older of the two and drops it. +func (r *Router) reconcileBannedPeers(peerIDs []string, fetchedAt time.Time) { + banned := make(map[peer.ID]struct{}, len(peerIDs)) + for _, id := range peerIDs { + p, err := peer.Decode(id) + if err != nil { + logger.Warnf("[Federation] Ignoring undecodable banned peer ID %q: %v", id, err) + continue + } + banned[p] = struct{}{} + } + + // Drop bans the control plane no longer holds, except any that landed at + // or after the moment this answer was asked for. + r.bannedPeers.Range(func(key, value any) bool { + p, ok := key.(peer.ID) + if !ok { + return true + } + if _, still := banned[p]; still { + return true + } + if bannedAt, ok := value.(time.Time); ok && !bannedAt.Before(fetchedAt) { + return true + } + logger.Infof("[Federation] Peer %s is no longer banned, removing from blocklist", p) + r.bannedPeers.Delete(p) + return true + }) + + // Apply the ones it does, evicting anything already admitted. + for p := range banned { + if _, already := r.bannedPeers.Load(p); already { + continue + } + logger.Infof("[Federation] Peer %s is banned, adding to blocklist", p) + r.bannedPeers.Store(p, fetchedAt) + r.authenticatedPeers.Delete(p) + if r.Host != nil { + _ = r.Host.Network().ClosePeer(p) + } + } +} + func (r *Router) runFederationLoop() { defer r.wg.Done() ticker := time.NewTicker(30 * time.Second) @@ -872,6 +936,10 @@ func (r *Router) runFederationLoop() { func (r *Router) connectBootstrapRouters() { client := &http.Client{Timeout: 10 * time.Second} + // Taken before the request: anything banned after this point cannot be + // reflected in the answer, so reconciliation must not read its absence as + // an unban (see reconcileBannedPeers). + fetchedAt := time.Now() resp, err := client.Get(r.config.ControlPlaneURL + "/info") if err != nil { logger.Errorf("[Federation] Failed to fetch router info: %v", err) @@ -894,6 +962,10 @@ func (r *Router) connectBootstrapRouters() { return } + // /info carries the whole ban set, so this is also where a router that + // restarted or missed a MeshEvent_BANNED catches up. + r.reconcileBannedPeers(info.GetBannedPeerIds(), fetchedAt) + for _, addrStr := range info.RouterAddresses { ma, err := multiaddr.NewMultiaddr(addrStr) if err != nil { diff --git a/internal/router/router_test.go b/internal/router/router_test.go index 5798219d..71fe0628 100644 --- a/internal/router/router_test.go +++ b/internal/router/router_test.go @@ -873,3 +873,69 @@ func TestPerformMutualAuthAcceptsRotatedKey(t *testing.T) { t.Fatal("peer not recorded as authenticated") } } + +// reconcileBannedPeers replaces the blocklist with the control plane's ban +// set rather than merging into it. The removal half is the point: the control +// plane can unban a peer and there is no event for that, so a peer that drops +// off /info has to drop out of the blocklist too. +func TestReconcileBannedPeers(t *testing.T) { + newPeer := func(t *testing.T) peer.ID { + t.Helper() + priv, _, err := crypto.GenerateEd25519Key(rand.Reader) + if err != nil { + t.Fatal(err) + } + id, err := peer.IDFromPrivateKey(priv) + if err != nil { + t.Fatal(err) + } + return id + } + + stillBanned := newPeer(t) + unbanned := newPeer(t) + newlyBanned := newPeer(t) + + r := &Router{} + // Prior state, as a running router would hold it. + past := time.Now().Add(-time.Minute) + r.bannedPeers.Store(stillBanned, past) + r.bannedPeers.Store(unbanned, past) + r.authenticatedPeers.Store(newlyBanned, true) + + fetchedAt := time.Now() + r.reconcileBannedPeers([]string{stillBanned.String(), newlyBanned.String()}, fetchedAt) + + if _, banned := r.bannedPeers.Load(stillBanned); !banned { + t.Error("a peer still in the ban set must stay banned") + } + if _, banned := r.bannedPeers.Load(newlyBanned); !banned { + t.Error("a peer newly in the ban set must become banned") + } + if _, banned := r.bannedPeers.Load(unbanned); banned { + t.Error("a peer no longer in the ban set must be unbanned: /info is the only signal an unban has") + } + if _, admitted := r.authenticatedPeers.Load(newlyBanned); admitted { + t.Error("banning a peer must drop any prior admission") + } + + // An undecodable entry must be skipped, not abort the reconciliation. + r.reconcileBannedPeers([]string{"not-a-peer-id", stillBanned.String()}, time.Now()) + if _, banned := r.bannedPeers.Load(stillBanned); !banned { + t.Error("a malformed entry must not discard the valid ones") + } + + // A ban recorded in the same tick as the fetch must survive. The answer + // was issued no later than the ban, so its silence cannot mean unbanned, + // and a clock too coarse to separate the two -- Windows readily is -- must + // not be what decides it. This is the startup race, and a strict + // comparison here let it through: TestRouterGossipSubBannedEvent failed + // roughly one run in ten with the ban deleted at the instant it arrived. + simultaneous := newPeer(t) + tick := time.Now() + r.bannedPeers.Store(simultaneous, tick) + r.reconcileBannedPeers(nil, tick) + if _, banned := r.bannedPeers.Load(simultaneous); !banned { + t.Error("a ban recorded at the fetch instant must survive: the answer is not newer than the ban, so it cannot report it unbanned") + } +} diff --git a/mobile/sam-node-ffi/ffi/ffi.go b/mobile/sam-node-ffi/ffi/ffi.go index 478d1e6c..ea26a847 100644 --- a/mobile/sam-node-ffi/ffi/ffi.go +++ b/mobile/sam-node-ffi/ffi/ffi.go @@ -144,7 +144,7 @@ func StartNode(configJSON string) error { var routerAddrs []multiaddr.Multiaddr // Sync config from stored/synced configuration - storedPubKey, syncedAddrs, err := node.SyncMeshConfig(context.Background(), store) + storedPubKey, syncedAddrs, bannedPeerIDs, err := node.SyncMeshConfig(context.Background(), store) if err == nil && len(storedPubKey) > 0 { controlPlanePubKey = storedPubKey routerAddrs = syncedAddrs @@ -191,6 +191,7 @@ func StartNode(configJSON string) error { ControlPlanePubKey: controlPlanePubKey, RouterAddrs: routerAddrs, Store: store, + BannedPeerIDs: bannedPeerIDs, MeshID: meshID, DiscoveryInterval: discoveryInterval, ListenAddrs: listenAddrs, @@ -374,7 +375,7 @@ func EnrollNode(dataDir string, controlPlaneURL string, jwt string, allowLoopbac return fmt.Errorf("failed to save control plane URL: %w", err) } - _, _, err = node.SyncMeshConfig(enrollCtx, store) + _, _, _, err = node.SyncMeshConfig(enrollCtx, store) if err != nil { return fmt.Errorf("failed to sync mesh config post-enrollment: %w", err) } From c3c488947bfd5aeaf952ec06fe6932898c8e920f Mon Sep 17 00:00:00 2001 From: HosniBelfeki Date: Fri, 4 Sep 2026 23:42:32 +0100 Subject: [PATCH 3/3] node: remove the unused on-disk peer ban store The banned_peers bbolt bucket was created and read -- Store.IsBanned was wired into both connection gater methods, peerIsRevoked and the debug connect handler -- but nothing in the node ever wrote it, so IsBanned always answered false. The only writer was TestConnectionGater, which put a key in through store.db.Update directly, so the test proved the gater read the bucket while nothing proved anything filled it. Persisting bans locally was an early idea that was never finished, and it should not be: a ban on disk cannot be cleared by the control plane, since there is no unban event, and it says nothing about a node that was offline when the ban was published. Reconciling against the ban set in /info on every start handles both, and now does (previous commit). So this drops the bucket, IsBanned and its four call sites. peerIsRevoked keeps the revocation cache, which is now seeded from /info at startup, and the gater keeps its cache check -- the only one of its two checks that was ever able to answer true. TestConnectionGater's third case becomes what it can now assert: a peer the control plane does not ban stays reachable. TestGaterEnforcesSeededBans replaces what the store case was reaching for, checking the gater denies a peer seeded from the control plane's ban set and does not deny unrelated peers. TestStore_IsBanned goes with the method it covered. --- internal/node/debug_handlers.go | 3 -- internal/node/gate.go | 9 ------ internal/node/gate_test.go | 48 +++++++++++++++++++++++------- internal/node/identity_evidence.go | 8 ++--- internal/node/store.go | 28 ++++------------- internal/node/store_test.go | 44 --------------------------- 6 files changed, 47 insertions(+), 93 deletions(-) diff --git a/internal/node/debug_handlers.go b/internal/node/debug_handlers.go index 118345cb..ff0cf323 100644 --- a/internal/node/debug_handlers.go +++ b/internal/node/debug_handlers.go @@ -265,9 +265,6 @@ func (n *SamNode) connectPeer(ctx context.Context, peerAddr string) error { if n.revokedPeers != nil && n.revokedPeers.Contains(addrInfo.ID.String()) { return fmt.Errorf("failed to dial: failed to dial %s: gater disallows connection to peer", addrInfo.ID) } - if n.Store.IsBanned(addrInfo.ID) { - return fmt.Errorf("failed to dial: failed to dial %s: gater disallows connection to peer", addrInfo.ID) - } conns := n.Host.Network().ConnsToPeer(addrInfo.ID) connectedness := n.Host.Network().Connectedness(addrInfo.ID) logger.Debugf("[connect-peer] Target peer %s, connectedness: %v, active conns: %d", addrInfo.ID, connectedness, len(conns)) diff --git a/internal/node/gate.go b/internal/node/gate.go index de56f049..f83c38ac 100644 --- a/internal/node/gate.go +++ b/internal/node/gate.go @@ -44,11 +44,6 @@ func (g *nodeConnGate) InterceptPeerDial(p peer.ID) (allow bool) { logger.Warnf("[Gater] InterceptPeerDial denying %s: in revoked cache", p) return false } - banned := g.node.Store.IsBanned(p) - if banned { - logger.Warnf("[Gater] InterceptPeerDial denying %s: banned in store", p) - return false - } logger.Debugf("[Gater] InterceptPeerDial allowing %s", p) return true } @@ -70,10 +65,6 @@ func (g *nodeConnGate) InterceptSecured(dir network.Direction, p peer.ID, n netw logger.Warnf("[Gater] InterceptSecured denying %s: in revoked cache", p) return false } - if g.node.Store.IsBanned(p) { - logger.Warnf("[Gater] InterceptSecured denying %s: banned in store", p) - return false - } logger.Debugf("[Gater] InterceptSecured allowing %s", p) return true } diff --git a/internal/node/gate_test.go b/internal/node/gate_test.go index c10593bc..c187bbf6 100644 --- a/internal/node/gate_test.go +++ b/internal/node/gate_test.go @@ -27,7 +27,6 @@ import ( "github.com/libp2p/go-libp2p/core/peer" "github.com/libp2p/go-libp2p/core/protocol" "github.com/modelcontextprotocol/go-sdk/mcp" - "go.etcd.io/bbolt" ) func TestConnectionGater(t *testing.T) { @@ -82,20 +81,49 @@ func TestConnectionGater(t *testing.T) { t.Errorf("expected InterceptSecured to deny peer2 (in revoked cache)") } - // Case 3: Peer is in persistent store (banned) - err = store.db.Update(func(tx *bbolt.Tx) error { - b := tx.Bucket([]byte(bucketBannedPeers)) - return b.Put([]byte(peer3.String()), []byte("true")) - }) + // Case 3: a peer the control plane does not ban stays reachable, even + // after another peer has been banned. + if !gater.InterceptPeerDial(peer3) { + t.Errorf("expected InterceptPeerDial to allow peer3") + } + if !gater.InterceptSecured(network.DirInbound, peer3, nil) { + t.Errorf("expected InterceptSecured to allow peer3") + } +} + +// The revocation cache is seeded from the control plane's ban set at startup +// (Options.BannedPeerIDs, filled by SyncMeshConfig). Without that a restarted +// node would enforce no ban at all until the next MeshEvent_BANNED, which for +// a ban published while it was down never arrives. +func TestGaterEnforcesSeededBans(t *testing.T) { + priv, _, _ := crypto.GenerateEd25519Key(nil) + banned, err := peer.IDFromPrivateKey(priv) + if err != nil { + t.Fatal(err) + } + otherPriv, _, _ := crypto.GenerateEd25519Key(nil) + allowed, err := peer.IDFromPrivateKey(otherPriv) if err != nil { t.Fatal(err) } - if gater.InterceptPeerDial(peer3) { - t.Errorf("expected InterceptPeerDial to deny peer3 (in store)") + cache, err := lru.New[string, int64](100) + if err != nil { + t.Fatal(err) + } + node := &SamNode{revokedPeers: cache} + // Mirrors what NewSamNode does with Options.BannedPeerIDs. + node.revokedPeers.Add(banned.String(), time.Now().Unix()) + + gater := &nodeConnGate{node: node} + if gater.InterceptPeerDial(banned) { + t.Error("a peer in the control plane's ban set must not be dialled") + } + if gater.InterceptSecured(network.DirInbound, banned, nil) { + t.Error("a peer in the control plane's ban set must not be accepted") } - if gater.InterceptSecured(network.DirInbound, peer3, nil) { - t.Errorf("expected InterceptSecured to deny peer3 (in store)") + if !gater.InterceptPeerDial(allowed) { + t.Error("seeding a ban must not deny unrelated peers") } } diff --git a/internal/node/identity_evidence.go b/internal/node/identity_evidence.go index fe5f331e..cf47474a 100644 --- a/internal/node/identity_evidence.go +++ b/internal/node/identity_evidence.go @@ -295,11 +295,11 @@ func (n *SamNode) buildPeerEvidence(requested peer.ID, observation peerBiscuitOb }, nil } +// peerIsRevoked reports whether the peer is in the revocation cache, which is +// seeded from the control plane's ban set at startup (see SyncMeshConfig) and +// updated by MeshEvent_BANNED. func (n *SamNode) peerIsRevoked(peerID peer.ID) bool { - if n.revokedPeers != nil && n.revokedPeers.Contains(peerID.String()) { - return true - } - return n.Store != nil && n.Store.IsBanned(peerID) + return n.revokedPeers != nil && n.revokedPeers.Contains(peerID.String()) } type biscuitClaims struct { diff --git a/internal/node/store.go b/internal/node/store.go index c0d0dfda..0356ea1a 100644 --- a/internal/node/store.go +++ b/internal/node/store.go @@ -22,7 +22,6 @@ import ( "path/filepath" "time" - "github.com/libp2p/go-libp2p/core/peer" "go.etcd.io/bbolt" bbolterrors "go.etcd.io/bbolt/errors" ) @@ -79,9 +78,6 @@ func NewStore(dir string) (*Store, error) { if _, err := tx.CreateBucketIfNotExists([]byte(bucketIdentity)); err != nil { return err } - if _, err := tx.CreateBucketIfNotExists([]byte(bucketBannedPeers)); err != nil { - return err - } return nil }) @@ -308,22 +304,8 @@ func (s *Store) Close() error { return s.db.Close() } -const ( - bucketBannedPeers = "banned_peers" -) - -// IsBanned checks local store to see if this peer is banned. -func (s *Store) IsBanned(p peer.ID) bool { - var banned bool - _ = s.db.View(func(tx *bbolt.Tx) error { - b := tx.Bucket([]byte(bucketBannedPeers)) - if b == nil { - return nil - } - if b.Get([]byte(p.String())) != nil { - banned = true - } - return nil - }) - return banned -} +// Peer bans are deliberately not kept here. A ban on disk cannot be undone by +// the control plane -- there is no unban event -- and it says nothing about a +// node that was offline when the ban was published. Both are handled instead by +// reconciling against the ban set in /info on every start (see SyncMeshConfig), +// with MeshEvent_BANNED as the sub-second path for nodes that are already up. diff --git a/internal/node/store_test.go b/internal/node/store_test.go index d2a1f385..0546fac7 100644 --- a/internal/node/store_test.go +++ b/internal/node/store_test.go @@ -18,15 +18,11 @@ import ( "bytes" "crypto/ed25519" "errors" - "fmt" "os" "path/filepath" "reflect" "testing" "time" - - "github.com/libp2p/go-libp2p/core/peer" - "go.etcd.io/bbolt" ) func TestStore_NewStore_And_Close(t *testing.T) { @@ -416,46 +412,6 @@ func TestStore_ResetMeshIdentity(t *testing.T) { t.Errorf("expected private key to survive reset, got %v want %v", loadedKey, key) } } - -func TestStore_IsBanned(t *testing.T) { - store, err := NewStore(t.TempDir()) - if err != nil { - t.Fatalf("Failed to create store: %v", err) - } - defer func() { - if err := store.Close(); err != nil { - t.Errorf("failed to close store: %v", err) - } - }() - - pID, err := peer.Decode("12D3KooWBysiyDVxxj7Lq8KvhFnZVhqKdZHtwRaJu7hvGwSZFMNg") - if err != nil { - t.Fatalf("Failed to decode peer ID: %v", err) - } - - // Verify not banned initially - if store.IsBanned(pID) { - t.Errorf("Expected peer %s to not be banned", pID) - } - - // Manually add peer to banned bucket to test IsBanned - err = store.db.Update(func(tx *bbolt.Tx) error { - b := tx.Bucket([]byte("banned_peers")) - if b == nil { - return fmt.Errorf("banned_peers bucket not found") - } - return b.Put([]byte(pID.String()), []byte("banned")) - }) - if err != nil { - t.Fatalf("Failed to manually ban peer: %v", err) - } - - // Verify banned now - if !store.IsBanned(pID) { - t.Errorf("Expected peer %s to be banned", pID) - } -} - func TestGetDefaultDataDir(t *testing.T) { dir, err := GetDefaultDataDir() if err != nil {