multi: add, initialize and test mailbox RPC server

We now plug everything together and make sure that the auth mailbox
server works by adding a new integration test.

Author: guggero

cmd/commands/conn.go

@@ -20,6 +20,7 @@ import (
 	"github.com/lightninglabs/taproot-assets/taprpc"
 	"github.com/lightninglabs/taproot-assets/taprpc/assetwalletrpc"
 	wrpc "github.com/lightninglabs/taproot-assets/taprpc/assetwalletrpc"
+	"github.com/lightninglabs/taproot-assets/taprpc/authmailboxrpc"
 	"github.com/lightninglabs/taproot-assets/taprpc/mintrpc"
 	"github.com/lightninglabs/taproot-assets/taprpc/rfqrpc"
 	tchrpc "github.com/lightninglabs/taproot-assets/taprpc/tapchannelrpc"
@@ -71,6 +72,7 @@ type RpcClientsBundle interface {
 	tchrpc.TaprootAssetChannelsClient
 	universerpc.UniverseClient
 	tapdevrpc.TapDevClient
+	authmailboxrpc.MailboxClient
 }
 
 // getRpcClientBundle returns a bundle of all the RPC clients.
@@ -88,6 +90,7 @@ func getRpcClientBundle(ctx *cli.Context) (RpcClientsBundle, func()) {
 	channelsClient := tchrpc.NewTaprootAssetChannelsClient(conn)
 	universeClient := universerpc.NewUniverseClient(conn)
 	devClient := tapdevrpc.NewTapDevClient(conn)
+	authMailboxClient := authmailboxrpc.NewMailboxClient(conn)
 
 	// Use an inline struct to eliminate the need for defining an additional
 	// struct, reducing potential confusion with RpcClientsBundle.
@@ -99,6 +102,7 @@ func getRpcClientBundle(ctx *cli.Context) (RpcClientsBundle, func()) {
 		tchrpc.TaprootAssetChannelsClient
 		universerpc.UniverseClient
 		tapdevrpc.TapDevClient
+		authmailboxrpc.MailboxClient
 	}{
 		TaprootAssetsClient:        tapClient,
 		AssetWalletClient:          assetWalletClient,
@@ -107,6 +111,7 @@ func getRpcClientBundle(ctx *cli.Context) (RpcClientsBundle, func()) {
 		TaprootAssetChannelsClient: channelsClient,
 		UniverseClient:             universeClient,
 		TapDevClient:               devClient,
+		MailboxClient:              authMailboxClient,
 	}, cleanUp
 }
 

config.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/taproot-assets/address"
+	"github.com/lightninglabs/taproot-assets/authmailbox"
 	"github.com/lightninglabs/taproot-assets/monitoring"
 	"github.com/lightninglabs/taproot-assets/proof"
 	"github.com/lightninglabs/taproot-assets/rfq"
@@ -158,6 +159,8 @@ type Config struct {
 
 	SignalInterceptor signal.Interceptor
 
+	MboxServerConfig authmailbox.ServerConfig
+
 	ReOrgWatcher *tapgarden.ReOrgWatcher
 
 	AssetMinter tapgarden.Planter

itest/authmailbox_test.go

@@ -0,0 +1,171 @@
+package itest
+
+import (
+	"context"
+	"time"
+
+	"github.com/btcsuite/btcd/txscript"
+	"github.com/btcsuite/btcd/wire"
+	"github.com/lightninglabs/taproot-assets/asset"
+	"github.com/lightninglabs/taproot-assets/authmailbox"
+	"github.com/lightninglabs/taproot-assets/internal/test"
+	"github.com/lightninglabs/taproot-assets/proof"
+	"github.com/lightningnetwork/lnd/lntest/wait"
+	"github.com/stretchr/testify/require"
+)
+
+// testAuthMailboxStoreAndFetchMessage tests that we can store and fetch auth
+// mailbox messages using the auth mailbox server.
+func testAuthMailboxStoreAndFetchMessage(t *harnessTest) {
+	ctx := context.Background()
+
+	// We first need to create two on-chain outputs that we can use to
+	// send messages to the auth mailbox server.
+	internalKey := test.RandPubKey(t.t)
+	merkleRoot := test.RandBytes(32)
+
+	pkScriptBip86, err := txscript.PayToTaprootScript(
+		txscript.ComputeTaprootKeyNoScript(internalKey),
+	)
+	require.NoError(t.t, err)
+
+	pkScriptTapscript, err := txscript.PayToTaprootScript(
+		txscript.ComputeTaprootOutputKey(
+			internalKey, merkleRoot,
+		),
+	)
+	require.NoError(t.t, err)
+
+	txHash, err := t.lndHarness.Miner().SendOutputs([]*wire.TxOut{
+		{
+			Value:    100000,
+			PkScript: pkScriptBip86,
+		},
+		{
+			Value:    100000,
+			PkScript: pkScriptTapscript,
+		},
+	}, 10)
+	require.NoError(t.t, err)
+
+	blockHash := t.lndHarness.Miner().GenerateBlocks(1)[0]
+	bestBlockHash, blockHeight := t.lndHarness.Miner().GetBestBlock()
+	require.Equal(t.t, blockHash, bestBlockHash)
+
+	block := t.lndHarness.Miner().GetBlock(blockHash)
+
+	require.Len(t.t, block.Transactions, 2)
+	tx := block.Transactions[1]
+	require.Equal(t.t, tx.TxHash(), *txHash)
+
+	txMerkleProof, err := proof.NewTxMerkleProof(block.Transactions, 1)
+	require.NoError(t.t, err)
+
+	txProof1 := proof.TxProof{
+		MsgTx:       *tx,
+		BlockHeader: block.Header,
+		BlockHeight: uint32(blockHeight),
+		MerkleProof: *txMerkleProof,
+		ClaimedOutPoint: wire.OutPoint{
+			Hash:  *txHash,
+			Index: 0,
+		},
+		InternalKey: *internalKey,
+	}
+	txProof2 := proof.TxProof{
+		MsgTx:       *tx,
+		BlockHeader: block.Header,
+		BlockHeight: uint32(blockHeight),
+		MerkleProof: *txMerkleProof,
+		ClaimedOutPoint: wire.OutPoint{
+			Hash:  *txHash,
+			Index: 1,
+		},
+		InternalKey: *internalKey,
+		MerkleRoot:  merkleRoot,
+	}
+
+	// Before we store any message, we set up a listener to make sure we
+	// can receive messages.
+	lndClient, err := t.newLndClient(t.tapd.cfg.LndNode)
+	require.NoError(t.t, err)
+	receiverKey, err := lndClient.WalletKit.DeriveNextKey(
+		ctx, int32(asset.TaprootAssetsKeyFamily),
+	)
+	require.NoError(t.t, err)
+
+	mboxClient := authmailbox.NewClient(&authmailbox.ClientConfig{
+		ServerAddress: t.tapd.rpcHost(),
+		SkipTlsVerify: true,
+		Signer:        lndClient.Signer,
+		MinBackoff:    time.Second,
+		MaxBackoff:    time.Second,
+	})
+	require.NoError(t.t, mboxClient.Start())
+
+	inboundChan := make(chan *authmailbox.ReceivedMessages, 10)
+	subscription, err := mboxClient.StartAccountSubscription(
+		ctx, inboundChan, *receiverKey, authmailbox.MessageFilter{},
+	)
+	require.NoError(t.t, err)
+
+	require.Eventually(t.t, func() bool {
+		return subscription.IsSubscribed()
+	}, defaultTimeout, wait.PollInterval)
+
+	t.t.Cleanup(func() {
+		require.NoError(t.t, subscription.Stop())
+		require.NoError(t.t, mboxClient.Stop())
+	})
+
+	// Now we can start sending messages to the mailbox server.
+	id, err := mboxClient.SendMessage(
+		ctx, *receiverKey.PubKey, []byte("message 1"), txProof1, 1234,
+	)
+	require.NoError(t.t, err)
+	require.Greater(t.t, id, uint64(0))
+	id2, err := mboxClient.SendMessage(
+		ctx, *receiverKey.PubKey, []byte("message 2"), txProof2, 2345,
+	)
+	require.NoError(t.t, err)
+	require.Greater(t.t, id2, uint64(0))
+
+	// We check that we can't use the same tx proof again.
+	_, err = mboxClient.SendMessage(
+		ctx, *receiverKey.PubKey, []byte("message 3"), txProof1, 3456,
+	)
+	require.ErrorContains(t.t, err, proof.ErrTxMerkleProofExists.Error())
+
+	// We also make sure that the TX proof is properly validated.
+	txProof1.MerkleRoot = test.RandBytes(32)
+	_, err = mboxClient.SendMessage(
+		ctx, *receiverKey.PubKey, []byte("message 3"), txProof1, 3456,
+	)
+	require.ErrorContains(
+		t.t, err, "validating proof: claimed output pk script doesn't "+
+			"match constructed Taproot output key pk script",
+	)
+
+	// And now we should have two messages in the mailbox.
+	select {
+	case msgs := <-inboundChan:
+		require.Len(t.t, msgs.Messages, 1)
+		msg := msgs.Messages[0]
+		require.Equal(t.t, id, msg.MessageId)
+		require.Equal(t.t, []byte("message 1"), msg.EncryptedPayload)
+
+	case <-time.After(defaultTimeout):
+		require.Fail(t.t, "timed out waiting for message 1")
+	}
+
+	select {
+	case msgs := <-inboundChan:
+		require.Len(t.t, msgs.Messages, 1)
+		msg := msgs.Messages[0]
+		require.Equal(t.t, id2, msg.MessageId)
+		require.Equal(t.t, []byte("message 2"), msg.EncryptedPayload)
+
+	case <-time.After(defaultTimeout):
+		require.Fail(t.t, "timed out waiting for message 2")
+	}
+}

itest/tapd_harness.go

@@ -22,6 +22,7 @@ import (
 	"github.com/lightninglabs/taproot-assets/tapdb"
 	"github.com/lightninglabs/taproot-assets/taprpc"
 	"github.com/lightninglabs/taproot-assets/taprpc/assetwalletrpc"
+	"github.com/lightninglabs/taproot-assets/taprpc/authmailboxrpc"
 	"github.com/lightninglabs/taproot-assets/taprpc/mintrpc"
 	"github.com/lightninglabs/taproot-assets/taprpc/rfqrpc"
 	tchrpc "github.com/lightninglabs/taproot-assets/taprpc/tapchannelrpc"
@@ -102,6 +103,7 @@ type tapdHarness struct {
 	tchrpc.TaprootAssetChannelsClient
 	universerpc.UniverseClient
 	tapdevrpc.TapDevClient
+	authmailboxrpc.MailboxClient
 }
 
 // tapdConfig holds all configuration items that are required to start a tapd
@@ -442,6 +444,7 @@ func (hs *tapdHarness) start(expectErrExit bool) error {
 	)
 	hs.UniverseClient = universerpc.NewUniverseClient(rpcConn)
 	hs.TapDevClient = tapdevrpc.NewTapDevClient(rpcConn)
+	hs.MailboxClient = authmailboxrpc.NewMailboxClient(rpcConn)
 
 	return nil
 }

itest/test_list_on_test.go

@@ -347,6 +347,10 @@ var testCases = []*testCase{
 		name: "pre commit output",
 		test: testPreCommitOutput,
 	},
+	{
+		name: "auth mailbox message store and fetch",
+		test: testAuthMailboxStoreAndFetchMessage,
+	},
 }
 
 var optionalTestCases = []*testCase{

log.go

@@ -3,6 +3,7 @@ package taprootassets
 import (
 	"github.com/btcsuite/btclog/v2"
 	"github.com/lightninglabs/taproot-assets/address"
+	"github.com/lightninglabs/taproot-assets/authmailbox"
 	"github.com/lightninglabs/taproot-assets/commitment"
 	"github.com/lightninglabs/taproot-assets/monitoring"
 	"github.com/lightninglabs/taproot-assets/proof"
@@ -118,6 +119,9 @@ func SetupLoggers(root *build.SubLoggerManager,
 	AddSubLogger(
 		root, tapchannel.Subsystem, interceptor, tapchannel.UseLogger,
 	)
+	AddSubLogger(
+		root, authmailbox.Subsystem, interceptor, authmailbox.UseLogger,
+	)
 }
 
 // AddSubLogger is a helper method to conveniently create and register the

server.go

@@ -15,6 +15,7 @@ import (
 	proxy "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/taproot-assets/address"
+	"github.com/lightninglabs/taproot-assets/authmailbox"
 	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/monitoring"
 	"github.com/lightninglabs/taproot-assets/rfqmsg"
@@ -63,6 +64,7 @@ type Server struct {
 	cfg *Config
 
 	*rpcServer
+	mboxServer      *authmailbox.Server
 	macaroonService *lndclient.MacaroonService
 
 	quit chan struct{}
@@ -178,6 +180,8 @@ func (s *Server) initialize(interceptorChain *rpcperms.InterceptorChain) error {
 		return fmt.Errorf("unable to create rpc server: %w", err)
 	}
 
+	s.mboxServer = authmailbox.NewServer(&s.cfg.MboxServerConfig)
+
 	// First, we'll start the main batched asset minter.
 	if err := s.cfg.AssetMinter.Start(); err != nil {
 		return fmt.Errorf("unable to start asset minter: %w", err)
@@ -240,13 +244,20 @@ func (s *Server) initialize(interceptorChain *rpcperms.InterceptorChain) error {
 		return fmt.Errorf("unable to start RPC server: %w", err)
 	}
 
+	shutdownFuncs["rpcServer"] = s.rpcServer.Stop
+
+	if err := s.mboxServer.Start(); err != nil {
+		return fmt.Errorf("unable to start auth mailbox server: %w",
+			err)
+	}
+
 	// This does have no effect if starting the rpc server is the last step
 	// in this function, but its better to have it here in case we add more
 	// steps in the future.
 	//
 	// NOTE: if this is not the last step in the function, feel free to
 	// delete this comment.
-	shutdownFuncs["rpcServer"] = s.rpcServer.Stop
+	shutdownFuncs["mboxServer"] = s.mboxServer.Stop
 
 	shutdownFuncs = nil
 
@@ -360,6 +371,11 @@ func (s *Server) RunUntilShutdown(mainErrChan <-chan error) error {
 		return mkErr("error registering gRPC server: %v", err)
 	}
 
+	err = s.mboxServer.RegisterWithGrpcServer(grpcServer)
+	if err != nil {
+		return mkErr("error registering auth mailbox server: %v", err)
+	}
+
 	// All the necessary components have been registered, so we can
 	// actually start listening for requests.
 	err = startGrpcListen(s.cfg, grpcServer, grpcListeners)
@@ -371,7 +387,7 @@ func (s *Server) RunUntilShutdown(mainErrChan <-chan error) error {
 	// direct tapd to connect to its loopback address rather than a
 	// wildcard to prevent certificate issues when accessing the proxy
 	// externally.
-	stopProxy, err := startRestProxy(s.cfg, s.rpcServer)
+	stopProxy, err := startRestProxy(s.cfg, s.rpcServer, s.mboxServer)
 	if err != nil {
 		return mkErr("error starting REST proxy: %v", err)
 	}
@@ -511,7 +527,9 @@ func startGrpcListen(cfg *Config, grpcServer *grpc.Server,
 
 // startRestProxy starts the given REST proxy on the listeners found in the
 // config.
-func startRestProxy(cfg *Config, rpcServer *rpcServer) (func(), error) {
+func startRestProxy(cfg *Config, rpcServer *rpcServer,
+	mboxServer *authmailbox.Server) (func(), error) {
+
 	// We use the first RPC listener as the destination for our REST proxy.
 	// If the listener is set to listen on all interfaces, we replace it
 	// with localhost, as we cannot dial it directly.
@@ -579,6 +597,13 @@ func startRestProxy(cfg *Config, rpcServer *rpcServer) (func(), error) {
 		return nil, err
 	}
 
+	err = mboxServer.RegisterWithRestProxy(
+		ctx, mux, cfg.RestDialOpts, restProxyDest,
+	)
+	if err != nil {
+		return nil, err
+	}
+
 	// Wrap the default grpc-gateway handler with the WebSocket handler.
 	restHandler := lnrpc.NewWebSocketProxy(
 		mux, rpcsLog, cfg.WSPingInterval, cfg.WSPongWait,