invoices: add invoice HTLC interceptor

Author: ffranr

invoices_client.go

@@ -12,9 +12,55 @@ import (
 	"github.com/lightningnetwork/lnd/lnrpc"
 	"github.com/lightningnetwork/lnd/lnrpc/invoicesrpc"
 	"github.com/lightningnetwork/lnd/lntypes"
+	"github.com/lightningnetwork/lnd/lnwire"
 	"google.golang.org/grpc"
 )
 
+// InvoiceHtlcModifyRequest is a request to modify an HTLC that is attempting to
+// settle an invoice.
+type InvoiceHtlcModifyRequest struct {
+	// Invoice is the current state of the invoice, _before_ this HTLC is
+	// applied. Any HTLC in the invoice is a previously accepted/settled
+	// one.
+	Invoice *lnrpc.Invoice
+
+	// CircuitKey is the circuit key of the HTLC that is attempting to
+	// settle the invoice.
+	CircuitKey invpkg.CircuitKey
+
+	// ExitHtlcAmt is the amount of the HTLC that is attempting to settle
+	// the invoice.
+	ExitHtlcAmt lnwire.MilliSatoshi
+
+	// ExitHtlcExpiry is the expiry of the HTLC that is attempting to settle
+	// the invoice.
+	ExitHtlcExpiry uint32
+
+	// CurrentHeight is the current block height.
+	CurrentHeight uint32
+
+	// WireCustomRecords is the wire custom records of the HTLC that is
+	// attempting to settle the invoice.
+	WireCustomRecords lnwire.CustomRecords
+}
+
+// InvoiceHtlcModifyResponse is a response to an HTLC modification request.
+type InvoiceHtlcModifyResponse struct {
+	// CircuitKey is the circuit key the response is for.
+	CircuitKey invpkg.CircuitKey
+
+	// AmtPaid is the amount the HTLC contributes toward settling the
+	// invoice. This amount can be different from the on-chain amount of the
+	// HTLC in case of custom channels. To not modify the amount and use the
+	// on-chain amount, set this to 0.
+	AmtPaid lnwire.MilliSatoshi
+}
+
+// InvoiceHtlcModifyHandler is a function that handles an HTLC modification
+// request.
+type InvoiceHtlcModifyHandler func(context.Context,
+	InvoiceHtlcModifyRequest) (*InvoiceHtlcModifyResponse, error)
+
 // InvoicesClient exposes invoice functionality.
 type InvoicesClient interface {
 	SubscribeSingleInvoice(ctx context.Context, hash lntypes.Hash) (
@@ -26,6 +72,14 @@ type InvoicesClient interface {
 
 	AddHoldInvoice(ctx context.Context, in *invoicesrpc.AddInvoiceData) (
 		string, error)
+
+	// HtlcModifier is a bidirectional streaming RPC that allows a client to
+	// intercept and modify the HTLCs that attempt to settle the given
+	// invoice. The server will send HTLCs of invoices to the client and the
+	// client can modify some aspects of the HTLC in order to pass the
+	// invoice acceptance tests.
+	HtlcModifier(ctx context.Context,
+		handler InvoiceHtlcModifyHandler) error
 }
 
 // InvoiceUpdate contains a state update for an invoice.
@@ -38,6 +92,8 @@ type invoicesClient struct {
 	client     invoicesrpc.InvoicesClient
 	invoiceMac serializedMacaroon
 	timeout    time.Duration
+	quitOnce   sync.Once
+	quit       chan struct{}
 	wg         sync.WaitGroup
 }
 
@@ -48,10 +104,15 @@ func newInvoicesClient(conn grpc.ClientConnInterface,
 		client:     invoicesrpc.NewInvoicesClient(conn),
 		invoiceMac: invoiceMac,
 		timeout:    timeout,
+		quit:       make(chan struct{}),
 	}
 }
 
 func (s *invoicesClient) WaitForFinished() {
+	s.quitOnce.Do(func() {
+		close(s.quit)
+	})
+
 	s.wg.Wait()
 }
 
@@ -184,3 +245,144 @@ func fromRPCInvoiceState(state lnrpc.Invoice_InvoiceState) (
 
 	return 0, errors.New("unknown state")
 }
+
+// HtlcModifier is a bidirectional streaming RPC that allows a client to
+// intercept and modify the HTLCs that attempt to settle the given invoice. The
+// server will send HTLCs of invoices to the client and the client can modify
+// some aspects of the HTLC in order to pass the invoice acceptance tests.
+func (s *invoicesClient) HtlcModifier(ctx context.Context,
+	handler InvoiceHtlcModifyHandler) error {
+
+	// Create a child context that will be canceled when this function
+	// exits. We use this context to be able to cancel goroutines when we
+	// exit on errors, because the parent context won't be canceled in that
+	// case.
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	stream, err := s.client.HtlcModifier(
+		s.invoiceMac.WithMacaroonAuth(ctx),
+	)
+	if err != nil {
+		return err
+	}
+
+	// Create an error channel that we'll send errors on if any of our
+	// goroutines fail. We buffer by 1 so that the goroutine doesn't depend
+	// on the stream being read, and select on context cancellation and
+	// quit channel so that we do not block in the case where we exit with
+	// multiple errors.
+	errChan := make(chan error, 1)
+
+	sendErr := func(err error) {
+		select {
+		case errChan <- err:
+		case <-ctx.Done():
+		case <-s.quit:
+		}
+	}
+
+	// Start a goroutine that consumes interception requests from lnd and
+	// sends them into our requests channel for handling. The requests
+	// channel is not buffered because we expect all requests to be handled
+	// until this function exits, at which point we expect our context to
+	// be canceled or quit channel to be closed.
+	requestChan := make(chan InvoiceHtlcModifyRequest)
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+
+		for {
+			// Do a quick check whether our client context has been
+			// canceled so that we can exit sooner if needed.
+			if ctx.Err() != nil {
+				return
+			}
+
+			req, err := stream.Recv()
+			if err != nil {
+				sendErr(err)
+				return
+			}
+
+			wireCustomRecords := req.ExitHtlcWireCustomRecords
+			interceptReq := InvoiceHtlcModifyRequest{
+				Invoice: req.Invoice,
+				CircuitKey: invpkg.CircuitKey{
+					ChanID: lnwire.NewShortChanIDFromInt(
+						req.ExitHtlcCircuitKey.ChanId,
+					),
+					HtlcID: req.ExitHtlcCircuitKey.HtlcId,
+				},
+				ExitHtlcAmt: lnwire.MilliSatoshi(
+					req.ExitHtlcAmt,
+				),
+				ExitHtlcExpiry:    req.ExitHtlcExpiry,
+				CurrentHeight:     req.CurrentHeight,
+				WireCustomRecords: wireCustomRecords,
+			}
+
+			// Try to send our interception request, failing on
+			// context cancel or router exit.
+			select {
+			case requestChan <- interceptReq:
+
+			case <-s.quit:
+				sendErr(ErrRouterShuttingDown)
+				return
+
+			case <-ctx.Done():
+				sendErr(ctx.Err())
+				return
+			}
+		}
+	}()
+
+	for {
+		select {
+		case request := <-requestChan:
+			// Handle requests in a goroutine so that the handler
+			// provided to this function can be blocking. If we
+			// get an error, send it into our error channel to
+			// shut down the interceptor.
+			s.wg.Add(1)
+			go func() {
+				defer s.wg.Done()
+
+				// Get a response from handler, this may block
+				// for a while.
+				resp, err := handler(ctx, request)
+				if err != nil {
+					sendErr(err)
+					return
+				}
+
+				key := resp.CircuitKey
+				amtPaid := uint64(resp.AmtPaid)
+				rpcResp := &invoicesrpc.HtlcModifyResponse{
+					CircuitKey: &invoicesrpc.CircuitKey{
+						ChanId: key.ChanID.ToUint64(),
+						HtlcId: key.HtlcID,
+					},
+					AmtPaid: &amtPaid,
+				}
+
+				if err := stream.Send(rpcResp); err != nil {
+					sendErr(err)
+					return
+				}
+			}()
+
+		// If one of our goroutines fails, exit with the error that
+		// occurred.
+		case err := <-errChan:
+			return err
+
+		case <-s.quit:
+			return ErrRouterShuttingDown
+
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}

testdata/permissions.json

@@ -144,6 +144,14 @@
                 }
             ]
         },
+        "/invoicesrpc.Invoices/HtlcModifier": {
+            "permissions": [
+                {
+                    "entity": "invoices",
+                    "action": "write"
+                }
+            ]
+        },
         "/lnrpc.Lightning/AbandonChannel": {
             "permissions": [
                 {