CCQ/Node: Guardian Changes

node/cmd/guardiand/node.go

@@ -205,6 +205,12 @@ var (
 
 	chainGovernorEnabled *bool
 
+	ccqEnabled           *bool
+	ccqAllowedRequesters *string
+	ccqP2pPort           *uint
+	ccqP2pBootstrap      *string
+	ccqAllowedPeers      *string
+
 	gatewayRelayerContract      *string
 	gatewayRelayerKeyPath       *string
 	gatewayRelayerKeyPassPhrase *string
@@ -370,6 +376,12 @@ func init() {
 
 	chainGovernorEnabled = NodeCmd.Flags().Bool("chainGovernorEnabled", false, "Run the chain governor")
 
+	ccqEnabled = NodeCmd.Flags().Bool("ccqEnabled", false, "Enable cross chain query support")
+	ccqAllowedRequesters = NodeCmd.Flags().String("ccqAllowedRequesters", "", "Comma separated list of signers allowed to submit cross chain queries")
+	ccqP2pPort = NodeCmd.Flags().Uint("ccqP2pPort", 8996, "CCQ P2P UDP listener port")
+	ccqP2pBootstrap = NodeCmd.Flags().String("ccqP2pBootstrap", "", "CCQ P2P bootstrap peers (comma-separated)")
+	ccqAllowedPeers = NodeCmd.Flags().String("ccqAllowedPeers", "", "CCQ allowed P2P peers (comma-separated)")
+
 	gatewayRelayerContract = NodeCmd.Flags().String("gatewayRelayerContract", "", "Address of the smart contract on wormchain to receive relayed VAAs")
 	gatewayRelayerKeyPath = NodeCmd.Flags().String("gatewayRelayerKeyPath", "", "Path to gateway relayer private key for signing transactions")
 	gatewayRelayerKeyPassPhrase = NodeCmd.Flags().String("gatewayRelayerKeyPassPhrase", "", "Pass phrase used to unarmor the gateway relayer key file")
@@ -469,6 +481,7 @@ func runNode(cmd *cobra.Command, args []string) {
 
 		// Use the first guardian node as bootstrap
 		*p2pBootstrap = fmt.Sprintf("/dns4/guardian-0.guardian/udp/%d/quic/p2p/%s", *p2pPort, g0key.String())
+		*ccqP2pBootstrap = fmt.Sprintf("/dns4/guardian-0.guardian/udp/%d/quic/p2p/%s", *ccqP2pPort, g0key.String())
 
 		// Deterministic ganache ETH devnet address.
 		*ethContract = unsafeDevModeEvmContractAddress(*ethContract)
@@ -1422,8 +1435,9 @@ func runNode(cmd *cobra.Command, args []string) {
 		node.GuardianOptionAccountant(*accountantContract, *accountantWS, *accountantCheckEnabled, accountantWormchainConn),
 		node.GuardianOptionGovernor(*chainGovernorEnabled),
 		node.GuardianOptionGatewayRelayer(*gatewayRelayerContract, gatewayRelayerWormchainConn),
+		node.GuardianOptionQueryHandler(*ccqEnabled, *ccqAllowedRequesters),
 		node.GuardianOptionAdminService(*adminSocketPath, ethRPC, ethContract, rpcMap),
-		node.GuardianOptionP2P(p2pKey, *p2pNetworkID, *p2pBootstrap, *nodeName, *disableHeartbeatVerify, *p2pPort, ibc.GetFeatures),
+		node.GuardianOptionP2P(p2pKey, *p2pNetworkID, *p2pBootstrap, *nodeName, *disableHeartbeatVerify, *p2pPort, *ccqP2pBootstrap, *ccqP2pPort, *ccqAllowedPeers, ibc.GetFeatures),
 		node.GuardianOptionStatusServer(*statusAddr),
 		node.GuardianOptionProcessor(),
 	}

node/cmd/spy/spy.go

@@ -356,6 +356,12 @@ func runSpy(cmd *cobra.Command, args []string) {
 				components,
 				nil,   // ibc feature string
 				false, // gateway relayer enabled
+				false, // ccqEnabled
+				nil,   // query requests
+				nil,   // query responses
+				"",    // query bootstrap peers
+				0,     // query port
+				"",    // query allow list
 			)); err != nil {
 			return err
 		}

node/pkg/adminrpc/adminserver_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	ethcrypto "github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/event"
+	ethRpc "github.com/ethereum/go-ethereum/rpc"
 	"github.com/stretchr/testify/require"
 	"github.com/wormhole-foundation/wormhole/sdk/vaa"
 	"go.uber.org/zap"
@@ -68,6 +69,10 @@ func (m mockEVMConnector) RawCallContext(ctx context.Context, result interface{}
 	panic("unimplemented")
 }
 
+func (m mockEVMConnector) RawBatchCallContext(ctx context.Context, b []ethRpc.BatchElem) error {
+	panic("unimplemented")
+}
+
 func generateGS(num int) (keys []*ecdsa.PrivateKey, addrs []common.Address) {
 	for i := 0; i < num; i++ {
 		key, err := ethcrypto.GenerateKey()

node/pkg/common/mode.go

@@ -1,5 +1,10 @@
 package common
 
+import (
+	"fmt"
+	"strings"
+)
+
 type Environment string
 
 const (
@@ -9,3 +14,24 @@ const (
 	GoTest         Environment = "unit-test"
 	AccountantMock Environment = "accountant-mock" // Used for mocking accountant with a Wormchain connection
 )
+
+// ParseEnvironment parses a string into the corresponding Environment value, allowing various reasonable variations.
+func ParseEnvironment(str string) (Environment, error) {
+	str = strings.ToLower(str)
+	if str == "prod" || str == "mainnet" {
+		return MainNet, nil
+	}
+	if str == "test" || str == "testnet" {
+		return TestNet, nil
+	}
+	if str == "dev" || str == "devnet" || str == "unsafedevnet" {
+		return UnsafeDevNet, nil
+	}
+	if str == "unit-test" || str == "gotest" {
+		return GoTest, nil
+	}
+	if str == "accountant-mock" || str == "accountantmock" {
+		return AccountantMock, nil
+	}
+	return UnsafeDevNet, fmt.Errorf("invalid environment string: %s", str)
+}

node/pkg/common/mode_test.go

@@ -0,0 +1,51 @@
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseEnvironment(t *testing.T) {
+	type test struct {
+		input  string
+		output Environment
+		err    bool
+	}
+
+	tests := []test{
+		{input: "MainNet", output: MainNet, err: false},
+		{input: "Prod", output: MainNet, err: false},
+
+		{input: "TestNet", output: TestNet, err: false},
+		{input: "test", output: TestNet, err: false},
+
+		{input: "UnsafeDevNet", output: UnsafeDevNet, err: false},
+		{input: "devnet", output: UnsafeDevNet, err: false},
+		{input: "dev", output: UnsafeDevNet, err: false},
+
+		{input: "GoTest", output: GoTest, err: false},
+		{input: "unit-test", output: GoTest, err: false},
+
+		{input: "AccountantMock", output: AccountantMock, err: false},
+		{input: "accountant-mock", output: AccountantMock, err: false},
+
+		{input: "junk", output: UnsafeDevNet, err: true},
+		{input: "", output: UnsafeDevNet, err: true},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.input, func(t *testing.T) {
+			output, err := ParseEnvironment(tc.input)
+			if err != nil {
+				if tc.err == false {
+					assert.NoError(t, err)
+				}
+			} else if tc.err {
+				assert.Error(t, err)
+			} else {
+				assert.Equal(t, tc.output, output)
+			}
+		})
+	}
+}

node/pkg/node/node.go

@@ -11,7 +11,9 @@ import (
 	"github.com/certusone/wormhole/node/pkg/governor"
 	"github.com/certusone/wormhole/node/pkg/gwrelayer"
 	gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1"
+	"github.com/certusone/wormhole/node/pkg/query"
 	"github.com/certusone/wormhole/node/pkg/supervisor"
+	"github.com/wormhole-foundation/wormhole/sdk/vaa"
 
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
@@ -59,6 +61,7 @@ type G struct {
 	acct            *accountant.Accountant
 	gov             *governor.ChainGovernor
 	gatewayRelayer  *gwrelayer.GatewayRelayer
+	queryHandler    *query.QueryHandler
 	publicrpcServer *grpc.Server
 
 	// runnables
@@ -82,6 +85,12 @@ type G struct {
 	obsvReqSendC channelPair[*gossipv1.ObservationRequest]
 	// acctC is the channel where messages will be put after they reached quorum in the accountant.
 	acctC channelPair[*common.MessagePublication]
+
+	// Cross Chain Query Handler channels
+	chainQueryReqC            map[vaa.ChainID]chan *query.PerChainQueryInternal
+	signedQueryReqC           channelPair[*gossipv1.SignedQueryRequest]
+	queryResponseC            channelPair[*query.PerChainQueryResponseInternal]
+	queryResponsePublicationC channelPair[*query.QueryResponsePublication]
 }
 
 func NewGuardianNode(
@@ -108,6 +117,11 @@ func (g *G) initializeBasic(rootCtxCancel context.CancelFunc) {
 	g.obsvReqC = makeChannelPair[*gossipv1.ObservationRequest](observationRequestInboundBufferSize)
 	g.obsvReqSendC = makeChannelPair[*gossipv1.ObservationRequest](observationRequestOutboundBufferSize)
 	g.acctC = makeChannelPair[*common.MessagePublication](accountant.MsgChannelCapacity)
+	// Cross Chain Query Handler channels
+	g.chainQueryReqC = make(map[vaa.ChainID]chan *query.PerChainQueryInternal)
+	g.signedQueryReqC = makeChannelPair[*gossipv1.SignedQueryRequest](query.SignedQueryRequestChannelSize)
+	g.queryResponseC = makeChannelPair[*query.PerChainQueryResponseInternal](0)
+	g.queryResponsePublicationC = makeChannelPair[*query.QueryResponsePublication](0)
 
 	// Guardian set state managed by processor
 	g.gst = common.NewGuardianSetState(nil)
@@ -191,6 +205,13 @@ func (g *G) Run(rootCtxCancel context.CancelFunc, options ...*GuardianOption) su
 			}
 		}
 
+		if g.queryHandler != nil {
+			logger.Info("Starting query handler", zap.String("component", "ccq"))
+			if err := g.queryHandler.Start(ctx); err != nil {
+				logger.Fatal("failed to create chain governor", zap.Error(err), zap.String("component", "ccq"))
+			}
+		}
+
 		// Start any other runnables
 		for name, runnable := range g.runnables {
 			if err := supervisor.Run(ctx, name, runnable); err != nil {

node/pkg/node/node_test.go

@@ -190,7 +190,7 @@ func mockGuardianRunnable(t testing.TB, gs []*mockGuardian, mockGuardianIndex ui
 			GuardianOptionNoAccountant(), // disable accountant
 			GuardianOptionGovernor(true),
 			GuardianOptionGatewayRelayer("", nil), // disable gateway relayer
-			GuardianOptionP2P(gs[mockGuardianIndex].p2pKey, networkID, bootstrapPeers, nodeName, false, cfg.p2pPort, func() string { return "" }),
+			GuardianOptionP2P(gs[mockGuardianIndex].p2pKey, networkID, bootstrapPeers, nodeName, false, cfg.p2pPort, "", 0, "", func() string { return "" }),
 			GuardianOptionPublicRpcSocket(cfg.publicSocket, publicRpcLogDetail),
 			GuardianOptionPublicrpcTcpService(cfg.publicRpc, publicRpcLogDetail),
 			GuardianOptionPublicWeb(cfg.publicWeb, cfg.publicSocket, "", false, ""),

node/pkg/node/options.go

@@ -16,6 +16,7 @@ import (
 	"github.com/certusone/wormhole/node/pkg/p2p"
 	"github.com/certusone/wormhole/node/pkg/processor"
 	gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1"
+	"github.com/certusone/wormhole/node/pkg/query"
 	"github.com/certusone/wormhole/node/pkg/readiness"
 	"github.com/certusone/wormhole/node/pkg/supervisor"
 	"github.com/certusone/wormhole/node/pkg/watchers"
@@ -38,7 +39,7 @@ type GuardianOption struct {
 
 // GuardianOptionP2P configures p2p networking.
 // Dependencies: Accountant, Governor
-func GuardianOptionP2P(p2pKey libp2p_crypto.PrivKey, networkId string, bootstrapPeers string, nodeName string, disableHeartbeatVerify bool, port uint, ibcFeaturesFunc func() string) *GuardianOption {
+func GuardianOptionP2P(p2pKey libp2p_crypto.PrivKey, networkId string, bootstrapPeers string, nodeName string, disableHeartbeatVerify bool, port uint, ccqBootstrapPeers string, ccqPort uint, ccqAllowedPeers string, ibcFeaturesFunc func() string) *GuardianOption {
 	return &GuardianOption{
 		name:         "p2p",
 		dependencies: []string{"accountant", "governor", "gateway-relayer"},
@@ -72,6 +73,36 @@ func GuardianOptionP2P(p2pKey libp2p_crypto.PrivKey, networkId string, bootstrap
 				components,
 				ibcFeaturesFunc,
 				(g.gatewayRelayer != nil),
+				(g.queryHandler != nil),
+				g.signedQueryReqC.writeC,
+				g.queryResponsePublicationC.readC,
+				ccqBootstrapPeers,
+				ccqPort,
+				ccqAllowedPeers,
+			)
+
+			return nil
+		}}
+}
+
+// GuardianOptionQueryHandler configures the Cross Chain Query module.
+func GuardianOptionQueryHandler(ccqEnabled bool, allowedRequesters string) *GuardianOption {
+	return &GuardianOption{
+		name: "query",
+		f: func(ctx context.Context, logger *zap.Logger, g *G) error {
+			if !ccqEnabled {
+				logger.Info("ccq: cross chain query is disabled", zap.String("component", "ccq"))
+				return nil
+			}
+
+			g.queryHandler = query.NewQueryHandler(
+				logger,
+				g.env,
+				allowedRequesters,
+				g.signedQueryReqC.readC,
+				g.chainQueryReqC,
+				g.queryResponseC.readC,
+				g.queryResponsePublicationC.writeC,
 			)
 
 			return nil
@@ -301,6 +332,32 @@ func GuardianOptionWatchers(watcherConfigs []watchers.WatcherConfig, ibcWatcherC
 				}(chainMsgC[chainId], chainId)
 			}
 
+			// Per-chain query response channel
+			chainQueryResponseC := make(map[vaa.ChainID]chan *query.PerChainQueryResponseInternal)
+			// aggregate per-chain msgC into msgC.
+			// SECURITY defense-in-depth: This way we enforce that a watcher must set the msg.EmitterChain to its chainId, which makes the code easier to audit
+			for _, chainId := range vaa.GetAllNetworkIDs() {
+				chainQueryResponseC[chainId] = make(chan *query.PerChainQueryResponseInternal)
+				go func(c <-chan *query.PerChainQueryResponseInternal, chainId vaa.ChainID) {
+					for {
+						select {
+						case <-ctx.Done():
+							return
+						case response := <-c:
+							if response.ChainId != chainId {
+								// SECURITY: This should never happen. If it does, a watcher has been compromised.
+								logger.Fatal("SECURITY CRITICAL: Received query response from a chain that was not marked as originating from that chain",
+									zap.Uint16("responseChainId", uint16(response.ChainId)),
+									zap.Stringer("watcherChainId", chainId),
+								)
+							} else {
+								g.queryResponseC.writeC <- response
+							}
+						}
+					}
+				}(chainQueryResponseC[chainId], chainId)
+			}
+
 			watchers := make(map[watchers.NetworkID]interfaces.L1Finalizer)
 
 			for _, wc := range watcherConfigs {
@@ -316,6 +373,7 @@ func GuardianOptionWatchers(watcherConfigs []watchers.WatcherConfig, ibcWatcherC
 				}
 
 				chainObsvReqC[wc.GetChainID()] = make(chan *gossipv1.ObservationRequest, observationRequestPerChainBufferSize)
+				g.chainQueryReqC[wc.GetChainID()] = make(chan *query.PerChainQueryInternal, query.QueryRequestBufferSize)
 
 				if wc.RequiredL1Finalizer() != "" {
 					l1watcher, ok := watchers[wc.RequiredL1Finalizer()]
@@ -327,7 +385,7 @@ func GuardianOptionWatchers(watcherConfigs []watchers.WatcherConfig, ibcWatcherC
 					wc.SetL1Finalizer(l1watcher)
 				}
 
-				l1finalizer, runnable, err := wc.Create(chainMsgC[wc.GetChainID()], chainObsvReqC[wc.GetChainID()], g.setC.writeC, g.env)
+				l1finalizer, runnable, err := wc.Create(chainMsgC[wc.GetChainID()], chainObsvReqC[wc.GetChainID()], g.chainQueryReqC[wc.GetChainID()], chainQueryResponseC[wc.GetChainID()], g.setC.writeC, g.env)
 
 				if err != nil {
 					return fmt.Errorf("error creating watcher: %w", err)