From d2abb109b2cad97d92a90684e837015fd145f51c Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Tue, 8 Sep 2026 13:41:45 -0500 Subject: [PATCH 1/3] staticaddr/loopin: process sweep requests concurrently The server can issue many presigning requests at once, but the client handled notifications serially. Slow transports could leave later responses past the server signing deadline. Dispatch each request independently and drain active handlers during shutdown so notification intake and cleanup remain safe. --- docs/release-notes/release-notes-next.md | 4 + staticaddr/loopin/manager.go | 41 +++-- staticaddr/loopin/manager_test.go | 210 +++++++++++++++++++++++ 3 files changed, 242 insertions(+), 13 deletions(-) diff --git a/docs/release-notes/release-notes-next.md b/docs/release-notes/release-notes-next.md index 34fd3d750..b1421e44b 100644 --- a/docs/release-notes/release-notes-next.md +++ b/docs/release-notes/release-notes-next.md @@ -13,6 +13,10 @@ #### Bug Fixes +* Static Address loop-in sweep signing requests are now processed concurrently, + preventing slow connections from delaying later signing responses until the + server times them out. + * 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/manager.go b/staticaddr/loopin/manager.go index b011f33fd..17353c549 100644 --- a/staticaddr/loopin/manager.go +++ b/staticaddr/loopin/manager.go @@ -7,6 +7,7 @@ import ( "math" "slices" "sort" + "sync" "sync/atomic" "time" @@ -187,6 +188,15 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error { sweepReqs := m.cfg.NotificationManager. SubscribeStaticLoopInSweepRequests(ctx) + // Track active sweep handlers so shutdown cancels and drains them before + // their manager-owned resources can be released. + sweepCtx, cancelSweepHandlers := context.WithCancel(ctx) + var sweepHandlers sync.WaitGroup + defer func() { + cancelSweepHandlers() + sweepHandlers.Wait() + }() + // Communicate to the caller that the address manager has completed its // initialization. close(initChan) @@ -238,11 +248,18 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error { return fmt.Errorf("ntfnChan closed") } - err = m.handleLoopInSweepReq(ctx, sweepReq) - if err != nil { - log.Errorf("Error handling loop-in sweep "+ - "request: %v", err) - } + // Process each request independently so a slow signing flow + // doesn't block later sweep notifications. + sweepHandlers.Add(1) + go func(req *swapserverrpc.ServerStaticLoopInSweepNotification) { + defer sweepHandlers.Done() + + err := m.handleLoopInSweepReq(sweepCtx, req) + if err != nil { + log.Errorf("Error handling loop-in sweep "+ + "request: %v", err) + } + }(sweepReq) case <-ctx.Done(): return ctx.Err() @@ -284,26 +301,24 @@ func (m *Manager) handleLoopInSweepReq(ctx context.Context, 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 } ignoreUnknownOutpoints := false - deposits, err := m.cfg.DepositManager.DepositsForOutpoints( + _, err = m.cfg.DepositManager.DepositsForOutpoints( ctx, loopIn.DepositOutpoints, ignoreUnknownOutpoints, ) if err != nil { return err } - loopIn.Deposits = deposits reader := bytes.NewReader(req.SweepTxPsbt) sweepPacket, err := psbt.NewFromRawBytes(reader, false) @@ -333,7 +348,7 @@ func (m *Manager) handleLoopInSweepReq(ctx context.Context, // 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 } @@ -396,7 +411,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 diff --git a/staticaddr/loopin/manager_test.go b/staticaddr/loopin/manager_test.go index 5fec56a16..09e5d9d48 100644 --- a/staticaddr/loopin/manager_test.go +++ b/staticaddr/loopin/manager_test.go @@ -5,6 +5,7 @@ import ( "context" "errors" "testing" + "time" "github.com/btcsuite/btcd/btcec/v2/schnorr/musig2" "github.com/btcsuite/btcd/btcutil" @@ -24,6 +25,215 @@ import ( "github.com/stretchr/testify/require" ) +// 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) + }() + <-initChan + + // Start and hold the first request in the store lookup. + requests <- testSweepRequest(1) + require.Equal(t, byte(1), receiveOrFail(t, requestStarted)) + + // The second request must enter while the first remains blocked. + requests <- testSweepRequest(2) + require.Equal(t, byte(2), receiveOrFail(t, requestStarted)) + + // A failed second request must not stop subsequent notification intake. + requests <- testSweepRequest(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) + }() + <-initChan + + // Closing the subscription makes Run cancel the active handler. + requests <- testSweepRequest(1) + <-handlerStarted + close(requests) + <-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") +} + +// testSweepRequest returns a minimal sweep request identified by id. +func testSweepRequest( + id byte) *swapserverrpc.ServerStaticLoopInSweepNotification { + + swapHash := lntypes.Hash{id} + + return &swapserverrpc.ServerStaticLoopInSweepNotification{ + SwapHash: swapHash[:], + } +} + +// receiveOrFail returns the next channel value or fails when the test context +// ends. +func receiveOrFail[T any](t *testing.T, ch <-chan T) T { + t.Helper() + + select { + case value := <-ch: + return value + + case <-t.Context().Done(): + var zero T + t.Fatalf("waiting for test value: %v", t.Context().Err()) + + return zero + } +} + +// 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 From 8b28351e69528b0c1e248ca516191320721d96f6 Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Tue, 8 Sep 2026 16:26:33 -0500 Subject: [PATCH 2/3] loop: parallelize authenticated L402 calls Authenticated RPCs were serialized behind Aperture's token-payment mutex. Slow sweep responses could therefore exceed the server timeout. Use a paid token without the acquisition lock while keeping a serialized fallback for payment challenges. --- swap_server_client.go | 135 ++++++++++++++++++++++- swap_server_client_test.go | 214 +++++++++++++++++++++++++++++++++++++ 2 files changed, 347 insertions(+), 2 deletions(-) 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) { From 04a4e3aeba3204e296ea4cbd94490e3b78d0507e Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Tue, 8 Sep 2026 16:26:44 -0500 Subject: [PATCH 3/3] staticaddr/loopin: harden concurrent sweep signing Unbounded notification handlers could exhaust client and signer resources. Incomplete server prevout mappings could also crash or misstate a signing hash. Bound request processing with a fixed worker pool and verify prevouts and locally owned deposit data before signing. --- docs/release-notes/release-notes-next.md | 7 +- staticaddr/loopin/actions_test.go | 9 +- staticaddr/loopin/manager.go | 251 ++++++++--- staticaddr/loopin/manager_test.go | 509 ++++++++++++++++++++++- 4 files changed, 698 insertions(+), 78 deletions(-) diff --git a/docs/release-notes/release-notes-next.md b/docs/release-notes/release-notes-next.md index b1421e44b..14c6ced8c 100644 --- a/docs/release-notes/release-notes-next.md +++ b/docs/release-notes/release-notes-next.md @@ -13,9 +13,10 @@ #### Bug Fixes -* Static Address loop-in sweep signing requests are now processed concurrently, - preventing slow connections from delaying later signing responses until the - server times them out. +* 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 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 17353c549..0b85d9bd9 100644 --- a/staticaddr/loopin/manager.go +++ b/staticaddr/loopin/manager.go @@ -3,6 +3,7 @@ package loopin import ( "bytes" "context" + "errors" "fmt" "math" "slices" @@ -35,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 ( @@ -145,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. @@ -155,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) @@ -188,10 +198,21 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error { sweepReqs := m.cfg.NotificationManager. SubscribeStaticLoopInSweepRequests(ctx) - // Track active sweep handlers so shutdown cancels and drains them before - // their manager-owned resources can be released. + // 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() @@ -236,33 +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 } - // Process each request independently so a slow signing flow - // doesn't block later sweep notifications. - sweepHandlers.Add(1) - go func(req *swapserverrpc.ServerStaticLoopInSweepNotification) { - defer sweepHandlers.Done() - - err := m.handleLoopInSweepReq(sweepCtx, req) - if err != nil { - log.Errorf("Error handling loop-in sweep "+ - "request: %v", err) - } - }(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 } } } @@ -288,13 +323,26 @@ 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 { @@ -313,21 +361,13 @@ func (m *Manager) handleLoopInSweepReq(ctx context.Context, } ignoreUnknownOutpoints := false - _, err = m.cfg.DepositManager.DepositsForOutpoints( + deposits, err := m.cfg.DepositManager.DepositsForOutpoints( ctx, loopIn.DepositOutpoints, ignoreUnknownOutpoints, ) if err != nil { return err } - 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. if !loopIn.IsInState(Succeeded) { @@ -338,13 +378,6 @@ 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. @@ -360,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( @@ -471,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 09e5d9d48..d77c6ec55 100644 --- a/staticaddr/loopin/manager_test.go +++ b/staticaddr/loopin/manager_test.go @@ -4,14 +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" @@ -19,10 +23,13 @@ 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 @@ -74,18 +81,18 @@ func TestManagerRunHandlesSweepRequestsConcurrently(t *testing.T) { go func() { runErr <- mgr.Run(ctx, initChan) }() - <-initChan + receiveOrFail(t, initChan) // Start and hold the first request in the store lookup. - requests <- testSweepRequest(1) + requests <- testSweepRequest(t, 1) require.Equal(t, byte(1), receiveOrFail(t, requestStarted)) // The second request must enter while the first remains blocked. - requests <- testSweepRequest(2) + requests <- testSweepRequest(t, 2) require.Equal(t, byte(2), receiveOrFail(t, requestStarted)) // A failed second request must not stop subsequent notification intake. - requests <- testSweepRequest(3) + requests <- testSweepRequest(t, 3) require.Equal(t, byte(3), receiveOrFail(t, requestStarted)) // Release the blocked handler before stopping and draining the manager. @@ -136,13 +143,13 @@ func TestManagerRunWaitsForSweepRequestHandlers(t *testing.T) { go func() { runErr <- mgr.Run(t.Context(), initChan) }() - <-initChan + receiveOrFail(t, initChan) // Closing the subscription makes Run cancel the active handler. - requests <- testSweepRequest(1) - <-handlerStarted + requests <- testSweepRequest(t, 1) + receiveOrFail(t, handlerStarted) close(requests) - <-handlerCanceled + receiveOrFail(t, handlerCanceled) // Run must wait until the canceled handler is allowed to return. require.Never(t, func() bool { @@ -159,19 +166,361 @@ func TestManagerRunWaitsForSweepRequestHandlers(t *testing.T) { require.ErrorContains(t, receiveOrFail(t, runErr), "ntfnChan closed") } -// testSweepRequest returns a minimal sweep request identified by id. -func testSweepRequest( +// 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[:], + SwapHash: swapHash[:], + SweepTxPsbt: psbtBuffer.Bytes(), } } -// receiveOrFail returns the next channel value or fails when the test context -// ends. +// receiveOrFail returns the next channel value or fails after one second. func receiveOrFail[T any](t *testing.T, ch <-chan T) T { t.Helper() @@ -179,14 +528,144 @@ func receiveOrFail[T any](t *testing.T, ch <-chan T) T { case value := <-ch: return value - case <-t.Context().Done(): + case <-time.After(time.Second): var zero T - t.Fatalf("waiting for test value: %v", t.Context().Err()) + 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.