diff --git a/account/interface.go b/account/interface.go index 4db4968..61a5bdb 100644 --- a/account/interface.go +++ b/account/interface.go @@ -380,14 +380,16 @@ type Auctioneer interface { ModifyAccount(context.Context, *Account, []*wire.TxIn, []*wire.TxOut, []Modifier) ([]byte, error) - // SubscribeAccountUpdates opens a stream to the server and subscribes + // StartAccountSubscription opens a stream to the server and subscribes // to all updates that concern the given account, including all orders // that spend from that account. Only a single stream is ever open to // the server, so a second call to this method will send a second // subscription over the same stream, multiplexing all messages into the // same connection. A stream can be long-lived, so this can be called - // for every account as soon as it's confirmed open. - SubscribeAccountUpdates(context.Context, *keychain.KeyDescriptor) error + // for every account as soon as it's confirmed open. This method will + // return as soon as the authentication was successful. Messages sent + // from the server can then be received on the FromServerChan channel. + StartAccountSubscription(context.Context, *keychain.KeyDescriptor) error // Terms returns the current dynamic auctioneer terms like max account // size, max order duration in blocks and the auction fee schedule. diff --git a/account/manager.go b/account/manager.go index 9d0dbdd..9c6ca8d 100644 --- a/account/manager.go +++ b/account/manager.go @@ -647,7 +647,7 @@ func (m *Manager) resumeAccount(ctx context.Context, account *Account, // nolint // update state, as that state is ineligible for batch // execution. if account.State == StatePendingBatch { - err = m.cfg.Auctioneer.SubscribeAccountUpdates( + err = m.cfg.Auctioneer.StartAccountSubscription( ctx, account.TraderKey, ) if err != nil { @@ -832,7 +832,7 @@ func (m *Manager) handleStateOpen(ctx context.Context, account *Account) error { // level anyway. And we might end up executing multiple orders for the // same account in one batch. The messages from the server are received // and dispatched to the correct manager by the rpcServer. - err = m.cfg.Auctioneer.SubscribeAccountUpdates(ctx, account.TraderKey) + err = m.cfg.Auctioneer.StartAccountSubscription(ctx, account.TraderKey) if err != nil { return fmt.Errorf("unable to subscribe for account updates: %v", err) diff --git a/account/mock_test.go b/account/mock_test.go index dfb7be8..17a39c0 100644 --- a/account/mock_test.go +++ b/account/mock_test.go @@ -195,7 +195,7 @@ func (a *mockAuctioneer) ModifyAccount(_ context.Context, _ *Account, return []byte("auctioneer sig"), nil } -func (a *mockAuctioneer) SubscribeAccountUpdates(_ context.Context, +func (a *mockAuctioneer) StartAccountSubscription(_ context.Context, accountKey *keychain.KeyDescriptor) error { var traderKey [33]byte diff --git a/auctioneer/client.go b/auctioneer/client.go index 43adf09..05a5620 100644 --- a/auctioneer/client.go +++ b/auctioneer/client.go @@ -481,14 +481,16 @@ func (c *Client) OrderState(ctx context.Context, nonce order.Nonce) ( }) } -// SubscribeAccountUpdates opens a stream to the server and subscribes -// to all updates that concern the given account, including all orders -// that spend from that account. Only a single stream is ever open to -// the server, so a second call to this method will send a second -// subscription over the same stream, multiplexing all messages into the -// same connection. A stream can be long-lived, so this can be called -// for every account as soon as it's confirmed open. -func (c *Client) SubscribeAccountUpdates(ctx context.Context, +// StartAccountSubscription opens a stream to the server and subscribes to all +// updates that concern the given account, including all orders that spend from +// that account. Only a single stream is ever open to the server, so a second +// call to this method will send a second subscription over the same stream, +// multiplexing all messages into the same connection. A stream can be +// long-lived, so this can be called for every account as soon as it's confirmed +// open. This method will return as soon as the authentication was successful. +// Messages sent from the server can then be received on the FromServerChan +// channel. +func (c *Client) StartAccountSubscription(ctx context.Context, acctKey *keychain.KeyDescriptor) error { _, _, err := c.connectAndAuthenticate(ctx, acctKey, false) @@ -1135,7 +1137,7 @@ func (c *Client) HandleServerShutdown(err error) error { } c.subscribedAcctsMtx.Unlock() for _, acctKey := range acctKeys { - err := c.SubscribeAccountUpdates(context.Background(), acctKey) + err := c.StartAccountSubscription(context.Background(), acctKey) if err != nil { return err } diff --git a/clientdb/db.go b/clientdb/db.go index 8d01651..fd4e4cb 100644 --- a/clientdb/db.go +++ b/clientdb/db.go @@ -97,6 +97,10 @@ func initDB(filepath string, firstInit bool) (*bbolt.DB, error) { if err != nil { return err } + _, err = tx.CreateBucketIfNotExists(sidecarsBucketKey) + if err != nil { + return err + } _, err = tx.CreateBucketIfNotExists(batchBucketKey) if err != nil { return err diff --git a/clientdb/order.go b/clientdb/order.go index 3e00507..115b503 100644 --- a/clientdb/order.go +++ b/clientdb/order.go @@ -9,6 +9,7 @@ import ( "github.com/btcsuite/btcutil" "github.com/lightninglabs/pool/event" "github.com/lightninglabs/pool/order" + "github.com/lightninglabs/pool/sidecar" "github.com/lightningnetwork/lnd/tlv" "go.etcd.io/bbolt" ) @@ -17,6 +18,10 @@ const ( // bidSelfChanBalanceType is the tlv type we use to store the self // channel balance on bid orders. bidSelfChanBalanceType tlv.Type = 1 + + // bidSidecarTicketType is the tlv type we use to store the sidecar + // ticket on bid orders. + bidSidecarTicketType tlv.Type = 2 ) var ( @@ -631,14 +636,18 @@ func DeserializeOrder(nonce order.Nonce, r io.Reader) ( func deserializeOrderTlvData(r io.Reader, o order.Order) error { var ( selfChanBalance uint64 + sidecarTicket []byte ) // We'll add records for all possible additional order data fields here // but will check below which of them were actually set, depending on // the order type as well. - tlvStream, err := tlv.NewStream(tlv.MakePrimitiveRecord( - bidSelfChanBalanceType, &selfChanBalance, - )) + tlvStream, err := tlv.NewStream( + tlv.MakePrimitiveRecord( + bidSelfChanBalanceType, &selfChanBalance, + ), + tlv.MakePrimitiveRecord(bidSidecarTicketType, &sidecarTicket), + ) if err != nil { return err } @@ -659,6 +668,15 @@ func deserializeOrderTlvData(r io.Reader, o order.Order) error { selfChanBalance, ) } + + if t, ok := parsedTypes[bidSidecarTicketType]; ok && t == nil { + castOrder.SidecarTicket, err = sidecar.DeserializeTicket( + bytes.NewReader(sidecarTicket), + ) + if err != nil { + return err + } + } } return nil @@ -679,6 +697,20 @@ func serializeOrderTlvData(w io.Writer, o order.Order) error { bidSelfChanBalanceType, &selfChanBalance, )) } + + if castOrder.SidecarTicket != nil { + var buf bytes.Buffer + if err := sidecar.SerializeTicket( + &buf, castOrder.SidecarTicket, + ); err != nil { + return err + } + + sidecarBytes := buf.Bytes() + tlvRecords = append(tlvRecords, tlv.MakePrimitiveRecord( + bidSidecarTicketType, &sidecarBytes, + )) + } } tlvStream, err := tlv.NewStream(tlvRecords...) diff --git a/clientdb/order_test.go b/clientdb/order_test.go index 48b5386..1e551e2 100644 --- a/clientdb/order_test.go +++ b/clientdb/order_test.go @@ -9,6 +9,7 @@ import ( "github.com/btcsuite/btcutil" "github.com/davecgh/go-spew/spew" "github.com/lightninglabs/pool/order" + "github.com/lightninglabs/pool/sidecar" "github.com/lightningnetwork/lnd/keychain" "github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lnwallet/chainfee" @@ -26,6 +27,19 @@ func TestSubmitOrder(t *testing.T) { Kit: *dummyOrder(500000, 1337), MinNodeTier: 2, SelfChanBalance: 123, + SidecarTicket: &sidecar.Ticket{ + ID: [8]byte{11, 22, 33, 44, 55, 66, 77}, + State: sidecar.StateRegistered, + Offer: sidecar.Offer{ + Capacity: 1000000, + PushAmt: 200000, + LeaseDurationBlocks: 2016, + }, + Recipient: &sidecar.Recipient{ + MultiSigPubKey: testTraderKey, + MultiSigKeyIndex: 7, + }, + }, } o.Details().MinUnitsMatch = 10 err := store.SubmitOrder(o) diff --git a/clientdb/sidecar.go b/clientdb/sidecar.go new file mode 100644 index 0000000..b8ff7b4 --- /dev/null +++ b/clientdb/sidecar.go @@ -0,0 +1,173 @@ +package clientdb + +import ( + "bytes" + "errors" + "fmt" + + "github.com/btcsuite/btcd/btcec" + "github.com/lightninglabs/pool/sidecar" + "go.etcd.io/bbolt" +) + +var ( + // ErrNoSidecar is the error returned if no sidecar with the given + // multisig pubkey exists in the store. + ErrNoSidecar = errors.New("no sidecar found") + + // sidecarsBucketKey is a bucket that contains all sidecars that are + // currently pending or completed. This bucket is keyed by the ticket ID + // and offer signing pubkey of a sidecar. + sidecarsBucketKey = []byte("sidecars") +) + +const ( + // sidecarKeyLen is the length of a sidecar ticket's key. It is the + // length of the sidecar ID (8 bytes) plus the length of a compressed + // public key (33 bytes). + sidecarKeyLen = 8 + 33 +) + +// A compile time check to make sure we satisfy the sidecar.Store interface. +var _ sidecar.Store = (*DB)(nil) + +// getSidecarKey returns the key for a sidecar. +func getSidecarKey(id [8]byte, offerSignPubKey *btcec.PublicKey) ([]byte, + error) { + + if offerSignPubKey == nil { + return nil, fmt.Errorf("offer signing pubkey cannot be nil") + } + + var result [sidecarKeyLen]byte + + copy(result[:], id[:]) + copy(result[8:], offerSignPubKey.SerializeCompressed()) + + return result[:], nil +} + +// AddSidecar adds a record for the sidecar to the database. +func (db *DB) AddSidecar(ticket *sidecar.Ticket) error { + sidecarKey, err := getSidecarKey(ticket.ID, ticket.Offer.SignPubKey) + if err != nil { + return err + } + + return db.Update(func(tx *bbolt.Tx) error { + sidecarBucket, err := getBucket(tx, sidecarsBucketKey) + if err != nil { + return err + } + + sidecarValue := sidecarBucket.Get(sidecarKey) + if len(sidecarValue) != 0 { + return fmt.Errorf("sidecar for key %x already exists", + sidecarKey) + } + + return storeSidecar(sidecarBucket, sidecarKey, ticket) + }) +} + +// UpdateSidecar updates a sidecar in the database. +func (db *DB) UpdateSidecar(ticket *sidecar.Ticket) error { + sidecarKey, err := getSidecarKey(ticket.ID, ticket.Offer.SignPubKey) + if err != nil { + return err + } + + return db.Update(func(tx *bbolt.Tx) error { + sidecarBucket, err := getBucket(tx, sidecarsBucketKey) + if err != nil { + return err + } + + sidecarValue := sidecarBucket.Get(sidecarKey) + if len(sidecarValue) == 0 { + return ErrNoSidecar + } + + return storeSidecar(sidecarBucket, sidecarKey, ticket) + }) +} + +// Sidecar retrieves a specific sidecar by its ID and provider signing key +// (offer signature pubkey) or returns ErrNoSidecar if it's not found. +func (db *DB) Sidecar(id [8]byte, + offerSignPubKey *btcec.PublicKey) (*sidecar.Ticket, error) { + + sidecarKey, err := getSidecarKey(id, offerSignPubKey) + if err != nil { + return nil, err + } + + var s *sidecar.Ticket + err = db.View(func(tx *bbolt.Tx) error { + sidecarBucket, err := getBucket(tx, sidecarsBucketKey) + if err != nil { + return err + } + + s, err = readSidecar(sidecarBucket, sidecarKey) + return err + }) + if err != nil { + return nil, err + } + + return s, nil +} + +// Sidecars retrieves all known sidecars from the database. +func (db *DB) Sidecars() ([]*sidecar.Ticket, error) { + var res []*sidecar.Ticket + err := db.View(func(tx *bbolt.Tx) error { + sidecarBucket, err := getBucket(tx, sidecarsBucketKey) + if err != nil { + return err + } + + return sidecarBucket.ForEach(func(k, v []byte) error { + // We don't expect any sub-buckets with sidecars. + if v == nil { + return fmt.Errorf("nil value for key %x", k) + } + + s, err := readSidecar(sidecarBucket, k) + if err != nil { + return err + } + res = append(res, s) + + return nil + }) + }) + if err != nil { + return nil, err + } + + return res, nil +} + +func storeSidecar(targetBucket *bbolt.Bucket, key []byte, + ticket *sidecar.Ticket) error { + + var sidecarBuf bytes.Buffer + if err := sidecar.SerializeTicket(&sidecarBuf, ticket); err != nil { + return err + } + + return targetBucket.Put(key, sidecarBuf.Bytes()) +} + +func readSidecar(sourceBucket *bbolt.Bucket, id []byte) (*sidecar.Ticket, + error) { + + sidecarBytes := sourceBucket.Get(id) + if sidecarBytes == nil { + return nil, ErrNoSidecar + } + + return sidecar.DeserializeTicket(bytes.NewReader(sidecarBytes)) +} diff --git a/clientdb/sidecar_test.go b/clientdb/sidecar_test.go new file mode 100644 index 0000000..a9b7c0b --- /dev/null +++ b/clientdb/sidecar_test.go @@ -0,0 +1,72 @@ +package clientdb + +import ( + "testing" + + "github.com/lightninglabs/pool/order" + "github.com/lightninglabs/pool/sidecar" + "github.com/stretchr/testify/require" +) + +func assertSidecarExists(t *testing.T, db *DB, expected *sidecar.Ticket) { + t.Helper() + + found, err := db.Sidecar(expected.ID, expected.Offer.SignPubKey) + require.NoError(t, err) + + require.Equal(t, expected, found) +} + +// TestSidecars ensures that all database operations involving sidecars run as +// expected. +func TestSidecars(t *testing.T) { + t.Parallel() + + db, cleanup := newTestDB(t) + defer cleanup() + + // Create a test sidecar we'll use to interact with the database. + s := &sidecar.Ticket{ + ID: [8]byte{12, 34, 56}, + State: sidecar.StateRegistered, + Offer: sidecar.Offer{ + Capacity: 1000000, + PushAmt: 200000, + SignPubKey: testTraderKey, + LeaseDurationBlocks: 2016, + }, + Recipient: &sidecar.Recipient{ + MultiSigPubKey: testTraderKey, + MultiSigKeyIndex: 7, + }, + } + + // First, we'll add it to the database. We should be able to retrieve + // after. + err := db.AddSidecar(s) + require.NoError(t, err) + assertSidecarExists(t, db, s) + + // Transition the sidecar state from SidecarInitialized to + // SidecarExpectingChannel and add the required information for that + // state. + s.State = sidecar.StateExpectingChannel + s.Order = &sidecar.Order{ + BidNonce: order.Nonce{1, 2, 3}, + } + err = db.UpdateSidecar(s) + require.NoError(t, err) + assertSidecarExists(t, db, s) + + // Retrieving all sidecars should show that we only have one sidecar, + // the same one. + sidecars, err := db.Sidecars() + require.NoError(t, err) + require.Len(t, sidecars, 1) + require.Contains(t, sidecars, s) + + // Make sure we can query a sidecar ticket by its ID and offer pubkey. + updatedTicket, err := db.Sidecar([8]byte{12, 34, 56}, testTraderKey) + require.NoError(t, err) + require.Equal(t, s, updatedTicket) +} diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md index 452d7cf..973d5e4 100644 --- a/docs/SUMMARY.md +++ b/docs/SUMMARY.md @@ -8,5 +8,6 @@ * [Orders](orders.md) * [Channel Leases](channel_leases.md) * [Batch Execution](batch_execution.md) +* [Sidecar Channels](sidecar_channels.md) * [FAQs](faq.md) diff --git a/docs/sidecar_channels.md b/docs/sidecar_channels.md new file mode 100644 index 0000000..995d3bd --- /dev/null +++ b/docs/sidecar_channels.md @@ -0,0 +1,94 @@ +# Sidecar Channels + +## Goal + +Alice is a new user of LN and has no UTXOs but wants to use Lightning. She’s +willing to pay an onboarding fee to a wallet provider to get set up with both an +initial balance (=outbound capacity) and the ability to receive payments +(=inbound capacity). + +The example in this document assumes that Alice is aware of the costs of +onboarding and is paying for them herself (which happens out of band from the +perspective of the Pool server and is only mentioned explicitly for the +completeness of the example). + +With the APIs proposed to be implemented in the Pool server and client, it +should also be possible for a wallet provider to implement the flow in a way +that they cover the full costs of the onboarding and therefore “gift” the +channel and the balance to Alice. That would hide some of the parameters the API +needs to work, therefore we show the more involved example here. + +## Participants + +* Alice: New user, possibly using a mobile wallet, has no UTXOs but a wallet app + that has lnd and Pool integrated as part of a unified (mobile) binary (only + runs the Pool daemon, possibly even a "Pool light" daemon, does not need her + own Pool account). +* Charlie: A wallet/service provider that offers the sidecar bootstrap service + for a fee. Offers to convert fiat or BTC into an initial LN balance for new + users, minus said fee. Has a well funded Pool account that can be used to pay + for leases. +* Bob: An independent liquidity provider that is willing to open channels to new + users. Has a well funded Pool account that can be used to open channels. + +## Requirements + +* All three participants need to run the Pool client that implements the sidecar + functionality: + * Alice: Must be able to register with the auctioneer by providing the + multisig pubkey and be able to sign with it (explained later on). + * Charlie: Must be able to submit an order that refers to a multisig pubkey + of the receiver node. + * Bob: Must be able to accept orders that require to set a push amount when + opening the channel. +* No new functionality in lnd is required (though can be optimized later by + replacing the channel acceptor with a new push amount parameter in the funding + shim). + +## Example flow (command line only) + +1. Precondition: An agreement between Alice and Charlie is reached out-of-band. + For this example, we assume Charlie is going to buy **10 units** (1m satoshi) + of capacity from Pool for Alice. Of those 10 units, **2 units will be + pushed** to Alice for her to gain outbound liquidity as well. + The cost for the order submission fees, chain fees, the premium and the push + amount are all paid from Charlie's Pool account. Whether Alice reimburses + Charlie for those costs or not is not part of the protocol and irrelevant for + this example. +1. ```shell + charlie$ pool sidecar offer --capacity 1000000 --self_chan_balance 200000 + { + "sidecar_ticket": "sidecarAAQgHBgUEAwIBAAMBYwUBAgdxAAAAAAAAAwkAAAAAAA...." + } + ``` + The sidecar ticket now contains the offer from Charlie to buy a 1m satoshi + channel with a 200k initial self channel balance (=push amount) for Alice. +1. Alice now needs to add her node's information: + ```shell + alice$ pool sidecar register sidecarAAQgHBgUEAwIBAAMBYwUBAgdxAAAAA.... + { + "sidecar_ticket": "sidecarAAQgHBgUEAwIBAAMBYwUBAgdxAAAAAAAAAwkAAAAAAA...." + } + ``` + The sidecar ticket now contains the `Recipient` field which encodes Alice's + node pubkey and pubkey that'll be used for the channel funding. + + Rationale: Alice will be counter-signing the commitment transactions for the + new channel so she needs to have the private key for her 2-of-2 multisig + channel funding key. We derive that from her keychain. +1. Charlie can now create the bid order with the updated ticket he got from + Alice: + ```shell + charlie$ pool orders submit bid --acct_key \ + --interest_rate_percent xxx --sidecar_ticket sidecarAAQgHBgUEAwIBAAMB... + { + "order_nonce": "0011223344...", + "sidecar_ticket": "sidecarAAQgHBgUEAwIBAAMBYwUBAgdxAAAAAAAAAwkAAAAAAA..." + } + ``` + The sidecar ticket now contains the order's nonce in the `Order` part. + Charlie can give the final version of the ticket back to Alice. +1. Alice instructs her node to start expecting a channel. + ```shell + alice$ pool sidecar expect-channel sidecarAAQgHBgUEAwIBAAMBYwUBAg... + ``` diff --git a/funding/manager.go b/funding/manager.go index d7f2170..302fd49 100644 --- a/funding/manager.go +++ b/funding/manager.go @@ -4,9 +4,11 @@ import ( "context" "encoding/hex" "fmt" + "io" "net" "strings" "sync" + "sync/atomic" "time" "github.com/btcsuite/btcd/chaincfg/chainhash" @@ -20,6 +22,7 @@ import ( "github.com/lightningnetwork/lnd/input" "github.com/lightningnetwork/lnd/lnrpc" "github.com/lightningnetwork/lnd/routing/route" + "github.com/lightningnetwork/lnd/subscribe" "github.com/lightningnetwork/lnd/tor" "golang.org/x/sync/errgroup" "google.golang.org/grpc" @@ -27,6 +30,12 @@ import ( "google.golang.org/grpc/status" ) +var ( + // rpcCodeFundingFailed is the error code we send if the channel funding + // fails because of a timeout or another problem. + rpcCodeFundingFailed = auctioneerrpc.OrderReject_CHANNEL_FUNDING_FAILED +) + // MatchRejectErr is an error type that is returned from the funding manager if // the trader rejects certain orders instead of the whole batch. type MatchRejectErr struct { @@ -76,9 +85,18 @@ type BaseClient interface { // non-externally funded channels in dev build. AbandonChannel(ctx context.Context, in *lnrpc.AbandonChannelRequest, opts ...grpc.CallOption) (*lnrpc.AbandonChannelResponse, error) + + // SubscribeChannelEvents creates a uni-directional stream from the + // server to the client in which any updates relevant to the state of + // the channels are sent over. Events include new active channels, + // inactive channels, and closed channels. + SubscribeChannelEvents(ctx context.Context, + in *lnrpc.ChannelEventSubscription, opts ...grpc.CallOption) ( + lnrpc.Lightning_SubscribeChannelEventsClient, error) } -type Manager struct { +// ManagerConfig holds all the items passed into the funding manager externally. +type ManagerConfig struct { // DB is the client database. DB *clientdb.DB @@ -97,11 +115,6 @@ type Manager struct { // node doesn't already have channels with. NewNodesOnly bool - // PendingOpenChannels is a channel through which we'll receive - // notifications for pending open channels resulting from a successful - // batch. - PendingOpenChannels chan *lnrpc.ChannelEventUpdate_PendingOpenChannel - // BatchStepTimeout is the timeout the manager uses when executing a // single batch step. BatchStepTimeout time.Duration @@ -112,6 +125,151 @@ type Manager struct { NotifyShimCreated func(ourBid *order.Bid, pendingChanID [32]byte) } +// Manager is responsible for everything channel funding related during the +// match making process. +type Manager struct { + started uint32 // To be used atomically. + stopped uint32 // To be used atomically. + + cfg *ManagerConfig + + wg sync.WaitGroup + quit chan struct{} + + pendingOpenChanCancel func() + pendingOpenChanServer *subscribe.Server + pendingOpenChanClient *subscribe.Client +} + +// NewManager creates a new funding manager from the given config. +func NewManager(cfg *ManagerConfig) *Manager { + return &Manager{ + cfg: cfg, + quit: make(chan struct{}), + pendingOpenChanServer: subscribe.NewServer(), + } +} + +// Start starts the rpcServer, making it ready to accept incoming requests. +func (m *Manager) Start() error { + if !atomic.CompareAndSwapUint32(&m.started, 0, 1) { + return nil + } + + log.Infof("Starting funding manager") + + // Subscribe to pending open channel notifications. This will be useful + // when we're creating channels with a matched order as part of a batch. + streamCtx, streamCancel := context.WithCancel(context.Background()) + m.pendingOpenChanCancel = streamCancel + subStream, err := m.cfg.BaseClient.SubscribeChannelEvents( + streamCtx, &lnrpc.ChannelEventSubscription{}, + ) + if err != nil { + return err + } + + if err := m.pendingOpenChanServer.Start(); err != nil { + return fmt.Errorf("error starting pending chan subscription "+ + "server: %v", err) + } + + // We want to make sure we don't miss any channel updates as long as we + // are running. But we might not be the only manager interested in the + // updates, that's why we are a client to our own server. + m.pendingOpenChanClient, err = m.SubscribePendingOpenChan() + if err != nil { + return fmt.Errorf("error subscribing to pending open "+ + "channel events: %v", err) + } + + m.wg.Add(1) + go m.consumePendingOpenChannels(subStream) + + log.Infof("Funding manager is now active") + + return nil +} + +// Stop stops the server. +func (m *Manager) Stop() error { + if !atomic.CompareAndSwapUint32(&m.stopped, 0, 1) { + return nil + } + + log.Info("Funding manager stopping") + + close(m.quit) + + // We call this before Wait to ensure the goroutine is stopped by this + // call. + m.pendingOpenChanClient.Cancel() + if err := m.pendingOpenChanServer.Stop(); err != nil { + return fmt.Errorf("error stopping pending chan subscription "+ + "server: %v", err) + } + m.pendingOpenChanCancel() + + m.wg.Wait() + + log.Info("Stopped funding manager") + + return nil +} + +// consumePendingOpenChannels consumes pending open channel events from the +// stream and notifies them if the trader currently has an ongoing batch. +func (m *Manager) consumePendingOpenChannels( + subStream lnrpc.Lightning_SubscribeChannelEventsClient) { + + defer m.wg.Done() + + for { + select { + case <-m.quit: + return + default: + } + + msg, err := subStream.Recv() + if err != nil { + select { + case <-m.quit: + return + default: + } + + log.Errorf("Unable to read channel event: %v", err) + + // If the lnd node shut down, there's no use continuing. + if err == io.EOF || err == io.ErrUnexpectedEOF || + status.Code(err) == codes.Unavailable { + + return + } + + continue + } + + // Skip any events other than the pending open channel one. + channel, ok := msg.Channel.(*lnrpc.ChannelEventUpdate_PendingOpenChannel) + if !ok { + continue + } + + update := channel.PendingOpenChannel + if err := m.pendingOpenChanServer.SendUpdate(update); err != nil { + log.Errorf("Error sending open channel update: %v", err) + } + } +} + +// SubscribePendingOpenChan creates a new subscription client to receive events +// for pending open channels from lnd. +func (m *Manager) SubscribePendingOpenChan() (*subscribe.Client, error) { + return m.pendingOpenChanServer.Subscribe() +} + // deriveFundingShim generates the proper funding shim that should be used by // the maker or taker to properly make a channel that stems off the main batch // funding transaction. @@ -155,7 +313,7 @@ func (m *Manager) deriveFundingShim(ourOrder order.Order, // scratch. ctxb := context.Background() ourKeyLocator := ourOrder.Details().MultiSigKeyLocator - ourMultiSigKey, err := m.WalletKit.DeriveKey( + ourMultiSigKey, err := m.cfg.WalletKit.DeriveKey( ctxb, &ourKeyLocator, ) if err != nil { @@ -223,7 +381,7 @@ func (m *Manager) registerFundingShim(ourBid *order.Bid, if err != nil { return err } - _, err = m.BaseClient.FundingStateStep( + _, err = m.cfg.BaseClient.FundingStateStep( ctxb, &lnrpc.FundingTransitionMsg{ Trigger: &lnrpc.FundingTransitionMsg_ShimRegister{ ShimRegister: fundingShim, @@ -237,8 +395,8 @@ func (m *Manager) registerFundingShim(ourBid *order.Bid, // In case there is a self chan balance involved, we need our channel // acceptor to be aware of the incoming order so it can verify the push // amount accordingly. - if m.NotifyShimCreated != nil { - m.NotifyShimCreated(ourBid, pendingChanID) + if m.cfg.NotifyShimCreated != nil { + m.cfg.NotifyShimCreated(ourBid, pendingChanID) } return nil @@ -247,14 +405,14 @@ func (m *Manager) registerFundingShim(ourBid *order.Bid, // PrepChannelFunding preps the backing node to either receive or initiate a // channel funding based on the items in the order batch. func (m *Manager) PrepChannelFunding(batch *order.Batch, - quit <-chan struct{}) error { + getOrder order.Fetcher) error { log.Infof("Batch(%x): preparing channel funding for %v orders", batch.ID[:], len(batch.MatchedOrders)) // As we need to change our behavior if the node has any Tor addresses, // we'll fetch the current state of our advertised addrs now. - nodeInfo, err := m.LightningClient.GetInfo(context.Background()) + nodeInfo, err := m.cfg.LightningClient.GetInfo(context.Background()) if err != nil { log.Errorf("error in GetInfo: %v", err) return err @@ -265,14 +423,14 @@ func (m *Manager) PrepChannelFunding(batch *order.Batch, // so we create a context that is valid for the whole funding step and // use that everywhere. setupCtx, cancel := context.WithTimeout( - context.Background(), m.BatchStepTimeout, + context.Background(), m.cfg.BatchStepTimeout, ) defer cancel() // Before we connect out to peers, we check that we don't get any new // channels from peers we already have channels with, in case this is // requested by the trader. - if m.NewNodesOnly { + if m.cfg.NewNodesOnly { fundingRejects, err := m.rejectDuplicateChannels(batch) if err != nil { return err @@ -291,7 +449,7 @@ func (m *Manager) PrepChannelFunding(batch *order.Batch, // all the funding shims we need to be able to respond connsInitiated := make(map[route.Vertex]struct{}) for ourOrderNonce, matchedOrders := range batch.MatchedOrders { - ourOrder, err := m.DB.GetOrder(ourOrderNonce) + ourOrder, err := getOrder(ourOrderNonce) if err != nil { return err } @@ -372,7 +530,7 @@ func (m *Manager) PrepChannelFunding(batch *order.Batch, // We need to wait for all connections to be established now. Otherwise // the asker won't be able to open the channel as it doesn't know the // connection details of the bidder. - return m.waitForPeerConnections(setupCtx, connsInitiated, batch, quit) + return m.waitForPeerConnections(setupCtx, connsInitiated, batch) } // BatchChannelSetup will attempt to establish new funding flows with all @@ -380,8 +538,8 @@ func (m *Manager) PrepChannelFunding(batch *order.Batch, // will block until the channel is considered pending. Once this phase is // complete, and the batch execution transaction broadcast, the channel will be // finalized and locked in. -func (m *Manager) BatchChannelSetup(batch *order.Batch, - quit <-chan struct{}) (map[wire.OutPoint]*chaninfo.ChannelInfo, error) { +func (m *Manager) BatchChannelSetup( + batch *order.Batch) (map[wire.OutPoint]*chaninfo.ChannelInfo, error) { var ( eg errgroup.Group @@ -396,7 +554,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, defer fundingRejectsMtx.Unlock() fundingRejects[nonce] = &auctioneerrpc.OrderReject{ - ReasonCode: auctioneerrpc.OrderReject_CHANNEL_FUNDING_FAILED, + ReasonCode: rpcCodeFundingFailed, Reason: reason, } @@ -410,7 +568,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, // so we create a context that is valid for the whole funding step and // use that everywhere. setupCtx, cancel := context.WithTimeout( - context.Background(), m.BatchStepTimeout, + context.Background(), m.cfg.BatchStepTimeout, ) defer cancel() @@ -421,7 +579,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, // flow, blocking until they all progress to the final state. batchTxHash := batch.BatchTX.TxHash() for ourOrderNonce, matchedOrders := range batch.MatchedOrders { - ourOrder, err := m.DB.GetOrder(ourOrderNonce) + ourOrder, err := m.cfg.DB.GetOrder(ourOrderNonce) if err != nil { return nil, err } @@ -480,7 +638,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, matchedOrderBid.SelfChanBalance, ), } - chanStream, err := m.BaseClient.OpenChannel( + chanStream, err := m.cfg.BaseClient.OpenChannel( setupCtx, fundingReq, ) @@ -510,7 +668,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, for { select { - case <-quit: + case <-m.quit: return fmt.Errorf("server " + "shutting down") default: @@ -549,12 +707,28 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, // we can report that together with the other errors. if err := eg.Wait(); err != nil { select { - case <-quit: + case <-m.quit: return nil, err default: } } + // We've kicked off all channel open streams. We'll now need to wait for + // either completion of the funding process or a timeout. + return m.waitForChannelOpen( + setupCtx, chanPoints, m.pendingOpenChanClient, fundingRejects, + ) +} + +// waitForChannelOpen waits until we get a pending open channel message for each +// of the channel outpoints provided. If we don't get all of them in time, we'll +// instead return a reject error with all the orders for which the funding +// failed. +func (m *Manager) waitForChannelOpen(ctx context.Context, + chanPoints map[wire.OutPoint]order.Nonce, chanUpdates *subscribe.Client, + fundingRejects map[order.Nonce]*auctioneerrpc.OrderReject) ( + map[wire.OutPoint]*chaninfo.ChannelInfo, error) { + // Once we've waited for the operations to complete, we'll wait to // receive each channel's pending open notification in order to retrieve // some keys from their SCB we'll need to submit to the auctioneer in @@ -571,29 +745,40 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, for { var chanPoint wire.OutPoint select { - case channel := <-m.PendingOpenChannels: - var hash chainhash.Hash - copy(hash[:], channel.PendingOpenChannel.Txid) - chanPoint = wire.OutPoint{ - Hash: hash, - Index: channel.PendingOpenChannel.OutputIndex, + case channel := <-chanUpdates.Updates(): + pendingChannel, ok := channel.(*lnrpc.PendingUpdate) + if !ok || pendingChannel == nil { + continue } - case <-setupCtx.Done(): + var hash chainhash.Hash + copy(hash[:], pendingChannel.Txid) + chanPoint = wire.OutPoint{ + Hash: hash, + Index: pendingChannel.OutputIndex, + } + + case <-ctx.Done(): // One or more of our peers timed out. At this point the // chanPoints map should only contain channels that // failed/timed out. So we can partially reject all of // them. const timeoutErrStr = "timed out waiting for pending " + "open channel notification" - for chanPoint, nonce := range chanPoints { - partialReject(nonce, timeoutErrStr, chanPoint) + for _, nonce := range chanPoints { + fundingRejects[nonce] = &auctioneerrpc.OrderReject{ + ReasonCode: rpcCodeFundingFailed, + Reason: timeoutErrStr, + } } return nil, &MatchRejectErr{ RejectedOrders: fundingRejects, } - case <-quit: + case <-m.pendingOpenChanClient.Quit(): + return nil, fmt.Errorf("server shutting down") + + case <-m.quit: return nil, fmt.Errorf("server shutting down") } @@ -607,7 +792,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, log.Debugf("Retrieving info for channel %v", chanPoint) chanInfo, err := chaninfo.GatherChannelInfo( - setupCtx, m.LightningClient, m.WalletKit, chanPoint, + ctx, m.cfg.LightningClient, m.cfg.WalletKit, chanPoint, ) if err != nil { return nil, err @@ -639,7 +824,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, // // NOTE: This is part of the auctioneer.BatchCleaner interface. func (m *Manager) DeletePendingBatch() error { - return m.DB.DeletePendingBatch() + return m.cfg.DB.DeletePendingBatch() } // RemovePendingBatchArtifacts removes any funding shims or pending channels @@ -653,7 +838,7 @@ func (m *Manager) RemovePendingBatchArtifacts( batchTx *wire.MsgTx) error { err := CancelPendingFundingShims( - matchedOrders, m.BaseClient, m.DB.GetOrder, + matchedOrders, m.cfg.BaseClient, m.cfg.DB.GetOrder, ) if err != nil { // CancelPendingFundingShims only returns hard errors that @@ -666,8 +851,8 @@ func (m *Manager) RemovePendingBatchArtifacts( // from a previous round of the same batch or a previous // batch that we didn't make it into the final round. err = AbandonCanceledChannels( - matchedOrders, batchTx, m.WalletKit, m.BaseClient, - m.DB.GetOrder, + matchedOrders, batchTx, m.cfg.WalletKit, m.cfg.BaseClient, + m.cfg.DB.GetOrder, ) if err != nil { // AbandonCanceledChannels also only returns hard errors that @@ -685,7 +870,7 @@ func (m *Manager) connectToMatchedTrader(ctx context.Context, nodeKey [33]byte, addrs []net.Addr) { for _, addr := range addrs { - err := m.LightningClient.Connect( + err := m.cfg.LightningClient.Connect( ctx, nodeKey, addr.String(), false, ) if err != nil { @@ -714,14 +899,13 @@ func (m *Manager) connectToMatchedTrader(ctx context.Context, // NOTE: The passed context MUST have a timeout applied to it, otherwise this // method will block forever in case a connection doesn't succeed. func (m *Manager) waitForPeerConnections(ctx context.Context, - peers map[route.Vertex]struct{}, batch *order.Batch, - quit <-chan struct{}) error { + peers map[route.Vertex]struct{}, batch *order.Batch) error { // First of all, subscribe to new peer events so we certainly don't miss // an update while we look for the already connected peers. ctxc, cancel := context.WithCancel(ctx) defer cancel() - subscription, err := m.BaseClient.SubscribePeerEvents( + subscription, err := m.cfg.BaseClient.SubscribePeerEvents( ctxc, &lnrpc.PeerEventSubscription{}, ) if err != nil { @@ -731,7 +915,7 @@ func (m *Manager) waitForPeerConnections(ctx context.Context, // Query all connected peers. This only returns active peers so once a // node key appears in this list, we can be reasonably sure the // connection is established (flapping peers notwithstanding). - resp, err := m.BaseClient.ListPeers( + resp, err := m.cfg.BaseClient.ListPeers( ctx, &lnrpc.ListPeersRequest{}, ) if err != nil { @@ -771,7 +955,7 @@ func (m *Manager) waitForPeerConnections(ctx context.Context, } // We're shutting down, nothing more to do here. - case <-quit: + case <-m.quit: return nil default: @@ -828,11 +1012,11 @@ func (m *Manager) rejectDuplicateChannels( // We gather all peers from the open and pending channels. ctxb := context.Background() peers := make(map[route.Vertex]struct{}) - openChans, err := m.LightningClient.ListChannels(ctxb) + openChans, err := m.cfg.LightningClient.ListChannels(ctxb) if err != nil { return nil, fmt.Errorf("error listing open channels: %v", err) } - pendingChans, err := m.LightningClient.PendingChannels(ctxb) + pendingChans, err := m.cfg.LightningClient.PendingChannels(ctxb) if err != nil { return nil, fmt.Errorf("error listing pending channels: %v", err) diff --git a/funding/manager_test.go b/funding/manager_test.go index 98a439c..1111c44 100644 --- a/funding/manager_test.go +++ b/funding/manager_test.go @@ -85,13 +85,44 @@ func (i *peerEventStream) Recv() (*lnrpc.PeerEvent, error) { } } +type channelEventStream struct { + lnrpc.Lightning_SubscribeChannelEventsClient + + updateChan chan *lnrpc.ChannelEventUpdate + ctx context.Context + cancelSubscription chan struct{} + quit chan struct{} +} + +func (c *channelEventStream) Recv() (*lnrpc.ChannelEventUpdate, error) { + select { + case msg := <-c.updateChan: + return msg, nil + + // To mimic the real GRPC client, we'll return an error status. + case <-c.ctx.Done(): + return nil, status.Error( + codes.DeadlineExceeded, c.ctx.Err().Error(), + ) + + case <-c.cancelSubscription: + return nil, status.Error( + codes.Canceled, "subscription canceled", + ) + + case <-c.quit: + return nil, context.Canceled + } +} + type fundingBaseClientMock struct { lightningClient *test.MockLightning fundingShims map[[32]byte]*lnrpc.ChanPointShim - peerList map[route.Vertex]string - peerEvents chan *lnrpc.PeerEvent - cancelSub chan struct{} + peerList map[route.Vertex]string + peerEvents chan *lnrpc.PeerEvent + channelEvents chan *lnrpc.ChannelEventUpdate + cancelSub chan struct{} quit chan struct{} } @@ -157,6 +188,17 @@ func (m *fundingBaseClientMock) OpenChannel(ctx context.Context, return stream, nil } +func (m *fundingBaseClientMock) SubscribeChannelEvents(ctx context.Context, + _ *lnrpc.ChannelEventSubscription, _ ...grpc.CallOption) ( + lnrpc.Lightning_SubscribeChannelEventsClient, error) { + + return &channelEventStream{ + quit: m.quit, + ctx: ctx, + updateChan: m.channelEvents, + }, nil +} + func (m *fundingBaseClientMock) ListPeers(_ context.Context, _ *lnrpc.ListPeersRequest, _ ...grpc.CallOption) (*lnrpc.ListPeersResponse, error) { @@ -196,7 +238,6 @@ type managerHarness struct { tempDir string db *clientdb.DB quit chan struct{} - msgChan chan *lnrpc.ChannelEventUpdate_PendingOpenChannel lnMock *test.MockLightning baseClientMock *fundingBaseClientMock mgr *Manager @@ -213,7 +254,6 @@ func newManagerHarness(t *testing.T) *managerHarness { } quit := make(chan struct{}) - msgChan := make(chan *lnrpc.ChannelEventUpdate_PendingOpenChannel) lightningClient := test.NewMockLightning() walletKitClient := test.NewMockWalletKit() baseClientMock := &fundingBaseClientMock{ @@ -221,26 +261,30 @@ func newManagerHarness(t *testing.T) *managerHarness { fundingShims: make(map[[32]byte]*lnrpc.ChanPointShim), peerList: make(map[route.Vertex]string), peerEvents: make(chan *lnrpc.PeerEvent), + channelEvents: make(chan *lnrpc.ChannelEventUpdate), cancelSub: make(chan struct{}), quit: quit, } + mgr := NewManager(&ManagerConfig{ + DB: db, + WalletKit: walletKitClient, + LightningClient: lightningClient, + BaseClient: baseClientMock, + NewNodesOnly: true, + BatchStepTimeout: 400 * time.Millisecond, + }) + + err = mgr.Start() + require.NoError(t, err) + return &managerHarness{ t: t, tempDir: tempDir, db: db, quit: quit, - msgChan: msgChan, lnMock: lightningClient, baseClientMock: baseClientMock, - mgr: &Manager{ - DB: db, - WalletKit: walletKitClient, - LightningClient: lightningClient, - BaseClient: baseClientMock, - NewNodesOnly: true, - PendingOpenChannels: msgChan, - BatchStepTimeout: 400 * time.Millisecond, - }, + mgr: mgr, } } @@ -248,6 +292,7 @@ func (m *managerHarness) stop() { close(m.quit) require.NoError(m.t, m.db.Close()) require.NoError(m.t, os.RemoveAll(m.tempDir)) + require.NoError(m.t, m.mgr.Stop()) } // TestFundingManager tests that the two main steps of the funding manager (the @@ -335,7 +380,7 @@ func TestFundingManager(t *testing.T) { h.baseClientMock.peerList = map[route.Vertex]string{ node1Key: "1.1.1.1", } - err = h.mgr.PrepChannelFunding(batch, h.quit) + err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder) require.NoError(t, err) // Verify we have the expected connections and funding shims registered. @@ -377,11 +422,11 @@ func TestFundingManager(t *testing.T) { // Next, make sure we get a partial reject error if we enable the "new // nodes only" flag and already have a channel with the matched node. - h.mgr.NewNodesOnly = true + h.mgr.cfg.NewNodesOnly = true h.lnMock.Channels = append(h.lnMock.Channels, lndclient.ChannelInfo{ PubKeyBytes: node1Key, }) - err = h.mgr.PrepChannelFunding(batch, h.quit) + err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder) require.Error(t, err) expectedErr := &MatchRejectErr{ @@ -397,9 +442,9 @@ func TestFundingManager(t *testing.T) { // As a last check of the funding preparation, make sure we get a reject // error if the connections to the remote peers couldn't be established. - h.mgr.NewNodesOnly = false + h.mgr.cfg.NewNodesOnly = false h.baseClientMock.peerList = make(map[route.Vertex]string) - err = h.mgr.PrepChannelFunding(batch, h.quit) + err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder) require.Error(t, err) expectedErr = &MatchRejectErr{ @@ -418,27 +463,31 @@ func TestFundingManager(t *testing.T) { // message for the one where we are the asker so we simulate two msgs. go func() { timeout := time.After(time.Second) - msg := &lnrpc.ChannelEventUpdate_PendingOpenChannel{ - PendingOpenChannel: &lnrpc.PendingUpdate{ - Txid: txidHash[:], - OutputIndex: 0, + msg := &lnrpc.ChannelEventUpdate{ + Channel: &lnrpc.ChannelEventUpdate_PendingOpenChannel{ + PendingOpenChannel: &lnrpc.PendingUpdate{ + Txid: txidHash[:], + OutputIndex: 0, + }, }, } // Send the message for the first channel. select { - case h.msgChan <- msg: + case h.baseClientMock.channelEvents <- msg: case <-timeout: } // And again for the second channel. - msg2 := &lnrpc.ChannelEventUpdate_PendingOpenChannel{ - PendingOpenChannel: &lnrpc.PendingUpdate{ - Txid: txidHash[:], - OutputIndex: 1, + msg2 := &lnrpc.ChannelEventUpdate{ + Channel: &lnrpc.ChannelEventUpdate_PendingOpenChannel{ + PendingOpenChannel: &lnrpc.PendingUpdate{ + Txid: txidHash[:], + OutputIndex: 1, + }, }, } select { - case h.msgChan <- msg2: + case h.baseClientMock.channelEvents <- msg2: case <-timeout: } }() @@ -452,13 +501,13 @@ func TestFundingManager(t *testing.T) { ChannelPoint: fmt.Sprintf("%s:1", txidHash.String()), }) h.lnMock.ScbKeyRing.EncryptionKey.PubKey = pubKeyAsk - chanInfo, err := h.mgr.BatchChannelSetup(batch, h.quit) + chanInfo, err := h.mgr.BatchChannelSetup(batch) require.NoError(t, err) require.Equal(t, 2, len(chanInfo)) // Finally, make sure we get a timeout error if no channel open messages // are received. - _, err = h.mgr.BatchChannelSetup(batch, h.quit) + _, err = h.mgr.BatchChannelSetup(batch) require.Error(t, err) code := &auctioneerrpc.OrderReject{ @@ -493,9 +542,7 @@ func TestWaitForPeerConnections(t *testing.T) { node1Key: {}, node2Key: {}, } - err := h.mgr.waitForPeerConnections( - ctxt, expectedConnections, nil, h.quit, - ) + err := h.mgr.waitForPeerConnections(ctxt, expectedConnections, nil) require.NoError(t, err) // Next, make sure that connections established while waiting are @@ -524,9 +571,7 @@ func TestWaitForPeerConnections(t *testing.T) { node1Key: {}, node2Key: {}, } - err = h.mgr.waitForPeerConnections( - ctxt, expectedConnections, nil, h.quit, - ) + err = h.mgr.waitForPeerConnections(ctxt, expectedConnections, nil) require.NoError(t, err) // Final test, make sure we get the correct error message back if we @@ -561,9 +606,7 @@ func TestWaitForPeerConnections(t *testing.T) { node1Key: {}, node2Key: {}, } - err = h.mgr.waitForPeerConnections( - ctxt, expectedConnections, fakeBatch, h.quit, - ) + err = h.mgr.waitForPeerConnections(ctxt, expectedConnections, fakeBatch) require.Error(t, err) code := &auctioneerrpc.OrderReject{ @@ -588,9 +631,7 @@ func TestWaitForPeerConnections(t *testing.T) { close(h.baseClientMock.cancelSub) }() - err = h.mgr.waitForPeerConnections( - ctxt, expectedConnections, fakeBatch, h.quit, - ) + err = h.mgr.waitForPeerConnections(ctxt, expectedConnections, fakeBatch) require.Error(t, err) require.Equal(t, expectedErr, err) } diff --git a/order/interface.go b/order/interface.go index 158f2ff..fbd7a82 100644 --- a/order/interface.go +++ b/order/interface.go @@ -49,6 +49,16 @@ const ( // self channel balance field. Only orders with this version are allowed // to use the self channel balance field. VersionSelfChanBalance Version = 3 + + // VersionSidecarChannel is the order version that added sidecar + // channels for bid orders. Only orders with this version are allowed + // to set the sidecar ticket field on bid orders. Since sidecar orders + // also add the feature of push amounts on the leased channels, this + // affects makers as well. Makers that don't want to support leasing out + // channels with a push amount (because it might screw up their + // accounting or whatever) can opt out by explicitly submitting their + // ask orders with a version previous to this one. + VersionSidecarChannel Version = 4 ) // Type is the type of an order. We don't use iota for the constants due to the @@ -370,7 +380,7 @@ func (a *Ask) Digest() ([sha256.Size]byte, error) { } case VersionNodeTierMinMatch, VersionLeaseDurationBuckets, - VersionSelfChanBalance: + VersionSelfChanBalance, VersionSidecarChannel: err := lnwire.WriteElements( &msg, a.nonce[:], uint32(a.Version), a.FixedRate, @@ -520,6 +530,15 @@ type Bid struct { // to the channel resulting from matching this bid by moving additional // funds from the taker's account into the channel. SelfChanBalance btcutil.Amount + + // SidecarTicket indicates, if non-nil, that the channel being purchased + // with this bid should be opened to a node other than the caller's + // node. The lease recipient is another Pool (light) node that + // authenticates itself to the auctioneer using the information in this + // ticket (the information exchange between bidder and lease recipient + // happens out of band). This will only be used if the order version is + // VersionSidecarChannel or greater. + SidecarTicket *sidecar.Ticket } // Type returns the order type. @@ -570,6 +589,22 @@ func (b *Bid) Digest() ([sha256.Size]byte, error) { return result, err } + case VersionSidecarChannel: + var isSidecar uint8 + if b.SidecarTicket != nil { + isSidecar = 1 + } + + err := lnwire.WriteElements( + &msg, b.nonce[:], uint32(b.Version), b.FixedRate, + b.Amt, b.LeaseDuration, uint64(b.MaxBatchFeeRate), + uint32(b.MinNodeTier), uint32(b.MinUnitsMatch), + uint64(b.SelfChanBalance), isSidecar, + ) + if err != nil { + return result, err + } + default: return result, fmt.Errorf("unknown version %d", b.Kit.Version) } diff --git a/rpcserver.go b/rpcserver.go index 2ccefa0..cfec16f 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -6,7 +6,6 @@ import ( "encoding/hex" "errors" "fmt" - "io" "sync" "sync/atomic" "time" @@ -36,8 +35,6 @@ import ( "github.com/lightningnetwork/lnd/lnwallet/chainfee" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/signal" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" ) const ( @@ -63,12 +60,11 @@ type rpcServer struct { accountManager *account.Manager orderManager *order.Manager - quit chan struct{} - wg sync.WaitGroup - blockNtfnCancel func() - pendingOpenChannelStreamCancel func() - recoveryMutex sync.Mutex - recoveryPending bool + quit chan struct{} + wg sync.WaitGroup + blockNtfnCancel func() + recoveryMutex sync.Mutex + recoveryPending bool // wumboSupported is true if the backing lnd node supports wumbo // channels. @@ -167,17 +163,6 @@ func (s *rpcServer) Start() error { s.updateHeight(height) - // Subscribe to pending open channel notifications. This will be useful - // when we're creating channels with a matched order as part of a batch. - streamCtx, streamCancel := context.WithCancel(ctx) - s.pendingOpenChannelStreamCancel = streamCancel - subStream, err := s.lndClient.SubscribeChannelEvents( - streamCtx, &lnrpc.ChannelEventSubscription{}, - ) - if err != nil { - return err - } - // Start the auctioneer client first to establish a connection. if err := s.auctioneer.Start(); err != nil { return fmt.Errorf("unable to start auctioneer client: %v", err) @@ -190,73 +175,21 @@ func (s *rpcServer) Start() error { if err := s.orderManager.Start(); err != nil { return fmt.Errorf("unable to start order manager: %v", err) } + if err := s.server.fundingManager.Start(); err != nil { + return fmt.Errorf("unable to start funding manager: %v", err) + } if err := s.server.channelAcceptor.Start(blockErrChan); err != nil { return fmt.Errorf("unable to start channel acceptor: %v", err) } - s.wg.Add(2) + s.wg.Add(1) go s.serverHandler(blockChan, blockErrChan) - go s.consumePendingOpenChannels(subStream) rpcLog.Infof("Trader server is now active") return nil } -// consumePendingOpenChannels consumes pending open channel events from the -// stream and notifies them if the trader currently has an ongoing batch. -func (s *rpcServer) consumePendingOpenChannels( - subStream lnrpc.Lightning_SubscribeChannelEventsClient) { - - defer s.wg.Done() - - for { - select { - case <-s.quit: - return - default: - } - - msg, err := subStream.Recv() - if err != nil { - select { - case <-s.quit: - return - default: - } - - rpcLog.Errorf("Unable to read channel event: %v", err) - - // If the lnd node shut down, there's no use continuing. - if err == io.EOF || err == io.ErrUnexpectedEOF || - status.Code(err) == codes.Unavailable { - - return - } - - continue - } - - // Skip any events other than the pending open channel one. - channel, ok := msg.Channel.(*lnrpc.ChannelEventUpdate_PendingOpenChannel) - if !ok { - continue - } - - // If we don't have a pending batch, then there's no need to - // notify any pending open channels.. - if !s.orderManager.HasPendingBatch() { - continue - } - - select { - case s.server.fundingManager.PendingOpenChannels <- channel: - case <-s.quit: - return - } - } -} - // Stop stops the server. func (s *rpcServer) Stop() error { if !atomic.CompareAndSwapUint32(&s.stopped, 0, 1) { @@ -264,6 +197,9 @@ func (s *rpcServer) Stop() error { } rpcLog.Info("Trader server stopping") + if err := s.server.fundingManager.Stop(); err != nil { + rpcLog.Errorf("Error stopping funding manager: %v", err) + } s.server.channelAcceptor.Stop() s.accountManager.Stop() s.orderManager.Stop() @@ -273,10 +209,6 @@ func (s *rpcServer) Stop() error { close(s.quit) - // We call this before Wait to ensure the goroutine is stopped by this - // call. - s.pendingOpenChannelStreamCancel() - s.wg.Wait() s.blockNtfnCancel() @@ -416,7 +348,9 @@ func (s *rpcServer) handleServerMessage( // Before we accept the batch, we'll finish preparations on our // end which include applying any order match predicates, // connecting out to peers, and registering funding shim. - err = s.server.fundingManager.PrepChannelFunding(batch, s.quit) + err = s.server.fundingManager.PrepChannelFunding( + batch, s.server.db.GetOrder, + ) if err != nil { rpcLog.Warnf("Error preparing channel funding: %v", err) @@ -436,7 +370,7 @@ func (s *rpcServer) handleServerMessage( // once all channel partners have responded. batch := s.orderManager.PendingBatch() channelKeys, err := s.server.fundingManager.BatchChannelSetup( - batch, s.quit, + batch, ) if err != nil { rpcLog.Errorf("Error setting up channels: %v", err) @@ -1585,38 +1519,9 @@ func (s *rpcServer) sendSignBatch(batch *order.Batch, sigs order.BatchSignature, rpcSigs[key] = sig.Serialize() } - rpcChannelInfos := make(map[string]*auctioneerrpc.ChannelInfo, len(chanInfos)) - for chanPoint, chanInfo := range chanInfos { - var channelType auctioneerrpc.ChannelType - switch chanInfo.Version { - case chanbackup.TweaklessCommitVersion: - channelType = auctioneerrpc.ChannelType_TWEAKLESS - - // The AnchorsCommitVersion was never widely deployed (at least - // in mainnet) because the lnd version that included it guarded - // the anchor channels behind a config flag. Also, the two - // anchor versions only differ in the fee negotiation and not - // the commitment TX format, so we don't need to distinguish - // between them for our purpose. - case chanbackup.AnchorsCommitVersion, - chanbackup.AnchorsZeroFeeHtlcTxCommitVersion: - - channelType = auctioneerrpc.ChannelType_ANCHORS - default: - return fmt.Errorf("unknown channel type: %v", - chanInfo.Version) - } - rpcChannelInfos[chanPoint.String()] = &auctioneerrpc.ChannelInfo{ - Type: channelType, - LocalNodeKey: chanInfo.LocalNodeKey. - SerializeCompressed(), - RemoteNodeKey: chanInfo.RemoteNodeKey. - SerializeCompressed(), - LocalPaymentBasePoint: chanInfo.LocalPaymentBasePoint. - SerializeCompressed(), - RemotePaymentBasePoint: chanInfo.RemotePaymentBasePoint. - SerializeCompressed(), - } + rpcChannelInfos, err := marshallChannelInfo(chanInfos) + if err != nil { + return fmt.Errorf("error marshalling channel info: %v", err) } rpcLog.Infof("Sending OrderMatchSign for batch %x", batch.ID[:]) @@ -2341,3 +2246,47 @@ func unmarshallNodeTier(nodeTier auctioneerrpc.NodeTier) (order.NodeTier, return 0, fmt.Errorf("unknown node tier: %v", nodeTier) } } + +// marshallChannelInfo turns the given channel information map into its RPC +// counterpart. +func marshallChannelInfo(chanInfos map[wire.OutPoint]*chaninfo.ChannelInfo) ( + map[string]*auctioneerrpc.ChannelInfo, error) { + + rpcChannelInfos := make( + map[string]*auctioneerrpc.ChannelInfo, len(chanInfos), + ) + for chanPoint, chanInfo := range chanInfos { + var channelType auctioneerrpc.ChannelType + switch chanInfo.Version { + case chanbackup.TweaklessCommitVersion: + channelType = auctioneerrpc.ChannelType_TWEAKLESS + + // The AnchorsCommitVersion was never widely deployed (at least + // in mainnet) because the lnd version that included it guarded + // the anchor channels behind a config flag. Also, the two + // anchor versions only differ in the fee negotiation and not + // the commitment TX format, so we don't need to distinguish + // between them for our purpose. + case chanbackup.AnchorsCommitVersion, + chanbackup.AnchorsZeroFeeHtlcTxCommitVersion: + + channelType = auctioneerrpc.ChannelType_ANCHORS + default: + return nil, fmt.Errorf("unknown channel type: %v", + chanInfo.Version) + } + rpcChannelInfos[chanPoint.String()] = &auctioneerrpc.ChannelInfo{ + Type: channelType, + LocalNodeKey: chanInfo.LocalNodeKey. + SerializeCompressed(), + RemoteNodeKey: chanInfo.RemoteNodeKey. + SerializeCompressed(), + LocalPaymentBasePoint: chanInfo.LocalPaymentBasePoint. + SerializeCompressed(), + RemotePaymentBasePoint: chanInfo.RemotePaymentBasePoint. + SerializeCompressed(), + } + } + + return rpcChannelInfos, nil +} diff --git a/server.go b/server.go index 901f4c5..d14484c 100644 --- a/server.go +++ b/server.go @@ -461,18 +461,15 @@ func (s *Server) setupClient() error { // starting/stopping it though as all that logic is currently there for // the other managers as well. s.channelAcceptor = NewChannelAcceptor(s.lndServices.Client) - s.fundingManager = &funding.Manager{ - DB: s.db, - WalletKit: s.lndServices.WalletKit, - LightningClient: s.lndServices.Client, - BaseClient: s.lndClient, - BatchStepTimeout: order.DefaultBatchStepTimeout, - NewNodesOnly: s.cfg.NewNodesOnly, - PendingOpenChannels: make( - chan *lnrpc.ChannelEventUpdate_PendingOpenChannel, - ), + s.fundingManager = funding.NewManager(&funding.ManagerConfig{ + DB: s.db, + WalletKit: s.lndServices.WalletKit, + LightningClient: s.lndServices.Client, + BaseClient: s.lndClient, + BatchStepTimeout: order.DefaultBatchStepTimeout, + NewNodesOnly: s.cfg.NewNodesOnly, NotifyShimCreated: s.channelAcceptor.ShimRegistered, - } + }) // Create an instance of the auctioneer client library. clientCfg := &auctioneer.Config{ diff --git a/sidecar/codec_test.go b/sidecar/codec_test.go index fc27076..5074976 100644 --- a/sidecar/codec_test.go +++ b/sidecar/codec_test.go @@ -9,15 +9,15 @@ import ( ) var ( - hardcodedTicket = "sidecarAAQgHBgUEAwIBAAIBYwMBAgp5CwgAAAAAAAADCQwIAA" + - "AAAAAAA3gNIQLVLm5gAB7Vh7eEvz8Y3CkF2DsvNuSWJj8oOxp1iSh5xw5AAA" + - "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACwAAAAAAAAAAAAAAAAAAA" + - "AAAAAAAAAAAAAAAAAAAAAAFhRGFSEC1S5uYAAe1Ye3hL8_GNwpBdg7Lzbkli" + - "Y_KDsadYkoeccWIQMZ1hb2o3OyT-XUX7KWS-pAVBloIPQe8aCTqcjiNOV89x" + - "5kHyALFiEsAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACBAAAAAAAAAAA" + - "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAGMAAAAAAAAAAAAAAAAAAAAAAAAAAA" + - "AAAAAAAAAAAAAAISgiKSBjWE0AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + - "AAAA3VE-c=" + hardcodedTicket = "sidecarAAQgHBgUEAwIBAAIBYwMBAgp_CwgAAAAAAAADCQwIAA" + + "AAAAAAA3gNBAAAB-AOIQLVLm5gAB7Vh7eEvz8Y3CkF2DsvNuSWJj8oOxp1iS" + + "h5xw9AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACwAAAAAAAAAAA" + + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAFhRMFSEC1S5uYAAe1Ye3hL8_GNwpBd" + + "g7LzbkliY_KDsadYkoeccWIQMZ1hb2o3OyT-XUX7KWS-pAVBloIPQe8aCTqc" + + "jiNOV89xcEAAAAAB5kHyALFiEsAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + + "AAACBAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAGMAAAAAAAAAAA" + + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAISgiKSBjWE0AAAAAAAAAAAAAAAAAAA" + + "AAAAAAAAAAAAAAAAAAAHcU6Y4=" ) // TestEncodeDecode tests that a ticket can be encoded and decoded from/to a @@ -29,9 +29,10 @@ func TestEncodeDecode(t *testing.T) { Version: Version(99), State: StateRegistered, Offer: Offer{ - Capacity: 777, - PushAmt: 888, - SignPubKey: testPubKey, + Capacity: 777, + PushAmt: 888, + LeaseDurationBlocks: 2016, + SignPubKey: testPubKey, SigOfferDigest: &btcec.Signature{ R: new(big.Int).SetInt64(44), S: new(big.Int).SetInt64(22), diff --git a/sidecar/interface.go b/sidecar/interface.go index 9e49ad6..07ed241 100644 --- a/sidecar/interface.go +++ b/sidecar/interface.go @@ -102,6 +102,10 @@ type Offer struct { // reduce the matching chances somewhat. PushAmt btcutil.Amount + // LeaseDurationBlocks is the number of blocks the offered channel in + // this offer would be leased for. + LeaseDurationBlocks uint32 + // SignPubKey is the public key for corresponding to the private key // that signed the SigOfferDigest below and, in a later state, the // SigOrderDigest of the Order struct. @@ -124,6 +128,9 @@ type Recipient struct { // multisig keys of the channel funding transaction output and is // advertised in the bid order. MultiSigPubKey *btcec.PublicKey + + // MultiSigKeyIndex is the derivation index of the MultiSigPubKey. + MultiSigKeyIndex uint32 } // Order is a struct holding the information about the sidecar bid order after @@ -189,15 +196,16 @@ type Ticket struct { // NewTicket creates a new sidecar ticket with the given version and offer // information. func NewTicket(version Version, capacity, pushAmt btcutil.Amount, - offerPubKey *btcec.PublicKey) (*Ticket, error) { + duration uint32, offerPubKey *btcec.PublicKey) (*Ticket, error) { t := &Ticket{ Version: version, State: StateOffered, Offer: Offer{ - Capacity: capacity, - PushAmt: pushAmt, - SignPubKey: offerPubKey, + Capacity: capacity, + PushAmt: pushAmt, + LeaseDurationBlocks: duration, + SignPubKey: offerPubKey, }, } @@ -256,3 +264,21 @@ func (t *Ticket) OrderDigest() ([32]byte, error) { } return sha256.Sum256(msg.Bytes()), nil } + +// Store is the interface a persistent storage must implement for storing and +// retrieving sidecar tickets. +type Store interface { + // AddSidecar adds a record for the sidecar order to the database. + AddSidecar(sidecar *Ticket) error + + // UpdateSidecar updates a sidecar order in the database. + UpdateSidecar(sidecar *Ticket) error + + // Sidecar retrieves a specific sidecar by its ID and provider signing + // key (offer signature pubkey) or returns ErrNoSidecar if it's not + // found. + Sidecar(id [8]byte, offerSignPubKey *btcec.PublicKey) (*Ticket, error) + + // Sidecars retrieves all known sidecar orders from the database. + Sidecars() ([]*Ticket, error) +} diff --git a/sidecar/tlv.go b/sidecar/tlv.go index 21304ea..2724e79 100644 --- a/sidecar/tlv.go +++ b/sidecar/tlv.go @@ -19,12 +19,14 @@ const ( offerType tlv.Type = 10 capacityType tlv.Type = 11 pushAmtType tlv.Type = 12 - signPubKeyType tlv.Type = 13 - sigOfferDigestType tlv.Type = 14 + leaseDurationType tlv.Type = 13 + signPubKeyType tlv.Type = 14 + sigOfferDigestType tlv.Type = 15 - recipientType tlv.Type = 20 - nodePubKeyType tlv.Type = 21 - multiSigPubKeyType tlv.Type = 22 + recipientType tlv.Type = 20 + nodePubKeyType tlv.Type = 21 + multiSigPubKeyType tlv.Type = 22 + multiSigKeyIndexType tlv.Type = 23 orderType tlv.Type = 30 bidNonceType tlv.Type = 31 @@ -177,6 +179,9 @@ func serializeOffer(o Offer) ([]byte, error) { tlvRecords := []tlv.Record{ tlv.MakePrimitiveRecord(capacityType, &capacity), tlv.MakePrimitiveRecord(pushAmtType, &pushAmt), + tlv.MakePrimitiveRecord( + leaseDurationType, &o.LeaseDurationBlocks, + ), } if o.SignPubKey != nil { @@ -206,6 +211,9 @@ func deserializeOffer(offerBytes []byte) (Offer, error) { offerBytes, tlv.MakePrimitiveRecord(capacityType, &capacity), tlv.MakePrimitiveRecord(pushAmtType, &pushAmt), + tlv.MakePrimitiveRecord( + leaseDurationType, &o.LeaseDurationBlocks, + ), tlv.MakePrimitiveRecord(signPubKeyType, &o.SignPubKey), tlv.MakeStaticRecord( sigOfferDigestType, &o.SigOfferDigest, 64, ESig, DSig, @@ -236,6 +244,10 @@ func serializeRecipient(r Recipient) ([]byte, error) { )) } + tlvRecords = append(tlvRecords, tlv.MakePrimitiveRecord( + multiSigKeyIndexType, &r.MultiSigKeyIndex, + )) + return encodeBytes(tlvRecords...) } @@ -247,6 +259,9 @@ func deserializeRecipient(recipientBytes []byte) (Recipient, error) { recipientBytes, tlv.MakePrimitiveRecord(nodePubKeyType, &r.NodePubKey), tlv.MakePrimitiveRecord(multiSigPubKeyType, &r.MultiSigPubKey), + tlv.MakePrimitiveRecord( + multiSigKeyIndexType, &r.MultiSigKeyIndex, + ), ) } diff --git a/sidecar/verification.go b/sidecar/verification.go index 1904bac..78d79a5 100644 --- a/sidecar/verification.go +++ b/sidecar/verification.go @@ -1,11 +1,190 @@ package sidecar import ( + "context" "fmt" "github.com/btcsuite/btcutil" + "github.com/lightninglabs/lndclient" + "github.com/lightningnetwork/lnd/keychain" + "github.com/lightningnetwork/lnd/lnwire" ) +// SignOffer adds a signature over the offer digest to the given ticket. +func SignOffer(ctx context.Context, ticket *Ticket, + signingKeyLoc keychain.KeyLocator, signer lndclient.SignerClient) error { + + // The ticket needs to be in the correct state for us to sign it. + if ticket == nil || ticket.State < StateOffered { + return fmt.Errorf("ticket is in invalid state") + } + + // The ticket also needs to have a signed offer. + offer := ticket.Offer + if offer.SignPubKey == nil || offer.SigOfferDigest != nil { + return fmt.Errorf("offer in ticket is not in expected state " + + "to be signed") + } + + // Let's sign the offer part of the ticket with our node's identity key + // now. + offerDigest, err := ticket.OfferDigest() + if err != nil { + return fmt.Errorf("error digesting offer: %v", err) + } + rawSig, err := signer.SignMessage(ctx, offerDigest[:], signingKeyLoc) + if err != nil { + return fmt.Errorf("error signing offer: %v", err) + } + wireSig, err := lnwire.NewSigFromRawSignature(rawSig) + if err != nil { + return fmt.Errorf("error parsing raw signature: %v", err) + } + ecSig, err := wireSig.ToSignature() + if err != nil { + return fmt.Errorf("error parsing EC signature: %v", err) + } + ticket.Offer.SigOfferDigest = ecSig + + return nil +} + +// VerifyOffer verifies the state of a ticket to be in the offered state and +// also makes sure the offer signature is valid. +func VerifyOffer(ctx context.Context, ticket *Ticket, + signer lndclient.SignerClient) error { + + // The ticket needs to be in the correct state for us to verify it. + if ticket == nil || ticket.State < StateOffered { + return fmt.Errorf("ticket is in invalid state") + } + + // The ticket also needs to have a signed offer. + offer := ticket.Offer + if offer.SignPubKey == nil || offer.SigOfferDigest == nil { + return fmt.Errorf("offer in ticket is not signed") + } + + var offerPubKeyRaw [33]byte + copy(offerPubKeyRaw[:], ticket.Offer.SignPubKey.SerializeCompressed()) + + // Make sure the provider's signature over the offer is valid. + offerDigest, err := ticket.OfferDigest() + if err != nil { + return fmt.Errorf("error calculating offer digest: %v", err) + } + sigValid, err := signer.VerifyMessage( + ctx, offerDigest[:], offer.SigOfferDigest.Serialize(), + offerPubKeyRaw, + ) + if err != nil { + return fmt.Errorf("unable to verify offer signature: %v", err) + } + if !sigValid { + return fmt.Errorf("signature not valid for public key %x", + offerPubKeyRaw[:]) + } + + return nil +} + +// SignOrder adds the order part to a ticket and signs it, adding the signature +// as well. +func SignOrder(ctx context.Context, ticket *Ticket, bidNonce [32]byte, + signingKeyLoc keychain.KeyLocator, signer lndclient.SignerClient) error { + + // The ticket needs to be in the correct state for us to sign it. + if ticket == nil || ticket.State < StateRegistered { + return fmt.Errorf("ticket is in invalid state") + } + + // The ticket also needs to have a signed offer. + offer := ticket.Offer + if offer.SignPubKey == nil || offer.SigOfferDigest == nil { + return fmt.Errorf("offer in ticket is not signed") + } + + // Add the bid order's nonce to the order part of the ticket now. + if ticket.Order == nil { + ticket.Order = &Order{} + } + ticket.Order.BidNonce = bidNonce + ticket.State = StateOrdered + + // Let's sign the order part of the ticket with our node's identity key + // now. + orderDigest, err := ticket.OrderDigest() + if err != nil { + return fmt.Errorf("error digesting order: %v", err) + } + rawSig, err := signer.SignMessage(ctx, orderDigest[:], signingKeyLoc) + if err != nil { + return fmt.Errorf("error signing order: %v", err) + } + wireSig, err := lnwire.NewSigFromRawSignature(rawSig) + if err != nil { + return fmt.Errorf("error parsing raw signature: %v", err) + } + ecSig, err := wireSig.ToSignature() + if err != nil { + return fmt.Errorf("error parsing EC signature: %v", err) + } + ticket.Order.SigOrderDigest = ecSig + + return nil +} + +// VerifyOrder verifies the state of a ticket to be in the ordered state and +// also makes sure the order signature is valid. +func VerifyOrder(ctx context.Context, ticket *Ticket, + signer lndclient.SignerClient) error { + + // The ticket needs to be in the correct state for us to verify it. + if ticket == nil || ticket.State < StateOrdered { + return fmt.Errorf("ticket is in invalid state") + } + + // The ticket also needs to have a pubkey in the offer and needs to be + // signed. We don't need to verify the signature again, the provider + // wouldn't sign the order if the offer itself isn't valid. + offer := ticket.Offer + if offer.SignPubKey == nil || offer.SigOfferDigest == nil { + return fmt.Errorf("offer in ticket is not signed") + } + + order := ticket.Order + if order == nil || order.SigOrderDigest == nil { + return fmt.Errorf("order in ticket is not signed") + } + + // The nonce shouldn't be empty either. + if order.BidNonce == [32]byte{} { + return fmt.Errorf("nonce in order part of ticket is empty") + } + + var orderPubKeyRaw [33]byte + copy(orderPubKeyRaw[:], ticket.Offer.SignPubKey.SerializeCompressed()) + + // Make sure the provider's signature over the order is valid. + orderDigest, err := ticket.OrderDigest() + if err != nil { + return fmt.Errorf("error calculating order digest: %v", err) + } + sigValid, err := signer.VerifyMessage( + ctx, orderDigest[:], order.SigOrderDigest.Serialize(), + orderPubKeyRaw, + ) + if err != nil { + return fmt.Errorf("unable to verify order signature: %v", err) + } + if !sigValid { + return fmt.Errorf("signature not valid for public key %x", + orderPubKeyRaw[:]) + } + + return nil +} + // CheckOfferParams makes sure the offer parameters of a sidecar ticket are // valid and sane. func CheckOfferParams(capacity, pushAmt, baseSupplyUnit btcutil.Amount) error { diff --git a/sidecar/verification_test.go b/sidecar/verification_test.go new file mode 100644 index 0000000..70bd036 --- /dev/null +++ b/sidecar/verification_test.go @@ -0,0 +1,352 @@ +package sidecar + +import ( + "context" + "math/big" + "testing" + + "github.com/btcsuite/btcd/btcec" + "github.com/lightninglabs/pool/internal/test" + "github.com/lightningnetwork/lnd/keychain" + "github.com/stretchr/testify/require" +) + +var ( + _, providerPubKey = btcec.PrivKeyFromBytes(btcec.S256(), []byte{0x02}) + testOfferSig = &btcec.Signature{ + R: new(big.Int).SetInt64(44), + S: new(big.Int).SetInt64(22), + } +) + +// TestSignOffer makes sure that a sidecar ticket's offer part can be signed +// correctly. +func TestSignOffer(t *testing.T) { + t.Parallel() + + mockSigner := test.NewMockSigner() + mockSigner.Signature = testOfferSig.Serialize() + + testCases := []struct { + name string + ticket *Ticket + expectedErr string + }{{ + name: "no ticket", + expectedErr: "ticket is in invalid state", + }, { + name: "non offered ticket", + ticket: &Ticket{ + State: StateCreated, + }, + expectedErr: "ticket is in invalid state", + }, { + name: "offered ticket with no sign pubkey", + ticket: &Ticket{ + State: StateOffered, + Offer: Offer{}, + }, + expectedErr: "offer in ticket is not in expected state to be", + }, { + name: "offered ticket has signature", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOffered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + }, + expectedErr: "offer in ticket is not in expected state to be", + }, { + name: "all valid", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOffered, + Offer: Offer{ + SignPubKey: providerPubKey, + }, + }, + expectedErr: "", + }} + + for _, tc := range testCases { + tc := tc + + if tc.ticket != nil { + digest, err := tc.ticket.OfferDigest() + require.NoError(t, err) + mockSigner.SignatureMsg = string(digest[:]) + } + + err := SignOffer( + context.Background(), tc.ticket, keychain.KeyLocator{}, + mockSigner, + ) + + if tc.expectedErr == "" { + require.NoError(t, err) + require.Equal( + t, testOfferSig, tc.ticket.Offer.SigOfferDigest, + ) + } else { + require.Error(t, err) + require.Contains(t, err.Error(), tc.expectedErr) + } + } +} + +// TestVerifyOffer makes sure that a sidecar ticket's offer part can be verified +// correctly. +func TestVerifyOffer(t *testing.T) { + t.Parallel() + + mockSigner := test.NewMockSigner() + mockSigner.Signature = testOfferSig.Serialize() + + testCases := []struct { + name string + ticket *Ticket + expectedErr string + }{{ + name: "no ticket", + expectedErr: "ticket is in invalid state", + }, { + name: "non offered ticket", + ticket: &Ticket{ + State: StateCreated, + }, + expectedErr: "ticket is in invalid state", + }, { + name: "offered ticket with no sign pubkey", + ticket: &Ticket{ + State: StateOffered, + Offer: Offer{}, + }, + expectedErr: "offer in ticket is not signed", + }, { + name: "invalid sig", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOffered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: &btcec.Signature{ + R: new(big.Int).SetInt64(33), + S: new(big.Int).SetInt64(33), + }, + }, + }, + expectedErr: "signature not valid for public key", + }, { + name: "all valid", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOffered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + }, + expectedErr: "", + }} + + for _, tc := range testCases { + tc := tc + + if tc.ticket != nil { + digest, err := tc.ticket.OfferDigest() + require.NoError(t, err) + mockSigner.SignatureMsg = string(digest[:]) + } + + err := VerifyOffer( + context.Background(), tc.ticket, mockSigner, + ) + + if tc.expectedErr == "" { + require.NoError(t, err) + } else { + require.Error(t, err) + require.Contains(t, err.Error(), tc.expectedErr) + } + } +} + +// TestSignOrder makes sure that a sidecar ticket's order part can be signed +// correctly. +func TestSignOrder(t *testing.T) { + t.Parallel() + + mockSigner := test.NewMockSigner() + mockSigner.Signature = testOfferSig.Serialize() + + testCases := []struct { + name string + ticket *Ticket + expectedErr string + }{{ + name: "no ticket", + expectedErr: "ticket is in invalid state", + }, { + name: "non registered ticket", + ticket: &Ticket{ + State: StateCreated, + }, + expectedErr: "ticket is in invalid state", + }, { + name: "non signed offer", + ticket: &Ticket{ + State: StateRegistered, + Offer: Offer{ + SignPubKey: providerPubKey, + }, + }, + expectedErr: "offer in ticket is not signed", + }, { + name: "all valid", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateRegistered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + }, + expectedErr: "", + }} + + for _, tc := range testCases { + tc := tc + + if tc.ticket != nil { + digest, err := tc.ticket.OfferDigest() + require.NoError(t, err) + mockSigner.SignatureMsg = string(digest[:]) + } + + err := SignOrder( + context.Background(), tc.ticket, [32]byte{}, + keychain.KeyLocator{}, mockSigner, + ) + + if tc.expectedErr == "" { + require.NoError(t, err) + + require.NotNil(t, tc.ticket.Order) + require.Equal( + t, testOfferSig, tc.ticket.Order.SigOrderDigest, + ) + require.Equal(t, StateOrdered, tc.ticket.State) + } else { + require.Error(t, err) + require.Contains(t, err.Error(), tc.expectedErr) + } + } +} + +// TestVerifyOrder makes sure that a sidecar ticket's order part can be verified +// correctly. +func TestVerifyOrder(t *testing.T) { + t.Parallel() + + mockSigner := test.NewMockSigner() + mockSigner.Signature = testOfferSig.Serialize() + + testCases := []struct { + name string + ticket *Ticket + expectedErr string + }{{ + name: "no ticket", + expectedErr: "ticket is in invalid state", + }, { + name: "non ordered ticket", + ticket: &Ticket{ + State: StateOffered, + }, + expectedErr: "ticket is in invalid state", + }, { + name: "invalid sig", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOrdered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + Order: &Order{ + SigOrderDigest: &btcec.Signature{ + R: new(big.Int).SetInt64(33), + S: new(big.Int).SetInt64(33), + }, + BidNonce: [32]byte{1, 2, 3}, + }, + }, + expectedErr: "signature not valid for public key", + }, { + name: "ordered ticket with no signature", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOrdered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + Order: &Order{ + BidNonce: [32]byte{1, 2, 3}, + }, + }, + expectedErr: "order in ticket is not signed", + }, { + name: "empty nonce", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOrdered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + Order: &Order{ + SigOrderDigest: testOfferSig, + }, + }, + expectedErr: "nonce in order part of ticket is empty", + }, { + name: "all valid", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOrdered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + Order: &Order{ + SigOrderDigest: testOfferSig, + BidNonce: [32]byte{1, 2, 3}, + }, + }, + expectedErr: "", + }} + + for _, tc := range testCases { + tc := tc + + if tc.ticket != nil && tc.ticket.Order != nil { + digest, err := tc.ticket.OrderDigest() + require.NoError(t, err) + mockSigner.SignatureMsg = string(digest[:]) + } + + err := VerifyOrder( + context.Background(), tc.ticket, mockSigner, + ) + + if tc.expectedErr == "" { + require.NoError(t, err, tc.name) + } else { + require.Error(t, err, tc.name) + require.Contains(t, err.Error(), tc.expectedErr, tc.name) + } + } +}