itest: add interceptor and first hop data tests

itest/list_on_test.go

@@ -458,6 +458,14 @@ var allTestCases = []*lntest.TestCase{
 		Name:     "forward interceptor modified htlc",
 		TestFunc: testForwardInterceptorModifiedHtlc,
 	},
+	{
+		Name:     "forward interceptor wire records",
+		TestFunc: testForwardInterceptorWireRecords,
+	},
+	{
+		Name:     "forward interceptor restart",
+		TestFunc: testForwardInterceptorRestart,
+	},
 	{
 		Name:     "zero conf channel open",
 		TestFunc: testZeroConfChannelOpen,

itest/lnd_forward_interceptor_test.go

@@ -1,7 +1,9 @@
 package itest
 
 import (
+	"bytes"
 	"fmt"
+	"reflect"
 	"strings"
 	"time"
 
@@ -13,6 +15,7 @@ import (
 	"github.com/lightningnetwork/lnd/lntest/node"
 	"github.com/lightningnetwork/lnd/lntest/wait"
 	"github.com/lightningnetwork/lnd/lntypes"
+	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/routing/route"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/codes"
@@ -24,6 +27,7 @@ var (
 	customTestValue        = []byte{1, 3, 5}
 
 	actionResumeModify = routerrpc.ResolveHoldForwardAction_RESUME_MODIFIED
+	actionResume       = routerrpc.ResolveHoldForwardAction_RESUME
 )
 
 type interceptorTestCase struct {
@@ -436,33 +440,309 @@ func testForwardInterceptorModifiedHtlc(ht *lntest.HarnessTest) {
 	ht.CloseChannel(bob, cpBC)
 }
 
+// testForwardInterceptorWireRecords tests that the interceptor can read any
+// wire custom records provided by the sender of a payment as part of the
+// update_add_htlc message.
+func testForwardInterceptorWireRecords(ht *lntest.HarnessTest) {
+	// Initialize the test context with 3 connected nodes.
+	ts := newInterceptorTestScenario(ht)
+
+	alice, bob, carol, dave := ts.alice, ts.bob, ts.carol, ts.dave
+
+	// Open and wait for channels.
+	const chanAmt = btcutil.Amount(300000)
+	p := lntest.OpenChannelParams{Amt: chanAmt}
+	reqs := []*lntest.OpenChannelRequest{
+		{Local: alice, Remote: bob, Param: p},
+		{Local: bob, Remote: carol, Param: p},
+		{Local: carol, Remote: dave, Param: p},
+	}
+	resp := ht.OpenMultiChannelsAsync(reqs)
+	cpAB, cpBC, cpCD := resp[0], resp[1], resp[2]
+
+	// Make sure Alice is aware of channel Bob=>Carol.
+	ht.AssertTopologyChannelOpen(alice, cpBC)
+
+	// Connect an interceptor to Bob's node.
+	bobInterceptor, cancelBobInterceptor := bob.RPC.HtlcInterceptor()
+	defer cancelBobInterceptor()
+
+	// Also connect an interceptor on Carol's node to check whether we're
+	// relaying the TLVs send in update_add_htlc over Alice -> Bob on the
+	// Bob -> Carol link.
+	carolInterceptor, cancelCarolInterceptor := carol.RPC.HtlcInterceptor()
+	defer cancelCarolInterceptor()
+
+	req := &lnrpc.Invoice{ValueMsat: 1000}
+	addResponse := dave.RPC.AddInvoice(req)
+	invoice := dave.RPC.LookupInvoice(addResponse.RHash)
+
+	sendReq := &routerrpc.SendPaymentRequest{
+		PaymentRequest: invoice.PaymentRequest,
+		TimeoutSeconds: int32(wait.PaymentTimeout.Seconds()),
+		FeeLimitMsat:   noFeeLimitMsat,
+		FirstHopCustomRecords: map[uint64][]byte{
+			65537: []byte("test"),
+		},
+	}
+
+	_ = alice.RPC.SendPayment(sendReq)
+
+	// We start the htlc interceptor with a simple implementation that saves
+	// all intercepted packets. These packets are held to simulate a
+	// pending payment.
+	packet := ht.ReceiveHtlcInterceptor(bobInterceptor)
+
+	require.Len(ht, packet.InWireCustomRecords, 1)
+
+	val, ok := packet.InWireCustomRecords[65537]
+	require.True(ht, ok, "expected custom record")
+	require.Equal(ht, []byte("test"), val)
+
+	// TODO(guggero): Actually modify the amount once we have the invoice
+	// interceptor and can accept a lower amount.
+	newOutAmountMsat := packet.OutgoingAmountMsat
+
+	err := bobInterceptor.Send(&routerrpc.ForwardHtlcInterceptResponse{
+		IncomingCircuitKey: packet.IncomingCircuitKey,
+		OutAmountMsat:      newOutAmountMsat,
+		Action:             actionResumeModify,
+	})
+	require.NoError(ht, err, "failed to send request")
+
+	// Assert that the Alice -> Bob custom records in update_add_htlc are
+	// not propagated on the Bob -> Carol link.
+	packet = ht.ReceiveHtlcInterceptor(carolInterceptor)
+	require.Len(ht, packet.InWireCustomRecords, 0)
+
+	// Just resume the payment on Carol.
+	err = carolInterceptor.Send(&routerrpc.ForwardHtlcInterceptResponse{
+		IncomingCircuitKey: packet.IncomingCircuitKey,
+		Action:             actionResume,
+	})
+	require.NoError(ht, err, "carol interceptor response")
+
+	// Assert that the payment was successful.
+	var preimage lntypes.Preimage
+	copy(preimage[:], invoice.RPreimage)
+	ht.AssertPaymentStatus(
+		alice, preimage, lnrpc.Payment_SUCCEEDED,
+		func(p *lnrpc.Payment) error {
+			recordsEqual := reflect.DeepEqual(
+				p.FirstHopCustomRecords,
+				sendReq.FirstHopCustomRecords,
+			)
+			if !recordsEqual {
+				return fmt.Errorf("expected custom records to "+
+					"be equal, got %v expected %v",
+					p.FirstHopCustomRecords,
+					sendReq.FirstHopCustomRecords)
+			}
+
+			return nil
+		},
+	)
+
+	// Finally, close channels.
+	ht.CloseChannel(alice, cpAB)
+	ht.CloseChannel(bob, cpBC)
+	ht.CloseChannel(carol, cpCD)
+}
+
+// testForwardInterceptorRestart tests that the interceptor can read any wire
+// custom records provided by the sender of a payment as part of the
+// update_add_htlc message and that those records are persisted correctly and
+// re-sent on node restart.
+func testForwardInterceptorRestart(ht *lntest.HarnessTest) {
+	// Initialize the test context with 3 connected nodes.
+	ts := newInterceptorTestScenario(ht)
+
+	alice, bob, carol, dave := ts.alice, ts.bob, ts.carol, ts.dave
+
+	// Open and wait for channels.
+	const chanAmt = btcutil.Amount(300000)
+	p := lntest.OpenChannelParams{Amt: chanAmt}
+	reqs := []*lntest.OpenChannelRequest{
+		{Local: alice, Remote: bob, Param: p},
+		{Local: bob, Remote: carol, Param: p},
+		{Local: carol, Remote: dave, Param: p},
+	}
+	resp := ht.OpenMultiChannelsAsync(reqs)
+	cpAB, cpBC, cpCD := resp[0], resp[1], resp[2]
+
+	// Make sure Alice is aware of channels Bob=>Carol and Carol=>Dave.
+	ht.AssertTopologyChannelOpen(alice, cpBC)
+	ht.AssertTopologyChannelOpen(alice, cpCD)
+
+	// Connect an interceptor to Bob's node.
+	bobInterceptor, cancelBobInterceptor := bob.RPC.HtlcInterceptor()
+
+	// Also connect an interceptor on Carol's node to check whether we're
+	// relaying the TLVs send in update_add_htlc over Alice -> Bob on the
+	// Bob -> Carol link.
+	carolInterceptor, cancelCarolInterceptor := carol.RPC.HtlcInterceptor()
+	defer cancelCarolInterceptor()
+
+	req := &lnrpc.Invoice{ValueMsat: 50_000_000}
+	addResponse := dave.RPC.AddInvoice(req)
+	invoice := dave.RPC.LookupInvoice(addResponse.RHash)
+
+	customRecords := map[uint64][]byte{
+		65537: []byte("test"),
+	}
+
+	sendReq := &routerrpc.SendPaymentRequest{
+		PaymentRequest:        invoice.PaymentRequest,
+		TimeoutSeconds:        int32(wait.PaymentTimeout.Seconds()),
+		FeeLimitMsat:          noFeeLimitMsat,
+		FirstHopCustomRecords: customRecords,
+	}
+
+	_ = alice.RPC.SendPayment(sendReq)
+
+	// We start the htlc interceptor with a simple implementation that saves
+	// all intercepted packets. These packets are held to simulate a
+	// pending payment.
+	packet := ht.ReceiveHtlcInterceptor(bobInterceptor)
+
+	require.Len(ht, packet.InWireCustomRecords, 1)
+	require.Equal(ht, customRecords, packet.InWireCustomRecords)
+
+	// We accept the payment at Bob and resume it, so it gets to Carol.
+	// This means the HTLC should now be fully locked in on Alice's side and
+	// any restart of the node should cause the payment to be resumed and
+	// the data to be persisted across restarts.
+	err := bobInterceptor.Send(&routerrpc.ForwardHtlcInterceptResponse{
+		IncomingCircuitKey: packet.IncomingCircuitKey,
+		Action:             actionResume,
+	})
+	require.NoError(ht, err, "failed to send request")
+
+	// We don't resume the payment on Carol, so it should be held there.
+
+	// The payment should now be in flight.
+	var preimage lntypes.Preimage
+	copy(preimage[:], invoice.RPreimage)
+	ht.AssertPaymentStatus(alice, preimage, lnrpc.Payment_IN_FLIGHT)
+
+	// We don't resume the payment on Carol, so it should be held there.
+	// We now restart first Bob, then Alice, so we can make sure we've
+	// started the interceptor again on Bob before Alice resumes the
+	// payment.
+	cancelBobInterceptor()
+	restartBob := ht.SuspendNode(bob)
+	restartAlice := ht.SuspendNode(alice)
+
+	require.NoError(ht, restartBob(), "failed to restart bob")
+	bobInterceptor, cancelBobInterceptor = bob.RPC.HtlcInterceptor()
+	defer cancelBobInterceptor()
+
+	require.NoError(ht, restartAlice(), "failed to restart alice")
+
+	// We should get another notification about the held HTLC.
+	packet = ht.ReceiveHtlcInterceptor(bobInterceptor)
+
+	require.Len(ht, packet.InWireCustomRecords, 1)
+	require.Equal(ht, customRecords, packet.InWireCustomRecords)
+
+	err = carolInterceptor.Send(&routerrpc.ForwardHtlcInterceptResponse{
+		IncomingCircuitKey: packet.IncomingCircuitKey,
+		Action:             actionResume,
+	})
+	require.NoError(ht, err, "failed to send request")
+
+	// And now we forward the payment at Carol.
+	packet = ht.ReceiveHtlcInterceptor(carolInterceptor)
+	require.Len(ht, packet.InWireCustomRecords, 0)
+	err = carolInterceptor.Send(&routerrpc.ForwardHtlcInterceptResponse{
+		IncomingCircuitKey: packet.IncomingCircuitKey,
+		Action:             actionResume,
+	})
+	require.NoError(ht, err, "failed to send request")
+
+	// Assert that the payment was successful.
+	ht.AssertPaymentStatus(
+		alice, preimage, lnrpc.Payment_SUCCEEDED,
+		func(p *lnrpc.Payment) error {
+			recordsEqual := reflect.DeepEqual(
+				p.FirstHopCustomRecords,
+				sendReq.FirstHopCustomRecords,
+			)
+			if !recordsEqual {
+				return fmt.Errorf("expected custom records to "+
+					"be equal, got %v expected %v",
+					p.FirstHopCustomRecords,
+					sendReq.FirstHopCustomRecords)
+			}
+
+			if len(p.Htlcs) != 1 {
+				return fmt.Errorf("expected 1 htlc, got %d",
+					len(p.Htlcs))
+			}
+
+			htlc := p.Htlcs[0]
+			rt := htlc.Route
+			if rt.FirstHopAmountMsat != rt.TotalAmtMsat {
+				return fmt.Errorf("expected first hop amount "+
+					"to be %d, got %d", rt.TotalAmtMsat,
+					rt.FirstHopAmountMsat)
+			}
+
+			cr := lnwire.CustomRecords(p.FirstHopCustomRecords)
+			recordData, err := cr.Serialize()
+			if err != nil {
+				return err
+			}
+
+			if !bytes.Equal(rt.CustomChannelData, recordData) {
+				return fmt.Errorf("expected custom records to "+
+					"be equal, got %x expected %x",
+					rt.CustomChannelData, recordData)
+			}
+
+			return nil
+		},
+	)
+
+	// Finally, close channels.
+	ht.CloseChannel(alice, cpAB)
+	ht.CloseChannel(bob, cpBC)
+	ht.CloseChannel(carol, cpCD)
+}
+
 // interceptorTestScenario is a helper struct to hold the test context and
 // provide the needed functionality.
 type interceptorTestScenario struct {
-	ht                *lntest.HarnessTest
-	alice, bob, carol *node.HarnessNode
+	ht                      *lntest.HarnessTest
+	alice, bob, carol, dave *node.HarnessNode
 }
 
 // newInterceptorTestScenario initializes a new test scenario with three nodes
 // and connects them to have the following topology,
 //
-//	Alice --> Bob --> Carol
+//	Alice --> Bob --> Carol --> Dave
 //
 // Among them, Alice and Bob are standby nodes and Carol is a new node.
 func newInterceptorTestScenario(
 	ht *lntest.HarnessTest) *interceptorTestScenario {
 
 	alice, bob := ht.Alice, ht.Bob
 	carol := ht.NewNode("carol", nil)
+	dave := ht.NewNode("dave", nil)
 
 	ht.EnsureConnected(alice, bob)
 	ht.EnsureConnected(bob, carol)
+	ht.EnsureConnected(carol, dave)
+
+	// So that carol can open channels.
+	ht.FundCoins(btcutil.SatoshiPerBitcoin, carol)
 
 	return &interceptorTestScenario{
 		ht:    ht,
 		alice: alice,
 		bob:   bob,
 		carol: carol,
+		dave:  dave,
 	}
 }
 

lntest/harness_assertion.go

@@ -1600,13 +1600,16 @@ func (h *HarnessTest) findPayment(hn *node.HarnessNode,
 	return nil
 }
 
+// PaymentCheck is a function that checks a payment for a specific condition.
+type PaymentCheck func(*lnrpc.Payment) error
+
 // AssertPaymentStatus asserts that the given node list a payment with the
 // given preimage has the expected status. It also checks that the payment has
 // the expected preimage, which is empty when it's not settled and matches the
 // given preimage when it's succeeded.
 func (h *HarnessTest) AssertPaymentStatus(hn *node.HarnessNode,
-	preimage lntypes.Preimage,
-	status lnrpc.Payment_PaymentStatus) *lnrpc.Payment {
+	preimage lntypes.Preimage, status lnrpc.Payment_PaymentStatus,
+	checks ...PaymentCheck) *lnrpc.Payment {
 
 	var target *lnrpc.Payment
 	payHash := preimage.Hash()
@@ -1636,6 +1639,11 @@ func (h *HarnessTest) AssertPaymentStatus(hn *node.HarnessNode,
 			target.PaymentPreimage, "expected zero preimage")
 	}
 
+	// Perform any additional checks on the payment.
+	for _, check := range checks {
+		require.NoError(h, check(target))
+	}
+
 	return target
 }