gaspersic/go-ethereum
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

rpc: implement websockets with github.com/gorilla/websocket (#19866)

Browse Source 
* rpc: implement websockets with github.com/gorilla/websocket

This change makes package rpc use the github.com/gorilla/websocket
package for WebSockets instead of golang.org/x/net/websocket. The new
library is more robust and supports all WebSocket features including
continuation frames.

There are new tests for two issues with the previously-used library:

  - TestWebsocketClientPing checks handling of Ping frames.
  - TestWebsocketLargeCall checks whether the request size limit is
    applied correctly.

* rpc: raise HTTP/WebSocket request size limit to 5MB

* rpc: remove default origin for client connections

The client used to put the local hostname into the Origin header because
the server wanted an origin to accept the connection, but that's silly:
Origin is for browsers/websites. The nobody would whitelist a particular
hostname.

Now that the server doesn't need Origin anymore, don't bother setting
one for clients. Users who need an origin can use DialWebsocket to
create a client with arbitrary origin if needed.

* vendor: put golang.org/x/net/websocket back

* rpc: don't set Origin header for empty (default) origin

* rpc: add HTTP status code to handshake error

This makes it easier to debug failing connections.

* ethstats: use github.com/gorilla/websocket

* rpc: fix lint
This commit is contained in:
Felix Lange
2019-07-22 12:22:39 +02:00
committed by Péter Szilágyi
parent e8141e1685
commit 04e175b8ec
32 changed files with 3961 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
rpc/client.go
View File

@ -41,9 +41,8 @@ var (
const (
// Timeouts
tcpKeepAliveInterval = 30 * time.Second
defaultDialTimeout = 10 * time.Second // used if context has no deadline
subscribeTimeout = 5 * time.Second // overall timeout eth_subscribe, rpc_modules calls
defaultDialTimeout = 10 * time.Second // used if context has no deadline
subscribeTimeout = 5 * time.Second // overall timeout eth_subscribe, rpc_modules calls
)
const (

2
rpc/http.go
View File

@ -36,7 +36,7 @@ import (
)
const (
maxRequestContentLength = 1024 * 512
maxRequestContentLength = 1024 * 1024 * 5
contentType = "application/json"
)

2
rpc/ipc_unix.go
View File

@ -50,5 +50,5 @@ func ipcListen(endpoint string) (net.Listener, error) {
// newIPCConnection will connect to a Unix socket on the given endpoint.
func newIPCConnection(ctx context.Context, endpoint string) (net.Conn, error) {
return dialContext(ctx, "unix", endpoint)
return new(net.Dialer).DialContext(ctx, "unix", endpoint)
}

17
rpc/json.go
View File

@ -141,6 +141,11 @@ type Conn interface {
SetWriteDeadline(time.Time) error
}
type deadlineCloser interface {
io.Closer
SetWriteDeadline(time.Time) error
}
// ConnRemoteAddr wraps the RemoteAddr operation, which returns a description
// of the peer address of a connection. If a Conn also implements ConnRemoteAddr, this
// description is used in log messages.
@ -165,12 +170,10 @@ type jsonCodec struct {
decode func(v interface{}) error // decoder to allow multiple transports
encMu sync.Mutex // guards the encoder
encode func(v interface{}) error // encoder to allow multiple transports
conn Conn
conn deadlineCloser
}
// NewCodec creates a new RPC server codec with support for JSON-RPC 2.0 based
// on explicitly given encoding and decoding methods.
func NewCodec(conn Conn, encode, decode func(v interface{}) error) ServerCodec {
func newCodec(conn deadlineCloser, encode, decode func(v interface{}) error) ServerCodec {
codec := &jsonCodec{
closed: make(chan interface{}),
encode: encode,
@ -183,12 +186,14 @@ func NewCodec(conn Conn, encode, decode func(v interface{}) error) ServerCodec {
return codec
}
// NewJSONCodec creates a new RPC server codec with support for JSON-RPC 2.0.
// NewJSONCodec creates a codec that reads from the given connection. If conn implements
// ConnRemoteAddr, log messages will use it to include the remote address of the
// connection.
func NewJSONCodec(conn Conn) ServerCodec {
enc := json.NewEncoder(conn)
dec := json.NewDecoder(conn)
dec.UseNumber()
return NewCodec(conn, enc.Encode, dec.Decode)
return newCodec(conn, enc.Encode, dec.Decode)
}
func (c *jsonCodec) RemoteAddr() string {

203
rpc/websocket.go
View File

@ -17,78 +17,26 @@
package rpc
import (
"bytes"
"context"
"crypto/tls"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"net"
"net/http"
"net/url"
"os"
"strings"
"time"
"sync"
mapset "github.com/deckarep/golang-set"
"github.com/ethereum/go-ethereum/log"
"golang.org/x/net/websocket"
"github.com/gorilla/websocket"
)
// websocketJSONCodec is a custom JSON codec with payload size enforcement and
// special number parsing.
var websocketJSONCodec = websocket.Codec{
// Marshal is the stock JSON marshaller used by the websocket library too.
Marshal: func(v interface{}) ([]byte, byte, error) {
msg, err := json.Marshal(v)
return msg, websocket.TextFrame, err
},
// Unmarshal is a specialized unmarshaller to properly convert numbers.
Unmarshal: func(msg []byte, payloadType byte, v interface{}) error {
dec := json.NewDecoder(bytes.NewReader(msg))
dec.UseNumber()
const (
wsReadBuffer = 1024
wsWriteBuffer = 1024
)
return dec.Decode(v)
},
}
// WebsocketHandler returns a handler that serves JSON-RPC to WebSocket connections.
//
// allowedOrigins should be a comma-separated list of allowed origin URLs.
// To allow connections with any origin, pass "*".
func (s *Server) WebsocketHandler(allowedOrigins []string) http.Handler {
return websocket.Server{
Handshake: wsHandshakeValidator(allowedOrigins),
Handler: func(conn *websocket.Conn) {
codec := newWebsocketCodec(conn)
s.ServeCodec(codec, OptionMethodInvocation|OptionSubscriptions)
},
}
}
func newWebsocketCodec(conn *websocket.Conn) ServerCodec {
// Create a custom encode/decode pair to enforce payload size and number encoding
conn.MaxPayloadBytes = maxRequestContentLength
encoder := func(v interface{}) error {
return websocketJSONCodec.Send(conn, v)
}
decoder := func(v interface{}) error {
return websocketJSONCodec.Receive(conn, v)
}
rpcconn := Conn(conn)
if conn.IsServerConn() {
// Override remote address with the actual socket address because
// package websocket crashes if there is no request origin.
addr := conn.Request().RemoteAddr
if wsaddr := conn.RemoteAddr().(*websocket.Addr); wsaddr.URL != nil {
// Add origin if present.
addr += "(" + wsaddr.URL.String() + ")"
}
rpcconn = connWithRemoteAddr{conn, addr}
}
return NewCodec(rpcconn, encoder, decoder)
}
var wsBufferPool = new(sync.Pool)
// NewWSServer creates a new websocket RPC server around an API provider.
//
@ -97,10 +45,32 @@ func NewWSServer(allowedOrigins []string, srv *Server) *http.Server {
return &http.Server{Handler: srv.WebsocketHandler(allowedOrigins)}
}
// WebsocketHandler returns a handler that serves JSON-RPC to WebSocket connections.
//
// allowedOrigins should be a comma-separated list of allowed origin URLs.
// To allow connections with any origin, pass "*".
func (s *Server) WebsocketHandler(allowedOrigins []string) http.Handler {
var upgrader = websocket.Upgrader{
ReadBufferSize: wsReadBuffer,
WriteBufferSize: wsWriteBuffer,
WriteBufferPool: wsBufferPool,
CheckOrigin: wsHandshakeValidator(allowedOrigins),
}
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
log.Debug("WebSocket upgrade failed", "err", err)
return
}
codec := newWebsocketCodec(conn)
s.ServeCodec(codec, OptionMethodInvocation|OptionSubscriptions)
})
}
// wsHandshakeValidator returns a handler that verifies the origin during the
// websocket upgrade process. When a '*' is specified as an allowed origins all
// connections are accepted.
func wsHandshakeValidator(allowedOrigins []string) func(*websocket.Config, *http.Request) error {
func wsHandshakeValidator(allowedOrigins []string) func(*http.Request) bool {
origins := mapset.NewSet()
allowAllOrigins := false
@ -112,7 +82,6 @@ func wsHandshakeValidator(allowedOrigins []string) func(*websocket.Config, *http
origins.Add(strings.ToLower(origin))
}
}
// allow localhost if no allowedOrigins are specified.
if len(origins.ToSlice()) == 0 {
origins.Add("http://localhost")
@ -120,52 +89,39 @@ func wsHandshakeValidator(allowedOrigins []string) func(*websocket.Config, *http
origins.Add("http://" + strings.ToLower(hostname))
}
}
log.Debug(fmt.Sprintf("Allowed origin(s) for WS RPC interface %v", origins.ToSlice()))
f := func(cfg *websocket.Config, req *http.Request) error {
f := func(req *http.Request) bool {
// Skip origin verification if no Origin header is present. The origin check
// is supposed to protect against browser based attacks. Browsers always set
// Origin. Non-browser software can put anything in origin and checking it doesn't
// provide additional security.
if _, ok := req.Header["Origin"]; !ok {
return nil
return true
}
// Verify origin against whitelist.
origin := strings.ToLower(req.Header.Get("Origin"))
if allowAllOrigins || origins.Contains(origin) {
return nil
return true
}
log.Warn("Rejected WebSocket connection", "origin", origin)
return errors.New("origin not allowed")
return false
}
return f
}
func wsGetConfig(endpoint, origin string) (*websocket.Config, error) {
if origin == "" {
var err error
if origin, err = os.Hostname(); err != nil {
return nil, err
}
if strings.HasPrefix(endpoint, "wss") {
origin = "https://" + strings.ToLower(origin)
} else {
origin = "http://" + strings.ToLower(origin)
}
}
config, err := websocket.NewConfig(endpoint, origin)
if err != nil {
return nil, err
}
type wsHandshakeError struct {
err error
status string
}
if config.Location.User != nil {
b64auth := base64.StdEncoding.EncodeToString([]byte(config.Location.User.String()))
config.Header.Add("Authorization", "Basic "+b64auth)
config.Location.User = nil
func (e wsHandshakeError) Error() string {
s := e.err.Error()
if e.status != "" {
s += " (HTTP status " + e.status + ")"
}
return config, nil
return s
}
// DialWebsocket creates a new RPC client that communicates with a JSON-RPC server
@ -174,65 +130,46 @@ func wsGetConfig(endpoint, origin string) (*websocket.Config, error) {
// The context is used for the initial connection establishment. It does not
// affect subsequent interactions with the client.
func DialWebsocket(ctx context.Context, endpoint, origin string) (*Client, error) {
config, err := wsGetConfig(endpoint, origin)
endpoint, header, err := wsClientHeaders(endpoint, origin)
if err != nil {
return nil, err
}
dialer := websocket.Dialer{
ReadBufferSize: wsReadBuffer,
WriteBufferSize: wsWriteBuffer,
WriteBufferPool: wsBufferPool,
}
return newClient(ctx, func(ctx context.Context) (ServerCodec, error) {
conn, err := wsDialContext(ctx, config)
conn, resp, err := dialer.DialContext(ctx, endpoint, header)
if err != nil {
return nil, err
hErr := wsHandshakeError{err: err}
if resp != nil {
hErr.status = resp.Status
}
return nil, hErr
}
return newWebsocketCodec(conn), nil
})
}
func wsDialContext(ctx context.Context, config *websocket.Config) (*websocket.Conn, error) {
var conn net.Conn
var err error
switch config.Location.Scheme {
case "ws":
conn, err = dialContext(ctx, "tcp", wsDialAddress(config.Location))
case "wss":
dialer := contextDialer(ctx)
conn, err = tls.DialWithDialer(dialer, "tcp", wsDialAddress(config.Location), config.TlsConfig)
default:
err = websocket.ErrBadScheme
}
func wsClientHeaders(endpoint, origin string) (string, http.Header, error) {
endpointURL, err := url.Parse(endpoint)
if err != nil {
return nil, err
return endpoint, nil, err
}
ws, err := websocket.NewClient(config, conn)
if err != nil {
conn.Close()
return nil, err
header := make(http.Header)
if origin != "" {
header.Add("origin", origin)
}
return ws, err
if endpointURL.User != nil {
b64auth := base64.StdEncoding.EncodeToString([]byte(endpointURL.User.String()))
header.Add("authorization", "Basic "+b64auth)
endpointURL.User = nil
}
return endpointURL.String(), header, nil
}
var wsPortMap = map[string]string{"ws": "80", "wss": "443"}
func wsDialAddress(location *url.URL) string {
if _, ok := wsPortMap[location.Scheme]; ok {
if _, _, err := net.SplitHostPort(location.Host); err != nil {
return net.JoinHostPort(location.Host, wsPortMap[location.Scheme])
}
}
return location.Host
}
func dialContext(ctx context.Context, network, addr string) (net.Conn, error) {
d := &net.Dialer{KeepAlive: tcpKeepAliveInterval}
return d.DialContext(ctx, network, addr)
}
func contextDialer(ctx context.Context) *net.Dialer {
dialer := &net.Dialer{Cancel: ctx.Done(), KeepAlive: tcpKeepAliveInterval}
if deadline, ok := ctx.Deadline(); ok {
dialer.Deadline = deadline
} else {
dialer.Deadline = time.Now().Add(defaultDialTimeout)
}
return dialer
func newWebsocketCodec(conn *websocket.Conn) ServerCodec {
conn.SetReadLimit(maxRequestContentLength)
return newCodec(conn, conn.WriteJSON, conn.ReadJSON)
}

251
rpc/websocket_test.go
View File

@ -16,31 +16,244 @@
package rpc
import "testing"
import (
"context"
"net"
"net/http"
"net/http/httptest"
"reflect"
"strings"
"testing"
"time"
func TestWSGetConfigNoAuth(t *testing.T) {
config, err := wsGetConfig("ws://example.com:1234", "")
"github.com/gorilla/websocket"
)
func TestWebsocketClientHeaders(t *testing.T) {
t.Parallel()
endpoint, header, err := wsClientHeaders("wss://testuser:test-PASS_01@example.com:1234", "https://example.com")
if err != nil {
t.Fatalf("wsGetConfig failed: %s", err)
}
if config.Location.User != nil {
t.Fatalf("User should have been stripped from the URL")
}
if config.Location.Hostname() != "example.com" ||
config.Location.Port() != "1234" || config.Location.Scheme != "ws" {
t.Fatalf("Unexpected URL: %s", config.Location)
}
}
func TestWSGetConfigWithBasicAuth(t *testing.T) {
config, err := wsGetConfig("wss://testuser:test-PASS_01@example.com:1234", "")
if err != nil {
t.Fatalf("wsGetConfig failed: %s", err)
}
if config.Location.User != nil {
if endpoint != "wss://example.com:1234" {
t.Fatal("User should have been stripped from the URL")
}
if config.Header.Get("Authorization") != "Basic dGVzdHVzZXI6dGVzdC1QQVNTXzAx" {
if header.Get("authorization") != "Basic dGVzdHVzZXI6dGVzdC1QQVNTXzAx" {
t.Fatal("Basic auth header is incorrect")
}
if header.Get("origin") != "https://example.com" {
t.Fatal("Origin not set")
}
}
// This test checks that the server rejects connections from disallowed origins.
func TestWebsocketOriginCheck(t *testing.T) {
t.Parallel()
var (
srv = newTestServer()
httpsrv = httptest.NewServer(srv.WebsocketHandler([]string{"http://example.com"}))
wsURL = "ws:" + strings.TrimPrefix(httpsrv.URL, "http:")
)
defer srv.Stop()
defer httpsrv.Close()
client, err := DialWebsocket(context.Background(), wsURL, "http://ekzample.com")
if err == nil {
client.Close()
t.Fatal("no error for wrong origin")
}
wantErr := wsHandshakeError{websocket.ErrBadHandshake, "403 Forbidden"}
if !reflect.DeepEqual(err, wantErr) {
t.Fatalf("wrong error for wrong origin: %q", err)
}
// Connections without origin header should work.
client, err = DialWebsocket(context.Background(), wsURL, "")
if err != nil {
t.Fatal("error for empty origin")
}
client.Close()
}
// This test checks whether calls exceeding the request size limit are rejected.
func TestWebsocketLargeCall(t *testing.T) {
t.Parallel()
var (
srv = newTestServer()
httpsrv = httptest.NewServer(srv.WebsocketHandler([]string{"*"}))
wsURL = "ws:" + strings.TrimPrefix(httpsrv.URL, "http:")
)
defer srv.Stop()
defer httpsrv.Close()
client, err := DialWebsocket(context.Background(), wsURL, "")
if err != nil {
t.Fatalf("can't dial: %v", err)
}
defer client.Close()
// This call sends slightly less than the limit and should work.
var result Result
arg := strings.Repeat("x", maxRequestContentLength-200)
if err := client.Call(&result, "test_echo", arg, 1); err != nil {
t.Fatalf("valid call didn't work: %v", err)
}
if result.String != arg {
t.Fatal("wrong string echoed")
}
// This call sends twice the allowed size and shouldn't work.
arg = strings.Repeat("x", maxRequestContentLength*2)
err = client.Call(&result, "test_echo", arg)
if err == nil {
t.Fatal("no error for too large call")
}
}
// This test checks that client handles WebSocket ping frames correctly.
func TestClientWebsocketPing(t *testing.T) {
t.Parallel()
var (
sendPing = make(chan struct{})
server = wsPingTestServer(t, sendPing)
ctx, cancel = context.WithTimeout(context.Background(), 1*time.Second)
)
defer cancel()
defer server.Shutdown(ctx)
client, err := DialContext(ctx, "ws://"+server.Addr)
if err != nil {
t.Fatalf("client dial error: %v", err)
}
resultChan := make(chan int)
sub, err := client.EthSubscribe(ctx, resultChan, "foo")
if err != nil {
t.Fatalf("client subscribe error: %v", err)
}
// Wait for the context's deadline to be reached before proceeding.
// This is important for reproducing https://github.com/ethereum/go-ethereum/issues/19798
<-ctx.Done()
close(sendPing)
// Wait for the subscription result.
timeout := time.NewTimer(5 * time.Second)
for {
select {
case err := <-sub.Err():
t.Error("client subscription error:", err)
case result := <-resultChan:
t.Log("client got result:", result)
return
case <-timeout.C:
t.Error("didn't get any result within the test timeout")
return
}
}
}
// wsPingTestServer runs a WebSocket server which accepts a single subscription request.
// When a value arrives on sendPing, the server sends a ping frame, waits for a matching
// pong and finally delivers a single subscription result.
func wsPingTestServer(t *testing.T, sendPing <-chan struct{}) *http.Server {
var srv http.Server
shutdown := make(chan struct{})
srv.RegisterOnShutdown(func() {
close(shutdown)
})
srv.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Upgrade to WebSocket.
upgrader := websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool { return true },
}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
t.Errorf("server WS upgrade error: %v", err)
return
}
defer conn.Close()
// Handle the connection.
wsPingTestHandler(t, conn, shutdown, sendPing)
})
// Start the server.
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal("can't listen:", err)
}
srv.Addr = listener.Addr().String()
go srv.Serve(listener)
return &srv
}
func wsPingTestHandler(t *testing.T, conn *websocket.Conn, shutdown, sendPing <-chan struct{}) {
// Canned responses for the eth_subscribe call in TestClientWebsocketPing.
const (
subResp = `{"jsonrpc":"2.0","id":1,"result":"0x00"}`
subNotify = `{"jsonrpc":"2.0","method":"eth_subscription","params":{"subscription":"0x00","result":1}}`
)
// Handle subscribe request.
if _, _, err := conn.ReadMessage(); err != nil {
t.Errorf("server read error: %v", err)
return
}
if err := conn.WriteMessage(websocket.TextMessage, []byte(subResp)); err != nil {
t.Errorf("server write error: %v", err)
return
}
// Read from the connection to process control messages.
var pongCh = make(chan string)
conn.SetPongHandler(func(d string) error {
t.Logf("server got pong: %q", d)
pongCh <- d
return nil
})
go func() {
for {
typ, msg, err := conn.ReadMessage()
if err != nil {
return
}
t.Logf("server got message (%d): %q", typ, msg)
}
}()
// Write messages.
var (
sendResponse <-chan time.Time
wantPong string
)
for {
select {
case _, open := <-sendPing:
if !open {
sendPing = nil
}
t.Logf("server sending ping")
conn.WriteMessage(websocket.PingMessage, []byte("ping"))
wantPong = "ping"
case data := <-pongCh:
if wantPong == "" {
t.Errorf("unexpected pong")
} else if data != wantPong {
t.Errorf("got pong with wrong data %q", data)
}
wantPong = ""
sendResponse = time.NewTimer(200 * time.Millisecond).C
case <-sendResponse:
t.Logf("server sending response")
conn.WriteMessage(websocket.TextMessage, []byte(subNotify))
sendResponse = nil
case <-shutdown:
conn.Close()
return
}
}
}