node: Validate and sanitize RPC URLs

- Checks watcher's URLs against a deny-list of public RPC URLs. Logs an
  error when a public RPC URL is found for a delegated guardian. This
  should prevent problems where a delegate is not correct pointed at a
  full node.
- Sanitizes URLs in logs so that path parts are not included. (These can
  sometimes contain API keys or other sensitive strings).
- Normalizes logged field names across watchers for the 'watcher
  started' log

Author: johnsaigle

node/pkg/common/url_verification.go

@@ -40,3 +40,27 @@ func ValidateURL(urlStr string, validSchemes []string) bool {
 	}
 	return false
 }
+
+// SafeURLForLogging returns only the hostname for a URL-like string.
+// It intentionally omits userinfo, path, query, and fragment because those may contain credentials.
+func SafeURLForLogging(urlStr string) string {
+	if urlStr == "" {
+		return ""
+	}
+
+	parsedURL, err := url.Parse(urlStr)
+	if err != nil || parsedURL.Host == "" {
+		// Schemeless host:port strings parse as scheme:opaque. Parse them again
+		// as network-path references so URL.Hostname can extract the host.
+		parsedURL, err = url.Parse("//" + urlStr)
+	}
+	if err != nil {
+		return "<invalid-url>"
+	}
+
+	if host := parsedURL.Hostname(); host != "" {
+		return host
+	}
+
+	return "<invalid-url>"
+}

node/pkg/common/url_verification_test.go

@@ -38,3 +38,25 @@ func TestValidateURL(t *testing.T) {
 		assert.Equal(t, test.expected, result)
 	}
 }
+
+func TestSafeURLForLogging(t *testing.T) {
+	tests := []struct {
+		name     string
+		urlStr   string
+		expected string
+	}{
+		{name: "https with credentials", urlStr: "https://user:pass@example.com/path?api_key=secret", expected: "example.com"},
+		{name: "websocket", urlStr: "wss://rpc.example.com/websocket", expected: "rpc.example.com"},
+		{name: "host port without scheme", urlStr: "example.com:8080/path?token=secret", expected: "example.com"},
+		{name: "credentials without scheme", urlStr: "user:pass@example.com/path?api_key=secret", expected: "example.com"},
+		{name: "ipv6", urlStr: "http://[::1]:8545/path", expected: "::1"},
+		{name: "ipv6 without scheme", urlStr: "[::1]:8545/path", expected: "::1"},
+		{name: "empty", urlStr: "", expected: ""},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, test.expected, SafeURLForLogging(test.urlStr))
+		})
+	}
+}

node/pkg/processor/processor.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"slices"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -25,6 +26,7 @@ import (
 	"github.com/certusone/wormhole/node/pkg/gwrelayer"
 	gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1"
 	"github.com/certusone/wormhole/node/pkg/supervisor"
+	"github.com/certusone/wormhole/node/pkg/watchers"
 	"github.com/wormhole-foundation/wormhole/sdk/vaa"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -210,6 +212,9 @@ type Processor struct {
 
 	// managerC is the channel used to send signed VAAs to the manager service (nil if manager service is disabled)
 	managerC chan<- *vaa.VAA
+
+	// publicRpcLogged tracks which chains have already had a public RPC warning logged.
+	publicRpcLogged map[vaa.ChainID]struct{}
 }
 
 // updateVaaEntry is used to queue up a VAA to be written to the database.
@@ -284,6 +289,101 @@ var (
 // batchObsvPubChanSize specifies the size of the channel used to publish observation batches. Allow five seconds worth.
 const batchObsvPubChanSize = p2p.MaxObservationBatchSize * 5
 
+// publicRPCDenyList contains domains for known public RPC endpoints.
+// Guardians using these for watcher URLs are at risk of serving incorrect data,
+// which can lead to consensus issues when the guardian is part of the delegated guardian set.
+var publicRPCDenyList = map[string]struct{}{
+	"infura.io":       {},
+	"alchemy.com":     {},
+	"quiknode.pro":    {},
+	"ankr.com":        {},
+	"blastapi.io":     {},
+	"publicnode.com":  {},
+	"llamarpc.com":    {},
+	"1rpc.io":         {},
+	"nodereal.io":     {},
+	"bnbchain.org":    {},
+	"pocket.network":  {},
+	"monad.xyz":       {},
+	"monadinfra.com":  {},
+	"hyperliquid.xyz": {},
+	"t.conduit.xyz":   {},
+}
+
+func publicRPCDomain(host string) (string, bool) {
+	host = strings.ToLower(strings.TrimSuffix(host, "."))
+	for domain := host; domain != ""; {
+		if _, exists := publicRPCDenyList[domain]; exists {
+			return domain, true
+		}
+
+		_, parentDomain, foundParentDomain := strings.Cut(domain, ".")
+		if !foundParentDomain {
+			return "", false
+		}
+		domain = parentDomain
+	}
+
+	return "", false
+}
+
+// checkPublicRpcEndpoints checks all chains in the DGS for public RPC URLs.
+// It logs an error once per chain per run if a public RPC endpoint is detected.
+// This is a safety measure: delegating guardians that use public RPC endpoints
+// risk producing incorrect VAAs, which can cause consensus failures.
+func (p *Processor) checkPublicRpcEndpoints() {
+	if !p.delegatedGuardiansEnabled {
+		return
+	}
+	if p.dgc == nil {
+		p.logger.Warn("public RPC endpoint check skipped: delegated guardian config is nil")
+		return
+	}
+
+	allChains := p.dgc.ReadAll()
+	if len(allChains) == 0 {
+		p.logger.Warn("public RPC endpoint check skipped: delegated guardian config has no chains")
+		return
+	}
+
+	for chainID, cfg := range allChains {
+		// Only check chains where this guardian is a delegated guardian
+		if _, ok := cfg.KeyIndex(p.ourAddr); !ok {
+			continue
+		}
+
+		// Skip if we've already logged for this chain.
+		if _, logged := p.publicRpcLogged[chainID]; logged {
+			continue
+		}
+
+		rpcURLs := watchers.RPCURLs(chainID)
+		if len(rpcURLs) == 0 {
+			continue
+		}
+
+		for _, rpcURL := range rpcURLs {
+			hostname := common.SafeURLForLogging(rpcURL)
+			if hostname == "<invalid-url>" {
+				continue
+			}
+
+			if domain, ok := publicRPCDomain(hostname); ok {
+				p.logger.Error("public RPC endpoint detected for delegated guardian chain",
+					zap.Stringer("chainID", chainID),
+					zap.String("rpcHost", hostname),
+					zap.String("matchedDomain", domain),
+				)
+				p.publicRpcLogged[chainID] = struct{}{}
+			}
+
+			if _, logged := p.publicRpcLogged[chainID]; logged {
+				break
+			}
+		}
+	}
+}
+
 func NewProcessor(
 	ctx context.Context,
 	db *guardianDB.Database,
@@ -345,6 +445,7 @@ func NewProcessor(
 		dgc:                       dgc,
 		delegatedGuardiansEnabled: delegatedGuardiansEnabled,
 		managerC:                  managerC,
+		publicRpcLogged:           make(map[vaa.ChainID]struct{}),
 	}
 }
 
@@ -509,6 +610,8 @@ func (p *Processor) Run(ctx context.Context) error {
 
 			if err := p.dgc.Set(chains); err != nil {
 				p.logger.Error("delegate guardian config update failed", zap.Error(err))
+			} else {
+				p.checkPublicRpcEndpoints()
 			}
 			dgConfig.mu.Unlock()
 		case k := <-p.msgC:

node/pkg/processor/public_rpc_test.go

@@ -0,0 +1,31 @@
+package processor
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPublicRPCDomain(t *testing.T) {
+	tests := []struct {
+		name           string
+		host           string
+		expectedDomain string
+		expectedMatch  bool
+	}{
+		{name: "exact match", host: "publicnode.com", expectedDomain: "publicnode.com", expectedMatch: true},
+		{name: "subdomain match", host: "rpc.monad.xyz", expectedDomain: "monad.xyz", expectedMatch: true},
+		{name: "nested subdomain match", host: "rpc-plume-mainnet-1.t.conduit.xyz", expectedDomain: "t.conduit.xyz", expectedMatch: true},
+		{name: "case insensitive", host: "RPC.HYPERLIQUID.XYZ", expectedDomain: "hyperliquid.xyz", expectedMatch: true},
+		{name: "trailing dot", host: "polygon-mainnet.quiknode.pro.", expectedDomain: "quiknode.pro", expectedMatch: true},
+		{name: "no match", host: "guardian.example.com", expectedMatch: false},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			domain, ok := publicRPCDomain(test.host)
+			assert.Equal(t, test.expectedMatch, ok)
+			assert.Equal(t, test.expectedDomain, domain)
+		})
+	}
+}

node/pkg/watchers/algorand/config.go

@@ -37,5 +37,7 @@ func (wc *WatcherConfig) Create(
 	_ chan<- *common.GuardianSet,
 	_ common.Environment,
 ) (supervisor.Runnable, interfaces.Reobserver, error) {
+	watchers.RegisterRPCURL(wc.ChainID, wc.IndexerRPC)
+	watchers.RegisterRPCURL(wc.ChainID, wc.AlgodRPC)
 	return NewWatcher(wc.IndexerRPC, wc.IndexerToken, wc.AlgodRPC, wc.AlgodToken, wc.AppID, msgC, obsvReqC).Run, nil, nil
 }

node/pkg/watchers/algorand/watcher.go

@@ -221,15 +221,13 @@ func (e *Watcher) Run(ctx context.Context) error {
 
 	logger.Info("Starting watcher",
 		zap.String("watcher_name", "algorand"),
-		zap.String("indexerRPC", e.indexerRPC),
-		zap.String("indexerToken", e.indexerToken),
-		zap.String("algodRPC", e.algodRPC),
-		zap.String("algodToken", e.algodToken),
+		zap.String("indexerURL", common.SafeURLForLogging(e.indexerRPC)),
+		zap.String("algodURL", common.SafeURLForLogging(e.algodRPC)),
 		zap.Uint64("appid", e.appid),
 	)
 
-	logger.Info("Algorand watcher connecting to indexer  ", zap.String("url", e.indexerRPC))
-	logger.Info("Algorand watcher connecting to RPC node ", zap.String("url", e.algodRPC))
+	logger.Info("Algorand watcher connecting to indexer  ", zap.String("indexerURL", common.SafeURLForLogging(e.indexerRPC)))
+	logger.Info("Algorand watcher connecting to RPC node ", zap.String("algodURL", common.SafeURLForLogging(e.algodRPC)))
 
 	timer := time.NewTicker(time.Second * 1)
 	defer timer.Stop()

node/pkg/watchers/aptos/config.go

@@ -35,5 +35,6 @@ func (wc *WatcherConfig) Create(
 	_ chan<- *common.GuardianSet,
 	_ common.Environment,
 ) (supervisor.Runnable, interfaces.Reobserver, error) {
+	watchers.RegisterRPCURL(wc.ChainID, wc.Rpc)
 	return NewWatcher(wc.ChainID, wc.NetworkID, wc.Rpc, wc.Account, wc.Handle, msgC, obsvReqC).Run, nil, nil
 }

node/pkg/watchers/aptos/watcher.go

@@ -85,7 +85,7 @@ func (e *Watcher) Run(ctx context.Context) error {
 
 	logger.Info("Starting watcher",
 		zap.String("watcher_name", e.networkID),
-		zap.String("rpc", e.aptosRPC),
+		zap.String("rpcURL", common.SafeURLForLogging(e.aptosRPC)),
 		zap.String("account", e.aptosAccount),
 		zap.String("handle", e.aptosHandle),
 	)
@@ -99,9 +99,9 @@ func (e *Watcher) Run(ctx context.Context) error {
 	var aptosHealth = fmt.Sprintf(`%s/v1`, e.aptosRPC)
 
 	logger.Info("watcher connecting to RPC node ",
-		zap.String("url", e.aptosRPC),
-		zap.String("eventsQuery", eventsEndpoint),
-		zap.String("healthQuery", aptosHealth),
+		zap.String("rpcURL", common.SafeURLForLogging(e.aptosRPC)),
+		zap.String("eventsURL", common.SafeURLForLogging(eventsEndpoint)),
+		zap.String("healthURL", common.SafeURLForLogging(aptosHealth)),
 	)
 
 	// the events have sequence numbers associated with them in the aptos API

node/pkg/watchers/cosmwasm/config.go

@@ -35,5 +35,7 @@ func (wc *WatcherConfig) Create(
 	_ chan<- *common.GuardianSet,
 	env common.Environment,
 ) (supervisor.Runnable, interfaces.Reobserver, error) {
+	watchers.RegisterRPCURL(wc.ChainID, wc.Websocket)
+	watchers.RegisterRPCURL(wc.ChainID, wc.Lcd)
 	return NewWatcher(wc.Websocket, wc.Lcd, wc.Contract, msgC, obsvReqC, wc.ChainID, env).Run, nil, nil
 }

node/pkg/watchers/cosmwasm/watcher.go

@@ -153,13 +153,13 @@ func (e *Watcher) Run(ctx context.Context) error {
 
 	logger.Info("Starting watcher",
 		zap.String("watcher_name", "cosmwasm"),
-		zap.String("urlWS", e.urlWS),
-		zap.String("urlLCD", e.urlLCD),
+		zap.String("wsURL", common.SafeURLForLogging(e.urlWS)),
+		zap.String("lcdURL", common.SafeURLForLogging(e.urlLCD)),
 		zap.String("contract", e.contract),
 		zap.String("chainID", e.chainID.String()),
 	)
 
-	logger.Info("connecting to websocket", zap.String("network", e.networkName), zap.String("url", e.urlWS))
+	logger.Info("connecting to websocket", zap.String("network", e.networkName), zap.String("wsURL", common.SafeURLForLogging(e.urlWS)))
 
 	//nolint:bodyclose // The close is down below. The linter misses it.
 	c, _, err := websocket.Dial(ctx, e.urlWS, nil)