From 7ee02b41e77e8d08479e74fa75c83dd056292a21 Mon Sep 17 00:00:00 2001 From: Evan Gray <56235822+evan-gray@users.noreply.github.com> Date: Fri, 15 May 2026 14:56:15 -0400 Subject: [PATCH] feat(node): aptos reobserve with endpoint --- node/pkg/watchers/aptos/chain_config.go | 145 +++++++++++++ node/pkg/watchers/aptos/chain_config_test.go | 178 ++++++++++++++++ node/pkg/watchers/aptos/config.go | 5 +- node/pkg/watchers/aptos/reobserve.go | 93 ++++++++ node/pkg/watchers/aptos/reobserve_test.go | 213 +++++++++++++++++++ node/pkg/watchers/aptos/watcher.go | 87 +++----- 6 files changed, 657 insertions(+), 64 deletions(-) create mode 100644 node/pkg/watchers/aptos/chain_config.go create mode 100644 node/pkg/watchers/aptos/chain_config_test.go create mode 100644 node/pkg/watchers/aptos/reobserve.go create mode 100644 node/pkg/watchers/aptos/reobserve_test.go diff --git a/node/pkg/watchers/aptos/chain_config.go b/node/pkg/watchers/aptos/chain_config.go new file mode 100644 index 00000000000..fd75adb2b6a --- /dev/null +++ b/node/pkg/watchers/aptos/chain_config.go @@ -0,0 +1,145 @@ +package aptos + +// This file defines the set of Aptos-derived chains supported by the guardian watcher and their native chain IDs. +// The native chain ID is what the node returns in the `chain_id` field of its `/v1` endpoint. + +import ( + "context" + "errors" + "fmt" + "math" + "net/http" + "time" + + "github.com/certusone/wormhole/node/pkg/common" + "github.com/tidwall/gjson" + "github.com/wormhole-foundation/wormhole/sdk/vaa" + "go.uber.org/zap" +) + +type ( + // EnvEntry specifies the config data for a given chain / environment. + EnvEntry struct { + // AptosChainID is the expected native chain ID (the `chain_id` field returned by the `/v1` endpoint). + AptosChainID uint64 + } + + // EnvMap defines the config data for a given environment (mainnet or testnet). + EnvMap map[vaa.ChainID]EnvEntry +) + +const ( + aptosMainnetChainID uint64 = 1 + aptosTestnetChainID uint64 = 2 + movementMainnetChainID uint64 = 126 + movementTestnetChainID uint64 = 250 + + chainIDQueryTimeout = 15 * time.Second +) + +var ( + ErrInvalidEnv = errors.New("invalid environment") + ErrNotFound = errors.New("not found") + + mainnetChainConfig = EnvMap{ + vaa.ChainIDAptos: {AptosChainID: aptosMainnetChainID}, + vaa.ChainIDMovement: {AptosChainID: movementMainnetChainID}, + } + + testnetChainConfig = EnvMap{ + vaa.ChainIDAptos: {AptosChainID: aptosTestnetChainID}, + vaa.ChainIDMovement: {AptosChainID: movementTestnetChainID}, + } +) + +// GetAptosChainID returns the configured native chain ID for the specified environment / chain. +func GetAptosChainID(env common.Environment, chainID vaa.ChainID) (uint64, error) { + m, err := GetChainConfigMap(env) + if err != nil { + return 0, err + } + + entry, exists := m[chainID] + if !exists { + return 0, ErrNotFound + } + + return entry.AptosChainID, nil +} + +// GetChainConfigMap returns the configuration map for the specified environment. +func GetChainConfigMap(env common.Environment) (EnvMap, error) { + if env == common.MainNet { + return mainnetChainConfig, nil + } + + if env == common.TestNet { + return testnetChainConfig, nil + } + + return EnvMap{}, ErrInvalidEnv +} + +// verifyAptosChainID reads the native chain ID from the node and verifies that it matches the expected value +// (making sure we aren't connected to the wrong chain). +func (e *Watcher) verifyAptosChainID(ctx context.Context, logger *zap.Logger, url string) error { + // Don't bother to check in tilt. + if e.env == common.UnsafeDevNet { + return nil + } + + expected, err := GetAptosChainID(e.env, e.chainID) + if err != nil { + return fmt.Errorf("failed to look up aptos chain id: %w", err) + } + + timeout, cancel := context.WithTimeout(ctx, chainIDQueryTimeout) + defer cancel() + + actual, err := queryAptosChainID(timeout, url) + if err != nil { + return err + } + + logger.Info("queried aptos chain id", zap.Uint64("expected", expected), zap.Uint64("actual", actual)) + + if actual != expected { + return fmt.Errorf("aptos chain ID mismatch, expected %d, received %d", expected, actual) + } + + return nil +} + +// queryAptosChainID queries the specified RPC for the native Aptos chain ID returned by the `/v1` endpoint. +func queryAptosChainID(ctx context.Context, url string) (uint64, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("%s/v1", url), nil) + if err != nil { + return 0, fmt.Errorf("failed to build chain id request: %w", err) + } + + res, err := http.DefaultClient.Do(req) + if err != nil { + return 0, fmt.Errorf("failed to query aptos chain id: %w", err) + } + defer res.Body.Close() + + body, err := common.SafeRead(res.Body) + if err != nil { + return 0, fmt.Errorf("failed to read aptos chain id response: %w", err) + } + + if !gjson.Valid(string(body)) { + return 0, fmt.Errorf("invalid JSON in chain id response: %s", string(body)) + } + + id := gjson.GetBytes(body, "chain_id") + if !id.Exists() { + return 0, fmt.Errorf("chain_id field missing from response") + } + + v := id.Uint() + if v == 0 || v > math.MaxUint32 { + return 0, fmt.Errorf("chain_id %d out of expected range", v) + } + return v, nil +} diff --git a/node/pkg/watchers/aptos/chain_config_test.go b/node/pkg/watchers/aptos/chain_config_test.go new file mode 100644 index 00000000000..1475c6dd9e2 --- /dev/null +++ b/node/pkg/watchers/aptos/chain_config_test.go @@ -0,0 +1,178 @@ +package aptos + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "testing" + + "github.com/certusone/wormhole/node/pkg/common" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/wormhole-foundation/wormhole/sdk/vaa" + "go.uber.org/zap" +) + +func TestGetChainConfigMap(t *testing.T) { + m, err := GetChainConfigMap(common.MainNet) + require.NoError(t, err) + assert.Equal(t, mainnetChainConfig, m) + + m, err = GetChainConfigMap(common.TestNet) + require.NoError(t, err) + assert.Equal(t, testnetChainConfig, m) + + _, err = GetChainConfigMap(common.UnsafeDevNet) + require.ErrorIs(t, err, ErrInvalidEnv) +} + +func TestGetAptosChainID(t *testing.T) { + tests := []struct { + name string + env common.Environment + chainID vaa.ChainID + want uint64 + wantErr error + }{ + {"mainnet aptos", common.MainNet, vaa.ChainIDAptos, 1, nil}, + {"mainnet movement", common.MainNet, vaa.ChainIDMovement, 126, nil}, + {"testnet aptos", common.TestNet, vaa.ChainIDAptos, 2, nil}, + {"testnet movement", common.TestNet, vaa.ChainIDMovement, 250, nil}, + {"unknown chain", common.MainNet, vaa.ChainIDEthereum, 0, ErrNotFound}, + {"invalid env", common.UnsafeDevNet, vaa.ChainIDAptos, 0, ErrInvalidEnv}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got, err := GetAptosChainID(tc.env, tc.chainID) + if tc.wantErr != nil { + require.ErrorIs(t, err, tc.wantErr) + return + } + require.NoError(t, err) + assert.Equal(t, tc.want, got) + }) + } +} + +// chainIDServer returns an httptest server that responds to `GET /v1` with body. +func chainIDServer(body string) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1" { + fmt.Fprint(w, body) + } + })) +} + +func TestVerifyAptosChainID(t *testing.T) { + logger := zap.NewNop() + ctx := context.Background() + + t.Run("devnet bypass", func(t *testing.T) { + w := &Watcher{env: common.UnsafeDevNet, chainID: vaa.ChainIDAptos} + require.NoError(t, w.verifyAptosChainID(ctx, logger, "http://unused")) + }) + + t.Run("unknown chain lookup error", func(t *testing.T) { + w := &Watcher{env: common.MainNet, chainID: vaa.ChainIDEthereum} + err := w.verifyAptosChainID(ctx, logger, "http://unused") + require.ErrorContains(t, err, "failed to look up aptos chain id") + }) + + t.Run("request build error", func(t *testing.T) { + // Embedded control character makes the URL fail to parse in net/http. + w := &Watcher{env: common.MainNet, chainID: vaa.ChainIDAptos} + err := w.verifyAptosChainID(ctx, logger, "http://example.com\x00") + require.Error(t, err) + assert.Contains(t, err.Error(), "failed to build chain id request") + }) + + t.Run("connection error", func(t *testing.T) { + // Spin up and immediately close a server to get a guaranteed-unreachable URL. + s := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + s.Close() + w := &Watcher{env: common.MainNet, chainID: vaa.ChainIDAptos} + err := w.verifyAptosChainID(ctx, logger, s.URL) + require.Error(t, err) + assert.Contains(t, err.Error(), "failed to query aptos chain id") + }) + + t.Run("body read error", func(t *testing.T) { + // Server claims a longer Content-Length than it actually writes, then closes the + // connection. The client's read of the body will fail with unexpected EOF. + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hj, ok := w.(http.Hijacker) + require.True(t, ok, "server doesn't support hijacking") + conn, _, err := hj.Hijack() + require.NoError(t, err) + defer conn.Close() + _, _ = fmt.Fprint(conn, "HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n") + _, _ = fmt.Fprint(conn, "short") + })) + defer s.Close() + w := &Watcher{env: common.MainNet, chainID: vaa.ChainIDAptos} + err := w.verifyAptosChainID(ctx, logger, s.URL) + require.ErrorContains(t, err, "failed to read aptos chain id response") + }) + + t.Run("invalid JSON", func(t *testing.T) { + s := chainIDServer("not json") + defer s.Close() + w := &Watcher{env: common.MainNet, chainID: vaa.ChainIDAptos} + err := w.verifyAptosChainID(ctx, logger, s.URL) + require.ErrorContains(t, err, "invalid JSON") + }) + + t.Run("chain_id missing", func(t *testing.T) { + s := chainIDServer(`{"epoch":"123"}`) + defer s.Close() + w := &Watcher{env: common.MainNet, chainID: vaa.ChainIDAptos} + err := w.verifyAptosChainID(ctx, logger, s.URL) + require.ErrorContains(t, err, "chain_id field missing") + }) + + t.Run("chain_id zero", func(t *testing.T) { + s := chainIDServer(`{"chain_id":0}`) + defer s.Close() + w := &Watcher{env: common.MainNet, chainID: vaa.ChainIDAptos} + err := w.verifyAptosChainID(ctx, logger, s.URL) + require.ErrorContains(t, err, "out of expected range") + }) + + t.Run("chain_id too large", func(t *testing.T) { + s := chainIDServer(`{"chain_id":4294967296}`) // MaxUint32 + 1 + defer s.Close() + w := &Watcher{env: common.MainNet, chainID: vaa.ChainIDAptos} + err := w.verifyAptosChainID(ctx, logger, s.URL) + require.ErrorContains(t, err, "out of expected range") + }) + + t.Run("mismatch", func(t *testing.T) { + s := chainIDServer(`{"chain_id":99}`) + defer s.Close() + w := &Watcher{env: common.MainNet, chainID: vaa.ChainIDAptos} + err := w.verifyAptosChainID(ctx, logger, s.URL) + require.ErrorContains(t, err, "mismatch") + }) + + t.Run("success aptos mainnet", func(t *testing.T) { + s := chainIDServer(`{"chain_id":1}`) + defer s.Close() + w := &Watcher{env: common.MainNet, chainID: vaa.ChainIDAptos} + require.NoError(t, w.verifyAptosChainID(ctx, logger, s.URL)) + }) + + t.Run("success movement testnet", func(t *testing.T) { + s := chainIDServer(`{"chain_id":250}`) + defer s.Close() + w := &Watcher{env: common.TestNet, chainID: vaa.ChainIDMovement} + require.NoError(t, w.verifyAptosChainID(ctx, logger, s.URL)) + }) +} + +// Sanity check: the package-level sentinel errors should be distinct. +func TestSentinelErrors(t *testing.T) { + require.False(t, errors.Is(ErrInvalidEnv, ErrNotFound)) + require.False(t, errors.Is(ErrNotFound, ErrInvalidEnv)) +} diff --git a/node/pkg/watchers/aptos/config.go b/node/pkg/watchers/aptos/config.go index 4dbdbb5ce61..0898c518d66 100644 --- a/node/pkg/watchers/aptos/config.go +++ b/node/pkg/watchers/aptos/config.go @@ -33,7 +33,8 @@ func (wc *WatcherConfig) Create( _ <-chan *query.PerChainQueryInternal, _ chan<- *query.PerChainQueryResponseInternal, _ chan<- *common.GuardianSet, - _ common.Environment, + env common.Environment, ) (supervisor.Runnable, interfaces.Reobserver, error) { - return NewWatcher(wc.ChainID, wc.NetworkID, wc.Rpc, wc.Account, wc.Handle, msgC, obsvReqC).Run, nil, nil + watcher := NewWatcher(wc.ChainID, wc.NetworkID, env, wc.Rpc, wc.Account, wc.Handle, msgC, obsvReqC) + return watcher.Run, watcher, nil } diff --git a/node/pkg/watchers/aptos/reobserve.go b/node/pkg/watchers/aptos/reobserve.go new file mode 100644 index 00000000000..f9e9c84ab95 --- /dev/null +++ b/node/pkg/watchers/aptos/reobserve.go @@ -0,0 +1,93 @@ +package aptos + +import ( + "context" + "encoding/binary" + "fmt" + + "github.com/tidwall/gjson" + "github.com/wormhole-foundation/wormhole/sdk/vaa" + "go.uber.org/zap" +) + +// handleReobservationRequest performs a reobservation against the given Aptos RPC base URL +// and publishes any observed messages. Returns the number of messages successfully observed. +func (e *Watcher) handleReobservationRequest(logger *zap.Logger, chainID vaa.ChainID, txHash []byte, rpcURL string) (uint32, error) { + // The caller is expected to send us only requests for our chainID. + if chainID != e.chainID { + return 0, fmt.Errorf("unexpected chain id: %v", chainID) + } + + // Aptos's TxID is a uint64. Historically, all TxIDs used a fixed 32-byte hash type. + // This parsing is leftover from that time period. It should be possible to refactor + // this code such that the TxID received from p2p is exactly 8 bytes, which would + // obviate the need for the below bounds check and parsing. + // + // SECURITY: This acts as a bounds check for the BigEndian.Uint64 call below. + const AptosTxIDExpectedLen = 32 + if len(txHash) < AptosTxIDExpectedLen { + return 0, fmt.Errorf("invalid TxID: too short") + } + + // uint64 will read the *first* 8 bytes, but the sequence is stored in the *last* 8. + nativeSeq := binary.BigEndian.Uint64(txHash[24:]) + + logger.Info("Received obsv request", + zap.Uint64("tx_hash", nativeSeq), + zap.String("rpc", rpcURL), + ) + + // SECURITY: the API guarantees that we only get the events from the right contract. + eventsEndpoint := fmt.Sprintf(`%s/v1/accounts/%s/events/%s/event`, rpcURL, e.aptosAccount, e.aptosHandle) + s := fmt.Sprintf(`%s?start=%d&limit=1`, eventsEndpoint, nativeSeq) + + body, err := e.retrievePayload(s) + if err != nil { + return 0, fmt.Errorf("retrievePayload: %w", err) + } + + if !gjson.Valid(string(body)) { + return 0, fmt.Errorf("invalid JSON in reobservation response: %s", string(body)) + } + + var numObservations uint32 + for _, chunk := range gjson.ParseBytes(body).Array() { + newSeq := chunk.Get("sequence_number") + if !newSeq.Exists() { + break + } + + if newSeq.Uint() != nativeSeq { + return numObservations, fmt.Errorf("newSeq != nativeSeq") + } + + data := chunk.Get("data") + if !data.Exists() { + break + } + if e.observeData(logger, data, nativeSeq, true) { + numObservations++ + } + } + return numObservations, nil +} + +// Reobserve is the interface for reobserving using a custom URL. It performs the reobservation against that URL. +func (e *Watcher) Reobserve(ctx context.Context, chainID vaa.ChainID, txID []byte, customEndpoint string) (uint32, error) { + logger := e.logger + if logger == nil { + logger = zap.NewNop() + } + logger.Info("received a request to reobserve using a custom endpoint", + zap.Stringer("chainID", chainID), + zap.Any("txID", txID), + zap.String("url", customEndpoint), + ) + + // Verify that this endpoint is for the correct chain. + if err := e.verifyAptosChainID(ctx, logger, customEndpoint); err != nil { + return 0, fmt.Errorf("failed to verify aptos chain id: %w", err) + } + + return e.handleReobservationRequest(logger, chainID, txID, customEndpoint) +} diff --git a/node/pkg/watchers/aptos/reobserve_test.go b/node/pkg/watchers/aptos/reobserve_test.go new file mode 100644 index 00000000000..bc8f1583586 --- /dev/null +++ b/node/pkg/watchers/aptos/reobserve_test.go @@ -0,0 +1,213 @@ +package aptos + +import ( + "context" + "encoding/binary" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/certusone/wormhole/node/pkg/common" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/wormhole-foundation/wormhole/sdk/vaa" + "go.uber.org/zap" +) + +const ( + testAccount = "0xdeadbeef" + testHandle = "0xdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef::wormhole::WormholeMessageHandle" +) + +// txHashWithSeq returns a 32-byte buffer with seq in the last 8 bytes (matching the +// transport format used by the watcher and the production observation request flow). +func txHashWithSeq(seq uint64) []byte { + b := make([]byte, 32) + binary.BigEndian.PutUint64(b[24:], seq) + return b +} + +// eventsServer returns an httptest server that serves event responses for the configured +// account/handle and chain id responses for `/v1`. +func eventsServer(t *testing.T, eventsBody, chainIDBody string) *httptest.Server { + t.Helper() + eventsPath := fmt.Sprintf("/v1/accounts/%s/events/%s/event", testAccount, testHandle) + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case eventsPath: + fmt.Fprint(w, eventsBody) + case "/v1": + fmt.Fprint(w, chainIDBody) + default: + http.NotFound(w, r) + } + })) +} + +func newTestWatcher(msgC chan<- *common.MessagePublication) *Watcher { + return &Watcher{ + chainID: vaa.ChainIDAptos, + networkID: "aptos-test", + env: common.MainNet, + aptosAccount: testAccount, + aptosHandle: testHandle, + msgC: msgC, + } +} + +func validEventArray(seq uint64) string { + return fmt.Sprintf(`[{ + "sequence_number":"%d", + "data":{ + "sender":"1", + "payload":"0xdeadbeef", + "timestamp":"1000", + "nonce":"42", + "sequence":"7", + "consistency_level":"1" + } + }]`, seq) +} + +func TestHandleReobservationRequest_ChainIDMismatch(t *testing.T) { + w := newTestWatcher(nil) + n, err := w.handleReobservationRequest(zap.NewNop(), vaa.ChainIDEthereum, txHashWithSeq(1), "http://unused") + require.ErrorContains(t, err, "unexpected chain id") + assert.Zero(t, n) +} + +func TestHandleReobservationRequest_TxHashTooShort(t *testing.T) { + w := newTestWatcher(nil) + n, err := w.handleReobservationRequest(zap.NewNop(), w.chainID, make([]byte, 31), "http://unused") + require.ErrorContains(t, err, "too short") + assert.Zero(t, n) +} + +func TestHandleReobservationRequest_RetrievePayloadError(t *testing.T) { + // Spin up and immediately close to get a guaranteed-unreachable URL. + s := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + s.Close() + + w := newTestWatcher(nil) + n, err := w.handleReobservationRequest(zap.NewNop(), w.chainID, txHashWithSeq(1), s.URL) + require.ErrorContains(t, err, "retrievePayload") + assert.Zero(t, n) +} + +func TestHandleReobservationRequest_InvalidJSON(t *testing.T) { + s := eventsServer(t, "not json", "") + defer s.Close() + + w := newTestWatcher(nil) + n, err := w.handleReobservationRequest(zap.NewNop(), w.chainID, txHashWithSeq(1), s.URL) + require.ErrorContains(t, err, "invalid JSON") + assert.Zero(t, n) +} + +func TestHandleReobservationRequest_MissingSequenceNumber(t *testing.T) { + // Array with one entry that has no `sequence_number` — the loop should break and return cleanly. + s := eventsServer(t, `[{"data":{}}]`, "") + defer s.Close() + + w := newTestWatcher(nil) + n, err := w.handleReobservationRequest(zap.NewNop(), w.chainID, txHashWithSeq(1), s.URL) + require.NoError(t, err) + assert.Zero(t, n) +} + +func TestHandleReobservationRequest_SequenceMismatch(t *testing.T) { + s := eventsServer(t, validEventArray(99), "") + defer s.Close() + + w := newTestWatcher(nil) + n, err := w.handleReobservationRequest(zap.NewNop(), w.chainID, txHashWithSeq(1), s.URL) + require.ErrorContains(t, err, "newSeq != nativeSeq") + assert.Zero(t, n) +} + +func TestHandleReobservationRequest_MissingData(t *testing.T) { + // Entry has matching sequence_number but no `data` field. + s := eventsServer(t, `[{"sequence_number":"1"}]`, "") + defer s.Close() + + w := newTestWatcher(nil) + n, err := w.handleReobservationRequest(zap.NewNop(), w.chainID, txHashWithSeq(1), s.URL) + require.NoError(t, err) + assert.Zero(t, n) +} + +func TestHandleReobservationRequest_ObserveDataFails(t *testing.T) { + // Matching seq + data, but the data is missing required fields, so observeData returns false. + s := eventsServer(t, `[{"sequence_number":"1","data":{}}]`, "") + defer s.Close() + + msgC := make(chan *common.MessagePublication, 1) + w := newTestWatcher(msgC) + n, err := w.handleReobservationRequest(zap.NewNop(), w.chainID, txHashWithSeq(1), s.URL) + require.NoError(t, err) + assert.Zero(t, n) + assert.Empty(t, msgC) +} + +func TestHandleReobservationRequest_Success(t *testing.T) { + s := eventsServer(t, validEventArray(7), "") + defer s.Close() + + msgC := make(chan *common.MessagePublication, 1) + w := newTestWatcher(msgC) + n, err := w.handleReobservationRequest(zap.NewNop(), w.chainID, txHashWithSeq(7), s.URL) + require.NoError(t, err) + assert.Equal(t, uint32(1), n) + require.Len(t, msgC, 1) + msg := <-msgC + assert.True(t, msg.IsReobservation) + assert.Equal(t, vaa.ChainIDAptos, msg.EmitterChain) +} + +func TestReobserve_LoggerNil_DevnetBypass(t *testing.T) { + // Devnet skips the chain id verification, so this exercises the nil-logger fallback + // and a clean handleReobservationRequest call. + s := eventsServer(t, validEventArray(7), "") + defer s.Close() + + msgC := make(chan *common.MessagePublication, 1) + w := newTestWatcher(msgC) + w.env = common.UnsafeDevNet + w.logger = nil + + n, err := w.Reobserve(context.Background(), w.chainID, txHashWithSeq(7), s.URL) + require.NoError(t, err) + assert.Equal(t, uint32(1), n) +} + +func TestReobserve_VerifyChainIDFails(t *testing.T) { + s := eventsServer(t, validEventArray(7), `{"chain_id":99}`) + defer s.Close() + + w := newTestWatcher(nil) + w.logger = zap.NewNop() + + n, err := w.Reobserve(context.Background(), w.chainID, txHashWithSeq(7), s.URL) + require.Error(t, err) + assert.Contains(t, err.Error(), "failed to verify aptos chain id") + assert.Zero(t, n) +} + +func TestReobserve_Success(t *testing.T) { + s := eventsServer(t, validEventArray(7), `{"chain_id":1}`) + defer s.Close() + + msgC := make(chan *common.MessagePublication, 1) + w := newTestWatcher(msgC) + w.logger = zap.NewNop() + + n, err := w.Reobserve(context.Background(), w.chainID, txHashWithSeq(7), s.URL) + require.NoError(t, err) + assert.Equal(t, uint32(1), n) + require.Len(t, msgC, 1) + + // Sanity: log message produced when txID is the raw bytes; ensure no panic on the Any() call path. + assert.NotEmpty(t, strings.TrimSpace(string(txHashWithSeq(7)))) +} diff --git a/node/pkg/watchers/aptos/watcher.go b/node/pkg/watchers/aptos/watcher.go index eb678535d9b..51616e19d17 100644 --- a/node/pkg/watchers/aptos/watcher.go +++ b/node/pkg/watchers/aptos/watcher.go @@ -29,6 +29,7 @@ type ( Watcher struct { chainID vaa.ChainID networkID string + env common.Environment aptosRPC string aptosAccount string @@ -37,6 +38,8 @@ type ( msgC chan<- *common.MessagePublication obsvReqC <-chan *gossipv1.ObservationRequest readinessSync readiness.Component + + logger *zap.Logger } ) @@ -58,6 +61,7 @@ var ( func NewWatcher( chainID vaa.ChainID, networkID watchers.NetworkID, + env common.Environment, aptosRPC string, aptosAccount string, aptosHandle string, @@ -67,6 +71,7 @@ func NewWatcher( return &Watcher{ chainID: chainID, networkID: string(networkID), + env: env, aptosRPC: aptosRPC, aptosAccount: aptosAccount, aptosHandle: aptosHandle, @@ -82,6 +87,7 @@ func (e *Watcher) Run(ctx context.Context) error { }) logger := supervisor.Logger(ctx) + e.logger = logger logger.Info("Starting watcher", zap.String("watcher_name", e.networkID), @@ -90,6 +96,11 @@ func (e *Watcher) Run(ctx context.Context) error { zap.String("handle", e.aptosHandle), ) + // Verify that we are connecting to the correct chain. + if err := e.verifyAptosChainID(ctx, logger, e.aptosRPC); err != nil { + return fmt.Errorf("failed to verify aptos chain id: %w", err) + } + // Get the node version for troubleshooting e.logVersion(logger) @@ -127,61 +138,12 @@ func (e *Watcher) Run(ctx context.Context) error { panic("invalid chain ID") } - // Aptos's TxID is a uint64. Historically, all TxIDs used a fixed 32-byte hash type. - // This parsing is leftover from that time period. It should be possible to refactor - // this code such that the TxID received from p2p is exactly 8 bytes, which would - // obviate the need for the below bounds check and parsing. - // - // SECURITY: This acts as a bounds check for the BigEndian.Unint64 call below. - const AptosTxIDExpectedLen = 32 - if len(r.TxHash) < AptosTxIDExpectedLen { - logger.Error("invalid TxID: too short") - p2p.DefaultRegistry.AddErrorCount(e.chainID, 1) - continue - } - - // uint64 will read the *first* 8 bytes, but the sequence is stored in the *last* 8. - nativeSeq := binary.BigEndian.Uint64(r.TxHash[24:]) - - logger.Info("Received obsv request", zap.Uint64("tx_hash", nativeSeq)) - - s := fmt.Sprintf(`%s?start=%d&limit=1`, eventsEndpoint, nativeSeq) - - body, err := e.retrievePayload(s) - if err != nil { - logger.Error("retrievePayload", zap.Error(err)) + if _, err := e.handleReobservationRequest(logger, vaa.ChainID(r.ChainId), r.TxHash, e.aptosRPC); err != nil { + logger.Error("failed to process observation request", zap.Error(err)) p2p.DefaultRegistry.AddErrorCount(e.chainID, 1) continue } - if !gjson.Valid(string(body)) { - logger.Error("InvalidJson: " + string(body)) - p2p.DefaultRegistry.AddErrorCount(e.chainID, 1) - break - - } - - outcomes := gjson.ParseBytes(body) - - for _, chunk := range outcomes.Array() { - newSeq := chunk.Get("sequence_number") - if !newSeq.Exists() { - break - } - - if newSeq.Uint() != nativeSeq { - logger.Error("newSeq != nativeSeq") - break - - } - - data := chunk.Get("data") - if !data.Exists() { - break - } - e.observeData(logger, data, nativeSeq, true) - } - case <-timer.C: s := "" @@ -299,11 +261,11 @@ func (e *Watcher) retrievePayload(s string) ([]byte, error) { return body, err } -func (e *Watcher) observeData(logger *zap.Logger, data gjson.Result, nativeSeq uint64, isReobservation bool) { +func (e *Watcher) observeData(logger *zap.Logger, data gjson.Result, nativeSeq uint64, isReobservation bool) bool { em := data.Get("sender") if !em.Exists() { logger.Error("sender field missing") - return + return false } emitter := make([]byte, 8) @@ -321,53 +283,53 @@ func (e *Watcher) observeData(logger *zap.Logger, data gjson.Result, nativeSeq u v := data.Get("payload") if !v.Exists() { logger.Error("payload field missing") - return + return false } s := v.String() if !strings.HasPrefix(s, "0x") { logger.Error("payload missing 0x prefix", zap.String("payload", s)) - return + return false } pl, err := hex.DecodeString(strings.TrimPrefix(s, "0x")) if err != nil { logger.Error("payload decode", zap.Error(err)) - return + return false } ts := data.Get("timestamp") if !ts.Exists() { logger.Error("timestamp field missing") - return + return false } nonce := data.Get("nonce") if !nonce.Exists() { logger.Error("nonce field missing") - return + return false } sequence := data.Get("sequence") if !sequence.Exists() { logger.Error("sequence field missing") - return + return false } consistencyLevel := data.Get("consistency_level") if !consistencyLevel.Exists() { logger.Error("consistencyLevel field missing") - return + return false } if nonce.Uint() > math.MaxUint32 { logger.Error("nonce is larger than expected MaxUint32") - return + return false } if consistencyLevel.Uint() > math.MaxUint8 { logger.Error("consistency level is larger than expected MaxUint8") - return + return false } observation := &common.MessagePublication{ @@ -400,6 +362,7 @@ func (e *Watcher) observeData(logger *zap.Logger, data gjson.Result, nativeSeq u ) e.msgC <- observation // Note on channel capacity: The channel to the processor is buffered and shared across chains, if it backs up we should stop processing new observations + return true } // logVersion retrieves the Aptos node version and logs it