diff --git a/docs/release-notes/release-notes-next.md b/docs/release-notes/release-notes-next.md index 34fd3d750..14c6ced8c 100644 --- a/docs/release-notes/release-notes-next.md +++ b/docs/release-notes/release-notes-next.md @@ -13,6 +13,11 @@ #### Bug Fixes +* Static Address loop-in sweep signing requests and their authenticated + responses are now processed concurrently, preventing slow connections from + delaying later responses until the server times them out. Sweep work is + bounded, and server-provided prevouts are validated before signing. + * Instant Out now attempts to cancel server-side swaps when client initialization fails, allowing locked reservations to be released without waiting for the server timeout. diff --git a/staticaddr/loopin/actions_test.go b/staticaddr/loopin/actions_test.go index afc0085d8..2b3cf1d23 100644 --- a/staticaddr/loopin/actions_test.go +++ b/staticaddr/loopin/actions_test.go @@ -3324,7 +3324,12 @@ func TestUnlockDepositsActionReportsTransitionError(t *testing.T) { // mockAddressManager is a minimal AddressManager implementation used by the // test FSM setup. type mockAddressManager struct { + // params contains the address parameters returned to the test. params *script.Parameters + + // staticAddress contains the derived static address returned to the + // test. + staticAddress *script.StaticAddress } // GetStaticAddressParameters returns the configured address parameters. @@ -3334,11 +3339,11 @@ func (m *mockAddressManager) GetStaticAddressParameters(_ context.Context) ( return m.params, nil } -// GetStaticAddress is unused for this test and returns nil. +// GetStaticAddress returns the configured derived static address. func (m *mockAddressManager) GetStaticAddress(_ context.Context) ( *script.StaticAddress, error) { - return nil, nil + return m.staticAddress, nil } // noopDepositManager is a stub DepositManager used to satisfy FSM config. diff --git a/staticaddr/loopin/manager.go b/staticaddr/loopin/manager.go index b011f33fd..0b85d9bd9 100644 --- a/staticaddr/loopin/manager.go +++ b/staticaddr/loopin/manager.go @@ -3,10 +3,12 @@ package loopin import ( "bytes" "context" + "errors" "fmt" "math" "slices" "sort" + "sync" "sync/atomic" "time" @@ -34,6 +36,11 @@ const ( // SwapNotFinishedMsg is the message that is sent to the server if a // swap is not considered finished yet. SwapNotFinishedMsg = "swap not finished yet" + + // maxConcurrentSweepRequests is a generous protocol-sized worker limit + // that accommodates the server's fee variants without allowing an + // unbounded number of signing flows to consume client resources. + maxConcurrentSweepRequests = 128 ) var ( @@ -144,6 +151,9 @@ type Manager struct { // currentHeight stores the currently best known block height. currentHeight atomic.Uint32 + + // sweepRequestConcurrency is the number of sweepless request workers. + sweepRequestConcurrency int } // NewManager creates a new deposit withdrawal manager. @@ -154,9 +164,10 @@ func NewManager(cfg *Config, currentHeight uint32) (*Manager, error) { } m := &Manager{ - cfg: cfg, - newLoopInChan: make(chan *newSwapRequest), - exitChan: make(chan struct{}), + cfg: cfg, + newLoopInChan: make(chan *newSwapRequest), + exitChan: make(chan struct{}), + sweepRequestConcurrency: maxConcurrentSweepRequests, } m.currentHeight.Store(currentHeight) @@ -187,6 +198,26 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error { sweepReqs := m.cfg.NotificationManager. SubscribeStaticLoopInSweepRequests(ctx) + // A fixed worker pool gives the notification stream bounded + // backpressure while still processing the server's fee variants + // concurrently. + sweepCtx, cancelSweepHandlers := context.WithCancel(ctx) + var sweepHandlers sync.WaitGroup + sweepRequestsClosed := make(chan struct{}) + var sweepRequestsCloseOnce sync.Once + for range m.sweepRequestConcurrency { + sweepHandlers.Go(func() { + m.runSweepRequestWorker( + sweepCtx, sweepReqs, sweepRequestsClosed, + &sweepRequestsCloseOnce, + ) + }) + } + defer func() { + cancelSweepHandlers() + sweepHandlers.Wait() + }() + // Communicate to the caller that the address manager has completed its // initialization. close(initChan) @@ -226,26 +257,47 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error { return ctx.Err() } + case <-sweepRequestsClosed: + // The channel has been closed, we'll stop the loop-in + // manager. + log.Debugf("Stopping loop-in manager " + + "(ntfnChan closed)") + + close(m.exitChan) + + return fmt.Errorf("ntfnChan closed") + + case <-ctx.Done(): + return ctx.Err() + } + } +} + +// runSweepRequestWorker processes notifications until the subscription or +// manager context closes. +func (m *Manager) runSweepRequestWorker(ctx context.Context, + sweepReqs <-chan *swapserverrpc.ServerStaticLoopInSweepNotification, + sweepRequestsClosed chan struct{}, closeOnce *sync.Once) { + + for { + select { case sweepReq, ok := <-sweepReqs: if !ok { - // The channel has been closed, we'll stop the - // loop-in manager. - log.Debugf("Stopping loop-in manager " + - "(ntfnChan closed)") + closeOnce.Do(func() { + close(sweepRequestsClosed) + }) - close(m.exitChan) - - return fmt.Errorf("ntfnChan closed") + return } - err = m.handleLoopInSweepReq(ctx, sweepReq) + err := m.handleLoopInSweepReq(ctx, sweepReq) if err != nil { log.Errorf("Error handling loop-in sweep "+ "request: %v", err) } case <-ctx.Done(): - return ctx.Err() + return } } } @@ -271,27 +323,39 @@ func (m *Manager) notifyNotFinished(ctx context.Context, swapHash lntypes.Hash, func (m *Manager) handleLoopInSweepReq(ctx context.Context, req *swapserverrpc.ServerStaticLoopInSweepNotification) error { - // First we'll check if the loop-ins are known to us and in - // the expected state. + if req == nil { + return errors.New("sweep request is nil") + } + swapHash, err := lntypes.MakeHash(req.SwapHash) if err != nil { return err } + reader := bytes.NewReader(req.SweepTxPsbt) + sweepPacket, err := psbt.NewFromRawBytes(reader, false) + if err != nil { + return err + } + if sweepPacket.UnsignedTx == nil { + return errors.New("sweep PSBT has no unsigned transaction") + } + + sweepTx := sweepPacket.UnsignedTx + // Fetch the loop-in from the store. loopIn, err := m.cfg.Store.GetLoopInByHash(ctx, swapHash) if err != nil { return err } - loopIn.AddressParams, err = - m.cfg.AddressManager.GetStaticAddressParameters(ctx) - + addressParams, err := m.cfg.AddressManager. + GetStaticAddressParameters(ctx) if err != nil { return err } - loopIn.Address, err = m.cfg.AddressManager.GetStaticAddress(ctx) + staticAddress, err := m.cfg.AddressManager.GetStaticAddress(ctx) if err != nil { return err } @@ -303,15 +367,6 @@ func (m *Manager) handleLoopInSweepReq(ctx context.Context, if err != nil { return err } - loopIn.Deposits = deposits - - reader := bytes.NewReader(req.SweepTxPsbt) - sweepPacket, err := psbt.NewFromRawBytes(reader, false) - if err != nil { - return err - } - - sweepTx := sweepPacket.UnsignedTx // If the loop-in is not in the Succeeded state we return an // error. @@ -323,17 +378,10 @@ func (m *Manager) handleLoopInSweepReq(ctx context.Context, swapHash) } - // Perform a sanity check on the number of unsigned tx inputs and - // prevout info. - if len(sweepTx.TxIn) != len(req.PrevoutInfo) { - return fmt.Errorf("expected %v inputs, got %v", - len(req.PrevoutInfo), len(sweepTx.TxIn)) - } - // If the user selected an amount that is less than the total deposit // amount we'll check that the server sends us the correct change amount // back to our static address. - err = m.checkChange(ctx, sweepTx, loopIn.AddressParams) + err = m.checkChange(ctx, sweepTx, addressParams) if err != nil { return err } @@ -345,22 +393,19 @@ func (m *Manager) handleLoopInSweepReq(ctx context.Context, return err } - prevoutMap := make(map[wire.OutPoint]*wire.TxOut, len(req.PrevoutInfo)) - - // Set all the prevouts in the prevout map. - for _, prevout := range req.PrevoutInfo { - txid, err := chainhash.NewHash(prevout.TxidBytes) - if err != nil { - return err - } + prevoutMap, err := validateSweepPrevouts(sweepTx, req.PrevoutInfo) + if err != nil { + return err + } - prevoutMap[wire.OutPoint{ - Hash: *txid, - Index: prevout.OutputIndex, - }] = &wire.TxOut{ - Value: int64(prevout.Value), - PkScript: prevout.PkScript, - } + // The server supplies prevouts for the full batch, but deposit values + // and scripts signed by this client must match local wallet state. + err = validateSigningPrevouts( + req.DepositToNonces, deposits, addressParams.PkScript, + prevoutMap, + ) + if err != nil { + return err } prevOutputFetcher := txscript.NewMultiPrevOutFetcher( @@ -396,7 +441,7 @@ func (m *Manager) handleLoopInSweepReq(ctx context.Context, } musig2Session, err := staticutil.CreateMusig2Session( - ctx, m.cfg.Signer, loopIn.AddressParams, loopIn.Address, + ctx, m.cfg.Signer, addressParams, staticAddress, ) if err != nil { return err @@ -456,6 +501,111 @@ func (m *Manager) handleLoopInSweepReq(ctx context.Context, return err } +// validateSweepPrevouts verifies that the prevout list is an exact, unique +// mapping for every unsigned transaction input. +func validateSweepPrevouts(sweepTx *wire.MsgTx, + prevoutInfo []*swapserverrpc.PrevoutInfo) ( + map[wire.OutPoint]*wire.TxOut, error) { + + if len(sweepTx.TxIn) != len(prevoutInfo) { + return nil, fmt.Errorf("expected %v prevouts, got %v", + len(sweepTx.TxIn), len(prevoutInfo)) + } + + prevoutMap := make(map[wire.OutPoint]*wire.TxOut, len(prevoutInfo)) + for i, prevout := range prevoutInfo { + if prevout == nil { + return nil, fmt.Errorf("prevout %v is nil", i) + } + if prevout.Value > math.MaxInt64 { + return nil, fmt.Errorf( + "prevout %v value overflows int64", i, + ) + } + + txID, err := chainhash.NewHash(prevout.TxidBytes) + if err != nil { + return nil, fmt.Errorf("invalid prevout %v txid: %w", i, + err) + } + + outpoint := wire.OutPoint{ + Hash: *txID, + Index: prevout.OutputIndex, + } + if _, ok := prevoutMap[outpoint]; ok { + return nil, fmt.Errorf("duplicate prevout %v", outpoint) + } + + prevoutMap[outpoint] = &wire.TxOut{ + Value: int64(prevout.Value), + PkScript: bytes.Clone(prevout.PkScript), + } + } + + transactionInputs := make(map[wire.OutPoint]struct{}, len(sweepTx.TxIn)) + for _, txIn := range sweepTx.TxIn { + outpoint := txIn.PreviousOutPoint + if _, ok := transactionInputs[outpoint]; ok { + return nil, fmt.Errorf("duplicate transaction input %v", + outpoint) + } + transactionInputs[outpoint] = struct{}{} + + if _, ok := prevoutMap[outpoint]; !ok { + return nil, fmt.Errorf( + "missing prevout for transaction input %v", + outpoint, + ) + } + } + + return prevoutMap, nil +} + +// validateSigningPrevouts verifies locally known values and scripts for each +// deposit the server asks this client to sign. +func validateSigningPrevouts(depositToNonces map[string][]byte, + deposits []*deposit.Deposit, depositPkScript []byte, + prevoutMap map[wire.OutPoint]*wire.TxOut) error { + + depositsByOutpoint := make(map[string]*deposit.Deposit, len(deposits)) + for _, currentDeposit := range deposits { + if currentDeposit == nil { + return errors.New("deposit lookup returned nil deposit") + } + + depositsByOutpoint[currentDeposit.OutPoint.String()] = + currentDeposit + } + + for depositOutpoint := range depositToNonces { + currentDeposit, ok := depositsByOutpoint[depositOutpoint] + if !ok { + return fmt.Errorf( + "no local deposit for signing outpoint %v", + depositOutpoint, + ) + } + + prevout, ok := prevoutMap[currentDeposit.OutPoint] + if !ok { + return fmt.Errorf("missing signing prevout %v", + currentDeposit.OutPoint) + } + if prevout.Value != int64(currentDeposit.Value) { + return fmt.Errorf("signing prevout %v value mismatch", + currentDeposit.OutPoint) + } + if !bytes.Equal(prevout.PkScript, depositPkScript) { + return fmt.Errorf("signing prevout %v script mismatch", + currentDeposit.OutPoint) + } + } + + return nil +} + // checkChange ensures that the server sends us the correct change amount // back to our static address. An edge case arises if a batch contains two // swaps with identical change outputs. The client needs to ensure that any diff --git a/staticaddr/loopin/manager_test.go b/staticaddr/loopin/manager_test.go index 5fec56a16..d77c6ec55 100644 --- a/staticaddr/loopin/manager_test.go +++ b/staticaddr/loopin/manager_test.go @@ -4,13 +4,18 @@ import ( "bytes" "context" "errors" + "math" + "sync/atomic" "testing" + "time" + "github.com/btcsuite/btcd/btcec/v2" "github.com/btcsuite/btcd/btcec/v2/schnorr/musig2" "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/btcutil/psbt" "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/wire" + "github.com/lightninglabs/lndclient" "github.com/lightninglabs/loop" "github.com/lightninglabs/loop/fsm" "github.com/lightninglabs/loop/labels" @@ -18,12 +23,696 @@ import ( "github.com/lightninglabs/loop/staticaddr/script" "github.com/lightninglabs/loop/swap" "github.com/lightninglabs/loop/swapserverrpc" + "github.com/lightningnetwork/lnd/input" + "github.com/lightningnetwork/lnd/keychain" "github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/routing/route" "github.com/lightningnetwork/lnd/zpay32" "github.com/stretchr/testify/require" + "google.golang.org/grpc" ) +// TestManagerRunHandlesSweepRequestsConcurrently verifies that a blocked +// sweep request doesn't prevent other requests from being handled and that a +// request error doesn't stop notification intake. +func TestManagerRunHandlesSweepRequestsConcurrently(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + requests := make( + chan *swapserverrpc.ServerStaticLoopInSweepNotification, 3, + ) + requestStarted := make(chan byte, 3) + firstRequestRelease := make(chan struct{}) + requestErr := errors.New("request failed") + + store := &sweepRequestStore{ + mockStore: &mockStore{}, + getLoopInByHash: func(ctx context.Context, + swapHash lntypes.Hash) (*StaticAddressLoopIn, error) { + + requestStarted <- swapHash[0] + if swapHash[0] == 1 { + // Hold the first handler open while later requests + // enter the store lookup. + select { + case <-firstRequestRelease: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + + return nil, requestErr + }, + } + + mgr, err := NewManager(&Config{ + ChainNotifier: &silentBlockChainNotifier{}, + NotificationManager: &sweepRequestNotificationManager{ + requests: requests, + }, + Store: store, + }, 1) + require.NoError(t, err) + + // Wait until Run has subscribed before publishing requests. + initChan := make(chan struct{}) + runErr := make(chan error, 1) + go func() { + runErr <- mgr.Run(ctx, initChan) + }() + receiveOrFail(t, initChan) + + // Start and hold the first request in the store lookup. + requests <- testSweepRequest(t, 1) + require.Equal(t, byte(1), receiveOrFail(t, requestStarted)) + + // The second request must enter while the first remains blocked. + requests <- testSweepRequest(t, 2) + require.Equal(t, byte(2), receiveOrFail(t, requestStarted)) + + // A failed second request must not stop subsequent notification intake. + requests <- testSweepRequest(t, 3) + require.Equal(t, byte(3), receiveOrFail(t, requestStarted)) + + // Release the blocked handler before stopping and draining the manager. + close(firstRequestRelease) + cancel() + require.ErrorIs(t, receiveOrFail(t, runErr), context.Canceled) +} + +// TestManagerRunWaitsForSweepRequestHandlers verifies that all active sweep +// request handlers are canceled and drained before the manager exits. +func TestManagerRunWaitsForSweepRequestHandlers(t *testing.T) { + requests := make( + chan *swapserverrpc.ServerStaticLoopInSweepNotification, 1, + ) + handlerStarted := make(chan struct{}) + handlerCanceled := make(chan struct{}) + handlerRelease := make(chan struct{}) + + store := &sweepRequestStore{ + mockStore: &mockStore{}, + getLoopInByHash: func(ctx context.Context, + _ lntypes.Hash) (*StaticAddressLoopIn, error) { + + close(handlerStarted) + <-ctx.Done() + + // Report cancellation before holding the handler open so + // the test can distinguish cancellation from draining. + close(handlerCanceled) + <-handlerRelease + + return nil, ctx.Err() + }, + } + + mgr, err := NewManager(&Config{ + ChainNotifier: &silentBlockChainNotifier{}, + NotificationManager: &sweepRequestNotificationManager{ + requests: requests, + }, + Store: store, + }, 1) + require.NoError(t, err) + + // Wait until Run has subscribed before publishing the request. + initChan := make(chan struct{}) + runErr := make(chan error, 1) + go func() { + runErr <- mgr.Run(t.Context(), initChan) + }() + receiveOrFail(t, initChan) + + // Closing the subscription makes Run cancel the active handler. + requests <- testSweepRequest(t, 1) + receiveOrFail(t, handlerStarted) + close(requests) + receiveOrFail(t, handlerCanceled) + + // Run must wait until the canceled handler is allowed to return. + require.Never(t, func() bool { + select { + case <-runErr: + return true + default: + return false + } + }, 100*time.Millisecond, 10*time.Millisecond) + + // Releasing the handler lets Run finish its deferred drain. + close(handlerRelease) + require.ErrorContains(t, receiveOrFail(t, runErr), "ntfnChan closed") +} + +// TestManagerRunBoundsSweepRequestConcurrency verifies that the worker pool +// applies backpressure instead of creating an unbounded goroutine per request. +func TestManagerRunBoundsSweepRequestConcurrency(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + requests := make( + chan *swapserverrpc.ServerStaticLoopInSweepNotification, 3, + ) + requestStarted := make(chan byte, 3) + releaseRequest := make(chan struct{}) + + store := &sweepRequestStore{ + mockStore: &mockStore{}, + getLoopInByHash: func(ctx context.Context, + swapHash lntypes.Hash) (*StaticAddressLoopIn, error) { + + requestStarted <- swapHash[0] + select { + case <-releaseRequest: + return nil, errors.New("released request") + + case <-ctx.Done(): + return nil, ctx.Err() + } + }, + } + + mgr, err := NewManager(&Config{ + ChainNotifier: &silentBlockChainNotifier{}, + NotificationManager: &sweepRequestNotificationManager{ + requests: requests, + }, + Store: store, + }, 1) + require.NoError(t, err) + mgr.sweepRequestConcurrency = 2 + + initChan := make(chan struct{}) + runErr := make(chan error, 1) + go func() { + runErr <- mgr.Run(ctx, initChan) + }() + receiveOrFail(t, initChan) + + // Fill both worker slots and leave a third request queued. + requests <- testSweepRequest(t, 1) + requests <- testSweepRequest(t, 2) + requests <- testSweepRequest(t, 3) + receiveOrFail(t, requestStarted) + receiveOrFail(t, requestStarted) + + // The third store lookup cannot begin until one bounded worker exits. + require.Never(t, func() bool { + return len(requestStarted) != 0 + }, 100*time.Millisecond, 10*time.Millisecond) + + releaseRequest <- struct{}{} + require.Equal(t, byte(3), receiveOrFail(t, requestStarted)) + + cancel() + require.ErrorIs(t, receiveOrFail(t, runErr), context.Canceled) +} + +// TestHandleLoopInSweepReqReachesPushConcurrently verifies that two complete +// signing flows overlap at the final response RPC. +func TestHandleLoopInSweepReqReachesPushConcurrently(t *testing.T) { + clientKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + serverKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + staticAddress, err := script.NewStaticAddress( + input.MuSig2Version100RC2, 1_000, clientKey.PubKey(), + serverKey.PubKey(), + ) + require.NoError(t, err) + depositPkScript, err := staticAddress.StaticAddressScript() + require.NoError(t, err) + addressParams := &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + PkScript: depositPkScript, + } + + const confirmationHeight = 1 + currentDeposit := makeDeposit(1, 0, 10_000, confirmationHeight) + depositOutpoint := currentDeposit.OutPoint.String() + swapHash := lntypes.Hash{1} + loopIn := &StaticAddressLoopIn{ + SwapHash: swapHash, + DepositOutpoints: []string{depositOutpoint}, + Deposits: []*deposit.Deposit{currentDeposit}, + SelectedAmount: currentDeposit.Value, + } + loopIn.SetState(Succeeded) + + pushStarted := make(chan *swapserverrpc.PushStaticAddressSweeplessSigsRequest, + 2) + releasePushes := make(chan struct{}) + mgr := &Manager{ + cfg: &Config{ + Server: &blockingSweepResponseServer{ + pushStarted: pushStarted, + release: releasePushes, + }, + AddressManager: &mockAddressManager{ + params: addressParams, + staticAddress: staticAddress, + }, + DepositManager: &mockDepositManager{ + byOutpoint: map[string]*deposit.Deposit{ + depositOutpoint: currentDeposit, + }, + }, + Signer: &sweepRequestSigner{}, + Store: &mockStore{ + loopIns: map[lntypes.Hash]*StaticAddressLoopIn{ + swapHash: loopIn, + }, + mapIDs: map[lntypes.Hash][]deposit.ID{ + swapHash: {currentDeposit.ID}, + }, + }, + }, + } + + firstReq := successfulSweepRequest( + t, swapHash, currentDeposit, depositPkScript, 9_000, + ) + secondReq := successfulSweepRequest( + t, swapHash, currentDeposit, depositPkScript, 8_000, + ) + + handlerErr := make(chan error, 2) + go func() { + handlerErr <- mgr.handleLoopInSweepReq(t.Context(), firstReq) + }() + go func() { + handlerErr <- mgr.handleLoopInSweepReq(t.Context(), secondReq) + }() + + // Both responses must reach Push while neither RPC has returned. This + // covers validation, sighash construction, MuSig2 signing and response + // assembly rather than only the initial store lookup. + firstPush := receiveOrFail(t, pushStarted) + secondPush := receiveOrFail(t, pushStarted) + require.NotEqual(t, firstPush.Txid, secondPush.Txid) + require.Contains(t, firstPush.SigningInfo, depositOutpoint) + require.Contains(t, secondPush.SigningInfo, depositOutpoint) + + close(releasePushes) + require.NoError(t, receiveOrFail(t, handlerErr)) + require.NoError(t, receiveOrFail(t, handlerErr)) +} + +// TestValidateSweepPrevoutsRejectsMalformedMappings verifies malformed server +// prevout lists fail before btcd's sighash constructor can dereference them. +func TestValidateSweepPrevoutsRejectsMalformedMappings(t *testing.T) { + firstOutpoint := wire.OutPoint{Hash: chainhash.Hash{1}, Index: 0} + secondOutpoint := wire.OutPoint{Hash: chainhash.Hash{2}, Index: 1} + validScript := []byte{0x51} + + testCases := []struct { + name string + tx *wire.MsgTx + prevouts []*swapserverrpc.PrevoutInfo + expectedErr string + }{ + { + name: "exact mapping", + tx: makeSweepTx( + []wire.OutPoint{firstOutpoint}, nil, + ), + prevouts: []*swapserverrpc.PrevoutInfo{ + testPrevoutInfo(firstOutpoint, 1_000, validScript), + }, + }, + { + name: "missing transaction input", + tx: makeSweepTx( + []wire.OutPoint{firstOutpoint}, nil, + ), + prevouts: []*swapserverrpc.PrevoutInfo{ + testPrevoutInfo(secondOutpoint, 1_000, validScript), + }, + expectedErr: "missing prevout", + }, + { + name: "duplicate prevout", + tx: makeSweepTx( + []wire.OutPoint{firstOutpoint, secondOutpoint}, nil, + ), + prevouts: []*swapserverrpc.PrevoutInfo{ + testPrevoutInfo(firstOutpoint, 1_000, validScript), + testPrevoutInfo(firstOutpoint, 1_000, validScript), + }, + expectedErr: "duplicate prevout", + }, + { + name: "nil prevout", + tx: makeSweepTx( + []wire.OutPoint{firstOutpoint}, nil, + ), + prevouts: []*swapserverrpc.PrevoutInfo{nil}, + expectedErr: "prevout 0 is nil", + }, + { + name: "value overflow", + tx: makeSweepTx( + []wire.OutPoint{firstOutpoint}, nil, + ), + prevouts: []*swapserverrpc.PrevoutInfo{ + testPrevoutInfo( + firstOutpoint, math.MaxInt64+1, validScript, + ), + }, + expectedErr: "value overflows int64", + }, + { + name: "duplicate transaction input", + tx: makeSweepTx( + []wire.OutPoint{firstOutpoint, firstOutpoint}, nil, + ), + prevouts: []*swapserverrpc.PrevoutInfo{ + testPrevoutInfo(firstOutpoint, 1_000, validScript), + testPrevoutInfo(secondOutpoint, 1_000, validScript), + }, + expectedErr: "duplicate transaction input", + }, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + var err error + var prevoutMap map[wire.OutPoint]*wire.TxOut + + // In particular, an exact-length but mismatched mapping must + // return an error instead of reaching the prior nil dereference. + require.NotPanics(t, func() { + prevoutMap, err = validateSweepPrevouts( + testCase.tx, testCase.prevouts, + ) + }) + if testCase.expectedErr == "" { + require.NoError(t, err) + require.Len(t, prevoutMap, len(testCase.prevouts)) + + return + } + + require.ErrorContains(t, err, testCase.expectedErr) + }) + } +} + +// TestValidateSigningPrevoutsChecksLocalDeposit verifies that owned inputs use +// locally known values and scripts rather than trusting the server's prevouts. +func TestValidateSigningPrevoutsChecksLocalDeposit(t *testing.T) { + currentDeposit := makeDeposit(1, 0, 10_000, 1) + depositOutpoint := currentDeposit.OutPoint.String() + depositScript := []byte{0x51, 0x20} + + testCases := []struct { + name string + nonces map[string][]byte + prevout *wire.TxOut + expectedErr string + }{ + { + name: "local value and script", + nonces: map[string][]byte{depositOutpoint: nil}, + prevout: &wire.TxOut{ + Value: int64(currentDeposit.Value), + PkScript: bytes.Clone(depositScript), + }, + }, + { + name: "server value mismatch", + nonces: map[string][]byte{depositOutpoint: nil}, + prevout: &wire.TxOut{ + Value: int64(currentDeposit.Value) - 1, + PkScript: bytes.Clone(depositScript), + }, + expectedErr: "value mismatch", + }, + { + name: "server script mismatch", + nonces: map[string][]byte{depositOutpoint: nil}, + prevout: &wire.TxOut{ + Value: int64(currentDeposit.Value), + PkScript: []byte{0x52}, + }, + expectedErr: "script mismatch", + }, + { + name: "unknown signing deposit", + nonces: map[string][]byte{ + wire.OutPoint{Hash: chainhash.Hash{2}}.String(): nil, + }, + prevout: &wire.TxOut{ + Value: int64(currentDeposit.Value), + PkScript: bytes.Clone(depositScript), + }, + expectedErr: "no local deposit", + }, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + prevoutMap := map[wire.OutPoint]*wire.TxOut{ + currentDeposit.OutPoint: testCase.prevout, + } + + // These comparisons are the final trusted-data boundary + // before constructing the Taproot signature hash. + err := validateSigningPrevouts( + testCase.nonces, []*deposit.Deposit{currentDeposit}, + depositScript, prevoutMap, + ) + if testCase.expectedErr == "" { + require.NoError(t, err) + + return + } + + require.ErrorContains(t, err, testCase.expectedErr) + }) + } +} + +// testSweepRequest returns a parseable sweep request identified by id. +func testSweepRequest(t *testing.T, + id byte) *swapserverrpc.ServerStaticLoopInSweepNotification { + + t.Helper() + + swapHash := lntypes.Hash{id} + sweepTx := makeSweepTx( + []wire.OutPoint{{Hash: chainhash.Hash{id}}}, + []*wire.TxOut{{Value: int64(id) + 1}}, + ) + sweepPacket, err := psbt.NewFromUnsignedTx(sweepTx) + require.NoError(t, err) + + var psbtBuffer bytes.Buffer + require.NoError(t, sweepPacket.Serialize(&psbtBuffer)) + + return &swapserverrpc.ServerStaticLoopInSweepNotification{ + SwapHash: swapHash[:], + SweepTxPsbt: psbtBuffer.Bytes(), + } +} + +// receiveOrFail returns the next channel value or fails after one second. +func receiveOrFail[T any](t *testing.T, ch <-chan T) T { + t.Helper() + + select { + case value := <-ch: + return value + + case <-time.After(time.Second): + var zero T + t.Fatal("timed out waiting for test value") + + return zero + } +} + +// successfulSweepRequest creates a complete sweepless request for one deposit +// and varies the output value to produce distinct transaction IDs. +func successfulSweepRequest(t *testing.T, swapHash lntypes.Hash, + currentDeposit *deposit.Deposit, depositPkScript []byte, + outputValue int64) *swapserverrpc.ServerStaticLoopInSweepNotification { + + t.Helper() + + sweepTx := makeSweepTx( + []wire.OutPoint{currentDeposit.OutPoint}, + []*wire.TxOut{{Value: outputValue, PkScript: []byte{0x51}}}, + ) + sweepPacket, err := psbt.NewFromUnsignedTx(sweepTx) + require.NoError(t, err) + + var psbtBuffer bytes.Buffer + require.NoError(t, sweepPacket.Serialize(&psbtBuffer)) + + depositOutpoint := currentDeposit.OutPoint.String() + + return &swapserverrpc.ServerStaticLoopInSweepNotification{ + SwapHash: swapHash[:], + SweepTxPsbt: psbtBuffer.Bytes(), + DepositToNonces: map[string][]byte{ + depositOutpoint: make([]byte, musig2.PubNonceSize), + }, + PrevoutInfo: []*swapserverrpc.PrevoutInfo{ + testPrevoutInfo( + currentDeposit.OutPoint, uint64(currentDeposit.Value), + depositPkScript, + ), + }, + } +} + +// testPrevoutInfo returns protobuf prevout information for one transaction +// outpoint. +func testPrevoutInfo(outpoint wire.OutPoint, value uint64, + pkScript []byte) *swapserverrpc.PrevoutInfo { + + return &swapserverrpc.PrevoutInfo{ + Value: value, + PkScript: bytes.Clone(pkScript), + TxidBytes: outpoint.Hash[:], + OutputIndex: outpoint.Index, + } +} + +// blockingSweepResponseServer records response RPCs and blocks them until the +// test releases all calls. +type blockingSweepResponseServer struct { + // StaticAddressServerClient supplies methods unused by the test. + swapserverrpc.StaticAddressServerClient + + // pushStarted receives each response that reaches the RPC boundary. + pushStarted chan<- *swapserverrpc.PushStaticAddressSweeplessSigsRequest + + // release blocks response RPCs so their overlap can be observed. + release <-chan struct{} +} + +// PushStaticAddressSweeplessSigs records a response and waits for the test to +// release it. +func (s *blockingSweepResponseServer) PushStaticAddressSweeplessSigs( + ctx context.Context, + req *swapserverrpc.PushStaticAddressSweeplessSigsRequest, + _ ...grpc.CallOption) (*swapserverrpc.PushStaticAddressSweeplessSigsResponse, + error) { + + select { + case s.pushStarted <- req: + case <-ctx.Done(): + return nil, ctx.Err() + } + + select { + case <-s.release: + return &swapserverrpc.PushStaticAddressSweeplessSigsResponse{}, nil + + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// sweepRequestSigner supplies deterministic MuSig2 session data for complete +// sweep-request tests. +type sweepRequestSigner struct { + // SignerClient supplies signer methods unused by the test. + lndclient.SignerClient + + // nextSession gives concurrent signing flows distinct session IDs. + nextSession atomic.Uint32 +} + +// MuSig2CreateSession returns a deterministic, size-correct session. +func (s *sweepRequestSigner) MuSig2CreateSession(context.Context, + input.MuSig2Version, *keychain.KeyLocator, [][]byte, + ...lndclient.MuSig2SessionOpts) (*input.MuSig2SessionInfo, error) { + + sessionNumber := s.nextSession.Add(1) + var sessionID [32]byte + sessionID[0] = byte(sessionNumber) + var publicNonce [musig2.PubNonceSize]byte + publicNonce[0] = byte(sessionNumber) + + return &input.MuSig2SessionInfo{ + SessionID: sessionID, + PublicNonce: publicNonce, + }, nil +} + +// MuSig2RegisterNonces reports that the deterministic session has all nonces. +func (*sweepRequestSigner) MuSig2RegisterNonces(context.Context, [32]byte, + [][musig2.PubNonceSize]byte) (bool, error) { + + return true, nil +} + +// MuSig2Sign returns a deterministic size-correct partial signature. +func (*sweepRequestSigner) MuSig2Sign(context.Context, [32]byte, [32]byte, + bool) ([]byte, error) { + + return make([]byte, 32), nil +} + +// MuSig2Cleanup accepts cleanup of the deterministic session. +func (*sweepRequestSigner) MuSig2Cleanup(context.Context, [32]byte) error { + return nil +} + +// sweepRequestStore lets manager tests control loop-in lookups. +type sweepRequestStore struct { + // mockStore supplies the remaining store methods. + *mockStore + + // getLoopInByHash handles loop-in lookups for each test. + getLoopInByHash func(context.Context, lntypes.Hash) ( + *StaticAddressLoopIn, error) +} + +// GetLoopInByHash delegates the lookup to the test's configured function. +func (s *sweepRequestStore) GetLoopInByHash(ctx context.Context, + swapHash lntypes.Hash) (*StaticAddressLoopIn, error) { + + return s.getLoopInByHash(ctx, swapHash) +} + +// sweepRequestNotificationManager supplies sweep requests to manager tests. +type sweepRequestNotificationManager struct { + // requests is the test-controlled sweep request stream. + requests chan *swapserverrpc.ServerStaticLoopInSweepNotification +} + +// SubscribeStaticLoopInSweepRequests returns the test sweep request stream. +func (m *sweepRequestNotificationManager) SubscribeStaticLoopInSweepRequests( + context.Context, +) <-chan *swapserverrpc.ServerStaticLoopInSweepNotification { + + return m.requests +} + +// SubscribeStaticLoopInRiskAccepted is unused by these manager tests. +func (*sweepRequestNotificationManager) SubscribeStaticLoopInRiskAccepted( + context.Context, lntypes.Hash, +) <-chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification { + + return nil +} + +// SubscribeStaticLoopInRiskRejected is unused by these manager tests. +func (*sweepRequestNotificationManager) SubscribeStaticLoopInRiskRejected( + context.Context, lntypes.Hash, +) <-chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification { + + return nil +} + type testCase struct { name string deposits []*deposit.Deposit diff --git a/swap_server_client.go b/swap_server_client.go index 160226648..cfa49b5fd 100644 --- a/swap_server_client.go +++ b/swap_server_client.go @@ -16,6 +16,7 @@ import ( "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/wire" "github.com/lightninglabs/aperture/l402" + "github.com/lightninglabs/lndclient" "github.com/lightninglabs/loop/loopdb" "github.com/lightninglabs/loop/swapserverrpc" "github.com/lightningnetwork/lnd/lnrpc" @@ -28,6 +29,7 @@ import ( "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" + "gopkg.in/macaroon.v2" ) var ( @@ -151,6 +153,135 @@ type grpcSwapServerClient struct { wg sync.WaitGroup } +// l402ClientInterceptor permits authenticated unary calls to run concurrently +// while delegating token acquisition and stream authentication to Aperture's +// serialized interceptor. +type l402ClientInterceptor struct { + // tokenStore provides the currently paid L402 token for the concurrent + // unary fast path. + tokenStore l402.Store + + // callTimeout bounds each unary attempt made by the concurrent fast + // path. + callTimeout time.Duration + + // allowInsecure controls whether the L402 credential may be sent over + // an insecure transport. + allowInsecure bool + + // fallbackUnary serializes calls that may need to acquire an L402 + // token. + fallbackUnary grpc.UnaryClientInterceptor + + // fallbackStream handles L402 authentication for streaming RPCs. + fallbackStream grpc.StreamClientInterceptor + + // loadPaidMacaroon loads the credential used by the concurrent unary + // fast path. Tests replace it to exercise the fast path without relying + // on Aperture's private token serialization. + loadPaidMacaroon func() (*macaroon.Macaroon, bool, error) +} + +// newL402ClientInterceptor creates an interceptor that only serializes calls +// when they may need to acquire or resume payment for an L402 token. +func newL402ClientInterceptor(lnd *lndclient.LndServices, store l402.Store, + callTimeout time.Duration, maxCost, maxFee btcutil.Amount, + allowInsecure bool) *l402ClientInterceptor { + + fallback := l402.NewInterceptor( + lnd, store, callTimeout, maxCost, maxFee, allowInsecure, + ) + interceptor := &l402ClientInterceptor{ + tokenStore: store, + callTimeout: callTimeout, + allowInsecure: allowInsecure, + fallbackUnary: fallback.UnaryInterceptor, + fallbackStream: fallback.StreamInterceptor, + } + interceptor.loadPaidMacaroon = interceptor.currentPaidMacaroon + + return interceptor +} + +// currentPaidMacaroon returns the current token's paid macaroon. The boolean +// is false when the store has no token or its token payment is still pending. +func (i *l402ClientInterceptor) currentPaidMacaroon() (*macaroon.Macaroon, + bool, error) { + + token, err := i.tokenStore.CurrentToken() + switch { + case errors.Is(err, l402.ErrNoToken): + return nil, false, nil + + case err != nil: + return nil, false, err + + case token == nil: + return nil, false, errors.New( + "L402 token store returned nil token", + ) + + case token.Preimage == (lntypes.Preimage{}): + return nil, false, nil + } + + paidMacaroon, err := token.PaidMacaroon() + if err != nil { + return nil, false, err + } + + return paidMacaroon, true, nil +} + +// UnaryInterceptor runs calls with an existing paid token concurrently and +// falls back to serialized token handling when authentication may be needed. +func (i *l402ClientInterceptor) UnaryInterceptor(ctx context.Context, + method string, req, reply any, cc *grpc.ClientConn, + invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { + + // Aperture's file store does not make token promotion atomic. A + // concurrent read can therefore fail while the serialized fallback is + // storing a paid token. Retrying through that fallback waits for the + // token operation and reads the completed file. + paidMacaroon, paid, err := i.loadPaidMacaroon() + if err != nil || !paid { + return i.fallbackUnary( + ctx, method, req, reply, cc, invoker, opts..., + ) + } + + // Authenticate the common paid-token path directly so independent RPCs + // don't wait behind Aperture's token-acquisition mutex. + callOpts := append([]grpc.CallOption(nil), opts...) + callOpts = append(callOpts, grpc.PerRPCCredentials( + l402.NewMacaroonCredential(paidMacaroon, i.allowInsecure), + )) + rpcCtx, cancel := context.WithTimeout(ctx, i.callTimeout) + defer cancel() + + err = invoker(rpcCtx, method, req, reply, cc, callOpts...) + if !l402.IsPaymentRequired(err) { + return err + } + + // A challenge can mean the token must be acquired or resumed. Delegate + // that uncommon path to the serialized interceptor to prevent duplicate + // payments from concurrent calls. + return i.fallbackUnary( + ctx, method, req, reply, cc, invoker, opts..., + ) +} + +// StreamInterceptor delegates stream establishment to Aperture's serialized +// L402 interceptor. The lock is released once the stream has been established. +func (i *l402ClientInterceptor) StreamInterceptor(ctx context.Context, + desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, + streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, + error) { + + return i.fallbackStream(ctx, desc, cc, method, streamer, opts...) +} + // stop sends the signal for the server's goroutines to shutdown and waits for // them to complete. func (s *grpcSwapServerClient) stop() { @@ -168,7 +299,7 @@ func newSwapServerClient(cfg *ClientConfig, l402Store l402.Store) ( // Create the server connection with the interceptor that will handle // the L402 protocol for us. - clientInterceptor := l402.NewInterceptor( + clientInterceptor := newL402ClientInterceptor( cfg.Lnd, l402Store, serverRPCTimeout, cfg.MaxL402Cost, cfg.MaxL402Fee, false, ) @@ -922,7 +1053,7 @@ func rpcRouteCancel(details *outCancelDetails) ( // proxyAddr indicates that a SOCKS proxy found at the address should be used to // establish the connection. func getSwapServerConn(address, proxyAddress string, skipCertCheck bool, - tlsPath string, interceptor *l402.ClientInterceptor) (*grpc.ClientConn, + tlsPath string, interceptor *l402ClientInterceptor) (*grpc.ClientConn, error) { // Create a dial options array. diff --git a/swap_server_client_test.go b/swap_server_client_test.go index 81826f8c5..6f005fe56 100644 --- a/swap_server_client_test.go +++ b/swap_server_client_test.go @@ -1,12 +1,226 @@ package loop import ( + "context" + "errors" "testing" + "time" + "github.com/lightninglabs/aperture/l402" looptest "github.com/lightninglabs/loop/test" "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/status" + "gopkg.in/macaroon.v2" ) +// TestL402ClientInterceptorConcurrentPaidCalls verifies that calls using an +// existing paid token reach the transport concurrently. +func TestL402ClientInterceptorConcurrentPaidCalls(t *testing.T) { + t.Parallel() + + paidMacaroon, err := macaroon.New( + []byte("root key"), []byte("token id"), "test", + macaroon.LatestVersion, + ) + require.NoError(t, err) + + interceptor := &l402ClientInterceptor{ + callTimeout: time.Minute, + loadPaidMacaroon: func() (*macaroon.Macaroon, bool, error) { + return paidMacaroon, true, nil + }, + fallbackUnary: func(context.Context, string, any, any, + *grpc.ClientConn, grpc.UnaryInvoker, + ...grpc.CallOption) error { + + return errors.New("unexpected L402 fallback") + }, + } + + callStarted := make(chan int, 2) + releaseCalls := make(chan struct{}) + invoker := func(_ context.Context, _ string, req, _ any, + _ *grpc.ClientConn, opts ...grpc.CallOption) error { + + // A paid call must carry the original option plus the L402 + // credential before it reaches the transport. + if len(opts) != 2 { + return errors.New("paid call has unexpected " + + "call options") + } + + callStarted <- req.(int) + <-releaseCalls + + return nil + } + + callErr := make(chan error, 2) + for id := 1; id <= 2; id++ { + go func() { + callErr <- interceptor.UnaryInterceptor( + t.Context(), "test", id, nil, nil, invoker, + grpc.WaitForReady(true), + ) + }() + } + + // Both calls must arrive while the first remains blocked. This is the + // property the Aperture interceptor's unconditional mutex prevented. + started := map[int]bool{ + receiveOrTimeout(t, callStarted): true, + receiveOrTimeout(t, callStarted): true, + } + require.Equal(t, map[int]bool{1: true, 2: true}, started) + + close(releaseCalls) + require.NoError(t, receiveOrTimeout(t, callErr)) + require.NoError(t, receiveOrTimeout(t, callErr)) +} + +// TestL402ClientInterceptorPaymentChallengeFallback verifies that a challenge +// on the concurrent path is retried through serialized token handling. +func TestL402ClientInterceptorPaymentChallengeFallback(t *testing.T) { + t.Parallel() + + paidMacaroon, err := macaroon.New( + []byte("root key"), []byte("token id"), "test", + macaroon.LatestVersion, + ) + require.NoError(t, err) + + fallbackCalled := make(chan struct{}, 1) + interceptor := &l402ClientInterceptor{ + callTimeout: time.Minute, + loadPaidMacaroon: func() (*macaroon.Macaroon, bool, error) { + return paidMacaroon, true, nil + }, + fallbackUnary: func(context.Context, string, any, any, + *grpc.ClientConn, grpc.UnaryInvoker, + ...grpc.CallOption) error { + + fallbackCalled <- struct{}{} + + return nil + }, + } + + invoker := func(context.Context, string, any, any, + *grpc.ClientConn, ...grpc.CallOption) error { + + // Simulate the server rejecting the cached paid token so the + // payment-aware fallback owns any acquisition or retry. + return status.Error(l402.GRPCErrCode, l402.GRPCErrMessage) + } + + err = interceptor.UnaryInterceptor( + t.Context(), "test", nil, nil, nil, invoker, + ) + require.NoError(t, err) + receiveOrTimeout(t, fallbackCalled) +} + +// TestL402ClientInterceptorCurrentTokenFallback verifies that token states +// which cannot use the concurrent paid path enter serialized token handling. +func TestL402ClientInterceptorCurrentTokenFallback(t *testing.T) { + t.Parallel() + + storeErr := errors.New("token store failed") + testCases := []struct { + name string + token *l402.Token + tokenErr error + }{ + { + name: "no token", + tokenErr: l402.ErrNoToken, + }, + { + name: "pending token", + token: &l402.Token{}, + }, + { + name: "nil token", + }, + { + name: "store error", + tokenErr: storeErr, + }, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + fallbackCalled := false + interceptor := &l402ClientInterceptor{ + tokenStore: &staticL402Store{ + token: testCase.token, + err: testCase.tokenErr, + }, + fallbackUnary: func(context.Context, string, + any, any, *grpc.ClientConn, + grpc.UnaryInvoker, + ...grpc.CallOption) error { + + // Reaching this callback proves the + // unusable token did not enter the + // concurrent transport path. + fallbackCalled = true + + return nil + }, + } + interceptor.loadPaidMacaroon = + interceptor.currentPaidMacaroon + + invoker := func(context.Context, string, any, any, + *grpc.ClientConn, ...grpc.CallOption) error { + + return errors.New("unexpected concurrent call") + } + + err := interceptor.UnaryInterceptor( + t.Context(), "test", nil, nil, nil, invoker, + ) + require.NoError(t, err) + require.True(t, fallbackCalled) + }) + } +} + +// receiveOrTimeout returns the next channel value or fails after one second. +func receiveOrTimeout[T any](t *testing.T, values <-chan T) T { + t.Helper() + + select { + case value := <-values: + return value + + case <-time.After(time.Second): + var zero T + t.Fatal("timed out waiting for test value") + + return zero + } +} + +// staticL402Store returns a configured token result for interceptor tests. +type staticL402Store struct { + // Store supplies methods unused by the token-loading tests. + l402.Store + + // token is returned by CurrentToken. + token *l402.Token + + // err is returned by CurrentToken. + err error +} + +// CurrentToken returns the token result configured by the test. +func (s *staticL402Store) CurrentToken() (*l402.Token, error) { + return s.token, s.err +} + // TestParseServerPubKey ensures that parseServerPubKey accepts a valid // compressed public key and rejects keys with an invalid length or contents. func TestParseServerPubKey(t *testing.T) {