Add logging for debug and more whitespace protection in the isWebsocket check

Add unit tests to try and reproduce the issue as well
This commit is contained in:
Tate 2023-10-06 14:32:53 -06:00
parent 052355f5e2
commit 1df0e2ef64
No known key found for this signature in database
GPG key ID: 3D78D8779CCC7D71
3 changed files with 194 additions and 30 deletions

View file

@ -521,8 +521,10 @@ func (n *Node) wsServerForPort(port int, authenticated bool) *httpServer {
httpServer, wsServer = n.httpAuth, n.wsAuth httpServer, wsServer = n.httpAuth, n.wsAuth
} }
if n.config.HTTPHost == "" || httpServer.port == port { if n.config.HTTPHost == "" || httpServer.port == port {
n.log.Info("CONFIGURED HTTP SERVER INSTEAD OF WS SERVER")
return httpServer return httpServer
} }
n.log.Info("PROPERLY RETURNED THE WS SERVER")
return wsServer return wsServer
} }
@ -683,7 +685,7 @@ func (n *Node) HTTPEndpoint() string {
// WSEndpoint returns the current JSON-RPC over WebSocket endpoint. // WSEndpoint returns the current JSON-RPC over WebSocket endpoint.
func (n *Node) WSEndpoint() string { func (n *Node) WSEndpoint() string {
if n.http.wsAllowed() { if n.http.wsAllowed("WSEndpoint") {
return "ws://" + n.http.listenAddr() + n.http.wsConfig.prefix return "ws://" + n.http.listenAddr() + n.http.wsConfig.prefix
} }
return "ws://" + n.ws.listenAddr() + n.ws.wsConfig.prefix return "ws://" + n.ws.listenAddr() + n.ws.wsConfig.prefix
@ -696,7 +698,7 @@ func (n *Node) HTTPAuthEndpoint() string {
// WSAuthEndpoint returns the current authenticated JSON-RPC over WebSocket endpoint. // WSAuthEndpoint returns the current authenticated JSON-RPC over WebSocket endpoint.
func (n *Node) WSAuthEndpoint() string { func (n *Node) WSAuthEndpoint() string {
if n.httpAuth.wsAllowed() { if n.httpAuth.wsAllowed("WSAuthEndpoint") {
return "ws://" + n.httpAuth.listenAddr() + n.httpAuth.wsConfig.prefix return "ws://" + n.httpAuth.listenAddr() + n.httpAuth.wsConfig.prefix
} }
return "ws://" + n.wsAuth.listenAddr() + n.wsAuth.wsConfig.prefix return "ws://" + n.wsAuth.listenAddr() + n.wsAuth.wsConfig.prefix

View file

@ -24,12 +24,15 @@ import (
"net/http" "net/http"
"reflect" "reflect"
"strings" "strings"
"sync"
"testing" "testing"
"time"
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
"github.com/gorilla/websocket"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
) )
@ -479,6 +482,93 @@ func TestWebsocketHTTPOnSeparatePort_WSRequest(t *testing.T) {
} }
} }
func TestWebSocketRouting(t *testing.T) {
node := startHTTP(t, 8544, 8545)
defer node.Close()
// Start both the HTTP and WS servers
httpServer := node.http
wsServer := node.ws
// Define the number of requests you'll be sending to each server
numRequests := 1000
// Use wait groups to synchronize the end of our goroutines
var wg sync.WaitGroup
hs := "http://" + httpServer.endpoint
// HTTP requests
wg.Add(1)
go func() {
defer wg.Done()
for i := 0; i < numRequests; i++ {
resp, err := http.Get(hs)
if err != nil {
t.Error(err)
}
// Add additional checks on the response if necessary
resp.Body.Close()
}
}()
s := "ws://" + wsServer.endpoint
// WebSocket requests
wg.Add(1)
go func() {
defer wg.Done()
for i := 0; i < numRequests; i++ {
// Create a new WebSocket connection and make sure it's successful
conn, _, err := websocket.DefaultDialer.Dial(s, nil)
if err != nil {
t.Error(err)
}
conn.Close()
}
}()
wg.Wait()
}
func TestWebSocketStartup(t *testing.T) {
// Define the number of startup cycles
numCycles := 1000
for cycle := 0; cycle < numCycles; cycle++ {
testFunc(t, cycle)
fmt.Println("Cycle", cycle, "complete")
}
}
func testFunc(t *testing.T, cycle int) {
// Start the WS server
node := startHTTP(t, 8544, 8545)
defer node.Close()
// Start both the HTTP and WS servers
// httpServer := node.http
wsServer := node.ws
ws := "ws://" + wsServer.endpoint
// Introduce a slight delay before sending the first request
time.Sleep(50 * time.Millisecond) // Adjust this delay as necessary
// Define the number of requests you'll be sending to the WS server during each cycle
numRequests := 2
// Send requests to the WebSocket server
for i := 0; i < numRequests; i++ {
// Create a new WebSocket connection and make sure it's successful
conn, _, err := websocket.DefaultDialer.Dial(ws, nil)
if err != nil {
t.Errorf("Error during cycle %d, request %d: %v", cycle, i, err)
}
conn.Close()
}
// Stop the WS server
node.Close()
}
type rpcPrefixTest struct { type rpcPrefixTest struct {
httpPrefix, wsPrefix string httpPrefix, wsPrefix string
// These lists paths on which JSON-RPC should be served / not served. // These lists paths on which JSON-RPC should be served / not served.
@ -586,10 +676,12 @@ func (test rpcPrefixTest) check(t *testing.T, node *Node) {
func createNode(t *testing.T, httpPort, wsPort int) *Node { func createNode(t *testing.T, httpPort, wsPort int) *Node {
conf := &Config{ conf := &Config{
HTTPHost: "127.0.0.1", HTTPHost: "0.0.0.0",
HTTPPort: httpPort, HTTPPort: httpPort,
WSHost: "127.0.0.1", HTTPVirtualHosts: []string{"*"},
WSHost: "0.0.0.0",
WSPort: wsPort, WSPort: wsPort,
WSOrigins: []string{"*"},
HTTPTimeouts: rpc.DefaultHTTPTimeouts, HTTPTimeouts: rpc.DefaultHTTPTimeouts,
} }
node, err := New(conf) node, err := New(conf)

View file

@ -64,6 +64,7 @@ type rpcHandler struct {
} }
type httpServer struct { type httpServer struct {
name string
log log.Logger log log.Logger
timeouts rpc.HTTPTimeouts timeouts rpc.HTTPTimeouts
mux http.ServeMux // registered handlers go here mux http.ServeMux // registered handlers go here
@ -97,10 +98,23 @@ func newHTTPServer(log log.Logger, timeouts rpc.HTTPTimeouts) *httpServer {
h := &httpServer{log: log, timeouts: timeouts, handlerNames: make(map[string]string)} h := &httpServer{log: log, timeouts: timeouts, handlerNames: make(map[string]string)}
h.httpHandler.Store((*rpcHandler)(nil)) h.httpHandler.Store((*rpcHandler)(nil))
h.wsHandler.Store((*rpcHandler)(nil)) h.storeWsHandler((*rpcHandler)(nil), "newHTTPServer")
return h return h
} }
func (h *httpServer) storeWsHandler(val any, where string) {
h.wsHandler.Store(val)
if val != nil {
h.log.Info(fmt.Sprintf("storeWsHandler %s", where), "ws", val, "h", h)
}
}
func (h *httpServer) loadWsHandler(where string) *rpcHandler {
ws := h.wsHandler.Load().(*rpcHandler)
h.log.Info(fmt.Sprintf("loadWsHandler %s", where), "ws", ws, "h", h)
return ws
}
// setListenAddr configures the listening address of the server. // setListenAddr configures the listening address of the server.
// The address can only be set while the server isn't running. // The address can only be set while the server isn't running.
func (h *httpServer) setListenAddr(host string, port int) error { func (h *httpServer) setListenAddr(host string, port int) error {
@ -131,6 +145,7 @@ func (h *httpServer) listenAddr() string {
func (h *httpServer) start() error { func (h *httpServer) start() error {
h.mu.Lock() h.mu.Lock()
defer h.mu.Unlock() defer h.mu.Unlock()
h.log.Info("RUN start()")
if h.endpoint == "" || h.listener != nil { if h.endpoint == "" || h.listener != nil {
return nil // already running or not configured return nil // already running or not configured
@ -156,19 +171,31 @@ func (h *httpServer) start() error {
return err return err
} }
h.listener = listener h.listener = listener
go h.server.Serve(listener) // go func() {
// h.log.Info("STARTING RPC SERVER")
// h.server.Serve(listener)
// }()
if h.wsAllowed() { if h.wsAllowed("start") {
url := fmt.Sprintf("ws://%v", listener.Addr()) url := fmt.Sprintf("ws://%v", listener.Addr())
if h.wsConfig.prefix != "" { if h.wsConfig.prefix != "" {
url += h.wsConfig.prefix url += h.wsConfig.prefix
} }
h.log.Info("WebSocket enabled", "url", url) h.log.Info("WebSocket enabled", "url", url)
h.name = "ws"
// go func() {
// h.log.Info("STARTING RPC SERVER")
// h.server.Serve(listener)
// }()
} else {
h.name = "http"
} }
// if server is websocket only, return after logging
if !h.rpcAllowed() { go func() {
return nil h.log.Info("STARTING RPC SERVER")
} h.server.Serve(listener)
}()
// Log http endpoint. // Log http endpoint.
h.log.Info("HTTP server started", h.log.Info("HTTP server started",
"endpoint", listener.Addr(), "auth", (h.httpConfig.jwtSecret != nil), "endpoint", listener.Addr(), "auth", (h.httpConfig.jwtSecret != nil),
@ -177,6 +204,18 @@ func (h *httpServer) start() error {
"vhosts", strings.Join(h.httpConfig.Vhosts, ","), "vhosts", strings.Join(h.httpConfig.Vhosts, ","),
) )
// 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(), "auth", (h.httpConfig.jwtSecret != nil),
// "prefix", h.httpConfig.prefix,
// "cors", strings.Join(h.httpConfig.CorsAllowedOrigins, ","),
// "vhosts", strings.Join(h.httpConfig.Vhosts, ","),
// )
// Log all handlers mounted on server. // Log all handlers mounted on server.
var paths []string var paths []string
for path := range h.handlerNames { for path := range h.handlerNames {
@ -191,18 +230,35 @@ func (h *httpServer) start() error {
logged[name] = true logged[name] = true
} }
} }
return nil return nil
} }
func printHeaders(h *httpServer, headers http.Header) {
for key, values := range headers {
for _, value := range values {
h.log.Info("HEADER", key, value)
}
}
}
func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
h.log.Info("REQUEST", "r", r)
ws := h.loadWsHandler("ServeHTTP")
h.log.Info("TRIAGE", "name", h.name, "ws", ws, "h", h, "isWebsocket", isWebsocket(r), "upgrade", r.Header.Get("Upgrade"), "connection", r.Header.Get("Connection"))
printHeaders(h, r.Header)
// check if ws request and serve if ws enabled // check if ws request and serve if ws enabled
ws := h.wsHandler.Load().(*rpcHandler) // if ws != nil && isWebsocket(r) {
if ws != nil && isWebsocket(r) { if isWebsocket(r) {
if ws != nil {
if checkPath(r, h.wsConfig.prefix) { if checkPath(r, h.wsConfig.prefix) {
ws.ServeHTTP(w, r) ws.ServeHTTP(w, r)
} }
return return
} }
panic("TRIAGE FAILED WS")
}
// if http-rpc is enabled, try to serve request // if http-rpc is enabled, try to serve request
rpc := h.httpHandler.Load().(*rpcHandler) rpc := h.httpHandler.Load().(*rpcHandler)
@ -266,13 +322,13 @@ func (h *httpServer) doStop() {
// Shut down the server. // Shut down the server.
httpHandler := h.httpHandler.Load().(*rpcHandler) httpHandler := h.httpHandler.Load().(*rpcHandler)
wsHandler := h.wsHandler.Load().(*rpcHandler) wsHandler := h.loadWsHandler("doStop")
if httpHandler != nil { if httpHandler != nil {
h.httpHandler.Store((*rpcHandler)(nil)) h.httpHandler.Store((*rpcHandler)(nil))
httpHandler.server.Stop() httpHandler.server.Stop()
} }
if wsHandler != nil { if wsHandler != nil {
h.wsHandler.Store((*rpcHandler)(nil)) h.storeWsHandler((*rpcHandler)(nil), "doStop")
wsHandler.server.Stop() wsHandler.server.Stop()
} }
@ -330,7 +386,7 @@ func (h *httpServer) enableWS(apis []rpc.API, config wsConfig) error {
h.mu.Lock() h.mu.Lock()
defer h.mu.Unlock() defer h.mu.Unlock()
if h.wsAllowed() { if h.wsAllowed("enableWS") {
return fmt.Errorf("JSON-RPC over WebSocket is already enabled") return fmt.Errorf("JSON-RPC over WebSocket is already enabled")
} }
// Create RPC server and handler. // Create RPC server and handler.
@ -340,10 +396,14 @@ func (h *httpServer) enableWS(apis []rpc.API, config wsConfig) error {
return err return err
} }
h.wsConfig = config h.wsConfig = config
h.wsHandler.Store(&rpcHandler{
h.storeWsHandler(&rpcHandler{
Handler: NewWSHandlerStack(srv.WebsocketHandler(config.Origins), config.jwtSecret), Handler: NewWSHandlerStack(srv.WebsocketHandler(config.Origins), config.jwtSecret),
server: srv, server: srv,
}) }, "enableWS")
// ws := h.loadWsHandler("enableWS")
// h.log.Info("enableWS wsHandler = YES", "ws", ws)
// h.startRoutineToCheckWsHandler()
return nil return nil
} }
@ -359,11 +419,21 @@ func (h *httpServer) stopWS() {
} }
} }
// func (h *httpServer) startRoutineToCheckWsHandler() {
// // Goroutine to check the value every second
// go func() {
// for range time.Tick(1 * time.Second) {
// ws := h.loadWsHandler()
// h.log.Info("check wsHandler", "ws", ws, "h", h)
// }
// }()
// }
// disableWS disables the WebSocket handler. This is internal, the caller must hold h.mu. // disableWS disables the WebSocket handler. This is internal, the caller must hold h.mu.
func (h *httpServer) disableWS() bool { func (h *httpServer) disableWS() bool {
ws := h.wsHandler.Load().(*rpcHandler) ws := h.loadWsHandler("disableWS")
if ws != nil { if ws != nil {
h.wsHandler.Store((*rpcHandler)(nil)) h.storeWsHandler((*rpcHandler)(nil), "disableWS")
ws.server.Stop() ws.server.Stop()
} }
return ws != nil return ws != nil
@ -375,8 +445,8 @@ func (h *httpServer) rpcAllowed() bool {
} }
// wsAllowed returns true when JSON-RPC over WebSocket is enabled. // wsAllowed returns true when JSON-RPC over WebSocket is enabled.
func (h *httpServer) wsAllowed() bool { func (h *httpServer) wsAllowed(where string) bool {
return h.wsHandler.Load().(*rpcHandler) != nil return h.loadWsHandler(fmt.Sprintf("wsAllowed %s", where)) != nil
} }
// isWebsocket checks the header of an http request for a websocket upgrade request. // isWebsocket checks the header of an http request for a websocket upgrade request.