diff --git a/config/cosmosbase/agreement_test.go b/config/cosmosbase/agreement_test.go index 103adeb464..53e35d716e 100644 --- a/config/cosmosbase/agreement_test.go +++ b/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), diff --git a/config/cosmosbase/cosmosbase.go b/config/cosmosbase/cosmosbase.go index f524a03447..c35b14e4d4 100644 --- a/config/cosmosbase/cosmosbase.go +++ b/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. diff --git a/config/cosmosbase/cosmosbase_test.go b/config/cosmosbase/cosmosbase_test.go index b8199582fa..29ff424f3b 100644 --- a/config/cosmosbase/cosmosbase_test.go +++ b/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", }) } diff --git a/ratelimiter/method_bucket.go b/ratelimiter/method_bucket.go index e350c9aca5..ed02c1120d 100644 --- a/ratelimiter/method_bucket.go +++ b/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 diff --git a/ratelimiter/method_bucket_test.go b/ratelimiter/method_bucket_test.go index 100e3bb5ea..9a2778b2fd 100644 --- a/ratelimiter/method_bucket_test.go +++ b/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))) } diff --git a/sei-cosmos/baseapp/grpcserver.go b/sei-cosmos/baseapp/grpcserver.go index d00a56a6e3..805c5ce923 100644 --- a/sei-cosmos/baseapp/grpcserver.go +++ b/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...) +} diff --git a/sei-cosmos/baseapp/grpcserver_test.go b/sei-cosmos/baseapp/grpcserver_test.go new file mode 100644 index 0000000000..c9600750b8 --- /dev/null +++ b/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) +} diff --git a/sei-cosmos/server/config/config.go b/sei-cosmos/server/config/config.go index ee732b85c9..546ddb696f 100644 --- a/sei-cosmos/server/config/config.go +++ b/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"), diff --git a/sei-cosmos/server/config/config_fuzz_test.go b/sei-cosmos/server/config/config_fuzz_test.go index 04ef74c8a4..b80a8e28d9 100644 --- a/sei-cosmos/server/config/config_fuzz_test.go +++ b/sei-cosmos/server/config/config_fuzz_test.go @@ -1166,6 +1166,10 @@ func TestGetConfigAbsentSectionDivergences(t *testing.T) { "grpc.keepalive-permit-without-stream", cfg.GRPC.KeepalivePermitWithoutStream, def.GRPC.KeepalivePermitWithoutStream, false, }, + { + "grpc.rate-limiting-enabled", + cfg.GRPC.RateLimitingEnabled, def.GRPC.RateLimitingEnabled, false, + }, {"telemetry.service-name", cfg.Telemetry.ServiceName, def.Telemetry.ServiceName, false}, {"telemetry.enable-hostname", cfg.Telemetry.EnableHostname, def.Telemetry.EnableHostname, false}, { @@ -1354,10 +1358,10 @@ func TestBaseConfigManifestNamesEveryField(t *testing.T) { ) } -// grpcKeys covers the three [grpc] keys read as plain casts. +// grpcKeys covers the four [grpc] keys read as plain casts. // -// The section is where the guarding in this reader is most complete, which is why only three keys -// are rows. Eight others are read behind v.IsSet or through clampNonNegativeDuration and so resolve +// The section is where the guarding in this reader is most complete, which is why only four keys +// are rows. Eleven others are read behind v.IsSet or through clampNonNegativeDuration and so resolve // an absent key to the in-code default rather than to a zero; CheckRow would predict the wrong // resolution for each, so they are driven by FuzzGetConfigGRPCDurationClamps and // TestGetConfigGRPCAbsentReads and recorded by name below. @@ -1380,14 +1384,20 @@ var grpcKeys = []configtest.KeySpec{ Why: "whether a client may ping with no active stream; false either way, so this row states " + "the key is read rather than recording a divergence", }, + { + Key: "grpc.rate-limiting-enabled", Path: "RateLimitingEnabled", + Cast: configtest.CastBool, Unguarded: true, + Why: "the declared default is false and an absent key resolves false, so this row states " + + "the key is read rather than recording a divergence", + }, } // grpcKeysWithTargetsOfTheirOwn are the [grpc] keys whose resolution a row cannot describe, recorded // for their names alone. // -// Six are read behind v.IsSet, so an absent key keeps the in-code default. The other two, +// Nine are read behind v.IsSet, so an absent key keeps the in-code default. The other two, // max-connection-age and max-connection-age-grace, are read unconditionally through the clamp -// (config.go:551-552), and their absent value matches the declared default only because both +// (config.go:604-605), and their absent value matches the declared default only because both // defaults are 0. The clamp rescues a negative value and does nothing for an absent one, so they are // unguarded reads whose clobber is invisible. TestGetConfigGRPCAbsentReads holds the two groups // apart for that reason. @@ -1406,6 +1416,9 @@ var grpcKeysWithTargetsOfTheirOwn = []configtest.KeyName{ "grpc.keepalive-time", "grpc.keepalive-timeout", "grpc.keepalive-min-time", + "grpc.ip-rate-limit-rps", + "grpc.ip-rate-limit-burst", + "grpc.trusted-proxy-cidrs", } func readGRPC(t testing.TB) func(configtest.AppOpts) (any, error) { @@ -1419,6 +1432,7 @@ func FuzzGRPCConfig(f *testing.F) { seeds.AddRow(uint(0), fuzzing.KindBool, "", int64(0), true) seeds.AddRow(uint(1), fuzzing.KindString, "127.0.0.1:19090", int64(0), false) seeds.AddRow(uint(2), fuzzing.KindBool, "", int64(0), true) + seeds.AddRow(uint(3), fuzzing.KindBool, "", int64(0), true) configtest.CheckEveryRowHasADiscriminatingSeed(f, "grpc", readGRPC(f), grpcKeys, seeds, grpcKeysWithTargetsOfTheirOwn...) @@ -1429,10 +1443,10 @@ func FuzzGRPCConfig(f *testing.F) { }) } -// TestGRPCKeyNamesMatchTheRecordedNames pins all eleven [grpc] key names, the three rows and the -// eight driven elsewhere. +// TestGRPCKeyNamesMatchTheRecordedNames pins all fifteen [grpc] key names, the four rows and the +// eleven driven elsewhere. // -// The eight had no record before this, because their target carries a local struct rather than a +// The eleven had no record before this, because their target carries a local struct rather than a // KeySpec table, so nothing held their spelling. That is the gap this closes. func TestGRPCKeyNamesMatchTheRecordedNames(t *testing.T) { configtest.CheckKeyNames(t, "grpc", grpcKeys, grpcKeysWithTargetsOfTheirOwn...) @@ -1443,6 +1457,9 @@ func TestGRPCManifestNamesEveryField(t *testing.T) { // Guarded reads, so an absent key keeps the in-code default rather than clobbering it. "MaxRecvMsgSize", "MaxOpenConnections", + "IPRateLimitRPS", + "IPRateLimitBurst", + "TrustedProxyCIDRs", // Clamped reads: a negative resolves to the in-code default rather than passing through. "MaxConnectionIdle", "MaxConnectionAge", @@ -1485,6 +1502,8 @@ func TestGetConfigGRPCAbsentReads(t *testing.T) { {"grpc.keepalive-time", got.KeepaliveTime, def.KeepaliveTime}, {"grpc.keepalive-timeout", got.KeepaliveTimeout, def.KeepaliveTimeout}, {"grpc.keepalive-min-time", got.KeepaliveMinTime, def.KeepaliveMinTime}, + {"grpc.ip-rate-limit-rps", got.IPRateLimitRPS, def.IPRateLimitRPS}, + {"grpc.ip-rate-limit-burst", got.IPRateLimitBurst, def.IPRateLimitBurst}, } { if c.absent != c.declared { t.Errorf("an absent %s resolved to %v rather than the declared %v, so its v.IsSet guard "+ @@ -1493,6 +1512,10 @@ func TestGetConfigGRPCAbsentReads(t *testing.T) { } } + if got.TrustedProxyCIDRs != nil { + t.Errorf("an absent grpc.trusted-proxy-cidrs resolved to %v rather than nil", got.TrustedProxyCIDRs) + } + // Read unconditionally and clamped. Nothing guards these, so the assertion is on the coincidence // itself: the declared default is the getter's zero, which is why an absent key looks correct. for _, c := range []struct { diff --git a/sei-cosmos/server/config/config_test.go b/sei-cosmos/server/config/config_test.go index 38a1abc0b9..1d1f09072d 100644 --- a/sei-cosmos/server/config/config_test.go +++ b/sei-cosmos/server/config/config_test.go @@ -11,6 +11,7 @@ import ( "github.com/spf13/viper" "github.com/stretchr/testify/require" + "github.com/sei-protocol/sei-chain/ratelimiter" storetypes "github.com/sei-protocol/sei-chain/sei-cosmos/store/types" sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types" seidbconfig "github.com/sei-protocol/sei-chain/sei-db/config" @@ -81,6 +82,10 @@ func TestDefaultGRPCConfig(t *testing.T) { require.Equal(t, DefaultGRPCKeepaliveTimeout, cfg.GRPC.KeepaliveTimeout) require.Equal(t, DefaultGRPCKeepaliveMinTime, cfg.GRPC.KeepaliveMinTime) require.Equal(t, DefaultGRPCKeepalivePermitWithoutStream, cfg.GRPC.KeepalivePermitWithoutStream) + require.Equal(t, ratelimiter.DefaultRPS, cfg.GRPC.IPRateLimitRPS) + require.Equal(t, ratelimiter.DefaultBurst, cfg.GRPC.IPRateLimitBurst) + require.False(t, cfg.GRPC.RateLimitingEnabled) + require.Nil(t, cfg.GRPC.TrustedProxyCIDRs) } // seedViperWithDefaultConfig renders the default app config template and reads diff --git a/sei-cosmos/server/config/testdata/grpc.keys.golden b/sei-cosmos/server/config/testdata/grpc.keys.golden index f3bb885cbe..b06deb44ed 100644 --- a/sei-cosmos/server/config/testdata/grpc.keys.golden +++ b/sei-cosmos/server/config/testdata/grpc.keys.golden @@ -1,6 +1,7 @@ "grpc.enable" "grpc.address" "grpc.keepalive-permit-without-stream" +"grpc.rate-limiting-enabled" # keys with a target of their own "grpc.max-recv-msg-size" "grpc.max-open-connections" @@ -10,3 +11,6 @@ "grpc.keepalive-time" "grpc.keepalive-timeout" "grpc.keepalive-min-time" +"grpc.ip-rate-limit-rps" +"grpc.ip-rate-limit-burst" +"grpc.trusted-proxy-cidrs" diff --git a/sei-cosmos/server/config/testdata/server_config.golden b/sei-cosmos/server/config/testdata/server_config.golden index b9076e3169..5cc2a887f1 100644 --- a/sei-cosmos/server/config/testdata/server_config.golden +++ b/sei-cosmos/server/config/testdata/server_config.golden @@ -38,6 +38,10 @@ GRPC.KeepaliveTime = time.Duration(2h0m0s) GRPC.KeepaliveTimeout = time.Duration(20s) GRPC.KeepaliveMinTime = time.Duration(5m0s) GRPC.KeepalivePermitWithoutStream = bool(false) +GRPC.IPRateLimitRPS = float64(200) +GRPC.IPRateLimitBurst = int(400) +GRPC.RateLimitingEnabled = bool(false) +GRPC.TrustedProxyCIDRs = Rosetta.Address = string(":8080") Rosetta.Blockchain = string("app") Rosetta.Network = string("network") diff --git a/sei-cosmos/server/config/toml.go b/sei-cosmos/server/config/toml.go index eb7d7f4ea0..c61b666726 100644 --- a/sei-cosmos/server/config/toml.go +++ b/sei-cosmos/server/config/toml.go @@ -260,6 +260,21 @@ keepalive-min-time = "{{ .GRPC.KeepaliveMinTime }}" # KeepalivePermitWithoutStream defines whether the server allows keepalive pings even when there are no active streams. keepalive-permit-without-stream = {{ .GRPC.KeepalivePermitWithoutStream }} +# ip-rate-limit-rps is the per-IP sustained request rate in requests/second for native gRPC (:9090). +# Zero disables per-IP throttling; set rate-limiting-enabled = false for a full bypass. +ip-rate-limit-rps = {{ .GRPC.IPRateLimitRPS }} + +# ip-rate-limit-burst is the maximum per-IP burst above the sustained rate. +# Zero disables per-IP throttling (same effect as ip-rate-limit-rps = 0). +ip-rate-limit-burst = {{ .GRPC.IPRateLimitBurst }} + +# rate-limiting-enabled is the master switch for gRPC rate-limit admission interceptors. +rate-limiting-enabled = {{ .GRPC.RateLimitingEnabled }} + +# trusted-proxy-cidrs lists CIDRs whose x-forwarded-for metadata is trusted when +# resolving the client IP for rate limiting. Empty means trust no proxy. +trusted-proxy-cidrs = [{{- range $i, $c := .GRPC.TrustedProxyCIDRs }}{{- if $i }}, {{ end }}"{{ $c }}"{{- end }}] + ############################################################################### ### gRPC Web Configuration (Auto-managed) ### ############################################################################### diff --git a/sei-cosmos/server/grpc/rate_limit.go b/sei-cosmos/server/grpc/rate_limit.go new file mode 100644 index 0000000000..f9e5bee2a1 --- /dev/null +++ b/sei-cosmos/server/grpc/rate_limit.go @@ -0,0 +1,39 @@ +package grpc + +import ( + "context" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/sei-protocol/sei-chain/ratelimiter" +) + +var errRateLimited = status.Error(codes.ResourceExhausted, "too many requests") + +// UnaryRateLimitInterceptor returns a server interceptor that applies per-IP +// token-bucket rate limiting before the handler runs. registry must be non-nil. +func UnaryRateLimitInterceptor(registry *ratelimiter.Registry) grpc.UnaryServerInterceptor { + return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { + ip := registry.IPFromGRPCContext(ctx) + if !registry.Allow(ctx, ip, ratelimiter.PlaneGRPC, info.FullMethod) { + return nil, errRateLimited + } + return handler(ctx, req) + } +} + +// StreamRateLimitInterceptor returns a server interceptor that applies per-IP +// token-bucket rate limiting when a client stream is established. registry must +// be non-nil. +func StreamRateLimitInterceptor(registry *ratelimiter.Registry) grpc.StreamServerInterceptor { + return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { + ctx := ss.Context() + ip := registry.IPFromGRPCContext(ctx) + if !registry.Allow(ctx, ip, ratelimiter.PlaneGRPC, info.FullMethod) { + return errRateLimited + } + return handler(srv, ss) + } +} diff --git a/sei-cosmos/server/grpc/rate_limit_dispatch_test.go b/sei-cosmos/server/grpc/rate_limit_dispatch_test.go new file mode 100644 index 0000000000..18e92b0e79 --- /dev/null +++ b/sei-cosmos/server/grpc/rate_limit_dispatch_test.go @@ -0,0 +1,76 @@ +package grpc + +import ( + "context" + "net" + "testing" + + "github.com/stretchr/testify/require" + dbm "github.com/tendermint/tm-db" + "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/ratelimiter" + "github.com/sei-protocol/sei-chain/sei-cosmos/baseapp" + codectypes "github.com/sei-protocol/sei-chain/sei-cosmos/codec/types" + "github.com/sei-protocol/sei-chain/sei-cosmos/testutil" + "github.com/sei-protocol/sei-chain/sei-cosmos/testutil/testdata" + sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types" +) + +// serveRateLimitedQuery exposes the testdata Query service behind registry on a +// real grpc.Server and returns a client dialled to it. +// +// Registration goes through BaseApp.RegisterGRPCServer, the same call +// StartGRPCServer makes, so the interceptor is admitted the way it is on a live +// node rather than by invoking its closure directly. +func serveRateLimitedQuery(t *testing.T, registry *ratelimiter.Registry) testdata.QueryClient { + t.Helper() + + app := baseapp.NewBaseApp(t.Name(), dbm.NewMemDB(), nil, nil, &testutil.TestAppOpts{}) + app.MountStores(sdk.NewKVStoreKey("test")) + require.NoError(t, app.LoadLatestVersion()) + + interfaceRegistry := codectypes.NewInterfaceRegistry() + testdata.RegisterInterfaces(interfaceRegistry) + app.SetInterfaceRegistry(interfaceRegistry) + testdata.RegisterQueryServer(app.GRPCQueryRouter(), testdata.QueryImpl{}) + + srv := grpc.NewServer(grpc.ChainUnaryInterceptor(UnaryRateLimitInterceptor(registry))) + 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 TestUnaryRateLimitInterceptor_RegisteredQueryServiceAllowThenReject(t *testing.T) { + client := serveRateLimitedQuery(t, mustNewRegistry(t, cfg(0.001, 1))) + + res, err := client.Echo(context.Background(), &testdata.EchoRequest{Message: "hello"}) + require.NoError(t, err) + require.Equal(t, "hello", res.Message) + + _, err = client.Echo(context.Background(), &testdata.EchoRequest{Message: "hello"}) + require.Error(t, err) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) +} + +func TestUnaryRateLimitInterceptor_RegisteredQueryServiceUnlimited(t *testing.T) { + client := serveRateLimitedQuery(t, mustNewRegistry(t, cfg(1000, 1000))) + + for i := 0; i < 5; i++ { + res, err := client.Echo(context.Background(), &testdata.EchoRequest{Message: "hello"}) + require.NoError(t, err) + require.Equal(t, "hello", res.Message) + } +} diff --git a/sei-cosmos/server/grpc/rate_limit_test.go b/sei-cosmos/server/grpc/rate_limit_test.go new file mode 100644 index 0000000000..56c9e8656b --- /dev/null +++ b/sei-cosmos/server/grpc/rate_limit_test.go @@ -0,0 +1,162 @@ +package grpc + +import ( + "context" + "net" + "testing" + + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/peer" + "google.golang.org/grpc/status" + + "github.com/sei-protocol/sei-chain/ratelimiter" +) + +func cfg(rps float64, burst int, cidrs ...string) ratelimiter.Config { + return ratelimiter.Config{RPS: rps, Burst: burst, TrustedProxyCIDRs: cidrs} +} + +func mustNewRegistry(t *testing.T, c ratelimiter.Config) *ratelimiter.Registry { + t.Helper() + r, err := ratelimiter.New(c) + require.NoError(t, err) + return r +} + +type mockAddr string + +func (a mockAddr) Network() string { return "tcp" } +func (a mockAddr) String() string { return string(a) } + +func grpcCtx(ctx context.Context, peerAddr string, xff ...string) context.Context { + ctx = peer.NewContext(ctx, &peer.Peer{Addr: mockAddr(peerAddr)}) + if len(xff) > 0 { + md := metadata.MD{"x-forwarded-for": xff} + ctx = metadata.NewIncomingContext(ctx, md) + } + return ctx +} + +func TestUnaryRateLimitInterceptor_AllowThenReject(t *testing.T) { + reg := mustNewRegistry(t, cfg(0.001, 1)) + ic := UnaryRateLimitInterceptor(reg) + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return "ok", nil + } + info := &grpc.UnaryServerInfo{FullMethod: "/cosmos.bank.v1beta1.Query/Balance"} + ctx := grpcCtx(t.Context(), "10.0.0.1:9000") + + _, err := ic(ctx, nil, info, handler) + require.NoError(t, err) + + _, err = ic(ctx, nil, info, handler) + require.Error(t, err) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) +} + +func TestUnaryRateLimitInterceptor_PerIPIsolation(t *testing.T) { + reg := mustNewRegistry(t, cfg(0.001, 1)) + ic := UnaryRateLimitInterceptor(reg) + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return "ok", nil + } + info := &grpc.UnaryServerInfo{FullMethod: "/cosmos.bank.v1beta1.Query/Balance"} + + _, err := ic(grpcCtx(t.Context(), "1.1.1.1:9000"), nil, info, handler) + require.NoError(t, err) + _, err = ic(grpcCtx(t.Context(), "1.1.1.1:9000"), nil, info, handler) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) + + _, err = ic(grpcCtx(t.Context(), "2.2.2.2:9000"), nil, info, handler) + require.NoError(t, err) +} + +func TestUnaryRateLimitInterceptor_TrustedProxyXFF(t *testing.T) { + reg := mustNewRegistry(t, cfg(0.001, 1, "10.0.0.0/8")) + ic := UnaryRateLimitInterceptor(reg) + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return "ok", nil + } + info := &grpc.UnaryServerInfo{FullMethod: "/cosmos.bank.v1beta1.Query/Balance"} + ctx := grpcCtx(t.Context(), "10.0.0.2:9000", "203.0.113.5, 10.0.0.2") + + _, err := ic(ctx, nil, info, handler) + require.NoError(t, err) + _, err = ic(ctx, nil, info, handler) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) +} + +type mockServerStream struct { + grpc.ServerStream + ctx context.Context +} + +func (m mockServerStream) Context() context.Context { return m.ctx } + +func TestStreamRateLimitInterceptor_AllowThenReject(t *testing.T) { + reg := mustNewRegistry(t, cfg(0.001, 1)) + ic := StreamRateLimitInterceptor(reg) + handler := func(srv interface{}, stream grpc.ServerStream) error { + return nil + } + info := &grpc.StreamServerInfo{FullMethod: "/cosmos.bank.v1beta1.Query/Balance"} + stream := mockServerStream{ctx: grpcCtx(t.Context(), "10.0.0.1:9000")} + + require.NoError(t, ic(nil, stream, info, handler)) + + stream = mockServerStream{ctx: grpcCtx(t.Context(), "10.0.0.1:9000")} + err := ic(nil, stream, info, handler) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) +} + +func TestUnaryRateLimitInterceptor_ZeroRPSBypassesBucket(t *testing.T) { + reg := mustNewRegistry(t, cfg(0, 10)) + ic := UnaryRateLimitInterceptor(reg) + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return "ok", nil + } + info := &grpc.UnaryServerInfo{FullMethod: "/cosmos.bank.v1beta1.Query/Balance"} + ctx := grpcCtx(t.Context(), "10.0.0.1:9000") + + for range 10 { + _, err := ic(ctx, nil, info, handler) + require.NoError(t, err) + } +} + +func TestUnaryRateLimitInterceptor_NoPeerStillRateLimited(t *testing.T) { + reg := mustNewRegistry(t, cfg(0.001, 1)) + ic := UnaryRateLimitInterceptor(reg) + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return "ok", nil + } + info := &grpc.UnaryServerInfo{FullMethod: "/cosmos.bank.v1beta1.Query/Balance"} + ctx := t.Context() + + _, err := ic(ctx, nil, info, handler) + require.NoError(t, err) + _, err = ic(ctx, nil, info, handler) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) +} + +func TestUnaryRateLimitInterceptor_RealTCPAddr(t *testing.T) { + reg := mustNewRegistry(t, cfg(0.001, 1)) + ic := UnaryRateLimitInterceptor(reg) + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return "ok", nil + } + info := &grpc.UnaryServerInfo{FullMethod: "/cosmos.bank.v1beta1.Query/Balance"} + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = ln.Close() }) + ctx := peer.NewContext(t.Context(), &peer.Peer{Addr: ln.Addr()}) + + _, err = ic(ctx, nil, info, handler) + require.NoError(t, err) + _, err = ic(ctx, nil, info, handler) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) +} diff --git a/sei-cosmos/server/grpc/server.go b/sei-cosmos/server/grpc/server.go index cf24007c15..e7550692a8 100644 --- a/sei-cosmos/server/grpc/server.go +++ b/sei-cosmos/server/grpc/server.go @@ -10,6 +10,7 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/keepalive" + "github.com/sei-protocol/sei-chain/ratelimiter" "github.com/sei-protocol/sei-chain/sei-cosmos/client" "github.com/sei-protocol/sei-chain/sei-cosmos/server/config" "github.com/sei-protocol/sei-chain/sei-cosmos/server/grpc/gogoreflection" @@ -25,7 +26,7 @@ func StartGRPCServer(clientCtx client.Context, app types.Application, cfg config maxRecvMsgSize = config.DefaultGRPCMaxRecvMsgSize } - grpcSrv := grpc.NewServer( + serverOpts := []grpc.ServerOption{ grpc.MaxConcurrentStreams(100), // MaxRecvMsgSize bounds per-request memory allocation before the rate // limiter fires, preventing an oversized request from exhausting memory. @@ -41,7 +42,19 @@ func StartGRPCServer(clientCtx client.Context, app types.Application, cfg config MinTime: cfg.KeepaliveMinTime, PermitWithoutStream: cfg.KeepalivePermitWithoutStream, }), - ) + } + if cfg.RateLimitingEnabled { + rateLimitRegistry, err := ratelimiter.New(cfg.RateLimiterConfig()) + if err != nil { + return nil, fmt.Errorf("grpc rate limiter: %w", err) + } + serverOpts = append(serverOpts, + grpc.ChainUnaryInterceptor(UnaryRateLimitInterceptor(rateLimitRegistry)), + grpc.ChainStreamInterceptor(StreamRateLimitInterceptor(rateLimitRegistry)), + ) + } + + grpcSrv := grpc.NewServer(serverOpts...) app.RegisterGRPCServer(grpcSrv) // reflection allows consumers to build dynamic clients that can write // to any cosmos-sdk application without relying on application packages at compile time