security: fix audit data-race crashes + blunt block-height poisoning (A6/A8/B1-B3)

.github/workflows/release-artifacts.yml

@@ -22,7 +22,10 @@ jobs:
       GITHUB_TOKEN: ${{ github.token }}
     if: startsWith(github.ref, 'refs/tags/v')
     steps:
-      - uses: rymndhng/release-on-push-action@master
+      # Pinned to an immutable commit SHA (v0.28.0) rather than a mutable branch.
+      # A moving @master ref means a compromised upstream repo would get arbitrary
+      # code execution in CI with GITHUB_TOKEN.
+      - uses: rymndhng/release-on-push-action@aebba2bbce07a9474bf95e8710e5ee8a9e922fe2 # v0.28.0
         with:
           bump_version_scheme: patch
           release_body: ":rocket: Release Notes !:fireworks: "

cmd/main.go

@@ -302,6 +302,7 @@ func main() {
 		WebsocketMessageBufferSize: config.GetRouterConfig().WebsocketMessageBufferSize,
 		ObservationQueue:           observationQueue,
 		DomainCircuitBreaker:       domainCircuitBreaker,
+		WebsocketConnectionLimiter: gateway.NewWebsocketConnectionLimiter(config.GetRouterConfig().MaxConcurrentWebsocketConnections),
 	}
 
 	// Until all components are ready, the `/healthz` endpoint will return a 503 Service

cmd/metrics.go

@@ -14,6 +14,10 @@ func setupMetricsServer(logger polylog.Logger, addr string) (*metrics.Prometheus
 		Logger: logger,
 	}
 
+	// Wire the logger used for the one-time WARN a cardinality guard emits when
+	// it first saturates (otherwise a capped metric silently goes incomplete).
+	metrics.SetCardinalityGuardLogger(logger)
+
 	if err := pmr.ServeMetrics(addr); err != nil {
 		return nil, err
 	}

config/config.schema.yaml

@@ -663,6 +663,12 @@ properties:
       websocket_message_buffer_size:
         description: "Buffer size for websocket messages."
         type: integer
+      max_concurrent_websocket_connections:
+        description: "Maximum number of concurrent live websocket connections per gateway pod (defense-in-depth against goroutine/FD exhaustion). Defaults to 10000; set to a negative value to disable the limit."
+        type: integer
+      max_request_body_bytes:
+        description: "Maximum size in bytes of an HTTP request body PATH will read into memory (prevents OOM from oversized bodies). Defaults to 78643200 (75MB); set to a negative value to disable the limit."
+        type: integer
 
   # Concurrency Configuration (optional)
   concurrency_config:

config/config_test.go

@@ -93,13 +93,15 @@ func Test_LoadGatewayConfigFromYAML(t *testing.T) {
 					},
 				},
 				Router: RouterConfig{
-					Port:                            defaultPort,
-					MaxRequestHeaderBytes:           defaultMaxRequestHeaderBytes,
-					ReadTimeout:                     30 * time.Second,  // Matches example config
-					WriteTimeout:                    30 * time.Second,  // Matches example config
-					IdleTimeout:                     120 * time.Second, // Matches example config
-					SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration,
-					WebsocketMessageBufferSize:      8192, // Matches example config
+					Port:                              defaultPort,
+					MaxRequestHeaderBytes:             defaultMaxRequestHeaderBytes,
+					ReadTimeout:                       30 * time.Second,  // Matches example config
+					WriteTimeout:                      30 * time.Second,  // Matches example config
+					IdleTimeout:                       120 * time.Second, // Matches example config
+					SystemOverheadAllowanceDuration:   defaultSystemOverheadAllowanceDuration,
+					WebsocketMessageBufferSize:        8192, // Matches example config
+					MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections,
+					MaxRequestBodyBytes:               defaultMaxRequestBodyBytes,
 				},
 				Logger: LoggerConfig{
 					Level: "info", // Matches example config
@@ -191,13 +193,15 @@ logger_config:
 					},
 				},
 				Router: RouterConfig{
-					Port:                            defaultPort,
-					MaxRequestHeaderBytes:           defaultMaxRequestHeaderBytes,
-					ReadTimeout:                     defaultHTTPServerReadTimeout,
-					WriteTimeout:                    defaultHTTPServerWriteTimeout,
-					IdleTimeout:                     defaultHTTPServerIdleTimeout,
-					SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration,
-					WebsocketMessageBufferSize:      defaultWebsocketMessageBufferSize,
+					Port:                              defaultPort,
+					MaxRequestHeaderBytes:             defaultMaxRequestHeaderBytes,
+					ReadTimeout:                       defaultHTTPServerReadTimeout,
+					WriteTimeout:                      defaultHTTPServerWriteTimeout,
+					IdleTimeout:                       defaultHTTPServerIdleTimeout,
+					SystemOverheadAllowanceDuration:   defaultSystemOverheadAllowanceDuration,
+					WebsocketMessageBufferSize:        defaultWebsocketMessageBufferSize,
+					MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections,
+					MaxRequestBodyBytes:               defaultMaxRequestBodyBytes,
 				},
 				Logger: LoggerConfig{
 					Level: "debug",
@@ -404,13 +408,15 @@ logger_config:
 					},
 				},
 				Router: RouterConfig{
-					Port:                            defaultPort,
-					MaxRequestHeaderBytes:           defaultMaxRequestHeaderBytes,
-					ReadTimeout:                     defaultHTTPServerReadTimeout,
-					WriteTimeout:                    defaultHTTPServerWriteTimeout,
-					IdleTimeout:                     defaultHTTPServerIdleTimeout,
-					SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration,
-					WebsocketMessageBufferSize:      defaultWebsocketMessageBufferSize,
+					Port:                              defaultPort,
+					MaxRequestHeaderBytes:             defaultMaxRequestHeaderBytes,
+					ReadTimeout:                       defaultHTTPServerReadTimeout,
+					WriteTimeout:                      defaultHTTPServerWriteTimeout,
+					IdleTimeout:                       defaultHTTPServerIdleTimeout,
+					SystemOverheadAllowanceDuration:   defaultSystemOverheadAllowanceDuration,
+					WebsocketMessageBufferSize:        defaultWebsocketMessageBufferSize,
+					MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections,
+					MaxRequestBodyBytes:               defaultMaxRequestBodyBytes,
 				},
 				Logger: LoggerConfig{
 					Level: "info",

config/router.go

@@ -31,6 +31,25 @@ const (
 	// Reduced from 1000 to prevent OOM. At 100: 100 × ~3KB × 100 connections = ~30MB.
 	// Can be tuned based on expected concurrent websocket connections and message frequency.
 	defaultWebsocketMessageBufferSize = 100
+
+	// defaultMaxRequestBodyBytes caps the size of an HTTP request body PATH will
+	// read into memory. Without a limit, a single request with a multi-GB body
+	// OOMs the process (the body is buffered whole, and observation/detection
+	// paths read it before any downstream limit applies). 75MB leaves generous
+	// headroom for the largest legitimate payloads (big batches, large
+	// eth_getLogs/debug traces, contract deploys) while still bounding a single
+	// request's allocation. Reads past the limit fail (request rejected) rather
+	// than growing memory without bound.
+	defaultMaxRequestBodyBytes = 75 * 1024 * 1024
+
+	// defaultMaxConcurrentWebsocketConnections caps concurrent live websocket
+	// connections per gateway pod (defense-in-depth against goroutine/FD
+	// exhaustion). Each connection costs ~5-7 goroutines + 2 sockets + buffers
+	// (~50-100KB), so 10000 ≈ 0.5-1GB worst case — a generous ceiling well above
+	// normal load that still bounds catastrophic runaway. Tune to observed
+	// concurrent websocket load (Polygon is the primary websocket service).
+	// Set to a negative value in config to disable the limit entirely.
+	defaultMaxConcurrentWebsocketConnections = 10000
 )
 
 /* --------------------------------- Router Config Struct -------------------------------- */
@@ -48,6 +67,13 @@ type RouterConfig struct {
 	// Larger values use more memory but can handle higher message throughput.
 	// Default: 50 (prevents OOM while maintaining reasonable throughput)
 	WebsocketMessageBufferSize int `yaml:"websocket_message_buffer_size"`
+	// MaxConcurrentWebsocketConnections caps the number of concurrent live
+	// websocket connections per gateway pod. Default: 10000. A negative value
+	// disables the limit.
+	MaxConcurrentWebsocketConnections int `yaml:"max_concurrent_websocket_connections"`
+	// MaxRequestBodyBytes caps the size (in bytes) of an HTTP request body PATH
+	// will read into memory. Default: 10MB. A negative value disables the limit.
+	MaxRequestBodyBytes int64 `yaml:"max_request_body_bytes"`
 }
 
 /* --------------------------------- Router Config Private Helpers -------------------------------- */
@@ -76,6 +102,14 @@ func (c *RouterConfig) hydrateRouterDefaults() error {
 	if c.WebsocketMessageBufferSize == 0 {
 		c.WebsocketMessageBufferSize = defaultWebsocketMessageBufferSize
 	}
+	// Only an unset (zero) value takes the default; a negative value is preserved
+	// so operators can explicitly disable the limit.
+	if c.MaxConcurrentWebsocketConnections == 0 {
+		c.MaxConcurrentWebsocketConnections = defaultMaxConcurrentWebsocketConnections
+	}
+	if c.MaxRequestBodyBytes == 0 {
+		c.MaxRequestBodyBytes = defaultMaxRequestBodyBytes
+	}
 	if c.SystemOverheadAllowanceDuration >= c.ReadTimeout || c.SystemOverheadAllowanceDuration >= c.WriteTimeout {
 		return fmt.Errorf("system overhead allowance duration %v must be less than read timeout %v and write timeout %v", c.SystemOverheadAllowanceDuration, c.ReadTimeout, c.WriteTimeout)
 	}

config/router_test.go

@@ -61,13 +61,15 @@ func TestRouterConfig_hydrateRouterDefaults(t *testing.T) {
 			name: "should set all defaults",
 			cfg:  RouterConfig{},
 			want: RouterConfig{
-				Port:                            defaultPort,
-				MaxRequestHeaderBytes:           defaultMaxRequestHeaderBytes,
-				ReadTimeout:                     defaultHTTPServerReadTimeout,
-				WriteTimeout:                    defaultHTTPServerWriteTimeout,
-				IdleTimeout:                     defaultHTTPServerIdleTimeout,
-				SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration,
-				WebsocketMessageBufferSize:      defaultWebsocketMessageBufferSize,
+				Port:                              defaultPort,
+				MaxRequestHeaderBytes:             defaultMaxRequestHeaderBytes,
+				ReadTimeout:                       defaultHTTPServerReadTimeout,
+				WriteTimeout:                      defaultHTTPServerWriteTimeout,
+				IdleTimeout:                       defaultHTTPServerIdleTimeout,
+				SystemOverheadAllowanceDuration:   defaultSystemOverheadAllowanceDuration,
+				WebsocketMessageBufferSize:        defaultWebsocketMessageBufferSize,
+				MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections,
+				MaxRequestBodyBytes:               defaultMaxRequestBodyBytes,
 			},
 			wantErr: false,
 		},
@@ -77,13 +79,15 @@ func TestRouterConfig_hydrateRouterDefaults(t *testing.T) {
 				Port: 8080,
 			},
 			want: RouterConfig{
-				Port:                            8080,
-				MaxRequestHeaderBytes:           defaultMaxRequestHeaderBytes,
-				ReadTimeout:                     defaultHTTPServerReadTimeout,
-				WriteTimeout:                    defaultHTTPServerWriteTimeout,
-				IdleTimeout:                     defaultHTTPServerIdleTimeout,
-				SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration,
-				WebsocketMessageBufferSize:      defaultWebsocketMessageBufferSize,
+				Port:                              8080,
+				MaxRequestHeaderBytes:             defaultMaxRequestHeaderBytes,
+				ReadTimeout:                       defaultHTTPServerReadTimeout,
+				WriteTimeout:                      defaultHTTPServerWriteTimeout,
+				IdleTimeout:                       defaultHTTPServerIdleTimeout,
+				SystemOverheadAllowanceDuration:   defaultSystemOverheadAllowanceDuration,
+				WebsocketMessageBufferSize:        defaultWebsocketMessageBufferSize,
+				MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections,
+				MaxRequestBodyBytes:               defaultMaxRequestBodyBytes,
 			},
 			wantErr: false,
 		},
@@ -111,13 +115,51 @@ func TestRouterConfig_hydrateRouterDefaults(t *testing.T) {
 				WebsocketMessageBufferSize: 500,
 			},
 			want: RouterConfig{
-				Port:                            defaultPort,
-				MaxRequestHeaderBytes:           defaultMaxRequestHeaderBytes,
-				ReadTimeout:                     defaultHTTPServerReadTimeout,
-				WriteTimeout:                    defaultHTTPServerWriteTimeout,
-				IdleTimeout:                     defaultHTTPServerIdleTimeout,
-				SystemOverheadAllowanceDuration: defaultSystemOverheadAllowanceDuration,
-				WebsocketMessageBufferSize:      500, // Custom value should be preserved
+				Port:                              defaultPort,
+				MaxRequestHeaderBytes:             defaultMaxRequestHeaderBytes,
+				ReadTimeout:                       defaultHTTPServerReadTimeout,
+				WriteTimeout:                      defaultHTTPServerWriteTimeout,
+				IdleTimeout:                       defaultHTTPServerIdleTimeout,
+				SystemOverheadAllowanceDuration:   defaultSystemOverheadAllowanceDuration,
+				WebsocketMessageBufferSize:        500, // Custom value should be preserved
+				MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections,
+				MaxRequestBodyBytes:               defaultMaxRequestBodyBytes,
+			},
+			wantErr: false,
+		},
+		{
+			name: "should preserve a negative max websocket connections (limit disabled)",
+			cfg: RouterConfig{
+				MaxConcurrentWebsocketConnections: -1,
+			},
+			want: RouterConfig{
+				Port:                              defaultPort,
+				MaxRequestHeaderBytes:             defaultMaxRequestHeaderBytes,
+				ReadTimeout:                       defaultHTTPServerReadTimeout,
+				WriteTimeout:                      defaultHTTPServerWriteTimeout,
+				IdleTimeout:                       defaultHTTPServerIdleTimeout,
+				SystemOverheadAllowanceDuration:   defaultSystemOverheadAllowanceDuration,
+				WebsocketMessageBufferSize:        defaultWebsocketMessageBufferSize,
+				MaxConcurrentWebsocketConnections: -1, // Negative preserved: limit disabled
+				MaxRequestBodyBytes:               defaultMaxRequestBodyBytes,
+			},
+			wantErr: false,
+		},
+		{
+			name: "should preserve a negative max request body bytes (limit disabled)",
+			cfg: RouterConfig{
+				MaxRequestBodyBytes: -1,
+			},
+			want: RouterConfig{
+				Port:                              defaultPort,
+				MaxRequestHeaderBytes:             defaultMaxRequestHeaderBytes,
+				ReadTimeout:                       defaultHTTPServerReadTimeout,
+				WriteTimeout:                      defaultHTTPServerWriteTimeout,
+				IdleTimeout:                       defaultHTTPServerIdleTimeout,
+				SystemOverheadAllowanceDuration:   defaultSystemOverheadAllowanceDuration,
+				WebsocketMessageBufferSize:        defaultWebsocketMessageBufferSize,
+				MaxConcurrentWebsocketConnections: defaultMaxConcurrentWebsocketConnections,
+				MaxRequestBodyBytes:               -1, // Negative preserved: limit disabled
 			},
 			wantErr: false,
 		},

gateway/gateway.go

@@ -73,6 +73,10 @@ type Gateway struct {
 	// on initial attempts for a TTL window, avoiding wasted relay attempts.
 	// Optional - if nil, no cross-pod domain circuit breaking occurs.
 	DomainCircuitBreaker *DomainCircuitBreaker
+
+	// WebsocketConnectionLimiter bounds the number of concurrent live websocket
+	// connections held open by this gateway. Optional - if nil, no limit is applied.
+	WebsocketConnectionLimiter *WebsocketConnectionLimiter
 }
 
 // HandleServiceRequest implements PATH gateway's service request processing.
@@ -207,6 +211,26 @@ func (g Gateway) handleWebSocketRequest(
 ) {
 	logger := g.Logger.With("method", "handleWebSocketRequest")
 
+	// Bound the number of concurrent live websocket connections (defense-in-depth
+	// against goroutine/FD exhaustion). Reserve a slot up front; release it either
+	// immediately if setup fails, or when the connection terminates if it succeeds.
+	if !g.WebsocketConnectionLimiter.Acquire() {
+		logger.Warn().
+			Int64("active_websocket_connections", g.WebsocketConnectionLimiter.Active()).
+			Msg("⚠️ rejecting websocket connection: concurrent connection limit reached")
+		http.Error(w, "too many concurrent websocket connections", http.StatusServiceUnavailable)
+		return
+	}
+	// established gates the slot release: while false, any early return releases the
+	// slot immediately (connection never went live); once true, the slot is released
+	// asynchronously when the connection terminates (see the goroutine below).
+	established := false
+	defer func() {
+		if !established {
+			g.WebsocketConnectionLimiter.Release()
+		}
+	}()
+
 	// Use a cancellable background context for the long-lived Websocket connection lifecycle.
 	// Unlike HTTP requests, Websocket connections are long-lived and should not be tied to the HTTP request context.
 	// The HTTP request context gets canceled when the HTTP handler returns, which would stop the observation listener.
@@ -233,6 +257,9 @@ func (g Gateway) handleWebSocketRequest(
 		// Note: We do NOT close messageObservationsChan here because Websocket connections
 		// outlive the HTTP handler. The channel will be closed when the Websocket actually disconnects.
 		messageObservationsChan: make(chan *observation.RequestResponseObservations, websocketBufferSize),
+		// ready is closed once the protocol context is assigned; message
+		// processors block on it to avoid the setup-window race.
+		ready: make(chan struct{}),
 	}
 
 	// Initialize the websocket request context using the HTTP request.
@@ -268,10 +295,21 @@ func (g Gateway) handleWebSocketRequest(
 		return
 	}
 
-	// At this point, the Websocket connection has terminated and the bridge has shut down.
-	// The defer block above will now execute and broadcast connection observations with:
-	//   - Complete connection duration (from establishment to termination)
-	//   - Final connection status and termination reason
-	// This ensures we send only ONE connection observation per Websocket connection.
-	logger.Debug().Msg("Websocket connection and bridge shutdown complete, ready to broadcast final observations")
+	// The connection is now live and owned by background bridge/listener goroutines.
+	// Hand the concurrency slot off to the connection's lifecycle: websocketCtx is
+	// canceled when the connection fully terminates (websocketRequestContext.cancelCtx),
+	// so release the slot then. established=true disarms the immediate-release defer,
+	// guaranteeing exactly one Release per successful Acquire.
+	established = true
+	go func() {
+		<-websocketCtx.Done()
+		g.WebsocketConnectionLimiter.Release()
+	}()
+
+	// The connection is now live and self-managed by the background bridge and
+	// listener goroutines; this handler returns immediately (the upgraded conn is
+	// detached from the HTTP request lifecycle via websocketCtx). Connection
+	// establishment/closure observations are emitted by those goroutines over the
+	// lifetime of the connection, not here.
+	logger.Debug().Msg("Websocket connection established; bridge running in background")
 }