Merge pull request #1412 from lightninglabs/chan-open-courier-check

tapchannel: validate proof courier before opening or accepting channels

Author: guggero

docs/examples/basic-price-oracle/go.mod

@@ -99,6 +99,7 @@ require (
 	github.com/lightninglabs/neutrino/cache v1.1.2 // indirect
 	github.com/lightningnetwork/lightning-onion v1.2.1-0.20240712235311-98bd56499dfb // indirect
 	github.com/lightningnetwork/lnd v0.19.0-beta.rc1 // indirect
+	github.com/lightningnetwork/lnd/cert v1.2.2 // indirect
 	github.com/lightningnetwork/lnd/clock v1.1.1 // indirect
 	github.com/lightningnetwork/lnd/fn/v2 v2.0.8 // indirect
 	github.com/lightningnetwork/lnd/healthcheck v1.2.6 // indirect

internal/test/grpc.go

@@ -0,0 +1,103 @@
+package test
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/lightningnetwork/lnd/cert"
+	"github.com/lightningnetwork/lnd/lntest/port"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc"
+)
+
+var (
+	// ListenAddrTemplate is the template for the address the mock server
+	// listens on.
+	ListenAddrTemplate = "localhost:%d"
+
+	// StartupWaitTime is the time we wait for the server to start up.
+	StartupWaitTime = 50 * time.Millisecond
+)
+
+// StartMockGRPCServer starts a mock gRPC server on a free port and returns the
+// address it's listening on. The caller should clean up the server by calling
+// the cleanup function.
+func StartMockGRPCServer(t *testing.T, grpcServer *grpc.Server,
+	withTLS bool) (string, func(), error) {
+
+	nextPort := port.NextAvailablePort()
+	listenAddr := fmt.Sprintf(ListenAddrTemplate, nextPort)
+
+	grpcListener, err := net.Listen("tcp", listenAddr)
+	if err != nil {
+		return "", nil, fmt.Errorf("mock RPC server unable to listen "+
+			"on %s", listenAddr)
+	}
+
+	listener := grpcListener
+	if withTLS {
+		listener = tls.NewListener(grpcListener, genCert(t))
+	}
+
+	// Create an errgroup with an associated context. If the goroutine
+	// errors, the context is closed.
+	g, ctx := errgroup.WithContext(context.Background())
+
+	// Channel to signal that the Serve goroutine has started.
+	startupSignal := make(chan struct{})
+
+	g.Go(func() error {
+		// The goroutine has started, signal the main goroutine.
+		close(startupSignal)
+
+		err := grpcServer.Serve(listener)
+		if err != nil {
+			return fmt.Errorf("mock RPC server unable to serve "+
+				"on %s: %v", listenAddr, err)
+		}
+
+		return nil
+	})
+
+	// We wait until the goroutine has started before returning the
+	// listener address.
+	<-startupSignal
+
+	// Use a timeout to check for any immediate errors.
+	select {
+	case <-ctx.Done():
+		// If the context is canceled, an error occurred during startup.
+		return "", nil, ctx.Err()
+
+	case <-time.After(StartupWaitTime):
+		// No error was reported within the startup wait time, we can
+		// assume the server is running now.
+	}
+
+	// The cleanup function stops the server and waits for the goroutine to
+	// finish.
+	cleanup := func() {
+		grpcServer.Stop()
+		_ = listener.Close()
+		_ = g.Wait()
+	}
+
+	return listenAddr, cleanup, nil
+}
+
+func genCert(t *testing.T) *tls.Config {
+	certBytes, keyBytes, err := cert.GenCertPair(
+		"tapd autogenerated cert", nil, nil, false, time.Minute,
+	)
+	require.NoError(t, err)
+
+	tlsCert, _, err := cert.LoadCertFromBytes(certBytes, keyBytes)
+	require.NoError(t, err)
+
+	return cert.TLSConfFromCert(tlsCert)
+}

proof/courier.go

@@ -1267,7 +1267,12 @@ func (c *UniverseRpcCourier) ensureConnect(ctx context.Context) error {
 	c.client = unirpc.NewUniverseClient(conn)
 	c.rawConn = conn
 
-	return nil
+	// Make sure we initiate the connection. The GetInfo RPC method is in
+	// the base macaroon white list, so it doesn't require any
+	// authentication, independent of the universe's configuration.
+	_, err = c.client.Info(ctx, &unirpc.InfoRequest{})
+
+	return err
 }
 
 // DeliverProof attempts to delivery a proof file to the receiver.
@@ -1666,3 +1671,31 @@ func FetchProofProvenance(ctx context.Context, localArchive Archiver,
 
 	return proofFile, nil
 }
+
+// CheckUniverseRpcCourierConnection checks if the universe RPC courier at the
+// given URL is reachable. It returns an error if the connection cannot be
+// established within the given timeout duration.
+func CheckUniverseRpcCourierConnection(ctx context.Context,
+	timeout time.Duration, courierURL *url.URL) error {
+
+	// We now also make a quick test connection.
+	ctxt, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+	courier, err := NewUniverseRpcCourier(
+		ctxt, &UniverseRpcCourierCfg{}, nil, nil, courierURL,
+		false,
+	)
+	if err != nil {
+		return fmt.Errorf("unable to test connection proof courier "+
+			"'%v': %v", courierURL.String(), err)
+	}
+
+	err = courier.Close()
+	if err != nil {
+		// We only log any disconnect errors, as they're not critical.
+		log.Warnf("Unable to disconnect from proof courier '%v': %v",
+			courierURL.String(), err)
+	}
+
+	return nil
+}

proof/courier_test.go

@@ -3,12 +3,18 @@ package proof
 import (
 	"bytes"
 	"context"
+	"fmt"
+	"net/url"
 	"testing"
 
 	"github.com/lightninglabs/taproot-assets/asset"
 	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/internal/test"
+	"github.com/lightninglabs/taproot-assets/taprpc/universerpc"
+	"github.com/lightningnetwork/lnd/lntest/port"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 // TestUniverseRpcCourierLocalArchiveShortCut tests that the local archive is
@@ -72,3 +78,71 @@ func TestUniverseRpcCourierLocalArchiveShortCut(t *testing.T) {
 	})
 	require.ErrorContains(t, err, "is missing outpoint")
 }
+
+// TestCheckUniverseRpcCourierConnection tests that we can connect to the
+// universe rpc courier. We also test that we fail to connect to a
+// universe rpc courier that is not listening on the given address.
+func TestCheckUniverseRpcCourierConnection(t *testing.T) {
+	serverOpts := []grpc.ServerOption{
+		grpc.Creds(insecure.NewCredentials()),
+	}
+	grpcServer := grpc.NewServer(serverOpts...)
+
+	server := MockUniverseServer{}
+	universerpc.RegisterUniverseServer(grpcServer, &server)
+
+	// We also grab a port that is free to listen on for our negative test.
+	// Since we know the port is free, and we don't listen on it, we expect
+	// the connection to fail.
+	noConnectPort := port.NextAvailablePort()
+	noConnectAddr := fmt.Sprintf(test.ListenAddrTemplate, noConnectPort)
+
+	mockServerAddr, cleanup, err := test.StartMockGRPCServer(
+		t, grpcServer, true,
+	)
+	require.NoError(t, err)
+	t.Cleanup(cleanup)
+
+	tests := []struct {
+		name        string
+		courierAddr *url.URL
+		expectErr   string
+	}{
+		{
+			name: "valid universe rpc courier",
+			courierAddr: MockCourierURL(
+				t, UniverseRpcCourierType, mockServerAddr,
+			),
+		},
+		{
+			name: "valid universe rpc courier, but can't connect",
+			courierAddr: MockCourierURL(
+				t, UniverseRpcCourierType, noConnectAddr,
+			),
+			expectErr: "unable to connect to courier service",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// We use a short timeout here, since we don't want to
+			// wait for the full default timeout of the funding
+			// controller
+			ctxt, cancel := context.WithTimeout(
+				context.Background(), test.StartupWaitTime*2,
+			)
+			defer cancel()
+
+			err := CheckUniverseRpcCourierConnection(
+				ctxt, test.StartupWaitTime, tt.courierAddr,
+			)
+			if tt.expectErr != "" {
+				require.ErrorContains(t, err, tt.expectErr)
+
+				return
+			}
+
+			require.NoError(t, err)
+		})
+	}
+}

proof/mock.go

@@ -20,6 +20,7 @@ import (
 	"github.com/lightninglabs/taproot-assets/commitment"
 	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/internal/test"
+	"github.com/lightninglabs/taproot-assets/taprpc/universerpc"
 	"github.com/lightningnetwork/lnd/keychain"
 	"github.com/lightningnetwork/lnd/lnutils"
 	"github.com/lightningnetwork/lnd/lnwire"
@@ -1046,3 +1047,27 @@ func newMockIgnoreChecker(ignoreAll bool,
 func (m *mockIgnoreChecker) IsIgnored(assetPoint AssetPoint) bool {
 	return m.ignoreAll || m.ignoredAssetPoints.Contains(assetPoint)
 }
+
+// MockUniverseServer is a mock implementation of the UniverseServer
+// interface. It implements the GetInfo RPC method, which returns an empty
+// InfoResponse.
+type MockUniverseServer struct {
+	universerpc.UnimplementedUniverseServer
+}
+
+// Info is a mock implementation of the GetInfo RPC.
+func (m *MockUniverseServer) Info(context.Context,
+	*universerpc.InfoRequest) (*universerpc.InfoResponse, error) {
+
+	return &universerpc.InfoResponse{}, nil
+}
+
+// MockCourierURL creates a new mock proof courier URL for the given protocol
+// and address.
+func MockCourierURL(t *testing.T, protocol, addr string) *url.URL {
+	urlString := fmt.Sprintf("%s://%s", protocol, addr)
+	proofCourierAddr, err := ParseCourierAddress(urlString)
+	require.NoError(t, err)
+
+	return proofCourierAddr
+}

rfq/oracle_test.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"net"
 	"testing"
 	"time"
 
@@ -22,10 +21,6 @@ import (
 )
 
 const (
-	// testServiceAddress is the address of the mock RPC price oracle
-	// service.
-	testServiceAddress = "localhost:8095"
-
 	// testAssetRate is the asset units to BTC rate used in the test cases.
 	testAssetRate uint64 = 42_000
 )
@@ -109,15 +104,15 @@ func validateAssetRatesRequest(
 
 // startBackendRPC starts the given RPC server and blocks until the server is
 // shut down.
-func startBackendRPC(grpcServer *grpc.Server) error {
+func startBackendRPC(t *testing.T, grpcServer *grpc.Server) string {
 	server := mockRpcPriceOracleServer{}
 	priceoraclerpc.RegisterPriceOracleServer(grpcServer, &server)
-	grpcListener, err := net.Listen("tcp", testServiceAddress)
-	if err != nil {
-		return fmt.Errorf("RPC server unable to listen on %s",
-			testServiceAddress)
-	}
-	return grpcServer.Serve(grpcListener)
+	mockAddr, cleanup, err := test.StartMockGRPCServer(t, grpcServer, false)
+	require.NoError(t, err)
+
+	t.Cleanup(cleanup)
+
+	return mockAddr
 }
 
 // testCaseQueryAskPrice is a test case for the RPC price oracle client
@@ -140,11 +135,7 @@ func runQueryAskPriceTest(t *testing.T, tc *testCaseQueryAskPrice) {
 		grpc.Creds(insecure.NewCredentials()),
 	}
 	backendService := grpc.NewServer(serverOpts...)
-	go func() { _ = startBackendRPC(backendService) }()
-	defer backendService.Stop()
-
-	// Wait for the server to start.
-	time.Sleep(200 * time.Millisecond)
+	testServiceAddress := startBackendRPC(t, backendService)
 
 	// Create a new RPC price oracle client and connect to the mock service.
 	serviceAddr := fmt.Sprintf("rfqrpc://%s", testServiceAddress)
@@ -252,11 +243,7 @@ func runQueryBidPriceTest(t *testing.T, tc *testCaseQueryBidPrice) {
 		grpc.Creds(insecure.NewCredentials()),
 	}
 	backendService := grpc.NewServer(serverOpts...)
-	go func() { _ = startBackendRPC(backendService) }()
-	defer backendService.Stop()
-
-	// Wait for the server to start.
-	time.Sleep(2 * time.Second)
+	testServiceAddress := startBackendRPC(t, backendService)
 
 	// Create a new RPC price oracle client and connect to the mock service.
 	serviceAddr := fmt.Sprintf("rfqrpc://%s", testServiceAddress)

tapchannel/aux_closer.go

@@ -578,7 +578,7 @@ func (a *AuxChanCloser) ShutdownBlob(
 	return lfn.Some[lnwire.CustomRecords](records), nil
 }
 
-// shipChannelTxn takes a chanenl transaction, an output commitment, and the
+// shipChannelTxn takes a channel transaction, an output commitment, and the
 // set of vPackets used to make the output commitment and ships a complete
 // pre-singed package off to the porter. This'll insert a transfer for the
 // channel, send the final transaction to the network, and update any