GlcoinCore/pool
1
0
Fork 0
mirror of https://github.com/lightninglabs/pool.git synced 2026-08-18 13:08:33 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #242 from lightninglabs/sidecar-prepare-refactor

Browse source 
sidecar channels 2/3: non-functional preparations for sidecar channels
This commit is contained in:
Johan T. Halseth 2021-04-06 13:35:23 +02:00 • committed by GitHub
commit e480bbad38
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
21 changed files with 1429 additions and 256 deletions
Download patch file Download diff file Expand all files Collapse all files

8
account/interface.go
View file

@ -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.

4
account/manager.go
View file

@ -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)

2
account/mock_test.go
View file

@ -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

20
auctioneer/client.go
View file

@ -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
}

4
clientdb/db.go
View file

@ -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

38
clientdb/order.go
View file

@ -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...)

14
clientdb/order_test.go
View file

@ -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)

173
clientdb/sidecar.go Normal file
View file

@ -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))
}

72
clientdb/sidecar_test.go Normal file
View file

@ -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)
}

1
docs/SUMMARY.md
View file

@ -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)

94
docs/sidecar_channels.md Normal file
View file

@ -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 <charlie-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...
```

278
funding/manager.go
View file

@ -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)

129
funding/manager_test.go
View file

@ -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)
}

37
order/interface.go
View file

@ -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)
}

177
rpcserver.go
View file

@ -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
}

19
server.go
View file

@ -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{

25
sidecar/codec_test.go
View file

@ -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),

34
sidecar/interface.go
View file

@ -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)
}

25
sidecar/tlv.go
View file

@ -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,
),
)
}

179
sidecar/verification.go
View file

@ -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 {

352
sidecar/verification_test.go Normal file
View file

@ -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)
}
}
}