PLT-1072: Wire rate limiter into native gRPC (:9090) unary+stream interceptors

config/cosmosbase/agreement_test.go

@@ -110,6 +110,10 @@ func readerValues(t *testing.T) map[string]string {
 		"grpc.keepalive-timeout":               fmt.Sprint(cfg.GRPC.KeepaliveTimeout),
 		"grpc.keepalive-min-time":              fmt.Sprint(cfg.GRPC.KeepaliveMinTime),
 		"grpc.keepalive-permit-without-stream": fmt.Sprint(cfg.GRPC.KeepalivePermitWithoutStream),
+		"grpc.ip-rate-limit-rps":               fmt.Sprint(cfg.GRPC.IPRateLimitRPS),
+		"grpc.ip-rate-limit-burst":             fmt.Sprint(cfg.GRPC.IPRateLimitBurst),
+		"grpc.rate-limiting-enabled":           fmt.Sprint(cfg.GRPC.RateLimitingEnabled),
+		"grpc.trusted-proxy-cidrs":             fmt.Sprint(cfg.GRPC.TrustedProxyCIDRs),
 		"telemetry.service-name":               fmt.Sprint(cfg.Telemetry.ServiceName),
 		"telemetry.enabled":                    fmt.Sprint(cfg.Telemetry.Enabled),
 		"telemetry.enable-hostname":            fmt.Sprint(cfg.Telemetry.EnableHostname),

config/cosmosbase/cosmosbase.go

@@ -96,7 +96,7 @@ func apiDefaults(mode registry.Mode) any { return forMode(mode).API }
 // interface follows and for the same reason. The upstream default is on for every kind, so declaring that
 // would state an open interface on the nodes meant to expose the least.
 //
-// Six of these eleven keys are read only when the key is present. Two more are durations read through a
+// Nine of these fifteen keys are read only when the key is present. Two more are durations read through a
 // clamp that rescues a negative value and does nothing for an absent one, so those two are unguarded and
 // their clobber leaves no trace. The durations are declared as durations and written into a file as text,
 // which is the shape the reader parses back.

config/cosmosbase/cosmosbase_test.go

@@ -77,6 +77,8 @@ func TestTheGRPCKeysAreTheOnesItsReaderResolves(t *testing.T) {
 		"grpc.max-connection-idle", "grpc.max-connection-age", "grpc.max-connection-age-grace",
 		"grpc.keepalive-time", "grpc.keepalive-timeout", "grpc.keepalive-min-time",
 		"grpc.keepalive-permit-without-stream",
+		"grpc.ip-rate-limit-rps", "grpc.ip-rate-limit-burst", "grpc.rate-limiting-enabled",
+		"grpc.trusted-proxy-cidrs",
 	})
 }
 

ratelimiter/method_bucket.go

@@ -5,6 +5,8 @@ import "strings"
 const (
 	// PlaneCometBFT is the rate-limit plane label for Tendermint RPC HTTP.
 	PlaneCometBFT = "cometbft"
+	// PlaneGRPC is the rate-limit plane label for native gRPC (:9090).
+	PlaneGRPC = "grpc"
 
 	// rpcMethodBucketOther is the fallback label for unrecognized methods.
 	rpcMethodBucketOther = "other"
@@ -75,6 +77,33 @@ var knownCometBFTRPCMethods = map[string]struct{}{
 	"websocket":            {},
 }
 
+// knownGRPCServices lists protobuf service names registered on the native gRPC
+// server. Rejection metrics on PlaneGRPC record the service name rather than
+// the full /service/Method path, keeping OTel attribute cardinality bounded.
+var knownGRPCServices = map[string]struct{}{
+	"cosmos.auth.v1beta1.Query":                         {},
+	"cosmos.authz.v1beta1.Query":                        {},
+	"cosmos.bank.v1beta1.Query":                         {},
+	"cosmos.base.reflection.v1beta1.ReflectionService":  {},
+	"cosmos.base.reflection.v2alpha1.ReflectionService": {},
+	"cosmos.base.tendermint.v1beta1.Service":            {},
+	"cosmos.distribution.v1beta1.Query":                 {},
+	"cosmos.evidence.v1beta1.Query":                     {},
+	"cosmos.gov.v1beta1.Query":                          {},
+	"cosmos.params.v1beta1.Query":                       {},
+	"cosmos.slashing.v1beta1.Query":                     {},
+	"cosmos.staking.v1beta1.Query":                      {},
+	"cosmos.tx.v1beta1.Service":                         {},
+	"cosmos.upgrade.v1beta1.Query":                      {},
+	"cosmwasm.wasm.v1.Query":                            {},
+	"grpc.reflection.v1.ServerReflection":               {},
+	"seiprotocol.seichain.epoch.Query":                  {},
+	"seiprotocol.seichain.evm.Query":                    {},
+	"seiprotocol.seichain.mint.Query":                   {},
+	"seiprotocol.seichain.oracle.Query":                 {},
+	"seiprotocol.seichain.tokenfactory.Query":           {},
+}
+
 // bucketRPCMethod maps a raw JSON-RPC method name to a low-cardinality label
 // suitable for OTel/Prometheus metrics. Attacker-controlled method strings
 // collapse to rpcMethodBucketOther.
@@ -85,9 +114,35 @@ func bucketRPCMethod(plane, method string) string {
 	if plane == PlaneCometBFT {
 		return bucketCometBFTRPCMethod(method)
 	}
+	if plane == PlaneGRPC {
+		return bucketGRPCMethod(method)
+	}
 	return bucketNamespacedRPCMethod(method)
 }
 
+func bucketGRPCMethod(fullMethod string) string {
+	if fullMethod == "" || len(fullMethod) > maxRPCMethodLen {
+		return rpcMethodBucketOther
+	}
+	svc := grpcServiceFromFullMethod(fullMethod)
+	if svc == "" || len(svc) > maxRPCMethodLen {
+		return rpcMethodBucketOther
+	}
+	if _, ok := knownGRPCServices[svc]; ok {
+		return svc
+	}
+	return rpcMethodBucketOther
+}
+
+func grpcServiceFromFullMethod(fullMethod string) string {
+	method := strings.TrimPrefix(fullMethod, "/")
+	slash := strings.LastIndexByte(method, '/')
+	if slash <= 0 {
+		return ""
+	}
+	return method[:slash]
+}
+
 func bucketCometBFTRPCMethod(method string) string {
 	if method == "" || len(method) > maxRPCMethodLen {
 		return rpcMethodBucketOther

ratelimiter/method_bucket_test.go

@@ -61,4 +61,17 @@ func TestBucketRPCMethod_CometBFTUnknownMethods(t *testing.T) {
 func TestBucketRPCMethod_Invalid(t *testing.T) {
 	require.Equal(t, MethodInvalid, bucketRPCMethod("evm", MethodInvalid))
 	require.Equal(t, MethodInvalid, bucketRPCMethod(PlaneCometBFT, MethodInvalid))
+	require.Equal(t, MethodInvalid, bucketRPCMethod(PlaneGRPC, MethodInvalid))
+}
+
+func TestBucketRPCMethod_GrpcknownServices(t *testing.T) {
+	require.Equal(t, "cosmos.bank.v1beta1.Query", bucketRPCMethod(PlaneGRPC, "/cosmos.bank.v1beta1.Query/Balance"))
+	require.Equal(t, "cosmos.tx.v1beta1.Service", bucketRPCMethod(PlaneGRPC, "/cosmos.tx.v1beta1.Service/Simulate"))
+}
+
+func TestBucketRPCMethod_GrpcUnknownServices(t *testing.T) {
+	require.Equal(t, rpcMethodBucketOther, bucketRPCMethod(PlaneGRPC, ""))
+	require.Equal(t, rpcMethodBucketOther, bucketRPCMethod(PlaneGRPC, "/bogus.Service/Call"))
+	require.Equal(t, rpcMethodBucketOther, bucketRPCMethod(PlaneGRPC, "not-a-grpc-path"))
+	require.Equal(t, rpcMethodBucketOther, bucketRPCMethod(PlaneGRPC, strings.Repeat("a", maxRPCMethodLen+1)))
 }

sei-cosmos/baseapp/grpcserver.go

@@ -78,11 +78,8 @@ func (app *BaseApp) RegisterGRPCServer(server gogogrpc.Server) {
 			methodHandler := method.Handler
 			newMethods[i] = grpc.MethodDesc{
 				MethodName: method.MethodName,
-				Handler: func(srv interface{}, ctx context.Context, dec func(interface{}) error, _ grpc.UnaryServerInterceptor) (interface{}, error) {
-					return methodHandler(srv, ctx, dec, grpcmiddleware.ChainUnaryServer(
-						grpcrecovery.UnaryServerInterceptor(),
-						interceptor,
-					))
+				Handler: func(srv interface{}, ctx context.Context, dec func(interface{}) error, serverInterceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+					return methodHandler(srv, ctx, dec, chainQueryInterceptors(serverInterceptor, interceptor))
 				},
 			}
 		}
@@ -98,3 +95,16 @@ func (app *BaseApp) RegisterGRPCServer(server gogogrpc.Server) {
 		server.RegisterService(newDesc, data.handler)
 	}
 }
+
+// chainQueryInterceptors returns the chain a unary query runs through: panic
+// recovery, then serverInterceptor when non-nil, then queryCtx last so a rejected
+// call never costs a query context.
+func chainQueryInterceptors(serverInterceptor, queryCtx grpc.UnaryServerInterceptor) grpc.UnaryServerInterceptor {
+	chain := make([]grpc.UnaryServerInterceptor, 0, 3)
+	chain = append(chain, grpcrecovery.UnaryServerInterceptor())
+	if serverInterceptor != nil {
+		chain = append(chain, serverInterceptor)
+	}
+	chain = append(chain, queryCtx)
+	return grpcmiddleware.ChainUnaryServer(chain...)
+}

sei-cosmos/baseapp/grpcserver_test.go

@@ -0,0 +1,88 @@
+package baseapp
+
+import (
+	"context"
+	"net"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+
+	"github.com/sei-protocol/sei-chain/sei-cosmos/codec/types"
+	"github.com/sei-protocol/sei-chain/sei-cosmos/testutil/testdata"
+)
+
+// serveTestQuery registers the testdata Query service on app, exposes it on a real
+// grpc.Server built with serverOpts, and returns a client dialled to it.
+//
+// The service is registered through RegisterGRPCServer so calls travel grpc-go's
+// own dispatch path. An interceptor supplied through serverOpts reaches handlers
+// only if that path is intact, which invoking the interceptor closure directly
+// cannot show.
+func serveTestQuery(t *testing.T, app *BaseApp, serverOpts ...grpc.ServerOption) testdata.QueryClient {
+	t.Helper()
+
+	interfaceRegistry := types.NewInterfaceRegistry()
+	testdata.RegisterInterfaces(interfaceRegistry)
+	app.SetInterfaceRegistry(interfaceRegistry)
+	testdata.RegisterQueryServer(app.GRPCQueryRouter(), testdata.QueryImpl{})
+
+	srv := grpc.NewServer(serverOpts...)
+	app.RegisterGRPCServer(srv)
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	go func() { _ = srv.Serve(listener) }()
+	t.Cleanup(srv.Stop)
+
+	conn, err := grpc.Dial(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = conn.Close() })
+
+	return testdata.NewQueryClient(conn)
+}
+
+func TestRegisterGRPCServerAppliesServerInterceptor(t *testing.T) {
+	var calls atomic.Int64
+	spy := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+		calls.Add(1)
+		return handler(ctx, req)
+	}
+
+	client := serveTestQuery(t, setupBaseApp(t), grpc.ChainUnaryInterceptor(spy))
+
+	res, err := client.Echo(t.Context(), &testdata.EchoRequest{Message: "hello"})
+	require.NoError(t, err)
+	require.Equal(t, "hello", res.Message)
+	require.Equal(t, int64(1), calls.Load(), "server-level interceptor never reached the query handler")
+}
+
+func TestRegisterGRPCServerServerInterceptorCanRejectQuery(t *testing.T) {
+	var handlerCalls atomic.Int64
+	reject := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+		return nil, status.Error(codes.ResourceExhausted, "too many requests")
+	}
+	count := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+		handlerCalls.Add(1)
+		return handler(ctx, req)
+	}
+
+	client := serveTestQuery(t, setupBaseApp(t), grpc.ChainUnaryInterceptor(reject, count))
+
+	_, err := client.Echo(t.Context(), &testdata.EchoRequest{Message: "hello"})
+	require.Error(t, err)
+	require.Equal(t, codes.ResourceExhausted, status.Code(err))
+	require.Zero(t, handlerCalls.Load(), "rejected query still ran the rest of the chain")
+}
+
+func TestRegisterGRPCServerWithoutServerInterceptor(t *testing.T) {
+	client := serveTestQuery(t, setupBaseApp(t))
+
+	res, err := client.Echo(t.Context(), &testdata.EchoRequest{Message: "hello"})
+	require.NoError(t, err)
+	require.Equal(t, "hello", res.Message)
+}

sei-cosmos/server/config/config.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/sei-protocol/sei-chain/ratelimiter"
 	storetypes "github.com/sei-protocol/sei-chain/sei-cosmos/store/types"
 	"github.com/sei-protocol/sei-chain/sei-cosmos/telemetry"
 	sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types"
@@ -258,6 +259,33 @@ type GRPCConfig struct {
 	// KeepalivePermitWithoutStream defines whether the server allows keepalive
 	// pings even when there are no active streams.
 	KeepalivePermitWithoutStream bool `mapstructure:"keepalive-permit-without-stream"`
+
+	// IPRateLimitRPS is the per-IP sustained request rate in requests/second for
+	// native gRPC (:9090). Zero disables the token bucket (Allow always returns
+	// true) and does not bypass the admission interceptor when
+	// rate-limiting-enabled is true.
+	IPRateLimitRPS float64 `mapstructure:"ip-rate-limit-rps"`
+
+	// IPRateLimitBurst is the maximum per-IP burst size. Zero disables the token
+	// bucket (same effect as ip-rate-limit-rps = 0) and does not bypass the
+	// admission interceptor when rate-limiting-enabled is true.
+	IPRateLimitBurst int `mapstructure:"ip-rate-limit-burst"`
+
+	// RateLimitingEnabled is the master switch for gRPC rate-limit admission.
+	RateLimitingEnabled bool `mapstructure:"rate-limiting-enabled"`
+
+	// TrustedProxyCIDRs lists CIDRs whose x-forwarded-for metadata is trusted
+	// when resolving the client IP for rate limiting. Empty means trust no proxy.
+	TrustedProxyCIDRs []string `mapstructure:"trusted-proxy-cidrs"`
+}
+
+// RateLimiterConfig builds the ratelimiter.Config used by native gRPC admission.
+func (c GRPCConfig) RateLimiterConfig() ratelimiter.Config {
+	return ratelimiter.Config{
+		RPS:               c.IPRateLimitRPS,
+		Burst:             c.IPRateLimitBurst,
+		TrustedProxyCIDRs: c.TrustedProxyCIDRs,
+	}
 }
 
 // GRPCWebConfig defines configuration for the gRPC-web server.
@@ -386,6 +414,10 @@ func DefaultConfig() *Config {
 			KeepaliveTimeout:             DefaultGRPCKeepaliveTimeout,
 			KeepaliveMinTime:             DefaultGRPCKeepaliveMinTime,
 			KeepalivePermitWithoutStream: DefaultGRPCKeepalivePermitWithoutStream,
+			IPRateLimitRPS:               ratelimiter.DefaultRPS,
+			IPRateLimitBurst:             ratelimiter.DefaultBurst,
+			RateLimitingEnabled:          false,
+			TrustedProxyCIDRs:            nil,
 		},
 		Rosetta: RosettaConfig{
 			Enable:     false,
@@ -572,6 +604,19 @@ func GetConfig(v *viper.Viper) (Config, error) {
 	grpcMaxConnectionAge := clampNonNegativeDuration(v.GetDuration("grpc.max-connection-age"), DefaultGRPCMaxConnectionAge)
 	grpcMaxConnectionAgeGrace := clampNonNegativeDuration(v.GetDuration("grpc.max-connection-age-grace"), DefaultGRPCMaxConnectionAgeGrace)
 
+	grpcIPRateLimitRPS := ratelimiter.DefaultRPS
+	if v.IsSet("grpc.ip-rate-limit-rps") {
+		grpcIPRateLimitRPS = v.GetFloat64("grpc.ip-rate-limit-rps")
+	}
+	grpcIPRateLimitBurst := ratelimiter.DefaultBurst
+	if v.IsSet("grpc.ip-rate-limit-burst") {
+		grpcIPRateLimitBurst = v.GetInt("grpc.ip-rate-limit-burst")
+	}
+	grpcTrustedProxyCIDRs := []string(nil)
+	if v.IsSet("grpc.trusted-proxy-cidrs") {
+		grpcTrustedProxyCIDRs = v.GetStringSlice("grpc.trusted-proxy-cidrs")
+	}
+
 	cfg := Config{
 		BaseConfig: BaseConfig{
 			MinGasPrices:       v.GetString("minimum-gas-prices"),
@@ -627,6 +672,10 @@ func GetConfig(v *viper.Viper) (Config, error) {
 			KeepaliveTimeout:             grpcKeepaliveTimeout,
 			KeepaliveMinTime:             grpcKeepaliveMinTime,
 			KeepalivePermitWithoutStream: v.GetBool("grpc.keepalive-permit-without-stream"),
+			IPRateLimitRPS:               grpcIPRateLimitRPS,
+			IPRateLimitBurst:             grpcIPRateLimitBurst,
+			RateLimitingEnabled:          v.GetBool("grpc.rate-limiting-enabled"),
+			TrustedProxyCIDRs:            grpcTrustedProxyCIDRs,
 		},
 		GRPCWeb: GRPCWebConfig{
 			Enable:             v.GetBool("grpc-web.enable"),