fix(util): harden rate limiting, ACLs and RPC cache correctness

rpc/acl_boundary_test.go

@@ -0,0 +1,131 @@
+package rpc
+
+import (
+	"context"
+	"encoding/json"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/Conflux-Chain/confura/util/acl"
+	"github.com/Conflux-Chain/confura/util/rate"
+	"github.com/Conflux-Chain/confura/util/rpc/handlers"
+	"github.com/Conflux-Chain/confura/util/rpc/middlewares"
+	"github.com/ethereum/go-ethereum/common"
+	rpc "github.com/openweb3/go-rpc-provider"
+	wt "github.com/openweb3/web3go/types"
+	"github.com/stretchr/testify/require"
+)
+
+type ACLBoundaryAPI struct{}
+
+func (*ACLBoundaryAPI) GetLogs(context.Context, wt.FilterQuery) ([]wt.Log, error) {
+	return []wt.Log{{Address: common.HexToAddress("0x2222222222222222222222222222222222222222")}}, nil
+}
+func (*ACLBoundaryAPI) NewFilter(context.Context, wt.FilterQuery) (string, error) { return "0x1", nil }
+func (*ACLBoundaryAPI) GetFilterLogs(context.Context, rpc.ID) ([]wt.Log, error) {
+	return []wt.Log{{Address: common.HexToAddress("0x1111111111111111111111111111111111111111")}}, nil
+}
+func (api *ACLBoundaryAPI) GetFilterChanges(ctx context.Context, id rpc.ID) ([]wt.Log, error) {
+	return api.GetFilterLogs(ctx, id)
+}
+func (*ACLBoundaryAPI) Echo(ctx context.Context) (string, error) {
+	id, _ := handlers.GetAuthIdFromContext(ctx)
+	return id, nil
+}
+func (*ACLBoundaryAPI) Logs(ctx context.Context, f wt.FilterQuery) (*rpc.Subscription, error) {
+	n, _ := rpc.NotifierFromContext(ctx)
+	return n.CreateSubscription(), nil
+}
+
+func TestContractACLHTTPBatchAndWebSocket(t *testing.T) {
+	// Same production registration order; no config loading, external services, or real credentials.
+	rpc.HookHandleCallMsg(middlewares.Recover)
+	rpc.HookHandleCallMsg(middlewares.AntiInjection)
+	rpc.HookHandleCallMsg(middlewares.Authenticate)
+	rpc.HookHandleCallMsg(middlewares.Allowlists)
+	rpc.HookHandleCallMsg(middlewares.DailyMaxReqRateLimit)
+	rpc.HookHandleCallMsg(middlewares.QpsRateLimit)
+	rpc.HookHandleCallMsg(middlewares.Metrics)
+	rpc.HookHandleCallMsg(middlewares.Log)
+	rpc.HookHandleCallMsg(clientMiddleware)
+	rpc.HookHandleCallMsg(middlewares.UniformError)
+	rpc.HookHandleCallMsg(rpc.PreventMessagesWithouID)
+	const key = "FixtureProvisionedSynthetic0123456789"
+	loader := rate.NewKeyLoader(func(f *rate.KeysetFilter) ([]*rate.KeyInfo, error) {
+		if len(f.KeySet) == 0 || f.KeySet[0] == key {
+			return []*rate.KeyInfo{{Key: key, SID: 1, AclID: 1}}, nil
+		}
+		return nil, nil
+	})
+	r := rate.NewRegistry(loader, acl.NewEthValidator)
+	s := rate.NewStrategy(1, "default")
+	s.LimitOptions["rpc_all_qps"] = rate.NewTokenBucketOption(10000, 10000)
+	s.LimitOptions["rpc_all_daily"] = rate.FixedWindowOption{Interval: time.Hour, Quota: 10000}
+	config := &rate.Config{Strategies: map[uint32]*rate.Strategy{1: s}, AllowLists: map[uint32]*acl.AllowList{1: {ID: 1, Name: "default", ContractAddresses: []string{"0x1111111111111111111111111111111111111111"}}}}
+	go r.AutoReload(time.Hour, func() (*rate.Config, error) { return config, nil })
+	require.Eventually(t, func() bool {
+		return len(r.Strategies()) == 1 && r.Allow(acl.Context{Context: context.Background(), RpcMethod: "eth_getLogs", ExtractRpcParams: func() ([]interface{}, error) { return []interface{}{wt.FilterQuery{}}, nil }}) != nil
+	}, time.Second, time.Millisecond)
+	h := rpc.NewServer()
+	defer h.Stop()
+	require.NoError(t, h.RegisterName("eth", &ACLBoundaryAPI{}))
+	httpH := httpMiddleware("eth", r, nil)(h)
+	request := func(path, token, body string) string {
+		q := httptest.NewRequest("POST", "http://fixture.invalid"+path, strings.NewReader(body))
+		q.Header.Set("Content-Type", "application/json")
+		q.Header.Set("Access-Token", token)
+		w := httptest.NewRecorder()
+		httpH.ServeHTTP(w, q)
+		return w.Body.String()
+	}
+	for _, tc := range []struct{ name, path, token, body, want string }{
+		{"guest", "/", "", `{"jsonrpc":"2.0","id":1,"method":"eth_echo","params":[]}`, `"result":""`},
+		{"unknown", "/", "FixtureUnknownSynthetic0123456789", `{"jsonrpc":"2.0","id":1,"method":"eth_echo","params":[]}`, "invalid api key"},
+		{"provisioned", "/", key, `{"jsonrpc":"2.0","id":1,"method":"eth_echo","params":[]}`, key},
+		{"path beats header", "/FixtureUnknownSynthetic0123456789", key, `{"jsonrpc":"2.0","id":1,"method":"eth_echo","params":[]}`, "invalid api key"},
+		{"deny getLogs", "/", "", `{"jsonrpc":"2.0","id":1,"method":"eth_getLogs","params":[{"address":"0x2222222222222222222222222222222222222222"}]}`, "invalid contract address"},
+		{"malformed params", "/", "", `{"jsonrpc":"2.0","id":1,"method":"eth_getLogs","params":[9]}`, "bad RPC parameters"},
+		{"filter denied", "/", "", `{"jsonrpc":"2.0","id":1,"method":"eth_newFilter","params":[{"address":"0x2222222222222222222222222222222222222222"}]}`, "invalid contract address"},
+		{"batch", "/", "", `[{"jsonrpc":"2.0","id":1,"method":"eth_getLogs","params":[{}]},{"jsonrpc":"2.0","id":2,"method":"eth_newFilter","params":[{}]}]`, "invalid contract address"},
+		{"notification", "/", "", `{"jsonrpc":"2.0","method":"eth_echo","params":[]}`, "invalid request"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			out := request(tc.path, tc.token, tc.body)
+			require.Contains(t, out, tc.want)
+			if tc.name == "batch" {
+				require.NotContains(t, out, `"result"`)
+			}
+		})
+	}
+	t.Run("allowed filter remains readable", func(t *testing.T) {
+		created := request("/", key, `{"jsonrpc":"2.0","id":1,"method":"eth_newFilter","params":[{"address":"0x1111111111111111111111111111111111111111"}]}`)
+		var response struct {
+			Result string          `json:"result"`
+			Error  json.RawMessage `json:"error"`
+		}
+		require.NoError(t, json.Unmarshal([]byte(created), &response))
+		require.Empty(t, response.Error, created)
+		require.Equal(t, "0x1", response.Result)
+		for _, method := range []string{"eth_getFilterLogs", "eth_getFilterChanges"} {
+			out := request("/", key, `{"jsonrpc":"2.0","id":2,"method":"`+method+`","params":["`+response.Result+`"]}`)
+			require.Contains(t, out, `"result":[{"address":"0x1111111111111111111111111111111111111111"`, out)
+			require.NotContains(t, out, `"error"`, out)
+		}
+	})
+	// Real WebSocket upgrade and provider subscription dispatch, with identical HTTP context injector.
+	ws := httptest.NewServer(httpMiddleware("eth", r, nil)(h.WebsocketHandler([]string{"*"})))
+	defer ws.Close()
+	conn, err := rpc.Dial("ws" + strings.TrimPrefix(ws.URL, "http"))
+	require.NoError(t, err)
+	defer conn.Close()
+	logs := make(chan json.RawMessage, 1)
+	sub, err := conn.Subscribe(context.Background(), "eth", logs, "logs", map[string]interface{}{"address": "0x2222222222222222222222222222222222222222"})
+	require.ErrorContains(t, err, "invalid contract address")
+	require.Nil(t, sub)
+	sub, err = conn.Subscribe(context.Background(), "eth", logs, "logs", map[string]interface{}{"address": "0x1111111111111111111111111111111111111111"})
+	require.NoError(t, err)
+	sub.Unsubscribe()
+
+}

util/acl/validator.go

@@ -2,6 +2,7 @@ package acl
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"regexp"
 	"strings"
@@ -28,6 +29,7 @@ type Context struct {
 	context.Context
 
 	RpcMethod        string
+	RawParams        json.RawMessage
 	ExtractRpcParams func() ([]interface{}, error)
 }
 
@@ -194,15 +196,46 @@ func (v *validatorBase) validateContractAddresses(ctx Context) error {
 		return nil
 	}
 
-	if ctx.ExtractRpcParams == nil {
-		return nil
+	// Check filter criteria at creation. Filter-ID reads retain their existing
+	// access semantics; they do not carry addresses to validate here.
+	if ctx.RpcMethod == "eth_subscribe" || ctx.RpcMethod == "cfx_subscribe" {
+		var args []json.RawMessage
+		var kind string
+		if json.Unmarshal(ctx.RawParams, &args) != nil || len(args) == 0 || json.Unmarshal(args[0], &kind) != nil {
+			return errBadRpcParams
+		}
+		if kind != "logs" {
+			return nil
+		}
+		if len(args) != 2 {
+			return errBadRpcParams
+		}
+		raw := args[1]
+		if ctx.RpcMethod == "eth_subscribe" {
+			ctx.RpcMethod = "eth_getLogs"
+			ctx.ExtractRpcParams = func() ([]interface{}, error) {
+				var f web3Types.FilterQuery
+				err := json.Unmarshal(raw, &f)
+				return []interface{}{f}, err
+			}
+		} else {
+			ctx.RpcMethod = "cfx_getLogs"
+			ctx.ExtractRpcParams = func() ([]interface{}, error) {
+				var f cfxTypes.LogFilter
+				err := json.Unmarshal(raw, &f)
+				return []interface{}{f}, err
+			}
+		}
 	}
 
 	parseContractAddresses, ok := v.contractAddressParsers[ctx.RpcMethod]
 	if !ok {
 		return nil
 	}
 
+	if ctx.ExtractRpcParams == nil {
+		return errBadRpcParams
+	}
 	inputParams, err := ctx.ExtractRpcParams()
 	if err != nil {
 		return errBadRpcParams
@@ -242,6 +275,7 @@ func NewEthValidator(al *AllowList) Validator {
 		"eth_call":                        v.parseCallRequest,
 		"eth_estimateGas":                 v.parseCallRequest,
 		"eth_getLogs":                     v.parseFilterQuery,
+		"eth_newFilter":                   v.parseFilterQuery,
 		"eth_scanLogs":                    parseContractAddressProvider,
 		"eth_scanLogsWithPivotAssumption": parseContractAddressProvider,
 		"eth_getBalance":                  v.parseAddr,
@@ -266,7 +300,7 @@ func (v *EthValidator) parseCallRequest(params []interface{}) (res []string, ok
 	}
 
 	cr, ok := params[0].(web3Types.CallRequest)
-	if ok {
+	if ok && cr.To != nil {
 		res = append(res, cr.To.String())
 	}
 
@@ -329,6 +363,7 @@ func NewCfxValidator(al *AllowList) Validator {
 		"cfx_call":                        v.parseCallRequest,
 		"cfx_estimateGasAndCollateral":    v.parseCallRequest,
 		"cfx_getLogs":                     v.parseLogFilter,
+		"cfx_newFilter":                   v.parseLogFilter,
 		"cfx_scanLogs":                    parseContractAddressProvider,
 		"cfx_scanLogsWithPivotAssumption": parseContractAddressProvider,
 		"cfx_getBalance":                  v.parseAddr,
@@ -361,7 +396,7 @@ func (v *CfxValidator) parseCallRequest(params []interface{}) (res []string, ok
 	}
 
 	cr, ok := params[0].(cfxTypes.CallRequest)
-	if ok { // only non-verbose base32 address format is accepted
+	if ok && cr.To != nil { // only non-verbose base32 address format is accepted
 		res = append(res, cr.To.MustGetBase32Address())
 	}
 
@@ -390,7 +425,7 @@ func (v *CfxValidator) parseAddr(params []interface{}) (res []string, ok bool) {
 
 	ca, ok := params[0].(cfxTypes.Address)
 	if ok {
-		res = append(res, ca.MustGetVerboseBase32Address())
+		res = append(res, ca.MustGetBase32Address())
 	}
 
 	return

util/acl/validator_boundary_test.go

@@ -0,0 +1,93 @@
+package acl
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+
+	"github.com/Conflux-Chain/confura/util/rpc/handlers"
+	cfxTypes "github.com/Conflux-Chain/go-conflux-sdk/types"
+	"github.com/Conflux-Chain/go-conflux-sdk/types/cfxaddress"
+	"github.com/ethereum/go-ethereum/common"
+	web3Types "github.com/openweb3/web3go/types"
+	"github.com/stretchr/testify/require"
+)
+
+func TestContractACLFilterAndSubscriptionBoundaries(t *testing.T) {
+	v := NewEthValidator(&AllowList{ContractAddresses: []string{allowedContractAddr}})
+	for _, method := range []string{"eth_getLogs", "eth_newFilter"} {
+		ctx := ethGetLogsContext(web3Types.FilterQuery{Addresses: []common.Address{common.HexToAddress(allowedContractAddr)}})
+		ctx.RpcMethod = method
+		require.NoError(t, v.Validate(ctx))
+		ctx.ExtractRpcParams = nil
+		require.Error(t, v.Validate(ctx))
+	}
+	for _, tc := range []struct {
+		params  string
+		allowed bool
+	}{
+		{`["logs",{"address":"` + allowedContractAddr + `"}]`, true},
+		{`["logs",{"address":"` + notAllowedContractAddr + `"}]`, false},
+		{`["logs",{}]`, false}, {`["logs"]`, false}, {`["logs",null]`, false},
+		{`["newHeads"]`, true}, {`[null]`, true},
+	} {
+		ctx := Context{Context: context.Background(), RpcMethod: "eth_subscribe", RawParams: json.RawMessage(tc.params)}
+		err := v.Validate(ctx)
+		require.Equal(t, tc.allowed, err == nil, tc.params)
+	}
+}
+
+func TestContractACLRetainsFilterIDReads(t *testing.T) {
+	for _, tc := range []struct {
+		prefix  string
+		factory ValidatorFactory
+	}{
+		{"eth", NewEthValidator},
+		{"cfx", NewCfxValidator},
+	} {
+		t.Run(tc.prefix, func(t *testing.T) {
+			for _, suffix := range []string{"getFilterLogs", "getFilterChanges"} {
+				method := tc.prefix + "_" + suffix
+				ctx := Context{Context: context.Background(), RpcMethod: method, RawParams: json.RawMessage(`["0x1"]`)}
+				require.NoError(t, tc.factory(&AllowList{ContractAddresses: []string{allowedContractAddr}}).Validate(ctx))
+				require.NoError(t, tc.factory(&AllowList{}).Validate(ctx))
+				// Filter-ID reads still enforce method restrictions.
+				require.ErrorIs(t, tc.factory(&AllowList{ContractAddresses: []string{allowedContractAddr}, DisallowMethods: []string{method}}).Validate(ctx), errBadRpcMethod)
+			}
+		})
+	}
+}
+
+func TestCoreNewFilterContractACL(t *testing.T) {
+	allowed := cfxaddress.MustNew(allowedContractAddr, 1)
+	denied := cfxaddress.MustNew(notAllowedContractAddr, 1)
+	v := NewCfxValidator(&AllowList{ContractAddresses: []string{allowed.MustGetBase32Address()}})
+	for _, tc := range []struct {
+		addresses []cfxaddress.Address
+		allowed   bool
+	}{
+		{[]cfxaddress.Address{allowed}, true},
+		{[]cfxaddress.Address{denied}, false},
+		{[]cfxaddress.Address{allowed, denied}, false},
+		{nil, false},
+	} {
+		ctx := cfxGetLogsContext(cfxTypes.LogFilter{Address: tc.addresses})
+		ctx.RpcMethod = "cfx_newFilter"
+		require.Equal(t, tc.allowed, v.Validate(ctx) == nil)
+	}
+}
+
+func TestLiteralOriginAndCoreAddress(t *testing.T) {
+	v := NewEthValidator(&AllowList{Origins: []string{"https://app.example.com"}})
+	for _, origin := range []string{"https://app-example.com", "https://appXexample.com"} {
+		require.Error(t, v.Validate(Context{Context: context.WithValue(context.Background(), handlers.CtxKeyReqOrigin, origin)}))
+	}
+	require.NoError(t, v.Validate(Context{Context: context.WithValue(context.Background(), handlers.CtxKeyReqOrigin, "https://app.example.com")}))
+	addr := cfxaddress.MustNew(allowedContractAddr, 1)
+	for _, form := range []string{addr.MustGetBase32Address(), addr.MustGetVerboseBase32Address()} {
+		v := NewCfxValidator(&AllowList{ContractAddresses: []string{form}})
+		for _, method := range []string{"cfx_getBalance", "cfx_getCode", "cfx_getStorageAt", "cfx_getNextNonce"} {
+			require.NoError(t, v.Validate(Context{RpcMethod: method, ExtractRpcParams: func() ([]interface{}, error) { return []interface{}{addr}, nil }}))
+		}
+	}
+}

util/blockchain.go

@@ -171,7 +171,7 @@ func IsEip155Tx(tx *web3goTypes.TransactionDetail) bool {
 
 // IsLegacyEthTx check if the EVM transaction is legacy (pre EIP155)
 func IsLegacyEthTx(tx *web3goTypes.TransactionDetail) bool {
-	if tx.V != nil && tx.V.Uint64() == 27 || tx.V.Uint64() == 28 {
+	if tx != nil && tx.V != nil && (tx.V.Uint64() == 27 || tx.V.Uint64() == 28) {
 		return true
 	}
 
@@ -200,6 +200,7 @@ func MatchEthLogTopics(log *web3goTypes.Log, topics [][]common.Hash) bool {
 
 	for i := range topics {
 		if len(topics[i]) == 0 {
+			// Conflux eSpace wildcards also match missing topic positions.
 			continue
 		}