123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566 |
- // Copyright 2020 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 (
- "compress/gzip"
- "context"
- "fmt"
- "io"
- "io/ioutil"
- "net"
- "net/http"
- "sort"
- "strings"
- "sync"
- "sync/atomic"
- "github.com/ethereum/go-ethereum/log"
- "github.com/ethereum/go-ethereum/plugin/security"
- "github.com/ethereum/go-ethereum/rpc"
- "github.com/rs/cors"
- )
- // httpConfig is the JSON-RPC/HTTP configuration.
- type httpConfig struct {
- Modules []string
- CorsAllowedOrigins []string
- Vhosts []string
- prefix string // path prefix on which to mount http handler
- }
- // wsConfig is the JSON-RPC/Websocket configuration
- type wsConfig struct {
- Origins []string
- Modules []string
- prefix string // path prefix on which to mount ws handler
- }
- type rpcHandler struct {
- http.Handler
- server *rpc.Server
- }
- type httpServer struct {
- log log.Logger
- timeouts rpc.HTTPTimeouts
- mux http.ServeMux // registered handlers go here
- mu sync.Mutex
- server *http.Server
- listener net.Listener // non-nil when server is running
- // HTTP RPC handler things.
- httpConfig httpConfig
- httpHandler atomic.Value // *rpcHandler
- // WebSocket handler things.
- wsConfig wsConfig
- wsHandler atomic.Value // *rpcHandler
- // These are set by setListenAddr.
- endpoint string
- host string
- port int
- handlerNames map[string]string
- // Quorum
- // isMultitenant determines if the server supports mutlitenancy
- isMultitenant bool
- }
- func newHTTPServer(log log.Logger, timeouts rpc.HTTPTimeouts) *httpServer {
- h := &httpServer{log: log, timeouts: timeouts, handlerNames: make(map[string]string)}
- h.httpHandler.Store((*rpcHandler)(nil))
- h.wsHandler.Store((*rpcHandler)(nil))
- return h
- }
- // Quorum
- // withMultitenancy indicates if this server supports multitenancy
- func (h *httpServer) withMultitenancy(b bool) *httpServer {
- h.isMultitenant = b
- return h
- }
- // setListenAddr configures the listening address of the server.
- // The address can only be set while the server isn't running.
- func (h *httpServer) setListenAddr(host string, port int) error {
- h.mu.Lock()
- defer h.mu.Unlock()
- if h.listener != nil && (host != h.host || port != h.port) {
- return fmt.Errorf("HTTP server already running on %s", h.endpoint)
- }
- h.host, h.port = host, port
- h.endpoint = fmt.Sprintf("%s:%d", host, port)
- return nil
- }
- // listenAddr returns the listening address of the server.
- func (h *httpServer) listenAddr() string {
- h.mu.Lock()
- defer h.mu.Unlock()
- if h.listener != nil {
- return h.listener.Addr().String()
- }
- return h.endpoint
- }
- // Quorum - add tlsConfigSource used to start a listener with tls
- // start starts the HTTP server if it is enabled and not already running.
- func (h *httpServer) start(tlsConfigSource security.TLSConfigurationSource) error {
- h.mu.Lock()
- defer h.mu.Unlock()
- if h.endpoint == "" || h.listener != nil {
- return nil // already running or not configured
- }
- // Initialize the server.
- h.server = &http.Server{Handler: h}
- if h.timeouts != (rpc.HTTPTimeouts{}) {
- CheckTimeouts(&h.timeouts)
- h.server.ReadTimeout = h.timeouts.ReadTimeout
- h.server.WriteTimeout = h.timeouts.WriteTimeout
- h.server.IdleTimeout = h.timeouts.IdleTimeout
- }
- // Start the server.
- isTls, listener, err := startListener(h.endpoint, tlsConfigSource)
- if err != nil {
- // If the server fails to start, we need to clear out the RPC and WS
- // configuration so they can be configured another time.
- h.disableRPC()
- h.disableWS()
- return err
- }
- h.listener = listener
- go h.server.Serve(listener)
- if h.wsAllowed() {
- url := fmt.Sprintf("ws://%v", listener.Addr())
- if h.wsConfig.prefix != "" {
- url += h.wsConfig.prefix
- }
- h.log.Info("WebSocket enabled", "url", url)
- }
- // if server is websocket only, return after logging
- if !h.rpcAllowed() {
- return nil
- }
- // Log http endpoint.
- h.log.Info("HTTP server started",
- "endpoint", listener.Addr(),
- "prefix", h.httpConfig.prefix,
- "cors", strings.Join(h.httpConfig.CorsAllowedOrigins, ","),
- "vhosts", strings.Join(h.httpConfig.Vhosts, ","),
- "isTls", isTls,
- )
- // Log all handlers mounted on server.
- var paths []string
- for path := range h.handlerNames {
- paths = append(paths, path)
- }
- sort.Strings(paths)
- logged := make(map[string]bool, len(paths))
- for _, path := range paths {
- name := h.handlerNames[path]
- if !logged[name] {
- log.Info(name+" enabled", "url", "http://"+listener.Addr().String()+path)
- logged[name] = true
- }
- }
- return nil
- }
- func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
- // check if ws request and serve if ws enabled
- ws := h.wsHandler.Load().(*rpcHandler)
- if ws != nil && isWebsocket(r) {
- if checkPath(r, h.wsConfig.prefix) {
- ws.ServeHTTP(w, r)
- }
- return
- }
- // if http-rpc is enabled, try to serve request
- rpc := h.httpHandler.Load().(*rpcHandler)
- if rpc != nil {
- // First try to route in the mux.
- // Requests to a path below root are handled by the mux,
- // which has all the handlers registered via Node.RegisterHandler.
- // These are made available when RPC is enabled.
- muxHandler, pattern := h.mux.Handler(r)
- if pattern != "" {
- muxHandler.ServeHTTP(w, r)
- return
- }
- if checkPath(r, h.httpConfig.prefix) {
- rpc.ServeHTTP(w, r)
- return
- }
- }
- w.WriteHeader(http.StatusNotFound)
- }
- // checkPath checks whether a given request URL matches a given path prefix.
- func checkPath(r *http.Request, path string) bool {
- // if no prefix has been specified, request URL must be on root
- if path == "" {
- return r.URL.Path == "/"
- }
- // otherwise, check to make sure prefix matches
- return len(r.URL.Path) >= len(path) && r.URL.Path[:len(path)] == path
- }
- // validatePrefix checks if 'path' is a valid configuration value for the RPC prefix option.
- func validatePrefix(what, path string) error {
- if path == "" {
- return nil
- }
- if path[0] != '/' {
- return fmt.Errorf(`%s RPC path prefix %q does not contain leading "/"`, what, path)
- }
- if strings.ContainsAny(path, "?#") {
- // This is just to avoid confusion. While these would match correctly (i.e. they'd
- // match if URL-escaped into path), it's not easy to understand for users when
- // setting that on the command line.
- return fmt.Errorf("%s RPC path prefix %q contains URL meta-characters", what, path)
- }
- return nil
- }
- // stop shuts down the HTTP server.
- func (h *httpServer) stop() {
- h.mu.Lock()
- defer h.mu.Unlock()
- h.doStop()
- }
- func (h *httpServer) doStop() {
- if h.listener == nil {
- return // not running
- }
- // Shut down the server.
- httpHandler := h.httpHandler.Load().(*rpcHandler)
- wsHandler := h.httpHandler.Load().(*rpcHandler)
- if httpHandler != nil {
- h.httpHandler.Store((*rpcHandler)(nil))
- httpHandler.server.Stop()
- }
- if wsHandler != nil {
- h.wsHandler.Store((*rpcHandler)(nil))
- wsHandler.server.Stop()
- }
- h.server.Shutdown(context.Background())
- h.listener.Close()
- h.log.Info("HTTP server stopped", "endpoint", h.listener.Addr())
- // Clear out everything to allow re-configuring it later.
- h.host, h.port, h.endpoint = "", 0, ""
- h.server, h.listener = nil, nil
- }
- // Quorum - added argument `authManager` used to create protected server
- // enableRPC turns on JSON-RPC over HTTP on the server.
- func (h *httpServer) enableRPC(apis []rpc.API, config httpConfig, authManager security.AuthenticationManager) error {
- h.mu.Lock()
- defer h.mu.Unlock()
- if h.rpcAllowed() {
- return fmt.Errorf("JSON-RPC over HTTP is already enabled")
- }
- // Create RPC server and handler.
- srv := rpc.NewProtectedServer(authManager, h.isMultitenant)
- if err := RegisterApisFromWhitelist(apis, config.Modules, srv, false); err != nil {
- return err
- }
- h.httpConfig = config
- h.httpHandler.Store(&rpcHandler{
- Handler: NewHTTPHandlerStack(srv, config.CorsAllowedOrigins, config.Vhosts),
- server: srv,
- })
- return nil
- }
- // disableRPC stops the HTTP RPC handler. This is internal, the caller must hold h.mu.
- func (h *httpServer) disableRPC() bool {
- handler := h.httpHandler.Load().(*rpcHandler)
- if handler != nil {
- h.httpHandler.Store((*rpcHandler)(nil))
- handler.server.Stop()
- }
- return handler != nil
- }
- // Quorum - added argument `authManager` used to create protected server
- // enableWS turns on JSON-RPC over WebSocket on the server.
- func (h *httpServer) enableWS(apis []rpc.API, config wsConfig, authManager security.AuthenticationManager) error {
- h.mu.Lock()
- defer h.mu.Unlock()
- if h.wsAllowed() {
- return fmt.Errorf("JSON-RPC over WebSocket is already enabled")
- }
- // Create RPC server and handler.
- srv := rpc.NewProtectedServer(authManager, h.isMultitenant)
- if err := RegisterApisFromWhitelist(apis, config.Modules, srv, false); err != nil {
- return err
- }
- h.wsConfig = config
- h.wsHandler.Store(&rpcHandler{
- Handler: srv.WebsocketHandler(config.Origins),
- server: srv,
- })
- return nil
- }
- // stopWS disables JSON-RPC over WebSocket and also stops the server if it only serves WebSocket.
- func (h *httpServer) stopWS() {
- h.mu.Lock()
- defer h.mu.Unlock()
- if h.disableWS() {
- if !h.rpcAllowed() {
- h.doStop()
- }
- }
- }
- // disableWS disables the WebSocket handler. This is internal, the caller must hold h.mu.
- func (h *httpServer) disableWS() bool {
- ws := h.wsHandler.Load().(*rpcHandler)
- if ws != nil {
- h.wsHandler.Store((*rpcHandler)(nil))
- ws.server.Stop()
- }
- return ws != nil
- }
- // rpcAllowed returns true when JSON-RPC over HTTP is enabled.
- func (h *httpServer) rpcAllowed() bool {
- return h.httpHandler.Load().(*rpcHandler) != nil
- }
- // wsAllowed returns true when JSON-RPC over WebSocket is enabled.
- func (h *httpServer) wsAllowed() bool {
- return h.wsHandler.Load().(*rpcHandler) != nil
- }
- // isWebsocket checks the header of an http request for a websocket upgrade request.
- func isWebsocket(r *http.Request) bool {
- return strings.ToLower(r.Header.Get("Upgrade")) == "websocket" &&
- strings.Contains(strings.ToLower(r.Header.Get("Connection")), "upgrade")
- }
- // NewHTTPHandlerStack returns wrapped http-related handlers
- func NewHTTPHandlerStack(srv http.Handler, cors []string, vhosts []string) http.Handler {
- // Wrap the CORS-handler within a host-handler
- handler := newCorsHandler(srv, cors)
- handler = newVHostHandler(vhosts, handler)
- return newGzipHandler(handler)
- }
- func newCorsHandler(srv http.Handler, allowedOrigins []string) http.Handler {
- // disable CORS support if user has not specified a custom CORS configuration
- if len(allowedOrigins) == 0 {
- return srv
- }
- c := cors.New(cors.Options{
- AllowedOrigins: allowedOrigins,
- AllowedMethods: []string{http.MethodPost, http.MethodGet},
- AllowedHeaders: []string{"*"},
- MaxAge: 600,
- })
- return c.Handler(srv)
- }
- // virtualHostHandler is a handler which validates the Host-header of incoming requests.
- // Using virtual hosts can help prevent DNS rebinding attacks, where a 'random' domain name points to
- // the service ip address (but without CORS headers). By verifying the targeted virtual host, we can
- // ensure that it's a destination that the node operator has defined.
- type virtualHostHandler struct {
- vhosts map[string]struct{}
- next http.Handler
- }
- func newVHostHandler(vhosts []string, next http.Handler) http.Handler {
- vhostMap := make(map[string]struct{})
- for _, allowedHost := range vhosts {
- vhostMap[strings.ToLower(allowedHost)] = struct{}{}
- }
- return &virtualHostHandler{vhostMap, next}
- }
- // ServeHTTP serves JSON-RPC requests over HTTP, implements http.Handler
- func (h *virtualHostHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
- // if r.Host is not set, we can continue serving since a browser would set the Host header
- if r.Host == "" {
- h.next.ServeHTTP(w, r)
- return
- }
- host, _, err := net.SplitHostPort(r.Host)
- if err != nil {
- // Either invalid (too many colons) or no port specified
- host = r.Host
- }
- if ipAddr := net.ParseIP(host); ipAddr != nil {
- // It's an IP address, we can serve that
- h.next.ServeHTTP(w, r)
- return
- }
- // Not an IP address, but a hostname. Need to validate
- if _, exist := h.vhosts["*"]; exist {
- h.next.ServeHTTP(w, r)
- return
- }
- if _, exist := h.vhosts[host]; exist {
- h.next.ServeHTTP(w, r)
- return
- }
- http.Error(w, "invalid host specified", http.StatusForbidden)
- }
- var gzPool = sync.Pool{
- New: func() interface{} {
- w := gzip.NewWriter(ioutil.Discard)
- return w
- },
- }
- type gzipResponseWriter struct {
- io.Writer
- http.ResponseWriter
- }
- func (w *gzipResponseWriter) WriteHeader(status int) {
- w.Header().Del("Content-Length")
- w.ResponseWriter.WriteHeader(status)
- }
- func (w *gzipResponseWriter) Write(b []byte) (int, error) {
- return w.Writer.Write(b)
- }
- func newGzipHandler(next http.Handler) http.Handler {
- return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
- if !strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") {
- next.ServeHTTP(w, r)
- return
- }
- w.Header().Set("Content-Encoding", "gzip")
- gz := gzPool.Get().(*gzip.Writer)
- defer gzPool.Put(gz)
- gz.Reset(w)
- defer gz.Close()
- next.ServeHTTP(&gzipResponseWriter{ResponseWriter: w, Writer: gz}, r)
- })
- }
- type ipcServer struct {
- log log.Logger
- endpoint string
- mu sync.Mutex
- listener net.Listener
- srv *rpc.Server
- // Quorum
- // isMultitenant determines if the server supports mutlitenancy
- isMultitenant bool
- }
- func newIPCServer(log log.Logger, endpoint string) *ipcServer {
- return &ipcServer{log: log, endpoint: endpoint, isMultitenant: false}
- }
- // Quorum
- // withMultitenancy indicates if this server supports multitenancy
- func (is *ipcServer) withMultitenancy(b bool) *ipcServer {
- is.isMultitenant = b
- return is
- }
- // Start starts the httpServer's http.Server
- func (is *ipcServer) start(apis []rpc.API) error {
- is.mu.Lock()
- defer is.mu.Unlock()
- if is.listener != nil {
- return nil // already running
- }
- listener, srv, err := rpc.StartIPCEndpoint(is.endpoint, apis)
- if err != nil {
- is.log.Warn("IPC opening failed", "url", is.endpoint, "error", err)
- return err
- }
- srv.EnableMultitenancy(is.isMultitenant)
- is.log.Info("IPC endpoint opened", "url", is.endpoint, "isMultitenant", is.isMultitenant)
- is.listener, is.srv = listener, srv
- return nil
- }
- func (is *ipcServer) stop() error {
- is.mu.Lock()
- defer is.mu.Unlock()
- if is.listener == nil {
- return nil // not running
- }
- err := is.listener.Close()
- is.srv.Stop()
- is.listener, is.srv = nil, nil
- is.log.Info("IPC endpoint closed", "url", is.endpoint)
- return err
- }
- // RegisterApisFromWhitelist checks the given modules' availability, generates a whitelist based on the allowed modules,
- // and then registers all of the APIs exposed by the services.
- func RegisterApisFromWhitelist(apis []rpc.API, modules []string, srv *rpc.Server, exposeAll bool) error {
- if bad, available := checkModuleAvailability(modules, apis); len(bad) > 0 {
- log.Error("Unavailable modules in HTTP API list", "unavailable", bad, "available", available)
- }
- // Generate the whitelist based on the allowed modules
- whitelist := make(map[string]bool)
- for _, module := range modules {
- whitelist[module] = true
- }
- // Register all the APIs exposed by the services
- for _, api := range apis {
- if exposeAll || whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) {
- if err := srv.RegisterName(api.Namespace, api.Service); err != nil {
- return err
- }
- }
- }
- return nil
- }
|