From fd764d4ff7bd7e75dfdd7f733f50dd3805d4b3f2 Mon Sep 17 00:00:00 2001 From: Bas van Kervel Date: Tue, 16 Jun 2015 13:07:13 +0200 Subject: [PATCH] added comms http --- rpc/api/mergedapi.go | 4 + rpc/comms/comms.go | 7 +- rpc/comms/http.go | 65 ++++++++++++++++ rpc/comms/http_net.go | 167 ++++++++++++++++++++++++++++++++++++++++++ rpc/shared/types.go | 12 +++ 5 files changed, 254 insertions(+), 1 deletion(-) create mode 100644 rpc/comms/http.go create mode 100644 rpc/comms/http_net.go diff --git a/rpc/api/mergedapi.go b/rpc/api/mergedapi.go index b62477a140..8cf4ae96e2 100644 --- a/rpc/api/mergedapi.go +++ b/rpc/api/mergedapi.go @@ -1,6 +1,8 @@ package api import ( + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/rpc/shared" ) @@ -40,6 +42,8 @@ func (self *MergedApi) Methods() []string { // Call the correct API's Execute method for the given request func (self *MergedApi) Execute(req *shared.Request) (interface{}, error) { + glog.V(logger.Detail).Infof("rpc method: %s", req.Method) + if res, _ := self.handle(req); res != nil { return res, nil } diff --git a/rpc/comms/comms.go b/rpc/comms/comms.go index 3f458adfdc..2af63e55dc 100644 --- a/rpc/comms/comms.go +++ b/rpc/comms/comms.go @@ -11,6 +11,11 @@ import ( "github.com/ethereum/go-ethereum/rpc/shared" ) +const ( + jsonrpcver = "2.0" + maxHttpSizeReqLength = 1024 * 1024 // 1MB +) + type EthereumClient interface { Close() Send(interface{}) error @@ -26,7 +31,7 @@ func handle(conn net.Conn, api api.EthereumApi, c codec.Codec) { codec.Close() return } else if err != nil { - glog.V(logger.Error).Infof("IPC recv err - %v\n", err) + glog.V(logger.Error).Infof("comms recv err - %v\n", err) codec.Close() return } diff --git a/rpc/comms/http.go b/rpc/comms/http.go new file mode 100644 index 0000000000..1fea8dc1d4 --- /dev/null +++ b/rpc/comms/http.go @@ -0,0 +1,65 @@ +package comms + +import ( + "fmt" + "net/http" + "strings" + + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/rpc/api" + "github.com/ethereum/go-ethereum/rpc/codec" + "github.com/rs/cors" +) + +var ( + // main HTTP rpc listener + httpListener *stoppableTCPListener + listenerStoppedError = fmt.Errorf("Listener has stopped") +) + +type HttpConfig struct { + ListenAddress string + ListenPort uint + CorsDomain string +} + +func StartHttp(cfg HttpConfig, codec codec.Codec, apis ...api.EthereumApi) error { + if httpListener != nil { + if fmt.Sprintf("%s:%d", cfg.ListenAddress, cfg.ListenPort) != httpListener.Addr().String() { + return fmt.Errorf("RPC service already running on %s ", httpListener.Addr().String()) + } + return nil // RPC service already running on given host/port + } + + l, err := newStoppableTCPListener(fmt.Sprintf("%s:%d", cfg.ListenAddress, cfg.ListenPort)) + if err != nil { + glog.V(logger.Error).Infof("Can't listen on %s:%d: %v", cfg.ListenAddress, cfg.ListenPort, err) + return err + } + httpListener = l + + api := api.Merge(apis...) + var handler http.Handler + if len(cfg.CorsDomain) > 0 { + var opts cors.Options + opts.AllowedMethods = []string{"POST"} + opts.AllowedOrigins = strings.Split(cfg.CorsDomain, " ") + + c := cors.New(opts) + handler = newStoppableHandler(c.Handler(gethHttpHandler(codec, api)), l.stop) + } else { + handler = newStoppableHandler(gethHttpHandler(codec, api), l.stop) + } + + go http.Serve(l, handler) + + return nil +} + +func StopHttp() { + if httpListener != nil { + httpListener.Stop() + httpListener = nil + } +} diff --git a/rpc/comms/http_net.go b/rpc/comms/http_net.go new file mode 100644 index 0000000000..8d1bacc062 --- /dev/null +++ b/rpc/comms/http_net.go @@ -0,0 +1,167 @@ +package comms + +import ( + "encoding/json" + "fmt" + "io" + "io/ioutil" + "net" + "net/http" + "time" + + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/rpc/api" + "github.com/ethereum/go-ethereum/rpc/codec" + "github.com/ethereum/go-ethereum/rpc/shared" +) + +// When https://github.com/golang/go/issues/4674 is implemented this could be replaced +type stoppableTCPListener struct { + *net.TCPListener + stop chan struct{} // closed when the listener must stop +} + +func newStoppableTCPListener(addr string) (*stoppableTCPListener, error) { + wl, err := net.Listen("tcp", addr) + if err != nil { + return nil, err + } + + if tcpl, ok := wl.(*net.TCPListener); ok { + stop := make(chan struct{}) + return &stoppableTCPListener{tcpl, stop}, nil + } + + return nil, fmt.Errorf("Unable to create TCP listener for RPC service") +} + +// Stop the listener and all accepted and still active connections. +func (self *stoppableTCPListener) Stop() { + close(self.stop) +} + +func (self *stoppableTCPListener) Accept() (net.Conn, error) { + for { + self.SetDeadline(time.Now().Add(time.Duration(1 * time.Second))) + c, err := self.TCPListener.AcceptTCP() + + select { + case <-self.stop: + if c != nil { // accept timeout + c.Close() + } + self.TCPListener.Close() + return nil, listenerStoppedError + default: + } + + if err != nil { + if netErr, ok := err.(net.Error); ok && netErr.Timeout() && netErr.Temporary() { + continue // regular timeout + } + } + + return &closableConnection{c, self.stop}, err + } +} + +type closableConnection struct { + *net.TCPConn + closed chan struct{} +} + +func (self *closableConnection) Read(b []byte) (n int, err error) { + select { + case <-self.closed: + self.TCPConn.Close() + return 0, io.EOF + default: + return self.TCPConn.Read(b) + } +} + +// Wraps the default handler and checks if the RPC service was stopped. In that case it returns an +// error indicating that the service was stopped. This will only happen for connections which are +// kept open (HTTP keep-alive) when the RPC service was shutdown. +func newStoppableHandler(h http.Handler, stop chan struct{}) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + select { + case <-stop: + w.Header().Set("Content-Type", "application/json") + err := fmt.Errorf("RPC service stopped") + response := shared.NewRpcResponse(-1, jsonrpcver, nil, err) + httpSend(w, response) + default: + h.ServeHTTP(w, r) + } + }) +} + +func httpSend(writer io.Writer, v interface{}) (n int, err error) { + var payload []byte + payload, err = json.MarshalIndent(v, "", "\t") + if err != nil { + glog.V(logger.Error).Infoln("Error marshalling JSON", err) + return 0, err + } + glog.V(logger.Detail).Infof("Sending payload: %s", payload) + + return writer.Write(payload) +} + +func gethHttpHandler(codec codec.Codec, api api.EthereumApi) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + w.Header().Set("Content-Type", "application/json") + + // Limit request size to resist DoS + if req.ContentLength > maxHttpSizeReqLength { + err := fmt.Errorf("Request too large") + response := shared.NewRpcErrorResponse(-1, jsonrpcver, -32700, err) + httpSend(w, &response) + return + } + + defer req.Body.Close() + payload, err := ioutil.ReadAll(req.Body) + if err != nil { + err := fmt.Errorf("Could not read request body") + response := shared.NewRpcErrorResponse(-1, jsonrpcver, -32700, err) + httpSend(w, &response) + return + } + + c := codec.New(nil) + var rpcReq shared.Request + if err = c.Decode(payload, &rpcReq); err == nil { + reply, err := api.Execute(&rpcReq) + res := shared.NewRpcResponse(rpcReq.Id, rpcReq.Jsonrpc, reply, err) + httpSend(w, &res) + return + } + + var reqBatch []shared.Request + if err = c.Decode(payload, &reqBatch); err == nil { + resBatch := make([]*interface{}, len(reqBatch)) + resCount := 0 + + for i, rpcReq := range reqBatch { + reply, err := api.Execute(&rpcReq) + if rpcReq.Id != nil { // this leaves nil entries in the response batch for later removal + resBatch[i] = shared.NewRpcResponse(rpcReq.Id, rpcReq.Jsonrpc, reply, err) + resCount += 1 + } + } + + // make response omitting nil entries + resBatch = resBatch[:resCount] + httpSend(w, resBatch) + return + } + + // invalid request + err = fmt.Errorf("Could not decode request") + res := shared.NewRpcErrorResponse(-1, jsonrpcver, -32600, err) + httpSend(w, res) + }) +} diff --git a/rpc/shared/types.go b/rpc/shared/types.go index 6a29fa88e1..dd38a45598 100644 --- a/rpc/shared/types.go +++ b/rpc/shared/types.go @@ -42,6 +42,18 @@ type ErrorObject struct { // Data interface{} `json:"data"` } +// Create RPC error response, this allows for custom error codes +func NewRpcErrorResponse(id interface{}, jsonrpcver string, errCode int, err error) *interface{} { + var response interface{} + + jsonerr := &ErrorObject{errCode, err.Error()} + response = ErrorResponse{Jsonrpc: jsonrpcver, Id: id, Error: jsonerr} + + glog.V(logger.Detail).Infof("Generated error response: %s", response) + return &response +} + +// Create RPC response func NewRpcResponse(id interface{}, jsonrpcver string, reply interface{}, err error) *interface{} { var response interface{}