From 07913db5f7e831fd2cdf1b130495f976ffb2a112 Mon Sep 17 00:00:00 2001 From: Roshan <48975233+pythonberg1997@users.noreply.github.com> Date: Mon, 10 Nov 2025 15:34:19 +0800 Subject: [PATCH] feat: add domain-specific RPC method mappings (#19) --- config.go | 33 +-- example.config.toml | 14 ++ integration_tests/domain_routing_test.go | 193 ++++++++++++++++++ .../testdata/domain_routing.toml | 44 ++++ .../testdata/domain_routing_multigroup.toml | 47 +++++ integration_tests/util_test.go | 59 +++++- proxyd.go | 1 + server.go | 105 ++++++---- 8 files changed, 434 insertions(+), 62 deletions(-) create mode 100644 integration_tests/domain_routing_test.go create mode 100644 integration_tests/testdata/domain_routing.toml create mode 100644 integration_tests/testdata/domain_routing_multigroup.toml diff --git a/config.go b/config.go index 7b7206d..cc98619 100644 --- a/config.go +++ b/config.go @@ -223,22 +223,23 @@ type EthCallOverrideConfig struct { } type Config struct { - WSBackendGroup string `toml:"ws_backend_group"` - Server ServerConfig `toml:"server"` - Cache CacheConfig `toml:"cache"` - Redis RedisConfig `toml:"redis"` - Metrics MetricsConfig `toml:"metrics"` - RateLimit RateLimitConfig `toml:"rate_limit"` - BackendOptions BackendOptions `toml:"backend"` - Backends BackendsConfig `toml:"backends"` - BatchConfig BatchConfig `toml:"batch"` - Authentication map[string]string `toml:"authentication"` - BackendGroups BackendGroupsConfig `toml:"backend_groups"` - RPCMethodMappings map[string]string `toml:"rpc_method_mappings"` - WSMethodWhitelist []string `toml:"ws_method_whitelist"` - WhitelistErrorMessage string `toml:"whitelist_error_message"` - SenderRateLimit SenderRateLimitConfig `toml:"sender_rate_limit"` - EthCallOverride EthCallOverrideConfig `toml:"eth_call_override"` + WSBackendGroup string `toml:"ws_backend_group"` + Server ServerConfig `toml:"server"` + Cache CacheConfig `toml:"cache"` + Redis RedisConfig `toml:"redis"` + Metrics MetricsConfig `toml:"metrics"` + RateLimit RateLimitConfig `toml:"rate_limit"` + BackendOptions BackendOptions `toml:"backend"` + Backends BackendsConfig `toml:"backends"` + BatchConfig BatchConfig `toml:"batch"` + Authentication map[string]string `toml:"authentication"` + BackendGroups BackendGroupsConfig `toml:"backend_groups"` + RPCMethodMappings map[string]string `toml:"rpc_method_mappings"` + DomainRPCMethodMappings map[string]map[string]string `toml:"domain_rpc_method_mappings"` + WSMethodWhitelist []string `toml:"ws_method_whitelist"` + WhitelistErrorMessage string `toml:"whitelist_error_message"` + SenderRateLimit SenderRateLimitConfig `toml:"sender_rate_limit"` + EthCallOverride EthCallOverrideConfig `toml:"eth_call_override"` } func ReadFromEnvOrConfig(value string) (string, error) { diff --git a/example.config.toml b/example.config.toml index c8b246c..7192894 100644 --- a/example.config.toml +++ b/example.config.toml @@ -136,6 +136,20 @@ eth_call = "query" eth_estimateGas = "query" eth_sendRawTransaction = "multicall" +# Domain-specific RPC method mappings (optional) +# Different domains can have different routing rules +# If no domain-specific mapping is found, it will fallback to rpc_method_mappings above +# [domain_rpc_method_mappings] +# [domain_rpc_method_mappings."domain1.example.com"] +# eth_blockNumber = "query" +# eth_sendRawTransaction = "multicall" +# eth_call = "query" +# +# [domain_rpc_method_mappings."domain2.example.com"] +# eth_blockNumber = "query" +# eth_sendRawTransaction = "query" +# eth_call = "multicall" + [eth_call_override] # 48Club [[eth_call_override.rules]] diff --git a/integration_tests/domain_routing_test.go b/integration_tests/domain_routing_test.go new file mode 100644 index 0000000..76efe4a --- /dev/null +++ b/integration_tests/domain_routing_test.go @@ -0,0 +1,193 @@ +package integration_tests + +import ( + "os" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/ethereum-optimism/infra/proxyd" +) + +func TestDomainRPCMethodMappings(t *testing.T) { + goodBackend1 := NewMockBackend(BatchedResponseHandler(200, goodResponse)) + defer goodBackend1.Close() + + goodBackend2 := NewMockBackend(BatchedResponseHandler(200, goodResponse)) + defer goodBackend2.Close() + + require.NoError(t, os.Setenv("GOOD_BACKEND_RPC_URL_1", goodBackend1.URL())) + require.NoError(t, os.Setenv("GOOD_BACKEND_RPC_URL_2", goodBackend2.URL())) + + config := ReadConfig("domain_routing") + _, shutdown, err := proxyd.Start(config) + require.NoError(t, err) + defer shutdown() + + t.Run("default domain uses default mappings", func(t *testing.T) { + // Reset counters + goodBackend1.Reset() + goodBackend2.Reset() + + client := NewProxydClient("http://127.0.0.1:8545") + res, statusCode, err := client.SendRPC("eth_blockNumber", nil) + require.NoError(t, err) + require.Equal(t, 200, statusCode) + require.NotNil(t, res) + + // eth_blockNumber should route to backend1 based on default rpc_method_mappings + require.Equal(t, 1, len(goodBackend1.Requests())) + require.Equal(t, 0, len(goodBackend2.Requests())) + }) + + t.Run("domain1 uses custom mappings", func(t *testing.T) { + // Reset counters + goodBackend1.Reset() + goodBackend2.Reset() + + client := NewProxydClient("http://127.0.0.1:8545") + // Set X-Forwarded-Host header to match domain1.example.com + req := NewRPCReq("1", "eth_blockNumber", nil) + res, statusCode, err := client.SendRequestWithHeaders(req, map[string]string{ + "X-Forwarded-Host": "domain1.example.com", + }) + require.NoError(t, err) + require.Equal(t, 200, statusCode) + require.NotNil(t, res) + + // For domain1.example.com, eth_blockNumber should route to backend2 + require.Equal(t, 0, len(goodBackend1.Requests())) + require.Equal(t, 1, len(goodBackend2.Requests())) + }) + + t.Run("domain2 uses custom mappings", func(t *testing.T) { + // Reset counters + goodBackend1.Reset() + goodBackend2.Reset() + + client := NewProxydClient("http://127.0.0.1:8545") + req := NewRPCReq("1", "eth_call", []interface{}{ + map[string]interface{}{"to": "0x1234"}, + "latest", + }) + res, statusCode, err := client.SendRequestWithHeaders(req, map[string]string{ + "X-Forwarded-Host": "domain2.example.com", + }) + require.NoError(t, err) + require.Equal(t, 200, statusCode) + require.NotNil(t, res) + + // For domain2.example.com, eth_call should route to backend1 + require.Equal(t, 1, len(goodBackend1.Requests())) + require.Equal(t, 0, len(goodBackend2.Requests())) + }) + + t.Run("unknown domain falls back to default mappings", func(t *testing.T) { + // Reset counters + goodBackend1.Reset() + goodBackend2.Reset() + + client := NewProxydClient("http://127.0.0.1:8545") + req := NewRPCReq("1", "eth_blockNumber", nil) + res, statusCode, err := client.SendRequestWithHeaders(req, map[string]string{ + "X-Forwarded-Host": "unknown.example.com", + }) + require.NoError(t, err) + require.Equal(t, 200, statusCode) + require.NotNil(t, res) + + // Unknown domain should use default mappings (backend1) + require.Equal(t, 1, len(goodBackend1.Requests())) + require.Equal(t, 0, len(goodBackend2.Requests())) + }) + + t.Run("batch requests use domain-specific mappings", func(t *testing.T) { + // Reset counters + goodBackend1.Reset() + goodBackend2.Reset() + + client := NewProxydClient("http://127.0.0.1:8545") + batch := []*proxyd.RPCReq{ + NewRPCReq("1", "eth_blockNumber", nil), + NewRPCReq("2", "eth_chainId", nil), + } + + res, statusCode, err := client.SendBatchRequestWithHeaders(batch, map[string]string{ + "X-Forwarded-Host": "domain1.example.com", + }) + require.NoError(t, err) + require.Equal(t, 200, statusCode) + require.NotNil(t, res) + + // For domain1.example.com, both methods should route to backend2 + require.Equal(t, 0, len(goodBackend1.Requests())) + require.Equal(t, 1, len(goodBackend2.Requests())) // batched into one request + }) +} + +func TestDomainRPCMethodMappingsWithMultipleBackendGroups(t *testing.T) { + backend1 := NewMockBackend(BatchedResponseHandler(200, goodResponse)) + defer backend1.Close() + + backend2 := NewMockBackend(BatchedResponseHandler(200, goodResponse)) + defer backend2.Close() + + backend3 := NewMockBackend(BatchedResponseHandler(200, goodResponse)) + defer backend3.Close() + + require.NoError(t, os.Setenv("BACKEND_1_URL", backend1.URL())) + require.NoError(t, os.Setenv("BACKEND_2_URL", backend2.URL())) + require.NoError(t, os.Setenv("BACKEND_3_URL", backend3.URL())) + + config := ReadConfig("domain_routing_multigroup") + _, shutdown, err := proxyd.Start(config) + require.NoError(t, err) + defer shutdown() + + t.Run("different domains route to different backend groups", func(t *testing.T) { + // Reset counters + backend1.Reset() + backend2.Reset() + backend3.Reset() + + client := NewProxydClient("http://127.0.0.1:8545") + + // Domain A: eth_call -> group_a (backend1) + req1 := NewRPCReq("1", "eth_call", []interface{}{ + map[string]interface{}{"to": "0x1234"}, + "latest", + }) + res1, statusCode1, err := client.SendRequestWithHeaders(req1, map[string]string{ + "X-Forwarded-Host": "domainA.example.com", + }) + require.NoError(t, err) + require.Equal(t, 200, statusCode1) + require.NotNil(t, res1) + require.Equal(t, 1, len(backend1.Requests())) + + // Domain B: eth_call -> group_b (backend2) + backend1.Reset() + req2 := NewRPCReq("2", "eth_call", []interface{}{ + map[string]interface{}{"to": "0x1234"}, + "latest", + }) + res2, statusCode2, err := client.SendRequestWithHeaders(req2, map[string]string{ + "X-Forwarded-Host": "domainB.example.com", + }) + require.NoError(t, err) + require.Equal(t, 200, statusCode2) + require.NotNil(t, res2) + require.Equal(t, 1, len(backend2.Requests())) + + // Default: eth_call -> group_c (backend3) + backend2.Reset() + res3, statusCode3, err := client.SendRPC("eth_call", []interface{}{ + map[string]interface{}{"to": "0x1234"}, + "latest", + }) + require.NoError(t, err) + require.Equal(t, 200, statusCode3) + require.NotNil(t, res3) + require.Equal(t, 1, len(backend3.Requests())) + }) +} diff --git a/integration_tests/testdata/domain_routing.toml b/integration_tests/testdata/domain_routing.toml new file mode 100644 index 0000000..14a704f --- /dev/null +++ b/integration_tests/testdata/domain_routing.toml @@ -0,0 +1,44 @@ +[server] +rpc_port = 8545 + +[backend] +response_timeout_seconds = 1 + +[backends] +[backends.backend1] +rpc_url = "$GOOD_BACKEND_RPC_URL_1" + +[backends.backend2] +rpc_url = "$GOOD_BACKEND_RPC_URL_2" + +[backend_groups] +[backend_groups.group1] +backends = ["backend1"] + +[backend_groups.group2] +backends = ["backend2"] + +# Default RPC method mappings +[rpc_method_mappings] +eth_blockNumber = "group1" +eth_chainId = "group1" +eth_call = "group1" +eth_gasPrice = "group1" + +# Domain-specific RPC method mappings +[domain_rpc_method_mappings] + +# domain1.example.com routes to group2 +[domain_rpc_method_mappings."domain1.example.com"] +eth_blockNumber = "group2" +eth_chainId = "group2" +eth_call = "group2" +eth_gasPrice = "group2" + +# domain2.example.com has mixed routing +[domain_rpc_method_mappings."domain2.example.com"] +eth_blockNumber = "group2" +eth_chainId = "group2" +eth_call = "group1" +eth_gasPrice = "group1" + diff --git a/integration_tests/testdata/domain_routing_multigroup.toml b/integration_tests/testdata/domain_routing_multigroup.toml new file mode 100644 index 0000000..8de6075 --- /dev/null +++ b/integration_tests/testdata/domain_routing_multigroup.toml @@ -0,0 +1,47 @@ +[server] +rpc_port = 8545 + +[backend] +response_timeout_seconds = 1 + +[backends] +[backends.backend_a] +rpc_url = "$BACKEND_1_URL" + +[backends.backend_b] +rpc_url = "$BACKEND_2_URL" + +[backends.backend_c] +rpc_url = "$BACKEND_3_URL" + +[backend_groups] +[backend_groups.group_a] +backends = ["backend_a"] + +[backend_groups.group_b] +backends = ["backend_b"] + +[backend_groups.group_c] +backends = ["backend_c"] + +# Default RPC method mappings - use group_c +[rpc_method_mappings] +eth_call = "group_c" +eth_blockNumber = "group_c" +eth_chainId = "group_c" + +# Domain-specific RPC method mappings +[domain_rpc_method_mappings] + +# domainA.example.com routes to group_a +[domain_rpc_method_mappings."domainA.example.com"] +eth_call = "group_a" +eth_blockNumber = "group_a" +eth_chainId = "group_a" + +# domainB.example.com routes to group_b +[domain_rpc_method_mappings."domainB.example.com"] +eth_call = "group_b" +eth_blockNumber = "group_b" +eth_chainId = "group_b" + diff --git a/integration_tests/util_test.go b/integration_tests/util_test.go index 456b693..0b44a76 100644 --- a/integration_tests/util_test.go +++ b/integration_tests/util_test.go @@ -12,11 +12,10 @@ import ( "time" "github.com/BurntSushi/toml" + "github.com/ethereum/go-ethereum/log" "github.com/gorilla/websocket" "github.com/stretchr/testify/require" - "github.com/ethereum/go-ethereum/log" - "github.com/ethereum-optimism/infra/proxyd" ) @@ -75,6 +74,62 @@ func (p *ProxydHTTPClient) SendRequest(body []byte) ([]byte, int, error) { return resBody, code, nil } +func (p *ProxydHTTPClient) SendRequestWithHeaders(rpcReq *proxyd.RPCReq, headers map[string]string) ([]byte, int, error) { + body, err := json.Marshal(rpcReq) + if err != nil { + panic(err) + } + + req, err := http.NewRequest("POST", p.url, bytes.NewReader(body)) + if err != nil { + panic(err) + } + req.Header = p.headers.Clone() + for k, v := range headers { + req.Header.Set(k, v) + } + + res, err := http.DefaultClient.Do(req) + if err != nil { + return nil, -1, err + } + defer res.Body.Close() + code := res.StatusCode + resBody, err := io.ReadAll(res.Body) + if err != nil { + panic(err) + } + return resBody, code, nil +} + +func (p *ProxydHTTPClient) SendBatchRequestWithHeaders(reqs []*proxyd.RPCReq, headers map[string]string) ([]byte, int, error) { + body, err := json.Marshal(reqs) + if err != nil { + panic(err) + } + + req, err := http.NewRequest("POST", p.url, bytes.NewReader(body)) + if err != nil { + panic(err) + } + req.Header = p.headers.Clone() + for k, v := range headers { + req.Header.Set(k, v) + } + + res, err := http.DefaultClient.Do(req) + if err != nil { + return nil, -1, err + } + defer res.Body.Close() + code := res.StatusCode + resBody, err := io.ReadAll(res.Body) + if err != nil { + panic(err) + } + return resBody, code, nil +} + func RequireEqualJSON(t *testing.T, expected []byte, actual []byte) { expJSON := canonicalizeJSON(t, expected) actJSON := canonicalizeJSON(t, actual) diff --git a/proxyd.go b/proxyd.go index b826cf3..f311634 100644 --- a/proxyd.go +++ b/proxyd.go @@ -347,6 +347,7 @@ func Start(config *Config) (*Server, func(), error) { wsBackendGroup, NewStringSetFromStrings(config.WSMethodWhitelist), config.RPCMethodMappings, + config.DomainRPCMethodMappings, config.Server.MaxBodySizeBytes, resolvedAuth, secondsToDuration(config.Server.TimeoutSeconds), diff --git a/server.go b/server.go index 6c5da02..7a17b79 100644 --- a/server.go +++ b/server.go @@ -54,33 +54,34 @@ const ( var emptyArrayResponse = json.RawMessage("[]") type Server struct { - BackendGroups map[string]*BackendGroup - wsBackendGroup *BackendGroup - wsMethodWhitelist *StringSet - rpcMethodMappings map[string]string - maxBodySize int64 - enableRequestLog bool - maxRequestBodyLogLen int - authenticatedPaths map[string]string - timeout time.Duration - maxUpstreamBatchSize int - maxBatchSize int - enableServedByHeader bool - upgrader *websocket.Upgrader - mainLim FrontendRateLimiter - exemptLims map[string]FrontendRateLimiter - overrideLims map[string]FrontendRateLimiter - senderLim FrontendRateLimiter - allowedChainIds []*big.Int - limExemptOrigins []*regexp.Regexp - limExemptUserAgents []*regexp.Regexp - globallyLimitedMethods map[string]bool - rpcServer *http.Server - wsServer *http.Server - cache RPCCache - srvMu sync.Mutex - rateLimitHeader string - ethCallOverrideRules []EthCallRule + BackendGroups map[string]*BackendGroup + wsBackendGroup *BackendGroup + wsMethodWhitelist *StringSet + rpcMethodMappings map[string]string + domainRPCMethodMappings map[string]map[string]string + maxBodySize int64 + enableRequestLog bool + maxRequestBodyLogLen int + authenticatedPaths map[string]string + timeout time.Duration + maxUpstreamBatchSize int + maxBatchSize int + enableServedByHeader bool + upgrader *websocket.Upgrader + mainLim FrontendRateLimiter + exemptLims map[string]FrontendRateLimiter + overrideLims map[string]FrontendRateLimiter + senderLim FrontendRateLimiter + allowedChainIds []*big.Int + limExemptOrigins []*regexp.Regexp + limExemptUserAgents []*regexp.Regexp + globallyLimitedMethods map[string]bool + rpcServer *http.Server + wsServer *http.Server + cache RPCCache + srvMu sync.Mutex + rateLimitHeader string + ethCallOverrideRules []EthCallRule } type limiterFunc func(method string) bool @@ -92,6 +93,7 @@ func NewServer( wsBackendGroup *BackendGroup, wsMethodWhitelist *StringSet, rpcMethodMappings map[string]string, + domainRPCMethodMappings map[string]map[string]string, maxBodySize int64, authenticatedPaths map[string]string, timeout time.Duration, @@ -185,19 +187,20 @@ func NewServer( } return &Server{ - BackendGroups: backendGroups, - wsBackendGroup: wsBackendGroup, - wsMethodWhitelist: wsMethodWhitelist, - rpcMethodMappings: rpcMethodMappings, - maxBodySize: maxBodySize, - authenticatedPaths: authenticatedPaths, - timeout: timeout, - maxUpstreamBatchSize: maxUpstreamBatchSize, - enableServedByHeader: enableServedByHeader, - cache: cache, - enableRequestLog: enableRequestLog, - maxRequestBodyLogLen: maxRequestBodyLogLen, - maxBatchSize: maxBatchSize, + BackendGroups: backendGroups, + wsBackendGroup: wsBackendGroup, + wsMethodWhitelist: wsMethodWhitelist, + rpcMethodMappings: rpcMethodMappings, + domainRPCMethodMappings: domainRPCMethodMappings, + maxBodySize: maxBodySize, + authenticatedPaths: authenticatedPaths, + timeout: timeout, + maxUpstreamBatchSize: maxUpstreamBatchSize, + enableServedByHeader: enableServedByHeader, + cache: cache, + enableRequestLog: enableRequestLog, + maxRequestBodyLogLen: maxRequestBodyLogLen, + maxBatchSize: maxBatchSize, upgrader: &websocket.Upgrader{ HandshakeTimeout: defaultWSHandshakeTimeout, }, @@ -406,7 +409,7 @@ func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) { return } - batchRes, batchContainsCached, servedBy, err := s.handleBatchRPC(ctx, reqs, isLimited, true) + batchRes, batchContainsCached, servedBy, err := s.handleBatchRPC(ctx, reqs, isLimited, true, origin) if err == context.DeadlineExceeded { writeRPCError(ctx, w, nil, ErrGatewayTimeout) return @@ -429,7 +432,7 @@ func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) { } rawBody := json.RawMessage(body) - backendRes, cached, servedBy, err := s.handleBatchRPC(ctx, []json.RawMessage{rawBody}, isLimited, false) + backendRes, cached, servedBy, err := s.handleBatchRPC(ctx, []json.RawMessage{rawBody}, isLimited, false, origin) if err != nil { if errors.Is(err, ErrConsensusGetReceiptsCantBeBatched) || errors.Is(err, ErrConsensusGetReceiptsInvalidTarget) { @@ -446,7 +449,7 @@ func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) { writeRPCRes(ctx, w, backendRes[0]) } -func (s *Server) handleBatchRPC(ctx context.Context, reqs []json.RawMessage, isLimited limiterFunc, isBatch bool) ([]*RPCRes, bool, string, error) { +func (s *Server) handleBatchRPC(ctx context.Context, reqs []json.RawMessage, isLimited limiterFunc, isBatch bool, origin string) ([]*RPCRes, bool, string, error) { // A request set is transformed into groups of batches. // Each batch group maps to a forwarded JSON-RPC batch request (subject to maxUpstreamBatchSize constraints) // A groupID is used to decouple Requests that have duplicate ID so they're not part of the same batch that's @@ -458,6 +461,9 @@ func (s *Server) handleBatchRPC(ctx context.Context, reqs []json.RawMessage, isL backendGroup string } + // Get the origin from context to select the appropriate rpc_method_mappings + rpcMethodMappings := s.getRPCMethodMappings(origin) + responses := make([]*RPCRes, len(reqs)) batches := make(map[batchGroup][]batchElem) ids := make(map[string]int, len(reqs)) @@ -500,7 +506,7 @@ func (s *Server) handleBatchRPC(ctx context.Context, reqs []json.RawMessage, isL } } - group := s.rpcMethodMappings[parsedReq.Method] + group := rpcMethodMappings[parsedReq.Method] if group == "" { // use unknown below to prevent DOS vector that fills up memory // with arbitrary method names. @@ -740,6 +746,17 @@ func (s *Server) isGlobalLimit(method string) bool { return s.globallyLimitedMethods[method] } +func (s *Server) getRPCMethodMappings(origin string) map[string]string { + // Check if there's a domain-specific mapping for this origin + if origin != "" { + if mapping, ok := s.domainRPCMethodMappings[origin]; ok { + return mapping + } + } + // Fallback to default mappings + return s.rpcMethodMappings +} + func (s *Server) rateLimitSender(ctx context.Context, req *RPCReq) error { var params []string if err := json.Unmarshal(req.Params, ¶ms); err != nil {