Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions node/cmd/guardiand/node.go
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,7 @@ var (
movementHandle *string

suiRPC *string
suiRPCHeaders *[]string
suiMoveEventType *string

solanaRPC *string
Expand Down Expand Up @@ -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"})
Expand Down Expand Up @@ -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),
}
Expand Down
8 changes: 8 additions & 0 deletions node/cmd/txverifier/sui.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import (
// CLI args
var (
suiRPC *string
suiRPCHeaders *[]string
suiProcessWormholeScanEvents *bool
suiEnvironment *string
suiDigest *string
Expand All @@ -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")
Expand Down Expand Up @@ -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))
Expand Down
100 changes: 100 additions & 0 deletions node/pkg/suiclient/grpc_headers.go
Original file line number Diff line number Diff line change
@@ -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) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion: I think some of this logic could be replaced by using cobra's built-in parsing.

This is an example that parses key-value pairs:

var labels map[string]string

cmd.Flags().StringToStringVar(
    &labels,
    "label",
    nil,
    "key-value labels, e.g. --label env=prod,team=infra",
)

It might make more sense to put this in a the cmd/ code as well since this is related to CLI input parsing rather than Sui client code.

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))
}
125 changes: 125 additions & 0 deletions node/pkg/suiclient/grpc_headers_test.go
Original file line number Diff line number Diff line change
@@ -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)
}
2 changes: 2 additions & 0 deletions node/pkg/watchers/sui/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ type WatcherConfig struct {
NetworkID watchers.NetworkID // human readable name
ChainID vaa.ChainID
Rpc string
RpcHeaders []string
SuiMoveEventType string
TxVerifierEnabled bool
}
Expand All @@ -39,6 +40,7 @@ func (wc *WatcherConfig) Create(

watcher, err := NewWatcher(
wc.Rpc,
wc.RpcHeaders,
wc.SuiMoveEventType,
devMode,
msgC,
Expand Down
29 changes: 24 additions & 5 deletions node/pkg/watchers/sui/watcher.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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
Expand Down Expand Up @@ -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)
}
Expand All @@ -145,6 +158,7 @@ func NewWatcher(

return &Watcher{
suiRPC: suiRPC,
suiRPCHeaders: suiRPCHeaders,
suiMoveEventType: suiMoveEventType,
unsafeDevMode: unsafeDevMode,
msgChan: messageEvents,
Expand Down Expand Up @@ -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)
}
Expand Down
Loading
Loading