feat(node): aptos reobserve with endpoint

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
+}

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))
+}

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
 }