diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 18c98cfbc..2dcfb8dc6 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -66,7 +66,7 @@ jobs: run: make test-fast PKG_EXCLUDE_REGEX='^github\.com/erpc/erpc/(cmd|test|erpc)($$|/)' - name: Test erpc critical paths run: | - go test ./erpc -run '^(TestConsensusPolicy|TestConsensusPolicy_DSLScenarios)$' -count=1 -parallel 1 -timeout 3m -v + go test ./erpc -run '^(TestConsensusPolicy|TestConsensusPolicy_DSLScenarios|TestBridgeWebsocketsProxiesBidirectionally|TestParseUrlPathAcceptsWebsocketUpgrade|TestWebsocketOriginAllowedUsesProjectCORS|TestAuthorizeWebsocketFrameEnforcesMethodFilters|TestDialWebsocketUpstreamFailsOver|TestWebsocketManagerLimitsAndShutsDownConnections)$' -count=1 -parallel 1 -timeout 3m -v make test-fallback-config - name: Validate Go modules run: git diff --exit-code -- go.mod go.sum diff --git a/auth/registry.go b/auth/registry.go index 4244b6034..7704cd9df 100644 --- a/auth/registry.go +++ b/auth/registry.go @@ -44,6 +44,15 @@ func NewAuthRegistry(appCtx context.Context, logger *zerolog.Logger, projectId s // Authenticate checks the authentication payload against all registered strategies func (r *AuthRegistry) Authenticate(ctx context.Context, req *common.NormalizedRequest, method string, ap *AuthPayload) (*common.User, error) { + return r.authenticate(ctx, req, method, ap, false) +} + +// AuthenticateWebsocket authenticates only against strategies explicitly enabled for WebSocket access. +func (r *AuthRegistry) AuthenticateWebsocket(ctx context.Context, req *common.NormalizedRequest, method string, ap *AuthPayload) (*common.User, error) { + return r.authenticate(ctx, req, method, ap, true) +} + +func (r *AuthRegistry) authenticate(ctx context.Context, req *common.NormalizedRequest, method string, ap *AuthPayload, requireWebsocket bool) (*common.User, error) { if ap == nil { return nil, common.NewErrAuthUnauthorized("n/a", "auth payload is nil") } @@ -56,7 +65,11 @@ func (r *AuthRegistry) Authenticate(ctx context.Context, req *common.NormalizedR var errs []error for _, az := range r.strategies { - if !az.shouldApplyToMethod(method) { + if requireWebsocket && !az.cfg.AllowWebsocket { + continue + } + + if (!requireWebsocket || method != "websocket_connect") && !az.shouldApplyToMethod(method) { continue } @@ -69,6 +82,10 @@ func (r *AuthRegistry) Authenticate(ctx context.Context, req *common.NormalizedR errs = append(errs, err) continue } + if requireWebsocket && user != nil && user.AuthFailOpen { + errs = append(errs, common.NewErrAuthUnauthorized("database", "WebSocket access is disabled during database auth fail-open")) + continue + } // Attach user to the request early so downstream labels (user/agent) can be populated if user != nil && req != nil { diff --git a/auth/registry_websocket_test.go b/auth/registry_websocket_test.go new file mode 100644 index 000000000..3375b5047 --- /dev/null +++ b/auth/registry_websocket_test.go @@ -0,0 +1,47 @@ +package auth + +import ( + "context" + "testing" + + "github.com/erpc/erpc/common" + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" +) + +func TestAuthRegistryAuthenticateWebsocketRequiresExplicitAccess(t *testing.T) { + t.Parallel() + logger := zerolog.Nop() + registry, err := NewAuthRegistry(context.Background(), &logger, "test", &common.AuthConfig{ + Strategies: []*common.AuthStrategyConfig{ + { + Type: common.AuthTypeSecret, + Secret: &common.SecretStrategyConfig{Id: "http-only", Value: "denied-key"}, + }, + { + Type: common.AuthTypeSecret, + AllowWebsocket: true, + IgnoreMethods: []string{"*"}, + AllowMethods: []string{"eth_subscribe"}, + Secret: &common.SecretStrategyConfig{Id: "websocket", Value: "allowed-key"}, + }, + }, + }, nil) + require.NoError(t, err) + + req := common.NewNormalizedRequest([]byte(`{"jsonrpc":"2.0","id":1,"method":"eth_subscribe","params":["newHeads"]}`)) + + _, err = registry.AuthenticateWebsocket(context.Background(), req, "eth_subscribe", &AuthPayload{ + Type: common.AuthTypeSecret, Secret: &SecretPayload{Value: "denied-key"}, + }) + require.Error(t, err) + require.True(t, common.HasErrorCode(err, common.ErrCodeAuthUnauthorized)) + + user, err := registry.AuthenticateWebsocket(context.Background(), req, "eth_subscribe", &AuthPayload{ + Type: common.AuthTypeSecret, Secret: &SecretPayload{Value: "allowed-key"}, + }) + require.NoError(t, err) + require.Equal(t, "websocket", user.Id) + _, err = registry.AuthenticateWebsocket(context.Background(), req, "eth_sendRawTransaction", &AuthPayload{Type: common.AuthTypeSecret, Secret: &SecretPayload{Value: "allowed-key"}}) + require.Error(t, err) +} diff --git a/auth/strategy_database.go b/auth/strategy_database.go index 1e83383e0..72b6740dd 100644 --- a/auth/strategy_database.go +++ b/auth/strategy_database.go @@ -608,7 +608,7 @@ func (s *DatabaseStrategy) buildFailOpenUser() *common.User { if s.cfg == nil || s.cfg.FailOpen == nil || !s.cfg.FailOpen.Enabled { return nil } - u := &common.User{Id: s.cfg.FailOpen.UserId} + u := &common.User{Id: s.cfg.FailOpen.UserId, AuthFailOpen: true} if s.cfg.FailOpen.RateLimitBudget != "" { u.RateLimitBudget = s.cfg.FailOpen.RateLimitBudget } diff --git a/auth/strategy_database_test.go b/auth/strategy_database_test.go index 7afcc18c1..9e600f078 100644 --- a/auth/strategy_database_test.go +++ b/auth/strategy_database_test.go @@ -486,6 +486,7 @@ func TestAuthenticate_FastPathDuringOutage(t *testing.T) { require.NoError(t, err) require.NotNil(t, u) assert.Equal(t, "emergency-failopen", u.Id) + assert.True(t, u.AuthFailOpen) assert.Equal(t, int64(1), fc.getCalls.Load()) assert.True(t, s.connectorDown.Load(), "first failure must latch connectorDown") diff --git a/common/config.go b/common/config.go index e39d3d6c1..4b85c82bd 100644 --- a/common/config.go +++ b/common/config.go @@ -160,6 +160,7 @@ type ServerConfig struct { TrustedIPForwarders []string `yaml:"trustedIPForwarders,omitempty" json:"trustedIPForwarders"` TrustedIPHeaders []string `yaml:"trustedIPHeaders,omitempty" json:"trustedIPHeaders"` ResponseHeaders map[string]string `yaml:"responseHeaders,omitempty" json:"responseHeaders"` + Websocket *WebsocketServerConfig `yaml:"websocket,omitempty" json:"websocket,omitempty"` // ExecutionHeaders controls the per-request diagnostic headers // (X-ERPC-Attempts, X-ERPC-Upstreams-Tried, etc.) that expose how @@ -169,6 +170,13 @@ type ServerConfig struct { ExecutionHeaders *ExecutionHeadersMode `yaml:"executionHeaders,omitempty" json:"executionHeaders" tstype:"ExecutionHeadersMode"` } +type WebsocketServerConfig struct { + DialTimeout *Duration `yaml:"dialTimeout,omitempty" json:"dialTimeout" tstype:"Duration"` + IdleTimeout *Duration `yaml:"idleTimeout,omitempty" json:"idleTimeout" tstype:"Duration"` + MaxLifetime *Duration `yaml:"maxLifetime,omitempty" json:"maxLifetime" tstype:"Duration"` + MaxConnectionsPerUser *int `yaml:"maxConnectionsPerUser,omitempty" json:"maxConnectionsPerUser"` +} + // ExecutionHeadersMode controls how much per-request execution detail is // exposed in HTTP response headers. type ExecutionHeadersMode string @@ -823,6 +831,7 @@ type UpstreamConfig struct { VendorName string `yaml:"vendorName,omitempty" json:"vendorName"` Endpoint string `yaml:"endpoint,omitempty" json:"endpoint"` + WebsocketEndpoint string `yaml:"websocketEndpoint,omitempty" json:"websocketEndpoint,omitempty"` Evm *EvmUpstreamConfig `yaml:"evm,omitempty" json:"evm"` JsonRpc *JsonRpcUpstreamConfig `yaml:"jsonRpc,omitempty" json:"jsonRpc"` Grpc *GrpcUpstreamConfig `yaml:"grpc,omitempty" json:"grpc"` @@ -1139,10 +1148,12 @@ func (c *UpstreamIntegrityEthGetBlockReceiptsConfig) Copy() *UpstreamIntegrityEt func (u *UpstreamConfig) MarshalJSON() ([]byte, error) { type UJAlias UpstreamConfig return sonic.Marshal(&struct { - Endpoint string `json:"endpoint"` + Endpoint string `json:"endpoint"` + WebsocketEndpoint string `json:"websocketEndpoint,omitempty"` *UJAlias }{ - Endpoint: util.RedactEndpoint(u.Endpoint), + Endpoint: util.RedactEndpoint(u.Endpoint), + WebsocketEndpoint: util.RedactEndpoint(u.WebsocketEndpoint), UJAlias: (*UJAlias)(u), }) } @@ -1151,6 +1162,7 @@ func (u *UpstreamConfig) MarshalYAML() (interface{}, error) { type UYAlias UpstreamConfig cp := *u cp.Endpoint = util.RedactEndpoint(u.Endpoint) + cp.WebsocketEndpoint = util.RedactEndpoint(u.WebsocketEndpoint) return (*UYAlias)(&cp), nil } @@ -2783,6 +2795,7 @@ type AuthStrategyConfig struct { IgnoreMethods []string `yaml:"ignoreMethods,omitempty" json:"ignoreMethods,omitempty"` AllowMethods []string `yaml:"allowMethods,omitempty" json:"allowMethods,omitempty"` RateLimitBudget string `yaml:"rateLimitBudget,omitempty" json:"rateLimitBudget,omitempty"` + AllowWebsocket bool `yaml:"allowWebsocket,omitempty" json:"allowWebsocket,omitempty"` Type AuthType `yaml:"type" json:"type" tstype:"TsAuthType"` Network *NetworkStrategyConfig `yaml:"network,omitempty" json:"network,omitempty"` diff --git a/common/config_redaction_test.go b/common/config_redaction_test.go index 8f8d36a32..7b9c5eba5 100644 --- a/common/config_redaction_test.go +++ b/common/config_redaction_test.go @@ -43,3 +43,16 @@ func assertGrpcConfigSecretsRedacted(t *testing.T, text string) { assert.Contains(t, text, "REDACTED") assert.Contains(t, text, "redacted=") } + + +func TestUpstreamConfigRedactsWebsocketEndpoint(t *testing.T) { + cfg := &UpstreamConfig{Id: "websocket-secure", Endpoint: "https://example.internal/rpc?token=http-secret", WebsocketEndpoint: "wss://example.internal/private/ws?token=websocket-secret"} + jsonBytes, err := json.Marshal(cfg); require.NoError(t, err) + sonicBytes, err := SonicCfg.Marshal(cfg); require.NoError(t, err) + yamlBytes, err := yaml.Marshal(cfg); require.NoError(t, err) + for _, text := range []string{string(jsonBytes), string(sonicBytes), string(yamlBytes)} { + assert.NotContains(t, text, "http-secret") + assert.NotContains(t, text, "websocket-secret") + assert.Contains(t, text, "redacted=") + } +} diff --git a/common/defaults.go b/common/defaults.go index 55fac11c5..03397ddf2 100644 --- a/common/defaults.go +++ b/common/defaults.go @@ -897,6 +897,11 @@ func (s *ServerConfig) SetDefaults() error { if s.EnableGzip == nil { s.EnableGzip = util.BoolPtr(true) } + if s.Websocket == nil { s.Websocket = &WebsocketServerConfig{} } + if s.Websocket.DialTimeout == nil { d := Duration(10 * time.Second); s.Websocket.DialTimeout = &d } + if s.Websocket.IdleTimeout == nil { d := Duration(5 * time.Minute); s.Websocket.IdleTimeout = &d } + if s.Websocket.MaxLifetime == nil { d := Duration(24 * time.Hour); s.Websocket.MaxLifetime = &d } + if s.Websocket.MaxConnectionsPerUser == nil { s.Websocket.MaxConnectionsPerUser = util.IntPtr(10) } if s.WaitBeforeShutdown == nil { d := Duration(10 * time.Second) s.WaitBeforeShutdown = &d @@ -1757,6 +1762,9 @@ func (u *UpstreamConfig) ApplyDefaults(defaults *UpstreamConfig) error { if u.Endpoint == "" { u.Endpoint = defaults.Endpoint } + if u.WebsocketEndpoint == "" { + u.WebsocketEndpoint = defaults.WebsocketEndpoint + } if u.Type == "" { u.Type = defaults.Type } diff --git a/common/user.go b/common/user.go index 27c9e00d1..2081daae2 100644 --- a/common/user.go +++ b/common/user.go @@ -3,4 +3,5 @@ package common type User struct { Id string RateLimitBudget string + AuthFailOpen bool } diff --git a/common/validation.go b/common/validation.go index 4e01e8d46..240955f1a 100644 --- a/common/validation.go +++ b/common/validation.go @@ -110,6 +110,12 @@ func (s *ServerConfig) Validate() error { if s.MaxBatchConcurrency != nil && *s.MaxBatchConcurrency <= 0 { return fmt.Errorf("server.maxBatchConcurrency must be > 0") } + if s.Websocket != nil { + if s.Websocket.DialTimeout == nil || *s.Websocket.DialTimeout <= 0 { return fmt.Errorf("server.websocket.dialTimeout must be > 0") } + if s.Websocket.IdleTimeout == nil || *s.Websocket.IdleTimeout <= 0 { return fmt.Errorf("server.websocket.idleTimeout must be > 0") } + if s.Websocket.MaxLifetime == nil || *s.Websocket.MaxLifetime <= 0 { return fmt.Errorf("server.websocket.maxLifetime must be > 0") } + if s.Websocket.MaxConnectionsPerUser == nil || *s.Websocket.MaxConnectionsPerUser <= 0 { return fmt.Errorf("server.websocket.maxConnectionsPerUser must be > 0") } + } // Validate trusted IP forwarders if provided (IPs or CIDRs). Support legacy + new field for _, entry := range s.TrustedIPForwarders { @@ -966,6 +972,12 @@ func (u *UpstreamConfig) Validate(c *Config, skipEndpointCheck bool) error { if !skipEndpointCheck && u.Endpoint == "" { return fmt.Errorf("upstream.*.endpoint is required") } + if u.WebsocketEndpoint != "" { + parsed, err := url.Parse(u.WebsocketEndpoint) + if err != nil || (parsed.Scheme != "ws" && parsed.Scheme != "wss") { + return fmt.Errorf("upstream.*.websocketEndpoint must use ws:// or wss://") + } + } if err := validateCapabilityTags("upstream.*.capabilities", u.Capabilities); err != nil { return err } diff --git a/docs/pages/config/auth.mdx b/docs/pages/config/auth.mdx index d9a882010..bf47bcdae 100644 --- a/docs/pages/config/auth.mdx +++ b/docs/pages/config/auth.mdx @@ -201,6 +201,7 @@ Authenticate fast paths (in evaluation order): | `strategies[*].ignoreMethods` | `[]string` | `nil` | Wildcard patterns (supports `*`, `\|`, `&`, `!`). Applied before `allowMethods`. | | `strategies[*].allowMethods` | `[]string` | `nil` | Overrides `ignoreMethods`; any matching allow re-enables the strategy for that method. | | `strategies[*].rateLimitBudget` | `string` | `""` | Strategy-level budget ID. Overridden by per-user budget when non-empty. | +| `strategies[*].allowWebsocket` | `bool` | `false` | Explicitly permits this strategy to authenticate WebSocket upgrades. Keep false for HTTP-only API keys. WebSocket access is denied unless a matching strategy opts in. | **`secret` strategy — `SecretStrategyConfig`** diff --git a/docs/pages/config/projects/upstreams.mdx b/docs/pages/config/projects/upstreams.mdx index 4c85f6d83..cfc40f352 100644 --- a/docs/pages/config/projects/upstreams.mdx +++ b/docs/pages/config/projects/upstreams.mdx @@ -97,6 +97,31 @@ https://docs.erpc.cloud/config/projects/upstreams.llms.txt`} +### WebSocket subscriptions + +WebSocket transport is opt-in at both ends. Configure `websocketEndpoint` on an upstream that supports `eth_subscribe`, then set `allowWebsocket: true` only on the auth strategies/API keys that may open WebSocket connections. Other keys continue to receive HTTP access only. + +```yaml +projects: + - id: main + auth: + strategies: + - type: secret + allowWebsocket: true + secret: + id: subscriptions-client + value: ${SUBSCRIPTIONS_API_KEY} + - type: secret + secret: + id: http-only-client + value: ${HTTP_API_KEY} + upstreams: + - endpoint: https://eth-mainnet.example/v2/key + websocketEndpoint: wss://eth-mainnet.example/v2/key +``` + +WebSocket connections use the normal `///` route with the same `secret`, `X-ERPC-Secret-Token`, Basic, Bearer, SIWE, or network credentials as HTTP. eRPC checks method filters and project/auth rate limits on every client frame, including each request in a JSON-RPC batch; binary and malformed frames are rejected. It tries eligible `eth_subscribe` upstreams in current policy order until one connects. WebSocket traffic is stateful: it is not cached, retried, hedged, or failed over after connection establishment. Connections are bounded by per-user concurrency, idle, maximum-lifetime, and dial limits and are closed during graceful shutdown. Database-auth fail-open users are rejected for WebSocket upgrades. + ### How it works **Config-time pipeline.** When eRPC loads a project, each upstream in `upstreams[]` diff --git a/docs/pages/config/server.mdx b/docs/pages/config/server.mdx index a09ca48c1..a5fec7c19 100644 --- a/docs/pages/config/server.mdx +++ b/docs/pages/config/server.mdx @@ -140,6 +140,10 @@ All fields live under `server.` unless noted. `Duration` accepts Go duration str | `server.readTimeout` | `*Duration` | `30s` | `http.Server.ReadTimeout` — covers reading headers and body. | | `server.writeTimeout` | `*Duration` | `120s` | `http.Server.WriteTimeout` — covers writing the response. The entire response is buffered by `TimeoutHandler` before reaching the socket, so this only matters at final flush. | | `server.enableGzip` | `*bool` | `true` | Wraps handler in `gzipHandler` for response compression. Inbound gzip is always accepted regardless of this flag. | +| `server.websocket.dialTimeout` | `*Duration` | `10s` | Per-upstream WebSocket dial timeout before trying the next eligible candidate. | +| `server.websocket.idleTimeout` | `*Duration` | `5m` | Closes a WebSocket connection when neither side sends a frame for this duration. | +| `server.websocket.maxLifetime` | `*Duration` | `24h` | Hard lifetime for a WebSocket connection. | +| `server.websocket.maxConnectionsPerUser` | `*int` | `10` | Maximum concurrent WebSocket connections per authenticated user/API-key ID. | | `server.tls.enabled` | `bool` | `false` | When true, both listeners use `ListenAndServeTLS` with TLS 1.2 minimum; gRPC also uses TLS. Disables h2c on the shared port. | | `server.tls.certFile` | `string` | `""` | PEM cert path. Load failure → "failed to load TLS certificate and key". | | `server.tls.keyFile` | `string` | `""` | PEM key path. | diff --git a/erpc/http_server.go b/erpc/http_server.go index e437fe4e7..6874cadff 100644 --- a/erpc/http_server.go +++ b/erpc/http_server.go @@ -57,6 +57,7 @@ type HttpServer struct { trustedForwarderIPs map[string]struct{} trustedIPHeaders []string resolvedResponseHeaders map[string]string + websocketManager *websocketManager } func logHttpRequestBody(lg *zerolog.Logger, body []byte) { @@ -145,7 +146,8 @@ func NewHttpServer( adminCfg: adminCfg, erpc: erpc, draining: &draining, - gzipPool: gzipPool, + gzipPool: gzipPool, + websocketManager: newWebsocketManager(cfg.Websocket), } if cfg != nil { @@ -200,16 +202,25 @@ func NewHttpServer( } } - h := srv.createRequestHandler() + requestHandler := srv.createRequestHandler() + normalHandler := requestHandler if cfg.EnableGzip != nil && *cfg.EnableGzip { - h = gzipHandler(h) + normalHandler = gzipHandler(normalHandler) } - // Create handler with timeout - httpHandler := TimeoutHandler(logger, h, reqMaxTimeout) - handlerV4 := httpHandler - handlerV6 := httpHandler + // WebSocket upgrades must bypass the buffering timeout and gzip wrappers, + // which do not implement http.Hijacker. + httpHandler := TimeoutHandler(logger, normalHandler, reqMaxTimeout) + websocketAwareHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if websocketUpgradeRequested(r) { + requestHandler.ServeHTTP(w, r) + return + } + httpHandler.ServeHTTP(w, r) + }) + handlerV4 := http.Handler(websocketAwareHandler) + handlerV6 := http.Handler(websocketAwareHandler) if grpcSharesHttpV4(cfg) { sharedGrpcServer, err := NewGrpcServer(ctx, logger, cfg, erpc) @@ -222,7 +233,7 @@ func NewHttpServer( sharedGrpcServer.server.ServeHTTP(w, r) return } - httpHandler.ServeHTTP(w, r) + websocketAwareHandler.ServeHTTP(w, r) }) if cfg.TLS == nil || !cfg.TLS.Enabled { handlerV4 = h2c.NewHandler(handlerV4, &http2.Server{}) @@ -399,6 +410,11 @@ func (s *HttpServer) createRequestHandler() http.Handler { } } + if websocketUpgradeRequested(r) { + s.handleWebsocket(httpCtx, w, r, project, architecture, chainId) + return + } + // Handle gzipped request bodies var bodyReader io.Reader = r.Body if r.Header.Get("Content-Encoding") == "gzip" { @@ -990,6 +1006,7 @@ func (s *HttpServer) parseUrlPath( isPost := r.Method == http.MethodPost isOptions := r.Method == http.MethodOptions + isWebsocket := websocketUpgradeRequested(r) // Initialize with preselected values projectId = preSelectedProjectId @@ -1006,7 +1023,7 @@ func (s *HttpServer) parseUrlPath( isHealthCheck = true segments = segments[:len(segments)-1] // Remove healthcheck segment } else if len(segments) == 0 || (len(segments) == 1 && segments[0] == "") { - if !(isPost || isOptions) { + if !(isPost || isOptions || isWebsocket) { isHealthCheck = true segments = nil } @@ -1162,7 +1179,7 @@ func (s *HttpServer) parseUrlPath( return "", "", "", false, false, common.NewErrInvalidUrlPath("architecture is not valid (must be 'evm')", ps) } - if !isPost && !isOptions { + if !isPost && !isOptions && !isWebsocket { isHealthCheck = true } @@ -1814,6 +1831,7 @@ func (s *HttpServer) createTLSConfig() (*tls.Config, error) { func (s *HttpServer) Shutdown(logger *zerolog.Logger) error { logger.Info().Msg("stopping http servers...") + if s.websocketManager != nil { s.websocketManager.shutdown() } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() diff --git a/erpc/projects.go b/erpc/projects.go index d9ef9d7c3..ab88527d2 100644 --- a/erpc/projects.go +++ b/erpc/projects.go @@ -98,6 +98,28 @@ func (p *PreparedProject) AuthenticateConsumer(ctx context.Context, req *common. return nil, nil } +func (p *PreparedProject) AuthenticateWebsocket(ctx context.Context, req *common.NormalizedRequest, method string, ap *auth.AuthPayload) (*common.User, error) { + if p.consumerAuthRegistry != nil { + return p.consumerAuthRegistry.AuthenticateWebsocket(ctx, req, method, ap) + } + return nil, common.NewErrAuthUnauthorized("n/a", "WebSocket access requires an explicitly enabled auth strategy") +} + +func (p *PreparedProject) SupportsMethod(method string) (bool, error) { + allowed := true + for _, pattern := range p.Config.IgnoreMethods { + match, err := common.WildcardMatch(pattern, method) + if err != nil { return false, err } + if match { allowed = false; break } + } + for _, pattern := range p.Config.AllowMethods { + match, err := common.WildcardMatch(pattern, method) + if err != nil { return false, err } + if match { allowed = true; break } + } + return allowed, nil +} + func (p *PreparedProject) Forward(ctx context.Context, networkId string, nq *common.NormalizedRequest) (*common.NormalizedResponse, error) { start := time.Now() ctx, span := common.StartDetailSpan(ctx, "Project.Forward") diff --git a/erpc/websocket_proxy.go b/erpc/websocket_proxy.go new file mode 100644 index 000000000..9ce79386a --- /dev/null +++ b/erpc/websocket_proxy.go @@ -0,0 +1,380 @@ +package erpc + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/url" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/coder/websocket" + "github.com/erpc/erpc/auth" + "github.com/erpc/erpc/common" +) + +const websocketConnectMethod = "websocket_connect" + +var errWebsocketPolicyViolation = errors.New("WebSocket request rejected") + +type websocketManager struct { + cfg *common.WebsocketServerConfig + mu sync.Mutex + connections map[*websocket.Conn]context.CancelFunc + byUser map[string]int +} + +func newWebsocketManager(cfg *common.WebsocketServerConfig) *websocketManager { + if cfg == nil { + cfg = &common.WebsocketServerConfig{} + } + if cfg.DialTimeout == nil { + d := common.Duration(10 * time.Second) + cfg.DialTimeout = &d + } + if cfg.IdleTimeout == nil { + d := common.Duration(5 * time.Minute) + cfg.IdleTimeout = &d + } + if cfg.MaxLifetime == nil { + d := common.Duration(24 * time.Hour) + cfg.MaxLifetime = &d + } + if cfg.MaxConnectionsPerUser == nil { + v := 10 + cfg.MaxConnectionsPerUser = &v + } + return &websocketManager{cfg: cfg, connections: make(map[*websocket.Conn]context.CancelFunc), byUser: make(map[string]int)} +} + +func (m *websocketManager) acquire(userID string) (func(), bool) { + m.mu.Lock() + defer m.mu.Unlock() + if m.byUser[userID] >= *m.cfg.MaxConnectionsPerUser { + return nil, false + } + m.byUser[userID]++ + return func() { + m.mu.Lock() + defer m.mu.Unlock() + m.byUser[userID]-- + if m.byUser[userID] == 0 { + delete(m.byUser, userID) + } + }, true +} + +func (m *websocketManager) register(conn *websocket.Conn, cancel context.CancelFunc) func() { + m.mu.Lock() + m.connections[conn] = cancel + m.mu.Unlock() + return func() { + m.mu.Lock() + delete(m.connections, conn) + m.mu.Unlock() + } +} + +func (m *websocketManager) shutdown() { + m.mu.Lock() + connections := make(map[*websocket.Conn]context.CancelFunc, len(m.connections)) + for conn, cancel := range m.connections { + connections[conn] = cancel + } + m.mu.Unlock() + for conn, cancel := range connections { + cancel() + if conn != nil { + _ = conn.CloseNow() + } + } +} + +func headerHasToken(header http.Header, name, want string) bool { + for _, value := range header.Values(name) { + for _, token := range strings.Split(value, ",") { + if strings.EqualFold(strings.TrimSpace(token), want) { + return true + } + } + } + return false +} + +func websocketUpgradeRequested(r *http.Request) bool { + return r != nil && r.Method == http.MethodGet && headerHasToken(r.Header, "Upgrade", "websocket") && headerHasToken(r.Header, "Connection", "upgrade") +} + +func (s *HttpServer) handleWebsocket(requestCtx context.Context, w http.ResponseWriter, r *http.Request, project *PreparedProject, architecture, chainID string) { + if project == nil { + http.Error(w, "project not found", http.StatusNotFound) + return + } + if !websocketOriginAllowed(r, project.Config.CORS) { + http.Error(w, "WebSocket origin denied", http.StatusForbidden) + return + } + + networkID := fmt.Sprintf("%s:%s", architecture, chainID) + network, err := project.GetNetwork(requestCtx, networkID) + if err != nil { + http.Error(w, "network not found", http.StatusNotFound) + return + } + candidates := websocketUpstreamCandidates(network) + if len(candidates) == 0 { + http.Error(w, "no eligible WebSocket upstream configured", http.StatusServiceUnavailable) + return + } + + nq := common.NewNormalizedRequest([]byte(`{"jsonrpc":"2.0","id":null,"method":"websocket_connect"}`)) + nq.SetClientIP(s.resolveRealClientIP(r)) + nq.SetNetwork(network) + payload, err := auth.NewPayloadFromHttp(websocketConnectMethod, r.RemoteAddr, r.Header, r.URL.Query()) + if err != nil { + http.Error(w, "invalid authentication", http.StatusUnauthorized) + return + } + user, err := project.AuthenticateWebsocket(requestCtx, nq, websocketConnectMethod, payload) + if err != nil || user == nil { + http.Error(w, "WebSocket access denied", http.StatusUnauthorized) + return + } + nq.SetUser(user) + + releaseUser, ok := s.websocketManager.acquire(user.Id) + if !ok { + http.Error(w, "WebSocket connection limit exceeded", http.StatusTooManyRequests) + return + } + defer releaseUser() + + controller := http.NewResponseController(w) + _ = controller.SetReadDeadline(time.Time{}) + _ = controller.SetWriteDeadline(time.Time{}) + clientConn, err := websocket.Accept(w, r, &websocket.AcceptOptions{InsecureSkipVerify: true, CompressionMode: websocket.CompressionDisabled}) + if err != nil { + return + } + defer clientConn.Close(websocket.StatusNormalClosure, "connection closed") + clientConn.SetReadLimit(8 * 1024 * 1024) + + connectionCtx, cancel := context.WithTimeout(s.appCtx, s.websocketManager.cfg.MaxLifetime.Duration()) + defer cancel() + unregister := s.websocketManager.register(clientConn, cancel) + defer unregister() + + upstreamConn, upstreamID, err := dialWebsocketUpstream(connectionCtx, candidates, s.websocketManager.cfg.DialTimeout.Duration()) + if err != nil { + s.logger.Warn().Err(err).Str("projectId", project.Config.Id).Str("networkId", networkID).Msg("failed to connect to WebSocket upstreams") + _ = clientConn.Close(websocket.StatusInternalError, "failed to connect to upstream") + return + } + defer upstreamConn.Close(websocket.StatusNormalClosure, "client disconnected") + upstreamConn.SetReadLimit(8 * 1024 * 1024) + + authorize := func(frameType websocket.MessageType, frame []byte) error { + if err := authorizeWebsocketFrame(connectionCtx, frameType, frame, project, network, payload, nq.ClientIP()); err != nil { + return fmt.Errorf("%w: %v", errWebsocketPolicyViolation, err) + } + return nil + } + if err := bridgeWebsockets(connectionCtx, clientConn, upstreamConn, s.websocketManager.cfg.IdleTimeout.Duration(), authorize); err != nil { + if errors.Is(err, errWebsocketPolicyViolation) { + _ = clientConn.Close(websocket.StatusPolicyViolation, "WebSocket request denied") + } + s.logger.Debug().Err(err).Str("projectId", project.Config.Id).Str("networkId", networkID).Str("upstreamId", upstreamID).Msg("WebSocket proxy closed") + } +} + +type websocketUpstreamCandidate struct { + id string + endpoint string + headers http.Header +} + +func websocketUpstreamCandidates(network *Network) []websocketUpstreamCandidate { + upstreams := network.AllUpstreams() + if len(upstreams) == 0 { + return nil + } + byID := make(map[string]int, len(upstreams)) + for i, upstream := range upstreams { + byID[upstream.Id()] = i + } + orderedIDs := network.PolicyOrderedUpstreams("eth_subscribe") + if len(orderedIDs) == 0 { + orderedIDs = make([]string, 0, len(upstreams)) + for _, upstream := range upstreams { + orderedIDs = append(orderedIDs, upstream.Id()) + } + } + candidates := make([]websocketUpstreamCandidate, 0, len(orderedIDs)) + for _, id := range orderedIDs { + index, ok := byID[id] + if !ok { + continue + } + upstream := upstreams[index] + cfg := upstream.Config() + if cfg.WebsocketEndpoint == "" || (cfg.Shadow != nil && cfg.Shadow.Enabled) || upstream.EvmSyncingState() == common.EvmSyncingStateSyncing { + continue + } + eligible, err := upstream.ShouldHandleMethod("eth_subscribe") + if err != nil || !eligible { + continue + } + parsed, err := url.Parse(cfg.WebsocketEndpoint) + if err != nil || (parsed.Scheme != "ws" && parsed.Scheme != "wss") { + continue + } + headers := make(http.Header) + if cfg.JsonRpc != nil { + for key, value := range cfg.JsonRpc.Headers { + headers.Set(key, value) + } + } + candidates = append(candidates, websocketUpstreamCandidate{id: id, endpoint: cfg.WebsocketEndpoint, headers: headers}) + } + return candidates +} + +func dialWebsocketUpstream(ctx context.Context, candidates []websocketUpstreamCandidate, timeout time.Duration) (*websocket.Conn, string, error) { + var errs []error + for _, candidate := range candidates { + dialCtx, cancel := context.WithTimeout(ctx, timeout) + conn, response, err := websocket.Dial(dialCtx, candidate.endpoint, &websocket.DialOptions{HTTPHeader: candidate.headers}) + cancel() + if err == nil { + return conn, candidate.id, nil + } + if response != nil { + _ = response.Body.Close() + } + errs = append(errs, fmt.Errorf("upstream %s: %w", candidate.id, err)) + } + return nil, "", errors.Join(errs...) +} + +func authorizeWebsocketFrame(ctx context.Context, messageType websocket.MessageType, frame []byte, project *PreparedProject, network *Network, payload *auth.AuthPayload, clientIP string) error { + if messageType != websocket.MessageText { + return fmt.Errorf("binary WebSocket frames are not supported") + } + trimmed := bytes.TrimSpace(frame) + if len(trimmed) == 0 { + return fmt.Errorf("empty WebSocket frame") + } + requests := []json.RawMessage{trimmed} + if trimmed[0] == byte(91) { + if err := common.SonicCfg.Unmarshal(trimmed, &requests); err != nil || len(requests) == 0 { + return fmt.Errorf("invalid JSON-RPC batch") + } + } + for _, raw := range requests { + nq := common.NewNormalizedRequest(raw) + if err := nq.Validate(); err != nil { + return err + } + method, err := nq.Method() + if err != nil { + return err + } + allowed, err := project.SupportsMethod(method) + if err != nil { + return err + } + if !allowed { + return fmt.Errorf("method not supported: %s", method) + } + nq.SetClientIP(clientIP) + nq.SetNetwork(network) + methodPayload := *payload + methodPayload.Method = method + user, err := project.AuthenticateWebsocket(ctx, nq, method, &methodPayload) + if err != nil { + return err + } + nq.SetUser(user) + if err := project.AcquireRateLimitPermit(ctx, nq); err != nil { + return err + } + } + return nil +} + +func websocketOriginAllowed(r *http.Request, cors *common.CORSConfig) bool { + origin := r.Header.Get("Origin") + if origin == "" { + return true + } + if cors == nil { + return false + } + for _, pattern := range cors.AllowedOrigins { + matched, err := common.WildcardMatch(pattern, origin) + if err == nil && matched { + return true + } + } + return false +} + +func bridgeWebsockets(ctx context.Context, clientConn, upstreamConn *websocket.Conn, idleTimeout time.Duration, authorize func(websocket.MessageType, []byte) error) error { + bridgeCtx, cancel := context.WithCancel(ctx) + defer cancel() + var lastActivity atomic.Int64 + lastActivity.Store(time.Now().UnixNano()) + errs := make(chan error, 3) + copyMessages := func(dst, src *websocket.Conn, authorizeFrame bool) { + for { + messageType, frame, err := src.Read(bridgeCtx) + if err != nil { + errs <- err + return + } + lastActivity.Store(time.Now().UnixNano()) + if authorizeFrame && authorize != nil { + if err := authorize(messageType, frame); err != nil { + errs <- err + return + } + } + if err := dst.Write(bridgeCtx, messageType, frame); err != nil { + errs <- err + return + } + } + } + go copyMessages(upstreamConn, clientConn, true) + go copyMessages(clientConn, upstreamConn, false) + go func() { + ticker := time.NewTicker(min(idleTimeout/2, time.Minute)) + defer ticker.Stop() + for { + select { + case <-bridgeCtx.Done(): + return + case <-ticker.C: + if time.Since(time.Unix(0, lastActivity.Load())) >= idleTimeout { + errs <- fmt.Errorf("WebSocket idle timeout exceeded") + return + } + } + } + }() + err := <-errs + cancel() + if errors.Is(err, context.Canceled) { + return nil + } + status := websocket.CloseStatus(err) + if status == websocket.StatusNormalClosure || status == websocket.StatusGoingAway { + return nil + } + return err +} diff --git a/erpc/websocket_proxy_test.go b/erpc/websocket_proxy_test.go new file mode 100644 index 000000000..ceb34671c --- /dev/null +++ b/erpc/websocket_proxy_test.go @@ -0,0 +1,126 @@ +package erpc + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/coder/websocket/wsjson" + "github.com/erpc/erpc/auth" + "github.com/erpc/erpc/common" + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" +) + +func TestBridgeWebsocketsProxiesBidirectionally(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { return } + defer conn.Close(websocket.StatusNormalClosure, "done") + var message map[string]any + require.NoError(t, wsjson.Read(r.Context(), conn, &message)) + require.NoError(t, wsjson.Write(r.Context(), conn, message)) + })) + defer upstream.Close() + proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + clientConn, err := websocket.Accept(w, r, nil) + if err != nil { return } + defer clientConn.Close(websocket.StatusNormalClosure, "done") + upstreamConn, _, err := websocket.Dial(r.Context(), "ws"+strings.TrimPrefix(upstream.URL, "http"), nil) + require.NoError(t, err) + defer upstreamConn.Close(websocket.StatusNormalClosure, "done") + require.NoError(t, bridgeWebsockets(r.Context(), clientConn, upstreamConn, time.Hour, nil)) + })) + defer proxy.Close() + client, _, err := websocket.Dial(ctx, "ws"+strings.TrimPrefix(proxy.URL, "http"), nil) + require.NoError(t, err) + defer client.Close(websocket.StatusNormalClosure, "done") + want := map[string]any{"jsonrpc": "2.0", "id": float64(1), "method": "eth_subscribe", "params": []any{"newHeads"}} + require.NoError(t, wsjson.Write(ctx, client, want)) + var got map[string]any + require.NoError(t, wsjson.Read(ctx, client, &got)) + require.Equal(t, want, got) +} + +func TestParseUrlPathAcceptsWebsocketUpgrade(t *testing.T) { + t.Parallel() + req := httptest.NewRequest(http.MethodGet, "/main/evm/1", nil) + req.Header.Set("Connection", "keep-alive, Upgrade") + req.Header.Set("Upgrade", "websocket") + projectID, architecture, chainID, isAdmin, isHealthCheck, err := (&HttpServer{}).parseUrlPath(req, "", "", "") + require.NoError(t, err) + require.Equal(t, "main", projectID) + require.Equal(t, "evm", architecture) + require.Equal(t, "1", chainID) + require.False(t, isAdmin) + require.False(t, isHealthCheck) +} + +func TestWebsocketOriginAllowedUsesProjectCORS(t *testing.T) { + t.Parallel() + req := httptest.NewRequest(http.MethodGet, "/main/evm/1", nil) + req.Header.Set("Origin", "https://app.example.com") + cors := &common.CORSConfig{AllowedOrigins: []string{"https://*.example.com"}} + require.True(t, websocketOriginAllowed(req, cors)) + req.Header.Set("Origin", "https://evil.example") + require.False(t, websocketOriginAllowed(req, cors)) +} + +func TestAuthorizeWebsocketFrameEnforcesMethodFilters(t *testing.T) { + t.Parallel() + logger := zerolog.Nop() + registry, err := auth.NewAuthRegistry(context.Background(), &logger, "test", &common.AuthConfig{Strategies: []*common.AuthStrategyConfig{{Type: common.AuthTypeSecret, AllowWebsocket: true, IgnoreMethods: []string{"*"}, AllowMethods: []string{"eth_subscribe"}, Secret: &common.SecretStrategyConfig{Id: "subscriptions", Value: "key"}}}}, nil) + require.NoError(t, err) + project := &PreparedProject{Config: &common.ProjectConfig{Id: "test", IgnoreMethods: []string{"*"}, AllowMethods: []string{"eth_subscribe"}}, consumerAuthRegistry: registry} + network := &Network{} + payload := &auth.AuthPayload{Type: common.AuthTypeSecret, Secret: &auth.SecretPayload{Value: "key"}} + allowed := []byte(`{"jsonrpc":"2.0","id":1,"method":"eth_subscribe","params":["newHeads"]}`) + require.NoError(t, authorizeWebsocketFrame(context.Background(), websocket.MessageText, allowed, project, network, payload, "127.0.0.1")) + denied := []byte(`{"jsonrpc":"2.0","id":2,"method":"eth_sendRawTransaction","params":["0x"]}`) + require.ErrorContains(t, authorizeWebsocketFrame(context.Background(), websocket.MessageText, denied, project, network, payload, "127.0.0.1"), "method not supported") + batch := []byte(`[{"jsonrpc":"2.0","id":3,"method":"eth_subscribe","params":["newHeads"]},{"jsonrpc":"2.0","id":4,"method":"eth_sendRawTransaction","params":["0x"]}]`) + require.Error(t, authorizeWebsocketFrame(context.Background(), websocket.MessageText, batch, project, network, payload, "127.0.0.1")) + require.Error(t, authorizeWebsocketFrame(context.Background(), websocket.MessageBinary, allowed, project, network, payload, "127.0.0.1")) +} + +func TestDialWebsocketUpstreamFailsOver(t *testing.T) { + t.Parallel() + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { return } + defer conn.Close(websocket.StatusNormalClosure, "done") + _, _, _ = conn.Read(context.Background()) + })) + defer upstream.Close() + conn, id, err := dialWebsocketUpstream(context.Background(), []websocketUpstreamCandidate{{id: "down", endpoint: "ws://127.0.0.1:1"}, {id: "healthy", endpoint: "ws"+strings.TrimPrefix(upstream.URL, "http")}}, 250*time.Millisecond) + require.NoError(t, err) + require.Equal(t, "healthy", id) + require.NoError(t, conn.Close(websocket.StatusNormalClosure, "done")) +} + +func TestWebsocketManagerLimitsAndShutsDownConnections(t *testing.T) { + t.Parallel() + max := 1 + dial, idle, lifetime := common.Duration(time.Second), common.Duration(time.Minute), common.Duration(time.Hour) + manager := newWebsocketManager(&common.WebsocketServerConfig{DialTimeout: &dial, IdleTimeout: &idle, MaxLifetime: &lifetime, MaxConnectionsPerUser: &max}) + release, ok := manager.acquire("user") + require.True(t, ok) + _, ok = manager.acquire("user") + require.False(t, ok) + release() + _, ok = manager.acquire("user") + require.True(t, ok) + ctx, cancel := context.WithCancel(context.Background()) + manager.mu.Lock() + manager.connections[nil] = cancel + manager.mu.Unlock() + manager.shutdown() + require.Eventually(t, func() bool { return ctx.Err() != nil }, time.Second, 10*time.Millisecond) +}