diff --git a/ethrpc/ethrpc.go b/ethrpc/ethrpc.go index 844115d4..dd33cbd2 100644 --- a/ethrpc/ethrpc.go +++ b/ethrpc/ethrpc.go @@ -29,6 +29,7 @@ type Provider struct { log *slog.Logger // TODO: not used.. nodeURL string nodeWSURL string + httpAuth rpc.HTTPAuth httpClient httpClient httpClientMu sync.RWMutex @@ -173,6 +174,11 @@ func (p *Provider) Do(ctx context.Context, calls ...Call) ([]byte, error) { } req = req.WithContext(ctx) req.Header.Set("Content-Type", "application/json") + if p.httpAuth != nil { + if err := p.httpAuth(req.Header); err != nil { + return nil, superr.Wrap(ErrRequestFail, fmt.Errorf("failed to set http auth header: %w", err)) + } + } if p.jwtToken != "" { req.Header.Set("Authorization", fmt.Sprintf("BEARER %s", p.jwtToken)) @@ -583,7 +589,11 @@ func (p *Provider) streamSubscribe(ctx context.Context, label string, subscribeF return nil, fmt.Errorf("ethrpc: provider instance has not enabled streaming") } - gethRPC, err := rpc.DialContext(ctx, p.nodeWSURL) + var opts []rpc.ClientOption + if p.httpAuth != nil { + opts = append(opts, rpc.WithHTTPAuth(p.httpAuth)) + } + gethRPC, err := rpc.DialOptions(ctx, p.nodeWSURL, opts...) if err != nil { return nil, fmt.Errorf("ethrpc: %s failed to connect to websocket: %w", label, err) } diff --git a/ethrpc/http_auth_test.go b/ethrpc/http_auth_test.go new file mode 100644 index 00000000..00545208 --- /dev/null +++ b/ethrpc/http_auth_test.go @@ -0,0 +1,145 @@ +package ethrpc_test + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/0xsequence/ethkit/ethrpc" + "github.com/0xsequence/ethkit/go-ethereum/core/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestWithHTTPAuthSetsRequestHeader(t *testing.T) { + authHeaders := make(chan string, 1) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + authHeaders <- r.Header.Get("Authorization") + body, _ := io.ReadAll(r.Body) + w.Header().Set("Content-Type", "application/json") + w.Write(chainIDResponse(body)) + })) + defer srv.Close() + + p, err := ethrpc.NewProvider(srv.URL, ethrpc.WithHTTPAuth(func(h http.Header) error { + h.Set("Authorization", "Bearer http-token") + return nil + })) + require.NoError(t, err) + + chainID, err := p.ChainID(context.Background()) + require.NoError(t, err) + assert.Equal(t, int64(1337), chainID.Int64()) + assert.Equal(t, "Bearer http-token", recvAuthHeader(t, authHeaders)) +} + +func TestWithHTTPAuthErrorAbortsRequest(t *testing.T) { + var hits atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + })) + defer srv.Close() + + authErr := errors.New("token expired") + p, err := ethrpc.NewProvider(srv.URL, ethrpc.WithHTTPAuth(func(h http.Header) error { + return authErr + })) + require.NoError(t, err) + + _, err = p.ChainID(context.Background()) + require.Error(t, err) + assert.ErrorIs(t, err, authErr) + assert.ErrorIs(t, err, ethrpc.ErrRequestFail) + assert.Equal(t, int32(0), hits.Load()) +} + +func TestWithHTTPAuthSetsWebSocketHandshakeHeader(t *testing.T) { + authHeaders := make(chan string, 1) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + authHeaders <- r.Header.Get("Authorization") + http.Error(w, "websocket upgrade not served in test", http.StatusBadRequest) + })) + defer srv.Close() + + p, err := ethrpc.NewProvider(srv.URL, + ethrpc.WithStreaming(srv.URL), + ethrpc.WithHTTPAuth(func(h http.Header) error { + h.Set("Authorization", "Bearer ws-token") + return nil + }), + ) + require.NoError(t, err) + + _, err = p.SubscribeNewHeads(context.Background(), make(chan *types.Header)) + require.Error(t, err) + assert.Equal(t, "Bearer ws-token", recvAuthHeader(t, authHeaders)) +} + +func TestWithHTTPAuthErrorAbortsWebSocketDial(t *testing.T) { + var hits atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + http.Error(w, "websocket upgrade not served in test", http.StatusBadRequest) + })) + defer srv.Close() + + authErr := errors.New("token expired") + p, err := ethrpc.NewProvider(srv.URL, + ethrpc.WithStreaming(srv.URL), + ethrpc.WithHTTPAuth(func(h http.Header) error { + return authErr + }), + ) + require.NoError(t, err) + + _, err = p.SubscribeNewHeads(context.Background(), make(chan *types.Header)) + require.Error(t, err) + assert.ErrorIs(t, err, authErr) + assert.Equal(t, int32(0), hits.Load()) +} + +func recvAuthHeader(t *testing.T, ch <-chan string) string { + t.Helper() + select { + case h := <-ch: + return h + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for request at httptest server") + return "" + } +} + +func chainIDResponse(body []byte) []byte { + type response struct { + JSONRPC string `json:"jsonrpc"` + ID json.RawMessage `json:"id"` + Result string `json:"result"` + } + + var single struct { + ID json.RawMessage `json:"id"` + } + if err := json.Unmarshal(body, &single); err == nil { + out, _ := json.Marshal(response{JSONRPC: "2.0", ID: single.ID, Result: "0x539"}) + return out + } + + var batch []struct { + ID json.RawMessage `json:"id"` + } + if err := json.Unmarshal(body, &batch); err != nil { + return nil + } + out := make([]response, 0, len(batch)) + for _, c := range batch { + out = append(out, response{JSONRPC: "2.0", ID: c.ID, Result: "0x539"}) + } + res, _ := json.Marshal(out) + return res +} diff --git a/ethrpc/option.go b/ethrpc/option.go index ea01bb9c..9381d0a9 100644 --- a/ethrpc/option.go +++ b/ethrpc/option.go @@ -5,6 +5,7 @@ import ( "net/http" "strings" + "github.com/0xsequence/ethkit/go-ethereum/rpc" "github.com/goware/breaker" ) @@ -29,6 +30,13 @@ func WithHTTPClient(c httpClient) Option { } } +// WithHTTPAuth sets headers on every HTTP request and WebSocket dial; it must be safe for concurrent use. +func WithHTTPAuth(a rpc.HTTPAuth) Option { + return func(p *Provider) { + p.httpAuth = a + } +} + func WithLogger(log *slog.Logger) Option { return func(p *Provider) { p.log = log