ethrpc: add WithHTTPAuth option for HTTP and WebSocket

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)
 	}

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
+}

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