plugeth/node/api.go
mmsqe f3314bb6df
rpc: add limit for batch request items and response size (#26681)
This PR adds server-side limits for JSON-RPC batch requests. Before this change, batches
were limited only by processing time. The server would pick calls from the batch and
answer them until the response timeout occurred, then stop processing the remaining batch
items.

Here, we are adding two additional limits which can be configured:

- the 'item limit': batches can have at most N items
- the 'response size limit': batches can contain at most X response bytes

These limits are optional in package rpc. In Geth, we set a default limit of 1000 items
and 25MB response size.

When a batch goes over the limit, an error response is returned to the client. However,
doing this correctly isn't always possible. In JSON-RPC, only method calls with a valid
`id` can be responded to. Since batches may also contain non-call messages or
notifications, the best effort thing we can do to report an error with the batch itself is
reporting the limit violation as an error for the first method call in the batch. If a batch is
too large, but contains only notifications and responses, the error will be reported with
a null `id`.

The RPC client was also changed so it can deal with errors resulting from too large
batches. An older client connected to the server code in this PR could get stuck
until the request timeout occurred when the batch is too large. **Upgrading to a version
of the RPC client containing this change is strongly recommended to avoid timeout issues.**

For some weird reason, when writing the original client implementation, @fjl worked off of
the assumption that responses could be distributed across batches arbitrarily. So for a
batch request containing requests `[A B C]`, the server could respond with `[A B C]` but
also with `[A B] [C]` or even `[A] [B] [C]` and it wouldn't make a difference to the
client.

So in the implementation of BatchCallContext, the client waited for all requests in the
batch individually. If the server didn't respond to some of the requests in the batch, the
client would eventually just time out (if a context was used).

With the addition of batch limits into the server, we anticipate that people will hit this
kind of error way more often. To handle this properly, the client now waits for a single
response batch and expects it to contain all responses to the requests.

---------

Co-authored-by: Felix Lange <fjl@twurst.com>
Co-authored-by: Martin Holst Swende <martin@swende.se>
2023-06-13 13:38:58 +02:00

338 lines
9.4 KiB
Go

// Copyright 2015 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package node
import (
"context"
"fmt"
"strings"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/internal/debug"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/rpc"
)
// apis returns the collection of built-in RPC APIs.
func (n *Node) apis() []rpc.API {
return []rpc.API{
{
Namespace: "admin",
Service: &adminAPI{n},
}, {
Namespace: "debug",
Service: debug.Handler,
}, {
Namespace: "web3",
Service: &web3API{n},
},
}
}
// adminAPI is the collection of administrative API methods exposed over
// both secure and unsecure RPC channels.
type adminAPI struct {
node *Node // Node interfaced by this API
}
// AddPeer requests connecting to a remote node, and also maintaining the new
// connection at all times, even reconnecting if it is lost.
func (api *adminAPI) AddPeer(url string) (bool, error) {
// Make sure the server is running, fail otherwise
server := api.node.Server()
if server == nil {
return false, ErrNodeStopped
}
// Try to add the url as a static peer and return
node, err := enode.Parse(enode.ValidSchemes, url)
if err != nil {
return false, fmt.Errorf("invalid enode: %v", err)
}
server.AddPeer(node)
return true, nil
}
// RemovePeer disconnects from a remote node if the connection exists
func (api *adminAPI) RemovePeer(url string) (bool, error) {
// Make sure the server is running, fail otherwise
server := api.node.Server()
if server == nil {
return false, ErrNodeStopped
}
// Try to remove the url as a static peer and return
node, err := enode.Parse(enode.ValidSchemes, url)
if err != nil {
return false, fmt.Errorf("invalid enode: %v", err)
}
server.RemovePeer(node)
return true, nil
}
// AddTrustedPeer allows a remote node to always connect, even if slots are full
func (api *adminAPI) AddTrustedPeer(url string) (bool, error) {
// Make sure the server is running, fail otherwise
server := api.node.Server()
if server == nil {
return false, ErrNodeStopped
}
node, err := enode.Parse(enode.ValidSchemes, url)
if err != nil {
return false, fmt.Errorf("invalid enode: %v", err)
}
server.AddTrustedPeer(node)
return true, nil
}
// RemoveTrustedPeer removes a remote node from the trusted peer set, but it
// does not disconnect it automatically.
func (api *adminAPI) RemoveTrustedPeer(url string) (bool, error) {
// Make sure the server is running, fail otherwise
server := api.node.Server()
if server == nil {
return false, ErrNodeStopped
}
node, err := enode.Parse(enode.ValidSchemes, url)
if err != nil {
return false, fmt.Errorf("invalid enode: %v", err)
}
server.RemoveTrustedPeer(node)
return true, nil
}
// PeerEvents creates an RPC subscription which receives peer events from the
// node's p2p.Server
func (api *adminAPI) PeerEvents(ctx context.Context) (*rpc.Subscription, error) {
// Make sure the server is running, fail otherwise
server := api.node.Server()
if server == nil {
return nil, ErrNodeStopped
}
// Create the subscription
notifier, supported := rpc.NotifierFromContext(ctx)
if !supported {
return nil, rpc.ErrNotificationsUnsupported
}
rpcSub := notifier.CreateSubscription()
go func() {
events := make(chan *p2p.PeerEvent)
sub := server.SubscribeEvents(events)
defer sub.Unsubscribe()
for {
select {
case event := <-events:
notifier.Notify(rpcSub.ID, event)
case <-sub.Err():
return
case <-rpcSub.Err():
return
case <-notifier.Closed():
return
}
}
}()
return rpcSub, nil
}
// StartHTTP starts the HTTP RPC API server.
func (api *adminAPI) StartHTTP(host *string, port *int, cors *string, apis *string, vhosts *string) (bool, error) {
api.node.lock.Lock()
defer api.node.lock.Unlock()
// Determine host and port.
if host == nil {
h := DefaultHTTPHost
if api.node.config.HTTPHost != "" {
h = api.node.config.HTTPHost
}
host = &h
}
if port == nil {
port = &api.node.config.HTTPPort
}
// Determine config.
config := httpConfig{
CorsAllowedOrigins: api.node.config.HTTPCors,
Vhosts: api.node.config.HTTPVirtualHosts,
Modules: api.node.config.HTTPModules,
rpcEndpointConfig: rpcEndpointConfig{
batchItemLimit: api.node.config.BatchRequestLimit,
batchResponseSizeLimit: api.node.config.BatchResponseMaxSize,
},
}
if cors != nil {
config.CorsAllowedOrigins = nil
for _, origin := range strings.Split(*cors, ",") {
config.CorsAllowedOrigins = append(config.CorsAllowedOrigins, strings.TrimSpace(origin))
}
}
if vhosts != nil {
config.Vhosts = nil
for _, vhost := range strings.Split(*host, ",") {
config.Vhosts = append(config.Vhosts, strings.TrimSpace(vhost))
}
}
if apis != nil {
config.Modules = nil
for _, m := range strings.Split(*apis, ",") {
config.Modules = append(config.Modules, strings.TrimSpace(m))
}
}
if err := api.node.http.setListenAddr(*host, *port); err != nil {
return false, err
}
if err := api.node.http.enableRPC(api.node.rpcAPIs, config); err != nil {
return false, err
}
if err := api.node.http.start(); err != nil {
return false, err
}
return true, nil
}
// StartRPC starts the HTTP RPC API server.
// Deprecated: use StartHTTP instead.
func (api *adminAPI) StartRPC(host *string, port *int, cors *string, apis *string, vhosts *string) (bool, error) {
log.Warn("Deprecation warning", "method", "admin.StartRPC", "use-instead", "admin.StartHTTP")
return api.StartHTTP(host, port, cors, apis, vhosts)
}
// StopHTTP shuts down the HTTP server.
func (api *adminAPI) StopHTTP() (bool, error) {
api.node.http.stop()
return true, nil
}
// StopRPC shuts down the HTTP server.
// Deprecated: use StopHTTP instead.
func (api *adminAPI) StopRPC() (bool, error) {
log.Warn("Deprecation warning", "method", "admin.StopRPC", "use-instead", "admin.StopHTTP")
return api.StopHTTP()
}
// StartWS starts the websocket RPC API server.
func (api *adminAPI) StartWS(host *string, port *int, allowedOrigins *string, apis *string) (bool, error) {
api.node.lock.Lock()
defer api.node.lock.Unlock()
// Determine host and port.
if host == nil {
h := DefaultWSHost
if api.node.config.WSHost != "" {
h = api.node.config.WSHost
}
host = &h
}
if port == nil {
port = &api.node.config.WSPort
}
// Determine config.
config := wsConfig{
Modules: api.node.config.WSModules,
Origins: api.node.config.WSOrigins,
// ExposeAll: api.node.config.WSExposeAll,
rpcEndpointConfig: rpcEndpointConfig{
batchItemLimit: api.node.config.BatchRequestLimit,
batchResponseSizeLimit: api.node.config.BatchResponseMaxSize,
},
}
if apis != nil {
config.Modules = nil
for _, m := range strings.Split(*apis, ",") {
config.Modules = append(config.Modules, strings.TrimSpace(m))
}
}
if allowedOrigins != nil {
config.Origins = nil
for _, origin := range strings.Split(*allowedOrigins, ",") {
config.Origins = append(config.Origins, strings.TrimSpace(origin))
}
}
// Enable WebSocket on the server.
server := api.node.wsServerForPort(*port, false)
if err := server.setListenAddr(*host, *port); err != nil {
return false, err
}
openApis, _ := api.node.getAPIs()
if err := server.enableWS(openApis, config); err != nil {
return false, err
}
if err := server.start(); err != nil {
return false, err
}
api.node.http.log.Info("WebSocket endpoint opened", "url", api.node.WSEndpoint())
return true, nil
}
// StopWS terminates all WebSocket servers.
func (api *adminAPI) StopWS() (bool, error) {
api.node.http.stopWS()
api.node.ws.stop()
return true, nil
}
// Peers retrieves all the information we know about each individual peer at the
// protocol granularity.
func (api *adminAPI) Peers() ([]*p2p.PeerInfo, error) {
server := api.node.Server()
if server == nil {
return nil, ErrNodeStopped
}
return server.PeersInfo(), nil
}
// NodeInfo retrieves all the information we know about the host node at the
// protocol granularity.
func (api *adminAPI) NodeInfo() (*p2p.NodeInfo, error) {
server := api.node.Server()
if server == nil {
return nil, ErrNodeStopped
}
return server.NodeInfo(), nil
}
// Datadir retrieves the current data directory the node is using.
func (api *adminAPI) Datadir() string {
return api.node.DataDir()
}
// web3API offers helper utils
type web3API struct {
stack *Node
}
// ClientVersion returns the node name
func (s *web3API) ClientVersion() string {
return s.stack.Server().Name
}
// Sha3 applies the ethereum sha3 implementation on the input.
// It assumes the input is hex encoded.
func (s *web3API) Sha3(input hexutil.Bytes) hexutil.Bytes {
return crypto.Keccak256(input)
}