itest: add custom channel integration test

Co-authored-by: Olaoluwa Osuntokun <laolu32@gmail.com>
Co-authored-by: Gijs van Dam <gijs@lightning.engineering>
Co-authored-by: George Tsagkarelis <george.tsagkarelis@gmail.com>

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/btcsuite/btcd/chaincfg/chainhash v1.1.0
 	github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c
 	github.com/btcsuite/btcwallet/walletdb v1.4.4
+	github.com/davecgh/go-spew v1.1.1
 	github.com/go-errors/errors v1.0.1
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0
 	github.com/improbable-eng/grpc-web v0.12.0
@@ -35,6 +36,7 @@ require (
 	github.com/urfave/cli v1.22.9
 	go.etcd.io/bbolt v1.3.11
 	golang.org/x/crypto v0.31.0
+	golang.org/x/exp v0.0.0-20240325151524-a685a6edb6d8
 	golang.org/x/net v0.27.0
 	golang.org/x/sync v0.10.0
 	google.golang.org/grpc v1.65.0
@@ -73,7 +75,6 @@ require (
 	github.com/coreos/go-systemd v0.0.0-20191104093116-d3cd4ed1dbcf // indirect
 	github.com/coreos/go-systemd/v22 v22.3.2 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.0 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/decred/dcrd/crypto/blake256 v1.0.1 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0 // indirect
 	github.com/decred/dcrd/lru v1.1.2 // indirect
@@ -201,7 +202,6 @@ require (
 	go.uber.org/mock v0.4.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
 	go.uber.org/zap v1.23.0 // indirect
-	golang.org/x/exp v0.0.0-20240325151524-a685a6edb6d8 // indirect
 	golang.org/x/mod v0.17.0 // indirect
 	golang.org/x/sys v0.28.0 // indirect
 	golang.org/x/term v0.27.0 // indirect

itest/assertions.go

@@ -3,14 +3,18 @@ package itest
 import (
 	"context"
 	"fmt"
+	"testing"
 
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
+	"github.com/lightninglabs/taproot-assets/taprpc"
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/lnrpc"
+	"github.com/lightningnetwork/lnd/lnrpc/walletrpc"
 	"github.com/lightningnetwork/lnd/lntest"
 	"github.com/lightningnetwork/lnd/lntest/wait"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
 )
 
 // shutdownAndAssert shuts down the given node and asserts that no errors
@@ -172,24 +176,25 @@ func assertChannelClosed(ctx context.Context, t *harnessTest,
 	// block.
 	block := mineBlocks(t, net, 1, 1)[0]
 
-	closingTxid, err := net.WaitForChannelClose(closeUpdates)
+	closingUpdate, err := net.WaitForChannelClose(closeUpdates)
 	require.NoError(t.t, err, "error while waiting for channel close")
 
+	closingTxid, err := chainhash.NewHash(closingUpdate.ClosingTxid)
+	require.NoError(t.t, err)
 	assertTxInBlock(t, block, closingTxid)
 
 	// Finally, the transaction should no longer be in the waiting close
 	// state as we've just mined a block that should include the closing
 	// transaction.
 	err = wait.Predicate(func() bool {
-		pendingChansRequest := &lnrpc.PendingChannelsRequest{}
-		pendingChanResp, err := node.PendingChannels(
-			ctx, pendingChansRequest,
+		resp, err := node.PendingChannels(
+			ctx, &lnrpc.PendingChannelsRequest{},
 		)
 		if err != nil {
 			return false
 		}
 
-		for _, pendingClose := range pendingChanResp.WaitingCloseChannels {
+		for _, pendingClose := range resp.WaitingCloseChannels {
 			if pendingClose.Channel.ChannelPoint == chanPointStr {
 				return false
 			}
@@ -203,3 +208,34 @@ func assertChannelClosed(ctx context.Context, t *harnessTest,
 
 	return closingTxid
 }
+
+func assertSweepExists(t *testing.T, node *HarnessNode,
+	witnessType walletrpc.WitnessType) {
+
+	ctxb := context.Background()
+	err := wait.NoError(func() error {
+		pendingSweeps, err := node.WalletKitClient.PendingSweeps(
+			ctxb, &walletrpc.PendingSweepsRequest{},
+		)
+		if err != nil {
+			return err
+		}
+
+		for _, sweep := range pendingSweeps.PendingSweeps {
+			if sweep.WitnessType == witnessType {
+				return nil
+			}
+		}
+
+		return fmt.Errorf("failed to find second level sweep: %v",
+			toProtoJSON(t, pendingSweeps))
+	}, defaultTimeout)
+	require.NoError(t, err)
+}
+
+func toProtoJSON(t *testing.T, resp proto.Message) string {
+	jsonBytes, err := taprpc.ProtoJSONMarshalOpts.Marshal(resp)
+	require.NoError(t, err)
+
+	return string(jsonBytes)
+}

itest/litd_accounts_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/lightninglabs/lightning-terminal/litrpc"
+	"github.com/lightninglabs/taproot-assets/taprpc/tapchannelrpc"
 	"github.com/lightningnetwork/lnd/lnrpc"
 	"github.com/lightningnetwork/lnd/lnrpc/routerrpc"
 	"github.com/lightningnetwork/lnd/lntest"
@@ -433,3 +434,44 @@ func getPaymentResult(stream routerrpc.Router_SendPaymentV2Client) (
 		}
 	}
 }
+
+func getAssetPaymentResult(
+	s tapchannelrpc.TaprootAssetChannels_SendPaymentClient,
+	isHodl bool) (*lnrpc.Payment, error) {
+
+	// No idea why it makes a difference whether we wait before calling
+	// s.Recv() or not, but it does. Without the sleep, the test will fail
+	// with "insufficient local balance"... ¯\_(ツ)_/¯
+	// Probably something weird within lnd itself.
+	time.Sleep(time.Second)
+
+	for {
+		msg, err := s.Recv()
+		if err != nil {
+			return nil, err
+		}
+
+		// Ignore RFQ quote acceptance messages read from the send
+		// payment stream, as they are not relevant.
+		quote := msg.GetAcceptedSellOrder()
+		if quote != nil {
+			continue
+		}
+
+		payment := msg.GetPaymentResult()
+		if payment == nil {
+			return nil, fmt.Errorf("unexpected message: %v", msg)
+		}
+
+		// If this is a hodl payment, then we'll return the first
+		// expected response. Otherwise, we'll wait until the in flight
+		// clears to we can observe the other payment states.
+		switch {
+		case isHodl:
+			return payment, nil
+
+		case payment.Status != lnrpc.Payment_IN_FLIGHT:
+			return payment, nil
+		}
+	}
+}

itest/litd_node.go

@@ -89,6 +89,9 @@ type LitNodeConfig struct {
 
 	LitPort     int
 	LitRESTPort int
+
+	// backupDBDir is the path where a database backup is stored, if any.
+	backupDBDir string
 }
 
 func (cfg *LitNodeConfig) LitAddr() string {
@@ -2087,3 +2090,38 @@ func connectLitRPC(ctx context.Context, hostPort, tlsCertPath,
 
 	return grpc.DialContext(ctx, hostPort, opts...)
 }
+
+// copyAll copies all files and directories from srcDir to dstDir recursively.
+// Note that this function does not support links.
+func copyAll(dstDir, srcDir string) error {
+	entries, err := os.ReadDir(srcDir)
+	if err != nil {
+		return err
+	}
+
+	for _, entry := range entries {
+		srcPath := filepath.Join(srcDir, entry.Name())
+		dstPath := filepath.Join(dstDir, entry.Name())
+
+		info, err := os.Stat(srcPath)
+		if err != nil {
+			return err
+		}
+
+		if info.IsDir() {
+			err := os.Mkdir(dstPath, info.Mode())
+			if err != nil && !os.IsExist(err) {
+				return err
+			}
+
+			err = copyAll(dstPath, srcPath)
+			if err != nil {
+				return err
+			}
+		} else if err := CopyFile(dstPath, srcPath); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

itest/litd_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/btcsuite/btclog"
 	"github.com/lightningnetwork/lnd/build"
 	"github.com/lightningnetwork/lnd/lntest"
 	"github.com/lightningnetwork/lnd/signal"
@@ -56,7 +57,8 @@ func TestLightningTerminal(t *testing.T) {
 
 			// Start a chain backend.
 			chainBackend, _, err := lntest.NewBackend(
-				lndHarness.Miner().P2PAddress(), harnessNetParams,
+				lndHarness.Miner().P2PAddress(),
+				harnessNetParams,
 			)
 			require.NoError(t1, err, "new backend")
 
@@ -130,6 +132,10 @@ func (h *harnessTest) setupLogging() {
 	require.NoError(h.t, err)
 	interceptor = &ic
 
+	UseLogger(build.NewSubLogger(Subsystem, func(tag string) btclog.Logger {
+		return logWriter.GenSubLogger(tag, func() {})
+	}))
+
 	err = build.ParseAndSetDebugLevels("debug", logWriter)
 	require.NoError(h.t, err)
 }

itest/litd_test_list_on_test.go

@@ -24,4 +24,56 @@ var allTestCases = []*testCase{
 		name: "test large http header",
 		test: testLargeHttpHeader,
 	},
+	{
+		name: "test custom channels",
+		test: testCustomChannels,
+	},
+	{
+		name: "test custom channels large",
+		test: testCustomChannelsLarge,
+	},
+	{
+		name: "test custom channels grouped asset",
+		test: testCustomChannelsGroupedAsset,
+	},
+	{
+		name: "test custom channels force close",
+		test: testCustomChannelsForceClose,
+	},
+	{
+		name: "test custom channels breach",
+		test: testCustomChannelsBreach,
+	},
+	{
+		name: "test custom channels liquidity",
+		test: testCustomChannelsLiquidityEdgeCases,
+	},
+	{
+		name: "test custom channels htlc force close",
+		test: testCustomChannelsHtlcForceClose,
+	},
+	{
+		name: "test custom channels balance consistency",
+		test: testCustomChannelsBalanceConsistency,
+	},
+	{
+		name: "test custom channels single asset multi input",
+		test: testCustomChannelsSingleAssetMultiInput,
+	},
+	{
+		name: "test custom channels oracle pricing",
+		test: testCustomChannelsOraclePricing,
+	},
+	{
+		name: "test custom channels fee",
+		test: testCustomChannelsFee,
+	},
+	{
+		name: "test custom channels forward bandwidth",
+		test: testCustomChannelsForwardBandwidth,
+	},
+	{
+		name: "test custom channels decode payreq",
+		test: testCustomChannelsDecodeAssetInvoice,
+	},
 }

itest/log.go

@@ -0,0 +1,24 @@
+package itest
+
+import (
+	"github.com/btcsuite/btclog"
+	"github.com/lightningnetwork/lnd/build"
+)
+
+const Subsystem = "ITST"
+
+// log is a logger that is initialized with no output filters. This means the
+// package will not perform any logging by default until the caller requests it.
+var log btclog.Logger
+
+// The default amount of logging is none.
+func init() {
+	UseLogger(build.NewSubLogger(Subsystem, nil))
+}
+
+// UseLogger uses a specified Logger to output package logging info.
+// This should be used in preference to SetLogWriter if the caller is also
+// using btclog.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}

itest/network_harness.go

@@ -46,7 +46,8 @@ type NetworkHarness struct {
 
 	// Miner is a reference to a running full node that can be used to create
 	// new blocks on the network.
-	Miner      *miner.HarnessMiner
+	Miner *miner.HarnessMiner
+
 	LNDHarness *lntest.HarnessTest
 
 	// server is an instance of the local Loop/Pool mock server.
@@ -435,6 +436,12 @@ tryconnect:
 					"finish syncing")
 			}
 		}
+
+		// Ignore "already connected to peer" errors.
+		if strings.Contains(err.Error(), "already connected to peer") {
+			return nil
+		}
+
 		return err
 	}
 
@@ -767,6 +774,58 @@ func (n *NetworkHarness) StopNode(node *HarnessNode) error {
 	return node.Stop()
 }
 
+// StopAndBackupDB backs up the database of the target node.
+func (n *NetworkHarness) StopAndBackupDB(node *HarnessNode) error {
+	restart, err := n.SuspendNode(node)
+	if err != nil {
+		return err
+	}
+
+	// Backup files.
+	tempDir, err := os.MkdirTemp("", "past-state")
+	if err != nil {
+		return fmt.Errorf("unable to create temp db folder: %w",
+			err)
+	}
+
+	if err := copyAll(tempDir, node.Cfg.DBDir()); err != nil {
+		return fmt.Errorf("unable to copy database files: %w",
+			err)
+	}
+
+	node.Cfg.backupDBDir = tempDir
+
+	return restart()
+}
+
+// StopAndRestoreDB stops the target node, restores the database from a backup
+// and starts the node again.
+func (n *NetworkHarness) StopAndRestoreDB(node *HarnessNode) error {
+	restart, err := n.SuspendNode(node)
+	if err != nil {
+		return err
+	}
+
+	// Restore files.
+	if node.Cfg.backupDBDir == "" {
+		return fmt.Errorf("no database backup created")
+	}
+
+	err = copyAll(node.Cfg.DBDir(), node.Cfg.backupDBDir)
+	if err != nil {
+		return fmt.Errorf("unable to copy database files: %w",
+			err)
+	}
+
+	if err := os.RemoveAll(node.Cfg.backupDBDir); err != nil {
+		return fmt.Errorf("unable to remove backup dir: %w",
+			err)
+	}
+	node.Cfg.backupDBDir = ""
+
+	return restart()
+}
+
 // OpenChannel attempts to open a channel between srcNode and destNode with the
 // passed channel funding parameters. If the passed context has a timeout, then
 // if the timeout is reached before the channel pending notification is
@@ -1053,8 +1112,11 @@ func (n *NetworkHarness) CloseChannel(lnNode *HarnessNode,
 		closeReq := &lnrpc.CloseChannelRequest{
 			ChannelPoint: cp,
 			Force:        force,
-			SatPerVbyte:  5,
 		}
+		if !force {
+			closeReq.SatPerVbyte = 5
+		}
+
 		closeRespStream, err = lnNode.CloseChannel(ctx, closeReq)
 		if err != nil {
 			return fmt.Errorf("unable to close channel: %v", err)
@@ -1097,7 +1159,8 @@ func (n *NetworkHarness) CloseChannel(lnNode *HarnessNode,
 // passed context has a timeout, then if the timeout is reached before the
 // notification is received then an error is returned.
 func (n *NetworkHarness) WaitForChannelClose(
-	closeChanStream lnrpc.Lightning_CloseChannelClient) (*chainhash.Hash, error) {
+	stream lnrpc.Lightning_CloseChannelClient) (*lnrpc.ChannelCloseUpdate,
+	error) {
 
 	ctxb := context.Background()
 	ctx, cancel := context.WithTimeout(ctxb, wait.ChannelCloseTimeout)
@@ -1106,13 +1169,14 @@ func (n *NetworkHarness) WaitForChannelClose(
 	errChan := make(chan error)
 	updateChan := make(chan *lnrpc.CloseStatusUpdate_ChanClose)
 	go func() {
-		closeResp, err := closeChanStream.Recv()
+		closeResp, err := stream.Recv()
 		if err != nil {
 			errChan <- err
 			return
 		}
 
-		closeFin, ok := closeResp.Update.(*lnrpc.CloseStatusUpdate_ChanClose)
+		update := closeResp.Update
+		closeFin, ok := update.(*lnrpc.CloseStatusUpdate_ChanClose)
 		if !ok {
 			errChan <- fmt.Errorf("expected channel close update, "+
 				"instead got %v", closeFin)
@@ -1130,7 +1194,7 @@ func (n *NetworkHarness) WaitForChannelClose(
 	case err := <-errChan:
 		return nil, err
 	case update := <-updateChan:
-		return chainhash.NewHash(update.ChanClose.ClosingTxid)
+		return update.ChanClose, nil
 	}
 }
 
@@ -1177,6 +1241,33 @@ func (n *NetworkHarness) AssertChannelExists(node *HarnessNode,
 	}, lntest.DefaultTimeout)
 }
 
+// AssertNodeKnown makes sure the given node knows about the target node in the
+// network graph.
+func (n *NetworkHarness) AssertNodeKnown(node, target *HarnessNode) error {
+	ctxb := context.Background()
+	ctxt, cancel := context.WithTimeout(ctxb, wait.DefaultTimeout)
+	defer cancel()
+
+	req := &lnrpc.NodeInfoRequest{
+		PubKey: hex.EncodeToString(
+			target.PubKey[:],
+		),
+	}
+	return wait.NoError(func() error {
+		info, err := node.GetNodeInfo(ctxt, req)
+		if err != nil {
+			return err
+		}
+
+		if info.Node == nil {
+			return fmt.Errorf("node %x has no info about %x",
+				node.PubKey[:], target.PubKey[:])
+		}
+
+		return nil
+	}, lntest.DefaultTimeout)
+}
+
 // DumpLogs reads the current logs generated by the passed node, and returns
 // the logs as a single string. This function is useful for examining the logs
 // of a particular node in the case of a test failure.

itest/oracle_test.go

@@ -0,0 +1,279 @@
+package itest
+
+import (
+	"context"
+	"crypto/tls"
+	"encoding/hex"
+	"fmt"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/lightninglabs/taproot-assets/asset"
+	"github.com/lightninglabs/taproot-assets/rfqmath"
+	"github.com/lightninglabs/taproot-assets/rfqmsg"
+	oraclerpc "github.com/lightninglabs/taproot-assets/taprpc/priceoraclerpc"
+	"github.com/lightningnetwork/lnd/cert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+)
+
+// oracleHarness is a basic integration test RPC price oracle server harness.
+type oracleHarness struct {
+	oraclerpc.UnimplementedPriceOracleServer
+
+	listenAddr string
+
+	grpcListener net.Listener
+	grpcServer   *grpc.Server
+
+	purchasePrices map[asset.ID]rfqmath.BigIntFixedPoint
+	salePrices     map[asset.ID]rfqmath.BigIntFixedPoint
+}
+
+func newOracleHarness(listenAddr string) *oracleHarness {
+	return &oracleHarness{
+		listenAddr:     listenAddr,
+		purchasePrices: make(map[asset.ID]rfqmath.BigIntFixedPoint),
+		salePrices:     make(map[asset.ID]rfqmath.BigIntFixedPoint),
+	}
+}
+
+func (o *oracleHarness) setPrice(assetID asset.ID, purchasePrice,
+	salePrice rfqmath.BigIntFixedPoint) {
+
+	o.purchasePrices[assetID] = purchasePrice
+	o.salePrices[assetID] = salePrice
+}
+
+func (o *oracleHarness) start(t *testing.T) {
+	// Start the mock RPC price oracle service.
+	//
+	// Generate self-signed certificate. This allows us to use TLS for the
+	// gRPC server.
+	tlsCert, err := generateSelfSignedCert()
+	require.NoError(t, err)
+
+	// Create the gRPC server with TLS
+	transportCredentials := credentials.NewTLS(&tls.Config{
+		Certificates: []tls.Certificate{tlsCert},
+	})
+	o.grpcServer = grpc.NewServer(grpc.Creds(transportCredentials))
+
+	serviceAddr := fmt.Sprintf("rfqrpc://%s", o.listenAddr)
+	log.Infof("Starting RPC price oracle service at address: %s\n",
+		serviceAddr)
+
+	oraclerpc.RegisterPriceOracleServer(o.grpcServer, o)
+
+	go func() {
+		var err error
+		o.grpcListener, err = net.Listen("tcp", o.listenAddr)
+		if err != nil {
+			log.Errorf("Error oracle listening: %v", err)
+			return
+		}
+		if err := o.grpcServer.Serve(o.grpcListener); err != nil {
+			log.Errorf("Error oracle serving: %v", err)
+		}
+	}()
+}
+
+func (o *oracleHarness) stop() {
+	if o.grpcServer != nil {
+		o.grpcServer.Stop()
+	}
+	if o.grpcListener != nil {
+		_ = o.grpcListener.Close()
+	}
+}
+
+// getAssetRates returns the asset rates for a given transaction type and
+// subject asset max amount.
+func (o *oracleHarness) getAssetRates(id asset.ID,
+	transactionType oraclerpc.TransactionType) (oraclerpc.AssetRates,
+	error) {
+
+	// Determine the rate based on the transaction type.
+	var subjectAssetRate rfqmath.BigIntFixedPoint
+	if transactionType == oraclerpc.TransactionType_PURCHASE {
+		rate, ok := o.purchasePrices[id]
+		if !ok {
+			return oraclerpc.AssetRates{}, fmt.Errorf("purchase "+
+				"price not found for asset ID=%v", id)
+		}
+		subjectAssetRate = rate
+	} else {
+		rate, ok := o.salePrices[id]
+		if !ok {
+			return oraclerpc.AssetRates{}, fmt.Errorf("sale "+
+				"price not found for asset ID=%v", id)
+		}
+		subjectAssetRate = rate
+	}
+
+	// Marshal subject asset rate to RPC format.
+	rpcSubjectAssetToBtcRate, err := oraclerpc.MarshalBigIntFixedPoint(
+		subjectAssetRate,
+	)
+	if err != nil {
+		return oraclerpc.AssetRates{}, err
+	}
+
+	// Marshal payment asset rate to RPC format.
+	rpcPaymentAssetToBtcRate, err := oraclerpc.MarshalBigIntFixedPoint(
+		rfqmsg.MilliSatPerBtc,
+	)
+	if err != nil {
+		return oraclerpc.AssetRates{}, err
+	}
+
+	expiry := time.Now().Add(5 * time.Minute).Unix()
+	return oraclerpc.AssetRates{
+		SubjectAssetRate: rpcSubjectAssetToBtcRate,
+		PaymentAssetRate: rpcPaymentAssetToBtcRate,
+		ExpiryTimestamp:  uint64(expiry),
+	}, nil
+}
+
+// QueryAssetRates queries the asset rates for a given transaction type, subject
+// asset, and payment asset. An asset rate is the number of asset units per
+// BTC.
+//
+// Example use case:
+//
+// Alice is trying to pay an invoice by spending an asset. Alice therefore
+// requests that Bob (her asset channel counterparty) purchase the asset from
+// her. Bob's payment, in BTC, will pay the invoice.
+//
+// Alice requests a bid quote from Bob. Her request includes an asset rates hint
+// (ask). Alice obtains the asset rates hint by calling this endpoint. She sets:
+// - `SubjectAsset` to the asset she is trying to sell.
+// - `SubjectAssetMaxAmount` to the max channel asset outbound.
+// - `PaymentAsset` to BTC.
+// - `TransactionType` to SALE.
+// - `AssetRateHint` to nil.
+//
+// Bob calls this endpoint to get the bid quote asset rates that he will send as
+// a response to Alice's request. He sets:
+// - `SubjectAsset` to the asset that Alice is trying to sell.
+// - `SubjectAssetMaxAmount` to the value given in Alice's quote request.
+// - `PaymentAsset` to BTC.
+// - `TransactionType` to PURCHASE.
+// - `AssetRateHint` to the value given in Alice's quote request.
+func (o *oracleHarness) QueryAssetRates(_ context.Context,
+	req *oraclerpc.QueryAssetRatesRequest) (
+	*oraclerpc.QueryAssetRatesResponse, error) {
+
+	// Ensure that the payment asset is BTC. We only support BTC as the
+	// payment asset in this example.
+	if !oraclerpc.IsAssetBtc(req.PaymentAsset) {
+		log.Infof("Payment asset is not BTC: %v", req.PaymentAsset)
+
+		return &oraclerpc.QueryAssetRatesResponse{
+			Result: &oraclerpc.QueryAssetRatesResponse_Error{
+				Error: &oraclerpc.QueryAssetRatesErrResponse{
+					Message: "unsupported payment asset, " +
+						"only BTC is supported",
+				},
+			},
+		}, nil
+	}
+
+	// Ensure that the subject asset is set correctly.
+	subjectAssetID, err := parseSubjectAsset(req.SubjectAsset)
+	if err != nil {
+		log.Errorf("Error parsing subject asset: %v", err)
+		return nil, fmt.Errorf("error parsing subject asset: %w", err)
+	}
+
+	_, hasPurchase := o.purchasePrices[subjectAssetID]
+	_, hasSale := o.salePrices[subjectAssetID]
+
+	log.Infof("Have for asset=%x, purchase=%v, sale=%v", subjectAssetID[:],
+		hasPurchase, hasSale)
+
+	// Ensure that the subject asset is supported.
+	if !hasPurchase || !hasSale {
+		log.Infof("Unsupported subject asset ID str: %v\n",
+			req.SubjectAsset)
+
+		return &oraclerpc.QueryAssetRatesResponse{
+			Result: &oraclerpc.QueryAssetRatesResponse_Error{
+				Error: &oraclerpc.QueryAssetRatesErrResponse{
+					Message: "unsupported subject asset",
+				},
+			},
+		}, nil
+	}
+
+	assetRates, err := o.getAssetRates(subjectAssetID, req.TransactionType)
+	if err != nil {
+		return nil, err
+	}
+
+	log.Infof("QueryAssetRates returning rates (subject_asset_rate=%v, "+
+		"payment_asset_rate=%v)", assetRates.SubjectAssetRate,
+		assetRates.PaymentAssetRate)
+
+	return &oraclerpc.QueryAssetRatesResponse{
+		Result: &oraclerpc.QueryAssetRatesResponse_Ok{
+			Ok: &oraclerpc.QueryAssetRatesOkResponse{
+				AssetRates: &assetRates,
+			},
+		},
+	}, nil
+}
+
+// parseSubjectAsset parses the subject asset from the given asset specifier.
+func parseSubjectAsset(subjectAsset *oraclerpc.AssetSpecifier) (asset.ID,
+	error) {
+
+	// Ensure that the subject asset is set.
+	if subjectAsset == nil {
+		return asset.ID{}, fmt.Errorf("subject asset is not set (nil)")
+	}
+
+	// Check the subject asset bytes if set.
+	var subjectAssetID asset.ID
+	switch {
+	case len(subjectAsset.GetAssetId()) > 0:
+		copy(subjectAssetID[:], subjectAsset.GetAssetId())
+
+	case len(subjectAsset.GetAssetIdStr()) > 0:
+		assetIDBytes, err := hex.DecodeString(
+			subjectAsset.GetAssetIdStr(),
+		)
+		if err != nil {
+			return asset.ID{}, fmt.Errorf("error decoding asset "+
+				"ID hex string: %w", err)
+		}
+
+		copy(subjectAssetID[:], assetIDBytes)
+
+	default:
+		return asset.ID{}, fmt.Errorf("subject asset ID bytes and ID " +
+			"str not set")
+	}
+
+	return subjectAssetID, nil
+}
+
+// generateSelfSignedCert generates a self-signed TLS certificate and private
+// key.
+func generateSelfSignedCert() (tls.Certificate, error) {
+	certBytes, keyBytes, err := cert.GenCertPair(
+		"itest price oracle", nil, nil, false, 24*time.Hour,
+	)
+	if err != nil {
+		return tls.Certificate{}, err
+	}
+
+	tlsCert, err := tls.X509KeyPair(certBytes, keyBytes)
+	if err != nil {
+		return tls.Certificate{}, err
+	}
+
+	return tlsCert, nil
+}