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/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/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/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/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/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/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 { 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/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 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) }