From f18adbec97fcdf21754d06b28f93e45ed0c1800a Mon Sep 17 00:00:00 2001 From: Dirk Brink Date: Fri, 3 Jul 2026 08:02:26 -0600 Subject: [PATCH] node: Add ability to pass custom headers to Sui gRPC --- node/cmd/guardiand/node.go | 3 + node/cmd/txverifier/sui.go | 8 ++ node/pkg/suiclient/grpc_headers.go | 100 +++++++++++++++++ node/pkg/suiclient/grpc_headers_test.go | 125 +++++++++++++++++++++ node/pkg/watchers/sui/config.go | 2 + node/pkg/watchers/sui/watcher.go | 29 ++++- node/pkg/watchers/sui/watcher_live_test.go | 44 +++++++- 7 files changed, 303 insertions(+), 8 deletions(-) create mode 100644 node/pkg/suiclient/grpc_headers.go create mode 100644 node/pkg/suiclient/grpc_headers_test.go diff --git a/node/cmd/guardiand/node.go b/node/cmd/guardiand/node.go index 6ded740aef1..90e2f83c64c 100644 --- a/node/cmd/guardiand/node.go +++ b/node/cmd/guardiand/node.go @@ -148,6 +148,7 @@ var ( movementHandle *string suiRPC *string + suiRPCHeaders *[]string suiMoveEventType *string solanaRPC *string @@ -418,6 +419,7 @@ func init() { movementHandle = NodeCmd.Flags().String("movementHandle", "", "movement handle") suiRPC = node.RegisterFlagWithValidationOrFail(NodeCmd, "suiRPC", "Sui gRPC endpoint", "sui:443", []string{""}) + suiRPCHeaders = NodeCmd.Flags().StringSlice("suiRPCHeaders", []string{}, "Sui gRPC headers as key=value pairs") suiMoveEventType = NodeCmd.Flags().String("suiMoveEventType", "", "Sui move event type for publish_message") solanaRPC = node.RegisterFlagWithValidationOrFail(NodeCmd, "solanaRPC", "Solana RPC URL (required)", "http://solana-devnet:8899", []string{"http", "https"}) @@ -1775,6 +1777,7 @@ func runNode(cmd *cobra.Command, args []string) { NetworkID: "sui", ChainID: vaa.ChainIDSui, Rpc: *suiRPC, + RpcHeaders: *suiRPCHeaders, SuiMoveEventType: *suiMoveEventType, TxVerifierEnabled: slices.Contains(txVerifierChains, vaa.ChainIDSui), } diff --git a/node/cmd/txverifier/sui.go b/node/cmd/txverifier/sui.go index 7effbefb502..671fa1993f9 100644 --- a/node/cmd/txverifier/sui.go +++ b/node/cmd/txverifier/sui.go @@ -26,6 +26,7 @@ import ( // CLI args var ( suiRPC *string + suiRPCHeaders *[]string suiProcessWormholeScanEvents *bool suiEnvironment *string suiDigest *string @@ -45,6 +46,7 @@ var TransferVerifierCmdSui = &cobra.Command{ // CLI parameters func init() { suiRPC = TransferVerifierCmdSui.Flags().String("suiRPC", "", "Sui gRPC endpoint, host:port (e.g. fullnode.mainnet.sui.io:443, or sui:443 in devnet)") + suiRPCHeaders = TransferVerifierCmdSui.Flags().StringSlice("suiRPCHeaders", []string{}, "Sui gRPC headers as key=value pairs") suiProcessWormholeScanEvents = TransferVerifierCmdSui.Flags().Bool("suiProcessWormholeScanEvents", false, "Indicate whether the Sui transfer verifier should process WormholeScan events") suiDigest = TransferVerifierCmdSui.Flags().String("suiDigest", "", "If provided, perform transaction verification on this single digest") suiEnvironment = TransferVerifierCmdSui.Flags().String("suiEnvironment", "mainnet", "The Sui environment to connect to. Supported values: mainnet, testnet and devnet") @@ -142,6 +144,12 @@ func runTransferVerifierSui(cmd *cobra.Command, args []string) { if *suiEnvironment == "devnet" { dialOpts = append(dialOpts, grpc.WithTransportCredentials(insecure.NewCredentials())) } + headerOpts, err := suiclient.GrpcHeaderDialOptions(*suiRPCHeaders) + if err != nil { + logger.Fatal("Invalid Sui gRPC headers", zap.Error(err)) + } + dialOpts = append(dialOpts, headerOpts...) + client, err := suiclient.NewSuiGrpcClient(*suiRPC, logger, dialOpts...) if err != nil { logger.Fatal("Failed to create Sui gRPC client", zap.Error(err)) diff --git a/node/pkg/suiclient/grpc_headers.go b/node/pkg/suiclient/grpc_headers.go new file mode 100644 index 00000000000..03d46ba974f --- /dev/null +++ b/node/pkg/suiclient/grpc_headers.go @@ -0,0 +1,100 @@ +package suiclient + +import ( + "context" + "fmt" + "strings" + + "google.golang.org/grpc" + "google.golang.org/grpc/metadata" +) + +// GrpcHeaderDialOptions converts key=value header specs from command line args +// into gRPC client interceptors that attach the headers as outgoing metadata +func GrpcHeaderDialOptions(headerSpecs []string) ([]grpc.DialOption, error) { + headers, err := parseGrpcHeaderSpecs(headerSpecs) + if err != nil { + return nil, err + } + if len(headers) == 0 { + return nil, nil + } + + headers = headers.Copy() + return []grpc.DialOption{ + grpc.WithChainUnaryInterceptor(grpcHeaderUnaryInterceptor(headers)), + grpc.WithChainStreamInterceptor(grpcHeaderStreamInterceptor(headers)), + }, nil +} + +func parseGrpcHeaderSpecs(headerSpecs []string) (metadata.MD, error) { + headers := metadata.MD{} + + for _, spec := range headerSpecs { + spec = strings.TrimSpace(spec) + if spec == "" { + continue + } + + key, value, ok := strings.Cut(spec, "=") + if !ok { + return nil, fmt.Errorf("invalid gRPC header: expected key=value") + } + + key = strings.ToLower(strings.TrimSpace(key)) + value = strings.TrimSpace(value) + + if err := validateGrpcMetadataKey(key); err != nil { + return nil, err + } + if value == "" { + return nil, fmt.Errorf("invalid gRPC header %q: value must not be empty", key) + } + if _, exists := headers[key]; exists { + return nil, fmt.Errorf("duplicate gRPC header key %q", key) + } + + headers.Append(key, value) + } + + return headers, nil +} + +func validateGrpcMetadataKey(key string) error { + if key == "" { + return fmt.Errorf("invalid gRPC header: key must not be empty") + } + + for _, r := range key { + switch { + case r >= 'a' && r <= 'z': + case r >= '0' && r <= '9': + case r == '-' || r == '_' || r == '.': + default: + return fmt.Errorf("invalid gRPC header key %q: keys may only contain letters, digits, '-', '_' or '.'", key) + } + } + + return nil +} + +func grpcHeaderUnaryInterceptor(headers metadata.MD) grpc.UnaryClientInterceptor { + return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { + return invoker(appendGrpcHeadersToContext(ctx, headers), method, req, reply, cc, opts...) + } +} + +func grpcHeaderStreamInterceptor(headers metadata.MD) grpc.StreamClientInterceptor { + return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) { + return streamer(appendGrpcHeadersToContext(ctx, headers), desc, cc, method, opts...) + } +} + +func appendGrpcHeadersToContext(ctx context.Context, headers metadata.MD) context.Context { + if len(headers) == 0 { + return ctx + } + + existing, _ := metadata.FromOutgoingContext(ctx) + return metadata.NewOutgoingContext(ctx, metadata.Join(existing, headers)) +} diff --git a/node/pkg/suiclient/grpc_headers_test.go b/node/pkg/suiclient/grpc_headers_test.go new file mode 100644 index 00000000000..a5b97f4cf16 --- /dev/null +++ b/node/pkg/suiclient/grpc_headers_test.go @@ -0,0 +1,125 @@ +package suiclient + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/metadata" +) + +func TestParseGrpcHeaderSpecs(t *testing.T) { + tests := []struct { + name string + specs []string + want metadata.MD + wantErr bool + }{ + { + name: "empty", + specs: nil, + want: metadata.MD{}, + }, + { + name: "valid headers", + specs: []string{"X-API-Key=secret", "chain-route=sui-mainnet"}, + want: metadata.MD{ + "x-api-key": []string{"secret"}, + "chain-route": []string{"sui-mainnet"}, + }, + }, + { + name: "duplicate key", + specs: []string{"X-API-Key=secret", "x-api-key=second"}, + wantErr: true, + }, + { + name: "trims whitespace", + specs: []string{" x-api-key = secret "}, + want: metadata.MD{ + "x-api-key": []string{"secret"}, + }, + }, + { + name: "missing separator", + specs: []string{"x-api-key"}, + wantErr: true, + }, + { + name: "empty key", + specs: []string{"=secret"}, + wantErr: true, + }, + { + name: "empty value", + specs: []string{"x-api-key="}, + wantErr: true, + }, + { + name: "invalid key", + specs: []string{"x api key=secret"}, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := parseGrpcHeaderSpecs(tt.specs) + if tt.wantErr { + require.Error(t, err) + return + } + + require.NoError(t, err) + require.Equal(t, tt.want, got) + }) + } +} + +func TestGrpcHeaderInterceptorsAppendMetadata(t *testing.T) { + headers := metadata.MD{ + "x-api-key": []string{"secret"}, + "chain-route": []string{"sui-mainnet"}, + } + + ctx := metadata.AppendToOutgoingContext(context.Background(), "existing-header", "existing-value") + + var unaryMetadata metadata.MD + unaryInvoker := func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, opts ...grpc.CallOption) error { + var ok bool + unaryMetadata, ok = metadata.FromOutgoingContext(ctx) + require.True(t, ok) + return nil + } + + err := grpcHeaderUnaryInterceptor(headers)(ctx, "/sui.rpc.v2.LedgerService/GetCheckpoint", nil, nil, nil, unaryInvoker) + require.NoError(t, err) + require.Equal(t, []string{"existing-value"}, unaryMetadata.Get("existing-header")) + require.Equal(t, []string{"secret"}, unaryMetadata.Get("x-api-key")) + require.Equal(t, []string{"sui-mainnet"}, unaryMetadata.Get("chain-route")) + + var streamMetadata metadata.MD + streamer := func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, opts ...grpc.CallOption) (grpc.ClientStream, error) { + var ok bool + streamMetadata, ok = metadata.FromOutgoingContext(ctx) + require.True(t, ok) + return nil, nil + } + + _, err = grpcHeaderStreamInterceptor(headers)(ctx, &grpc.StreamDesc{}, nil, "/sui.rpc.v2.SubscriptionService/SubscribeCheckpoints", streamer) + require.NoError(t, err) + require.Equal(t, []string{"existing-value"}, streamMetadata.Get("existing-header")) + require.Equal(t, []string{"secret"}, streamMetadata.Get("x-api-key")) + require.Equal(t, []string{"sui-mainnet"}, streamMetadata.Get("chain-route")) +} + +func TestGrpcHeaderDialOptions(t *testing.T) { + opts, err := GrpcHeaderDialOptions([]string{"x-api-key=secret"}) + require.NoError(t, err) + require.Len(t, opts, 2) + + opts, err = GrpcHeaderDialOptions(nil) + require.NoError(t, err) + require.Empty(t, opts) +} diff --git a/node/pkg/watchers/sui/config.go b/node/pkg/watchers/sui/config.go index 926e2a1a1d3..b3228d655c4 100644 --- a/node/pkg/watchers/sui/config.go +++ b/node/pkg/watchers/sui/config.go @@ -14,6 +14,7 @@ type WatcherConfig struct { NetworkID watchers.NetworkID // human readable name ChainID vaa.ChainID Rpc string + RpcHeaders []string SuiMoveEventType string TxVerifierEnabled bool } @@ -39,6 +40,7 @@ func (wc *WatcherConfig) Create( watcher, err := NewWatcher( wc.Rpc, + wc.RpcHeaders, wc.SuiMoveEventType, devMode, msgC, diff --git a/node/pkg/watchers/sui/watcher.go b/node/pkg/watchers/sui/watcher.go index f59353bbaaa..2ef4f29f6b9 100644 --- a/node/pkg/watchers/sui/watcher.go +++ b/node/pkg/watchers/sui/watcher.go @@ -35,17 +35,24 @@ import ( // suiGrpcDialOpts returns the gRPC dial options for connecting to the Sui endpoint. In unsafe // dev mode the local Sui node serves plaintext gRPC, so TLS is disabled; otherwise the default // (TLS) transport configured by suiclient.NewSuiGrpcClient is used. -func suiGrpcDialOpts(unsafeDevMode bool) []grpc.DialOption { +func suiGrpcDialOpts(unsafeDevMode bool, rpcHeaders []string) ([]grpc.DialOption, error) { + opts, err := suiclient.GrpcHeaderDialOptions(rpcHeaders) + if err != nil { + return nil, err + } + if unsafeDevMode { - return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())} + opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials())) } - return nil + + return opts, nil } type ( // Watcher is responsible for looking over Sui blockchain and reporting new transactions to the wormhole contract Watcher struct { suiRPC string + suiRPCHeaders []string suiMoveEventType string unsafeDevMode bool @@ -84,6 +91,7 @@ var ( // NewWatcher creates a new Sui appid watcher func NewWatcher( suiRPC string, + suiRPCHeaders []string, suiMoveEventType string, unsafeDevMode bool, messageEvents chan<- *common.MessagePublication, @@ -93,6 +101,11 @@ func NewWatcher( ) (*Watcher, error) { var suiTxVerifier *txverifier.SuiTransferVerifier + dialOpts, err := suiGrpcDialOpts(unsafeDevMode, suiRPCHeaders) + if err != nil { + return nil, fmt.Errorf("invalid Sui gRPC headers: %w", err) + } + if txVerifierEnabled { // Extracted from the suiMoveEventType passed to guardiand as CLI arg @@ -128,7 +141,7 @@ func NewWatcher( } // Create the Sui gRPC client used by the transfer verifier to query transactions and objects. - suiClient, err := suiclient.NewSuiGrpcClient(suiRPC, nil, suiGrpcDialOpts(unsafeDevMode)...) + suiClient, err := suiclient.NewSuiGrpcClient(suiRPC, nil, dialOpts...) if err != nil { return nil, fmt.Errorf("failed to create Sui gRPC client for transfer verifier: %w", err) } @@ -145,6 +158,7 @@ func NewWatcher( return &Watcher{ suiRPC: suiRPC, + suiRPCHeaders: suiRPCHeaders, suiMoveEventType: suiMoveEventType, unsafeDevMode: unsafeDevMode, msgChan: messageEvents, @@ -299,7 +313,12 @@ func (e *Watcher) Run(ctx context.Context) error { // concurrent goroutines below cannot observe a closed or nil client during shutdown. client := e.suiClient if client == nil { - grpcClient, err := suiclient.NewSuiGrpcClient(e.suiRPC, logger, suiGrpcDialOpts(e.unsafeDevMode)...) + dialOpts, err := suiGrpcDialOpts(e.unsafeDevMode, e.suiRPCHeaders) + if err != nil { + return fmt.Errorf("invalid Sui gRPC headers: %w", err) + } + + grpcClient, err := suiclient.NewSuiGrpcClient(e.suiRPC, logger, dialOpts...) if err != nil { return fmt.Errorf("failed to create Sui gRPC client: %w", err) } diff --git a/node/pkg/watchers/sui/watcher_live_test.go b/node/pkg/watchers/sui/watcher_live_test.go index b80f8810e4b..9a7a990900c 100644 --- a/node/pkg/watchers/sui/watcher_live_test.go +++ b/node/pkg/watchers/sui/watcher_live_test.go @@ -7,6 +7,7 @@ import ( "context" "encoding/hex" "encoding/json" + "errors" "fmt" "net/http" "os" @@ -39,12 +40,14 @@ import ( // SUI_WATCHER_RPC gRPC endpoint host:port (default fullnode.mainnet.sui.io:443) // SUI_WATCHER_JSONRPC JSON-RPC URL for cross-check (default https://fullnode.mainnet.sui.io) // SUI_WATCHER_EVENT_TYPE core bridge WormholeMessage (default mainnet core bridge type) +// SUI_WATCHER_RPC_HEADERS comma-separated gRPC headers (key=value,key=value) // SUI_WATCHER_SECONDS how long to observe (default 120) // SUI_WATCHER_TXVERIFIER set to any value to enable the transfer verifier func TestLiveSuiWatcher(t *testing.T) { rpc := liveEnv("SUI_WATCHER_RPC", suiclient.SuiRPCMainnet) jsonRPC := liveEnv("SUI_WATCHER_JSONRPC", "https://fullnode.mainnet.sui.io") eventType := liveEnv("SUI_WATCHER_EVENT_TYPE", "0x5306f64e312b581766351c07af79c72fcb1cd25147157fdc2f8ad76de9a3fb6a::publish_message::WormholeMessage") + rpcHeaders := liveHeaderEnv("SUI_WATCHER_RPC_HEADERS") seconds := 120 if s := os.Getenv("SUI_WATCHER_SECONDS"); s != "" { @@ -61,6 +64,7 @@ func TestLiveSuiWatcher(t *testing.T) { zap.String("rpc", rpc), zap.String("jsonRPC", jsonRPC), zap.String("eventType", eventType), + zap.Int("rpcHeaderCount", len(rpcHeaders)), zap.Int("seconds", seconds), zap.Bool("txVerifierEnabled", txVerifierEnabled), ) @@ -69,7 +73,7 @@ func TestLiveSuiWatcher(t *testing.T) { msgChan := make(chan *common.MessagePublication, 100) obsvReqC := make(chan *gossipv1.ObservationRequest, 10) - watcher, err := NewWatcher(rpc, eventType, false, msgChan, obsvReqC, common.MainNet, txVerifierEnabled) + watcher, err := NewWatcher(rpc, rpcHeaders, eventType, false, msgChan, obsvReqC, common.MainNet, txVerifierEnabled) require.NoError(t, err) rootCtx, cancel := context.WithTimeout(context.Background(), time.Duration(seconds)*time.Second) @@ -77,6 +81,7 @@ func TestLiveSuiWatcher(t *testing.T) { observed := 0 verified := 0 + watcherErrC := make(chan error, 1) var wg sync.WaitGroup wg.Add(1) @@ -104,11 +109,17 @@ func TestLiveSuiWatcher(t *testing.T) { } }() - if rerr := supervisor.Run(ctx, "sui", watcher.Run); rerr != nil { + if rerr := watcher.Run(ctx); rerr != nil { + if !isContextError(rerr) { + select { + case watcherErrC <- rerr: + default: + } + cancel() + } return rerr } - <-ctx.Done() return nil }, supervisor.WithPropagatePanic) @@ -116,6 +127,11 @@ func TestLiveSuiWatcher(t *testing.T) { // Wait for the drain goroutine to finish any in-flight verification before the test returns, // so it never calls t.Errorf after the test function has returned. wg.Wait() + select { + case err := <-watcherErrC: + require.NoError(t, err) + default: + } logger.Info("live Sui watcher finished", zap.Int("messagesObserved", observed), zap.Int("messagesVerified", verified)) } @@ -248,3 +264,25 @@ func liveEnv(key, def string) string { } return def } + +func liveHeaderEnv(key string) []string { + v := strings.TrimSpace(os.Getenv(key)) + if v == "" { + return nil + } + + parts := strings.Split(v, ",") + headers := make([]string, 0, len(parts)) + for _, part := range parts { + part = strings.TrimSpace(part) + if part != "" { + headers = append(headers, part) + } + } + + return headers +} + +func isContextError(err error) bool { + return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) +}