Skip to content

Commit 191ccd7

Browse files
evan-grayevan-gray
committed
chore(hack): broadcast
1 parent 94140f0 commit 191ccd7

1 file changed

Lines changed: 274 additions & 0 deletions

File tree

‎node/hack/broadcastvaa/broadcast_vaa.go‎

Lines changed: 274 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,274 @@
1+
// broadcastvaa is a one-off recovery tool. It takes a single, already-signed VAA
2+
// as a hex string, verifies it against the on-chain guardian set that it
3+
// references, and (when --broadcast is set) broadcasts it over the gossip
4+
// network as a `SignedVAAWithQuorum`.
5+
//
6+
// This is a hack tool and should not be imported by any production tooling.
7+
//
8+
// Example:
9+
//
10+
// go run . \
11+
// --vaa 01000000... \
12+
// --ethRPC https://ethereum-rpc.publicnode.com \
13+
// --broadcast
14+
package main
15+
16+
import (
17+
"context"
18+
"encoding/hex"
19+
"flag"
20+
"fmt"
21+
"log"
22+
"strings"
23+
"time"
24+
25+
"github.com/certusone/wormhole/node/pkg/common"
26+
"github.com/certusone/wormhole/node/pkg/p2p"
27+
gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1"
28+
"github.com/certusone/wormhole/node/pkg/supervisor"
29+
"github.com/certusone/wormhole/node/pkg/watchers/evm"
30+
31+
ethAbi "github.com/certusone/wormhole/node/pkg/watchers/evm/connectors/ethabi"
32+
ethBind "github.com/ethereum/go-ethereum/accounts/abi/bind"
33+
ethCommon "github.com/ethereum/go-ethereum/common"
34+
ethClient "github.com/ethereum/go-ethereum/ethclient"
35+
ethRpc "github.com/ethereum/go-ethereum/rpc"
36+
37+
"github.com/wormhole-foundation/wormhole/sdk/vaa"
38+
"go.uber.org/zap"
39+
"google.golang.org/protobuf/proto"
40+
)
41+
42+
const (
43+
// defaultP2PPort is the default UDP port the broadcast node listens on.
44+
defaultP2PPort = 8997
45+
// defaultBroadcastWait is how long to stay connected after broadcasting, to allow propagation.
46+
defaultBroadcastWait = 15 * time.Second
47+
// rpcTimeout bounds Ethereum RPC calls.
48+
rpcTimeout = 15 * time.Second
49+
)
50+
51+
var (
52+
vaaHex = flag.String("vaa", "", "The signed VAA to broadcast, as a hex string (with or without a 0x prefix)")
53+
ethRPC = flag.String("ethRPC", "", "Ethereum RPC endpoint used to read the guardian set (defaults to the Ethereum public RPC from the evm chain config for the env)")
54+
coreContract = flag.String("coreContract", "", "Core bridge contract address (defaults to the Ethereum core contract from the evm chain config for the env)")
55+
envStr = flag.String("env", "mainnet", "Environment: mainnet or testnet")
56+
wormholescan = flag.String("wormholescan", "https://api.wormholescan.io", "Wormholescan API base URL (used only to log the signed VAA link)")
57+
nodeKeyPath = flag.String("nodeKey", "/tmp/broadcastvaa_node.key", "Path to the libp2p node key (created if missing)")
58+
p2pPort = flag.Uint("port", defaultP2PPort, "P2P UDP port to listen on")
59+
p2pNetworkID = flag.String("network", "", "P2P network identifier (overrides the env default)")
60+
p2pBootstrap = flag.String("bootstrap", "", "P2P bootstrap peers (overrides the env default)")
61+
broadcastWait = flag.Duration("broadcastWait", defaultBroadcastWait, "How long to stay connected after broadcasting, to allow propagation")
62+
doBroadcast = flag.Bool("broadcast", false, "Broadcast the verified VAA over gossip. If not set, the tool only parses and verifies it.")
63+
)
64+
65+
func main() {
66+
flag.Parse()
67+
68+
if *vaaHex == "" {
69+
log.Fatal("--vaa is required")
70+
}
71+
72+
env, err := common.ParseEnvironment(*envStr)
73+
if err != nil || (env != common.MainNet && env != common.TestNet) {
74+
log.Fatalf("--env must be mainnet or testnet, got %q", *envStr)
75+
}
76+
77+
logger, err := zap.NewDevelopment()
78+
if err != nil {
79+
log.Fatalf("failed to create logger: %v", err)
80+
}
81+
82+
ctx := context.Background()
83+
84+
// Parse the VAA from hex.
85+
vaaBytes, err := hex.DecodeString(strings.TrimPrefix(strings.TrimSpace(*vaaHex), "0x"))
86+
if err != nil {
87+
log.Fatalf("failed to hex-decode --vaa: %v", err)
88+
}
89+
v, err := vaa.Unmarshal(vaaBytes)
90+
if err != nil {
91+
log.Fatalf("failed to parse VAA: %v", err)
92+
}
93+
logger.Info("parsed VAA",
94+
zap.String("messageID", v.MessageID()),
95+
zap.Uint32("guardianSetIndex", v.GuardianSetIndex),
96+
zap.Int("numSignatures", len(v.Signatures)),
97+
zap.String("digest", v.SigningDigest().Hex()),
98+
)
99+
100+
// Pull the Ethereum core contract address and public RPC from the evm chain config,
101+
// using them as defaults unless overridden on the command line.
102+
chainCfg, err := evm.GetChainConfigMap(env)
103+
if err != nil {
104+
log.Fatalf("failed to load evm chain config for env %s: %v", env, err)
105+
}
106+
ethCfg, ok := chainCfg[vaa.ChainIDEthereum]
107+
if !ok {
108+
log.Fatalf("no Ethereum entry in evm chain config for env %s", env)
109+
}
110+
111+
coreAddr := *coreContract
112+
if coreAddr == "" {
113+
coreAddr = ethCfg.ContractAddr
114+
}
115+
rpcURL := *ethRPC
116+
if rpcURL == "" {
117+
rpcURL = ethCfg.PublicRPC
118+
}
119+
logger.Info("using Ethereum core contract and RPC",
120+
zap.String("coreContract", coreAddr),
121+
zap.String("ethRPC", rpcURL),
122+
)
123+
124+
// Read the guardian set that the VAA references from the core contract, and verify
125+
// the VAA against it.
126+
gs, err := fetchGuardianSet(ctx, rpcURL, coreAddr, v.GuardianSetIndex)
127+
if err != nil {
128+
log.Fatalf("failed to fetch guardian set %d: %v", v.GuardianSetIndex, err)
129+
}
130+
logger.Info("fetched guardian set",
131+
zap.Uint32("index", gs.Index),
132+
zap.Int("numKeys", len(gs.Keys)),
133+
zap.Int("quorum", gs.Quorum()),
134+
)
135+
136+
if err := v.Verify(gs.Keys); err != nil {
137+
log.Fatalf("VAA failed verification against guardian set %d: %v", gs.Index, err)
138+
}
139+
logger.Info("VAA verified",
140+
zap.String("messageID", v.MessageID()),
141+
zap.Int("numSignatures", len(v.Signatures)),
142+
)
143+
144+
if !*doBroadcast {
145+
logger.Info("not broadcasting. Re-run with --broadcast to broadcast over gossip.")
146+
return
147+
}
148+
149+
if err := broadcast(ctx, logger, env, gs, v); err != nil {
150+
log.Fatalf("failed to broadcast VAA: %v", err)
151+
}
152+
}
153+
154+
// fetchGuardianSet reads the keys for the given guardian set index from the core contract.
155+
func fetchGuardianSet(ctx context.Context, rpcURL, coreAddr string, index uint32) (*common.GuardianSet, error) {
156+
timeout, cancel := context.WithTimeout(ctx, rpcTimeout)
157+
defer cancel()
158+
159+
rawClient, err := ethRpc.DialContext(timeout, rpcURL)
160+
if err != nil {
161+
return nil, fmt.Errorf("failed to connect to ethereum: %w", err)
162+
}
163+
client := ethClient.NewClient(rawClient)
164+
caller, err := ethAbi.NewAbiCaller(ethCommon.HexToAddress(coreAddr), client)
165+
if err != nil {
166+
return nil, fmt.Errorf("failed to create core contract caller: %w", err)
167+
}
168+
169+
result, err := caller.GetGuardianSet(&ethBind.CallOpts{Context: timeout}, index)
170+
if err != nil {
171+
return nil, fmt.Errorf("failed to read guardian set %d: %w", index, err)
172+
}
173+
174+
return common.NewGuardianSet(result.Keys, index), nil
175+
}
176+
177+
// broadcast starts a p2p node and publishes the verified VAA as a SignedVAAWithQuorum.
178+
func broadcast(
179+
ctx context.Context,
180+
logger *zap.Logger,
181+
env common.Environment,
182+
gs *common.GuardianSet,
183+
v *vaa.VAA,
184+
) error {
185+
networkID := *p2pNetworkID
186+
if networkID == "" {
187+
networkID = p2p.GetNetworkId(env)
188+
}
189+
bootstrap := *p2pBootstrap
190+
if bootstrap == "" {
191+
var err error
192+
bootstrap, err = p2p.GetBootstrapPeers(env)
193+
if err != nil {
194+
return fmt.Errorf("failed to determine bootstrap peers: %w", err)
195+
}
196+
}
197+
198+
priv, err := common.GetOrCreateNodeKey(logger, *nodeKeyPath)
199+
if err != nil {
200+
return fmt.Errorf("failed to load node key: %w", err)
201+
}
202+
203+
// Make the guardian set available to the p2p layer.
204+
gst := common.NewGuardianSetState(nil)
205+
gst.Set(gs)
206+
207+
marshaled, err := v.Marshal()
208+
if err != nil {
209+
return fmt.Errorf("failed to marshal VAA %s: %w", v.MessageID(), err)
210+
}
211+
w := gossipv1.GossipMessage{Message: &gossipv1.GossipMessage_SignedVaaWithQuorum{
212+
SignedVaaWithQuorum: &gossipv1.SignedVAAWithQuorum{Vaa: marshaled},
213+
}}
214+
msg, err := proto.Marshal(&w)
215+
if err != nil {
216+
return fmt.Errorf("failed to marshal gossip message for %s: %w", v.MessageID(), err)
217+
}
218+
219+
// Buffer the channel so the VAA can be queued without blocking.
220+
gossipVaaSendC := make(chan []byte, 1)
221+
gossipVaaSendC <- msg
222+
223+
// Log the Wormholescan signed VAA link so it can be polled to confirm the broadcast was picked up.
224+
logger.Info("broadcasting VAA",
225+
zap.String("messageID", v.MessageID()),
226+
zap.String("wormholescan", fmt.Sprintf("%s/v1/signed_vaa/%d/%s/%d",
227+
*wormholescan, v.EmitterChain, v.EmitterAddress, v.Sequence)),
228+
)
229+
230+
rootCtx, rootCtxCancel := context.WithCancel(ctx)
231+
defer rootCtxCancel()
232+
233+
logger.Info("starting p2p node to broadcast VAA",
234+
zap.String("networkID", networkID),
235+
zap.Duration("broadcastWait", *broadcastWait),
236+
)
237+
238+
supervisor.New(rootCtx, logger, func(supCtx context.Context) error {
239+
components := p2p.DefaultComponents()
240+
components.Port = *p2pPort
241+
242+
params, err := p2p.NewRunParams(
243+
bootstrap,
244+
networkID,
245+
priv,
246+
gst,
247+
rootCtxCancel,
248+
p2p.WithVAASender(gossipVaaSendC),
249+
p2p.WithComponents(components),
250+
)
251+
if err != nil {
252+
return err
253+
}
254+
255+
if err := supervisor.Run(supCtx, "p2p", p2p.Run(params)); err != nil {
256+
return err
257+
}
258+
259+
logger.Info("p2p started; broadcasting and waiting for propagation")
260+
261+
// The send loop drains gossipVaaSendC as soon as the topic is joined. Wait for
262+
// the configured duration to give the message time to propagate, then shut down.
263+
select {
264+
case <-supCtx.Done():
265+
case <-time.After(*broadcastWait):
266+
}
267+
rootCtxCancel()
268+
return nil
269+
}, supervisor.WithPropagatePanic)
270+
271+
<-rootCtx.Done()
272+
logger.Info("done broadcasting")
273+
return nil
274+
}

0 commit comments

Comments
 (0)