12430728843

Committed 20 Dec 2024 11:36AM UTC coverage: 61.336% (+2.6%) from 58.716%

Build # 12430728843

Build Type

Pull #8777

github

Committed by

ziggie1984

Commit Message

channeldb: fix typo.

Pull Request Pull Request #8777: multi: make reassignment of alias channel edge atomic

Run Details

161 of 213 new or added lines in 7 files covered. (75.59%)

70 existing lines in 17 files now uncovered.

23369 of 38100 relevant lines covered (61.34%)

115813.77 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.32

/watchtower/wtclient/session_negotiator.go

package wtclient

import (
        "errors"
        "fmt"
        "sync"
        "time"

        "github.com/btcsuite/btcd/chaincfg/chainhash"
        "github.com/btcsuite/btclog/v2"
        "github.com/lightningnetwork/lnd/keychain"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/watchtower/blob"
        "github.com/lightningnetwork/lnd/watchtower/wtdb"
        "github.com/lightningnetwork/lnd/watchtower/wtpolicy"
        "github.com/lightningnetwork/lnd/watchtower/wtserver"
        "github.com/lightningnetwork/lnd/watchtower/wtwire"
)

// SessionNegotiator is an interface for asynchronously requesting new sessions.
type SessionNegotiator interface {
        // RequestSession signals to the session negotiator that the client
        // needs another session. Once the session is negotiated, it should be
        // returned via NewSessions.
        RequestSession()

        // NewSessions is a read-only channel where newly negotiated sessions
        // will be delivered.
        NewSessions() <-chan *ClientSession

        // Start safely initializes the session negotiator.
        Start() error

        // Stop safely shuts down the session negotiator.
        Stop() error
}

// NegotiatorConfig provides access to the resources required by a
// SessionNegotiator to faithfully carry out its duties. All nil-able field must
// be initialized.
type NegotiatorConfig struct {
        // DB provides access to a persistent storage medium used by the tower
        // to properly allocate session ephemeral keys and record successfully
        // negotiated sessions.
        DB DB

        // SecretKeyRing allows the client to derive new session private keys
        // when attempting to negotiate session with a tower.
        SecretKeyRing ECDHKeyRing

        // Candidates is an abstract set of tower candidates that the negotiator
        // will traverse serially when attempting to negotiate a new session.
        Candidates TowerCandidateIterator

        // Policy defines the session policy that will be proposed to towers
        // when attempting to negotiate a new session. This policy will be used
        // across all negotiation proposals for the lifetime of the negotiator.
        Policy wtpolicy.Policy

        // Dial initiates an outbound brontide connection to the given address
        // using a specified private key. The peer is returned in the event of a
        // successful connection.
        Dial func(keychain.SingleKeyECDH, *lnwire.NetAddress) (wtserver.Peer,
                error)

        // SendMessage writes a wtwire message to remote peer.
        SendMessage func(wtserver.Peer, wtwire.Message) error

        // ReadMessage reads a message from a remote peer and returns the
        // decoded wtwire message.
        ReadMessage func(wtserver.Peer) (wtwire.Message, error)

        // ChainHash the genesis hash identifying the chain for any negotiated
        // sessions. Any state updates sent to that session should also
        // originate from this chain.
        ChainHash chainhash.Hash

        // MinBackoff defines the initial backoff applied by the session
        // negotiator after all tower candidates have been exhausted and
        // reattempting negotiation with the same set of candidates. Subsequent
        // backoff durations will grow exponentially.
        MinBackoff time.Duration

        // MaxBackoff defines the maximum backoff applied by the session
        // negotiator after all tower candidates have been exhausted and
        // reattempting negotiation with the same set of candidates. If the
        // exponential backoff produces a timeout greater than this value, the
        // backoff duration will be clamped to MaxBackoff.
        MaxBackoff time.Duration

        // Log specifies the desired log output, which should be prefixed by the
        // client type, e.g. anchor or legacy.
        Log btclog.Logger
}

// sessionNegotiator is concrete SessionNegotiator that is able to request new
// sessions from a set of candidate towers asynchronously and return successful
// sessions to the primary client.
type sessionNegotiator struct {
        started sync.Once
        stopped sync.Once

        localInit *wtwire.Init

        cfg *NegotiatorConfig
        log btclog.Logger

        dispatcher             chan struct{}
        newSessions            chan *ClientSession
        successfulNegotiations chan *ClientSession

        wg   sync.WaitGroup
        quit chan struct{}
}

// Compile-time constraint to ensure a *sessionNegotiator implements the
// SessionNegotiator interface.
var _ SessionNegotiator = (*sessionNegotiator)(nil)

// newSessionNegotiator initializes a fresh sessionNegotiator instance.
func newSessionNegotiator(cfg *NegotiatorConfig) *sessionNegotiator {
        // Generate the set of features the negotiator will present to the tower
        // upon connection.
        features := cfg.Policy.FeatureBits()

        localInit := wtwire.NewInitMessage(
                lnwire.NewRawFeatureVector(features...),
                cfg.ChainHash,
        )

        return &sessionNegotiator{
                cfg:                    cfg,
                log:                    cfg.Log,
                localInit:              localInit,
                dispatcher:             make(chan struct{}, 1),
                newSessions:            make(chan *ClientSession),
                successfulNegotiations: make(chan *ClientSession),
                quit:                   make(chan struct{}),
        }
}

// Start safely starts up the sessionNegotiator.
func (n *sessionNegotiator) Start() error {
        n.started.Do(func() {
                n.log.Debugf("Starting session negotiator")

                n.wg.Add(1)
                go n.negotiationDispatcher()
        })

        return nil
}

// Stop safely shuts down the sessionNegotiator.
func (n *sessionNegotiator) Stop() error {
        n.stopped.Do(func() {
                n.log.Debugf("Stopping session negotiator")

                close(n.quit)
                n.wg.Wait()
        })

        return nil
}

// NewSessions returns a receive-only channel from which newly negotiated
// sessions will be returned.
func (n *sessionNegotiator) NewSessions() <-chan *ClientSession {
        return n.newSessions
}

// RequestSession sends a request to the sessionNegotiator to begin requesting a
// new session. If one is already in the process of being negotiated, the
// request will be ignored.
func (n *sessionNegotiator) RequestSession() {
        select {
        case n.dispatcher <- struct{}{}:
        default:
        }
}

// negotiationDispatcher acts as the primary event loop for the
// sessionNegotiator, coordinating requests for more sessions and dispatching
// attempts to negotiate them from a list of candidates.
func (n *sessionNegotiator) negotiationDispatcher() {
        defer n.wg.Done()

        var pendingNegotiations int
        for {
                select {
                case <-n.dispatcher:
                        pendingNegotiations++

                        if pendingNegotiations > 1 {
                                n.log.Debugf("Already negotiating session, " +
                                        "waiting for existing negotiation to " +
                                        "complete")
                                continue
                        }

                        // TODO(conner): consider reusing good towers

                        n.log.Debugf("Dispatching session negotiation")

                        n.wg.Add(1)
                        go n.negotiate()

                case session := <-n.successfulNegotiations:
                        select {
                        case n.newSessions <- session:
                                pendingNegotiations--
                        case <-n.quit:
                                return
                        }

                        if pendingNegotiations > 0 {
                                n.log.Debugf("Dispatching pending session " +
                                        "negotiation")

                                n.wg.Add(1)
                                go n.negotiate()
                        }

                case <-n.quit:
                        return
                }
        }
}

// negotiate handles the process of iterating through potential tower candidates
// and attempting to negotiate a new session until a successful negotiation
// occurs. If the candidate iterator becomes exhausted because none were
// successful, this method will back off exponentially up to the configured max
// backoff. This method will continue trying until a negotiation is successful
// before returning the negotiated session to the dispatcher via the succeed
// channel.
//
// NOTE: This method MUST be run as a goroutine.
func (n *sessionNegotiator) negotiate() {
        defer n.wg.Done()

        // On the first pass, initialize the backoff to our configured min
        // backoff.
        var backoff time.Duration

        // Create a closure to update the backoff upon failure such that it
        // stays within our min and max backoff parameters.
        updateBackoff := func() {
                if backoff == 0 {
                        backoff = n.cfg.MinBackoff
                } else {
                        backoff *= 2
                        if backoff > n.cfg.MaxBackoff {
                                backoff = n.cfg.MaxBackoff
                        }
                }
        }

retryWithBackoff:
        // If we are retrying, wait out the delay before continuing.
        if backoff > 0 {
                select {
                case <-time.After(backoff):
                case <-n.quit:
                        return
                }
        }

tryNextCandidate:
        for {
                select {
                case <-n.quit:
                        return
                default:
                }

                // Pull the next candidate from our list of addresses.
                tower, err := n.cfg.Candidates.Next()
                if err != nil {
                        // We've run out of addresses, update our backoff.
                        updateBackoff()

                        n.log.Debugf("Unable to get new tower candidate, "+
                                "retrying after %v -- reason: %v", backoff, err)

                        // Only reset the iterator once we've exhausted all
                        // candidates. Doing so allows us to load balance
                        // sessions better amongst all of the tower candidates.
                        if err == ErrTowerCandidatesExhausted {
                                n.cfg.Candidates.Reset()
                        }

                        goto retryWithBackoff
                }

                towerPub := tower.IdentityKey.SerializeCompressed()
                n.log.Debugf("Attempting session negotiation with tower=%x",
                        towerPub)

                var forceNextKey bool
                for {
                        // Before proceeding, we will reserve a session key
                        // index to use with this specific tower. If one is
                        // already reserved, the existing index will be
                        // returned.
                        keyIndex, err := n.cfg.DB.NextSessionKeyIndex(
                                tower.ID, n.cfg.Policy.BlobType, forceNextKey,
                        )
                        if err != nil {
                                n.log.Debugf("Unable to reserve session key "+
                                        "index for tower=%x: %v", towerPub, err)

                                goto tryNextCandidate
                        }

                        // We'll now attempt the CreateSession dance with the
                        // tower to get a new session, trying all addresses if
                        // necessary.
                        err = n.createSession(tower, keyIndex)
                        if err == nil {
                                return
                        } else if errors.Is(err, ErrSessionKeyAlreadyUsed) {
                                forceNextKey = true
                                continue
                        }

                        // An unexpected error occurred, update our backoff.
                        updateBackoff()

                        n.log.Debugf("Session negotiation with tower=%x "+
                                "failed, trying again -- reason: %v", towerPub,
                                err)

                        goto retryWithBackoff
                }
        }
}

// createSession takes a tower and attempts to negotiate a session using any of
// its stored addresses. This method returns after the first successful
// negotiation, or after all addresses have failed with ErrFailedNegotiation.
func (n *sessionNegotiator) createSession(tower *Tower, keyIndex uint32) error {
        sessionKeyDesc, err := n.cfg.SecretKeyRing.DeriveKey(
                keychain.KeyLocator{
                        Family: keychain.KeyFamilyTowerSession,
                        Index:  keyIndex,
                },
        )
        if err != nil {
                return err
        }
        sessionKey := keychain.NewPubKeyECDH(
                sessionKeyDesc, n.cfg.SecretKeyRing,
        )

        addr := tower.Addresses.PeekAndLock()
        for {
                lnAddr := &lnwire.NetAddress{
                        IdentityKey: tower.IdentityKey,
                        Address:     addr,
                }

                err = n.tryAddress(sessionKey, keyIndex, tower, lnAddr)
                tower.Addresses.ReleaseLock(addr)
                switch {
                case errors.Is(err, ErrSessionKeyAlreadyUsed):
                        return err

                case errors.Is(err, ErrPermanentTowerFailure):
                        // TODO(conner): report to iterator? can then be reset
                        // with restart
                        fallthrough

                case err != nil:
                        n.log.Debugf("Request for session negotiation with "+
                                "tower=%s failed, trying again -- reason: "+
                                "%v", lnAddr, err)

                        // Get the next tower address if there is one.
                        addr, err = tower.Addresses.NextAndLock()
                        if err == ErrAddressesExhausted {
                                tower.Addresses.Reset()

                                return ErrFailedNegotiation
                        }

                        continue

                default:
                        return nil
                }
        }
}

// tryAddress executes a single create session dance using the given address.
// The address should belong to the tower's set of addresses. This method only
// returns true if all steps succeed and the new session has been persisted, and
// fails otherwise.
func (n *sessionNegotiator) tryAddress(sessionKey keychain.SingleKeyECDH,
        keyIndex uint32, tower *Tower, lnAddr *lnwire.NetAddress) error {

        // Connect to the tower address using our generated session key.
        conn, err := n.cfg.Dial(sessionKey, lnAddr)
        if err != nil {
                return err
        }

        // Send local Init message.
        err = n.cfg.SendMessage(conn, n.localInit)
        if err != nil {
                return fmt.Errorf("unable to send Init: %w", err)
        }

        // Receive remote Init message.
        remoteMsg, err := n.cfg.ReadMessage(conn)
        if err != nil {
                return fmt.Errorf("unable to read Init: %w", err)
        }

        // Check that returned message is wtwire.Init.
        remoteInit, ok := remoteMsg.(*wtwire.Init)
        if !ok {
                return fmt.Errorf("expected Init, got %T in reply", remoteMsg)
        }

        // Verify the watchtower's remote Init message against our own.
        err = n.localInit.CheckRemoteInit(remoteInit, wtwire.FeatureNames)
        if err != nil {
                return err
        }

        policy := n.cfg.Policy
        createSession := &wtwire.CreateSession{
                BlobType:     policy.BlobType,
                MaxUpdates:   policy.MaxUpdates,
                RewardBase:   policy.RewardBase,
                RewardRate:   policy.RewardRate,
                SweepFeeRate: policy.SweepFeeRate,
        }

        // Send CreateSession message.
        err = n.cfg.SendMessage(conn, createSession)
        if err != nil {
                return fmt.Errorf("unable to send CreateSession: %w", err)
        }

        // Receive CreateSessionReply message.
        remoteMsg, err = n.cfg.ReadMessage(conn)
        if err != nil {
                return fmt.Errorf("unable to read CreateSessionReply: %w", err)
        }

        // Check that returned message is wtwire.CreateSessionReply.
        createSessionReply, ok := remoteMsg.(*wtwire.CreateSessionReply)
        if !ok {
                return fmt.Errorf("expected CreateSessionReply, got %T in "+
                        "reply", remoteMsg)
        }

        switch createSessionReply.Code {
        case wtwire.CodeOK:
                // TODO(conner): validate reward address
                rewardPkScript := createSessionReply.Data

                sessionID := wtdb.NewSessionIDFromPubKey(sessionKey.PubKey())
                dbClientSession := &wtdb.ClientSession{
                        ClientSessionBody: wtdb.ClientSessionBody{
                                TowerID:        tower.ID,
                                KeyIndex:       keyIndex,
                                Policy:         n.cfg.Policy,
                                RewardPkScript: rewardPkScript,
                        },
                        ID: sessionID,
                }

                err = n.cfg.DB.CreateClientSession(dbClientSession)
                if err != nil {
                        return fmt.Errorf("unable to persist ClientSession: %w",
                                err)
                }

                n.log.Debugf("New session negotiated with %s, policy: %s",
                        lnAddr, dbClientSession.Policy)

                clientSession := &ClientSession{
                        ID:                sessionID,
                        ClientSessionBody: dbClientSession.ClientSessionBody,
                        Tower:             tower,
                        SessionKeyECDH:    sessionKey,
                }

                // We have a newly negotiated session, return it to the
                // dispatcher so that it can update how many outstanding
                // negotiation requests we have.
                select {
                case n.successfulNegotiations <- clientSession:
                        return nil
                case <-n.quit:
                        return ErrNegotiatorExiting
                }

        case wtwire.CreateSessionCodeAlreadyExists:
                // TODO(conner): use the last-applied in the create session
                //  reply to handle case where we lose state, session already
                //  exists, and we want to possibly resume using the session.
                //  NOTE that this should not be done until the server code
                //  has been adapted to first check that the CreateSession
                //  request is for the same blob-type as the initial session.

                return ErrSessionKeyAlreadyUsed

        // TODO(conner): handle error codes properly
        case wtwire.CreateSessionCodeRejectBlobType:
                return fmt.Errorf("tower rejected blob type: %v",
                        policy.BlobType)

        case wtwire.CreateSessionCodeRejectMaxUpdates:
                return fmt.Errorf("tower rejected max updates: %v",
                        policy.MaxUpdates)

        case wtwire.CreateSessionCodeRejectRewardRate:
                // The tower rejected the session because of the reward rate. If
                // we didn't request a reward session, we'll treat this as a
                // permanent tower failure.
                if !policy.BlobType.Has(blob.FlagReward) {
                        return ErrPermanentTowerFailure
                }

                return fmt.Errorf("tower rejected reward rate: %v",
                        policy.RewardRate)

        case wtwire.CreateSessionCodeRejectSweepFeeRate:
                return fmt.Errorf("tower rejected sweep fee rate: %v",
                        policy.SweepFeeRate)

        default:
                return fmt.Errorf("received unhandled error code: %v",
                        createSessionReply.Code)
        }
}

1	package wtclient
2
3	import (
4	"errors"
5	"fmt"
6	"sync"
7	"time"
8
9	"github.com/btcsuite/btcd/chaincfg/chainhash"
10	"github.com/btcsuite/btclog/v2"
11	"github.com/lightningnetwork/lnd/keychain"
12	"github.com/lightningnetwork/lnd/lnwire"
13	"github.com/lightningnetwork/lnd/watchtower/blob"
14	"github.com/lightningnetwork/lnd/watchtower/wtdb"
15	"github.com/lightningnetwork/lnd/watchtower/wtpolicy"
16	"github.com/lightningnetwork/lnd/watchtower/wtserver"
17	"github.com/lightningnetwork/lnd/watchtower/wtwire"
18	)
19
20	// SessionNegotiator is an interface for asynchronously requesting new sessions.
21	type SessionNegotiator interface {
22	// RequestSession signals to the session negotiator that the client
23	// needs another session. Once the session is negotiated, it should be
24	// returned via NewSessions.
25	RequestSession()
26
27	// NewSessions is a read-only channel where newly negotiated sessions
28	// will be delivered.
29	NewSessions() <-chan *ClientSession
30
31	// Start safely initializes the session negotiator.
32	Start() error
33
34	// Stop safely shuts down the session negotiator.
35	Stop() error
36	}
37
38	// NegotiatorConfig provides access to the resources required by a
39	// SessionNegotiator to faithfully carry out its duties. All nil-able field must
40	// be initialized.
41	type NegotiatorConfig struct {
42	// DB provides access to a persistent storage medium used by the tower
43	// to properly allocate session ephemeral keys and record successfully
44	// negotiated sessions.
45	DB DB
46
47	// SecretKeyRing allows the client to derive new session private keys
48	// when attempting to negotiate session with a tower.
49	SecretKeyRing ECDHKeyRing
50
51	// Candidates is an abstract set of tower candidates that the negotiator
52	// will traverse serially when attempting to negotiate a new session.
53	Candidates TowerCandidateIterator
54
55	// Policy defines the session policy that will be proposed to towers
56	// when attempting to negotiate a new session. This policy will be used
57	// across all negotiation proposals for the lifetime of the negotiator.
58	Policy wtpolicy.Policy
59
60	// Dial initiates an outbound brontide connection to the given address
61	// using a specified private key. The peer is returned in the event of a
62	// successful connection.
63	Dial func(keychain.SingleKeyECDH, *lnwire.NetAddress) (wtserver.Peer,
64	error)
65
66	// SendMessage writes a wtwire message to remote peer.
67	SendMessage func(wtserver.Peer, wtwire.Message) error
68
69	// ReadMessage reads a message from a remote peer and returns the
70	// decoded wtwire message.
71	ReadMessage func(wtserver.Peer) (wtwire.Message, error)
72
73	// ChainHash the genesis hash identifying the chain for any negotiated
74	// sessions. Any state updates sent to that session should also
75	// originate from this chain.
76	ChainHash chainhash.Hash
77
78	// MinBackoff defines the initial backoff applied by the session
79	// negotiator after all tower candidates have been exhausted and
80	// reattempting negotiation with the same set of candidates. Subsequent
81	// backoff durations will grow exponentially.
82	MinBackoff time.Duration
83
84	// MaxBackoff defines the maximum backoff applied by the session
85	// negotiator after all tower candidates have been exhausted and
86	// reattempting negotiation with the same set of candidates. If the
87	// exponential backoff produces a timeout greater than this value, the
88	// backoff duration will be clamped to MaxBackoff.
89	MaxBackoff time.Duration
90
91	// Log specifies the desired log output, which should be prefixed by the
92	// client type, e.g. anchor or legacy.
93	Log btclog.Logger
94	}
95
96	// sessionNegotiator is concrete SessionNegotiator that is able to request new
97	// sessions from a set of candidate towers asynchronously and return successful
98	// sessions to the primary client.
99	type sessionNegotiator struct {
100	started sync.Once
101	stopped sync.Once
102
103	localInit *wtwire.Init
104
105	cfg *NegotiatorConfig
106	log btclog.Logger
107
108	dispatcher chan struct{}
109	newSessions chan *ClientSession
110	successfulNegotiations chan *ClientSession
111
112	wg sync.WaitGroup
113	quit chan struct{}
114	}
115
116	// Compile-time constraint to ensure a *sessionNegotiator implements the
117	// SessionNegotiator interface.
118	var _ SessionNegotiator = (*sessionNegotiator)(nil)
119
120	// newSessionNegotiator initializes a fresh sessionNegotiator instance.
121	func newSessionNegotiator(cfg NegotiatorConfig) sessionNegotiator {	40✔
122	// Generate the set of features the negotiator will present to the tower	40✔
123	// upon connection.	40✔
124	features := cfg.Policy.FeatureBits()	40✔
125		40✔
126	localInit := wtwire.NewInitMessage(	40✔
127	lnwire.NewRawFeatureVector(features...),	40✔
128	cfg.ChainHash,	40✔
129	)	40✔
130		40✔
131	return &sessionNegotiator{	40✔
132	cfg: cfg,	40✔
133	log: cfg.Log,	40✔
134	localInit: localInit,	40✔
135	dispatcher: make(chan struct{}, 1),	40✔
136	newSessions: make(chan *ClientSession),	40✔
137	successfulNegotiations: make(chan *ClientSession),	40✔
138	quit: make(chan struct{}),	40✔
139	}	40✔
140	}	40✔
141
142	// Start safely starts up the sessionNegotiator.
143	func (n *sessionNegotiator) Start() error {	40✔
144	n.started.Do(func() {	80✔
145	n.log.Debugf("Starting session negotiator")	40✔
146		40✔
147	n.wg.Add(1)	40✔
148	go n.negotiationDispatcher()	40✔
149	})	40✔
150
151	return nil	40✔
152	}
153
154	// Stop safely shuts down the sessionNegotiator.
155	func (n *sessionNegotiator) Stop() error {	40✔
156	n.stopped.Do(func() {	80✔
157	n.log.Debugf("Stopping session negotiator")	40✔
158		40✔
159	close(n.quit)	40✔
160	n.wg.Wait()	40✔
161	})	40✔
162
163	return nil	40✔
164	}
165
166	// NewSessions returns a receive-only channel from which newly negotiated
167	// sessions will be returned.
168	func (n sessionNegotiator) NewSessions() <-chan ClientSession {	653✔
169	return n.newSessions	653✔
170	}	653✔
171
172	// RequestSession sends a request to the sessionNegotiator to begin requesting a
173	// new session. If one is already in the process of being negotiated, the
174	// request will be ignored.
175	func (n *sessionNegotiator) RequestSession() {	83✔
176	select {	83✔
177	case n.dispatcher <- struct{}{}:	83✔
178	default:	×
179	}
180	}
181
182	// negotiationDispatcher acts as the primary event loop for the
183	// sessionNegotiator, coordinating requests for more sessions and dispatching
184	// attempts to negotiate them from a list of candidates.
185	func (n *sessionNegotiator) negotiationDispatcher() {	40✔
186	defer n.wg.Done()	40✔
187		40✔
188	var pendingNegotiations int	40✔
189	for {	232✔
190	select {	192✔
191	case <-n.dispatcher:	83✔
192	pendingNegotiations++	83✔
193		83✔
194	if pendingNegotiations > 1 {	83✔
195	n.log.Debugf("Already negotiating session, " +	×
196	"waiting for existing negotiation to " +	×
197	"complete")	×
198	continue	×
199	}
200
201	// TODO(conner): consider reusing good towers
202
203	n.log.Debugf("Dispatching session negotiation")	83✔
204		83✔
205	n.wg.Add(1)	83✔
206	go n.negotiate()	83✔
207
208	case session := <-n.successfulNegotiations:	77✔
209	select {	77✔
210	case n.newSessions <- session:	77✔
211	pendingNegotiations--	77✔
212	case <-n.quit:	×
213	return	×
214	}
215
216	if pendingNegotiations > 0 {	77✔
217	n.log.Debugf("Dispatching pending session " +	×
218	"negotiation")	×
219		×
220	n.wg.Add(1)	×
221	go n.negotiate()	×
222	}	×
223
224	case <-n.quit:	40✔
225	return	40✔
226	}
227	}
228	}
229
230	// negotiate handles the process of iterating through potential tower candidates
231	// and attempting to negotiate a new session until a successful negotiation
232	// occurs. If the candidate iterator becomes exhausted because none were
233	// successful, this method will back off exponentially up to the configured max
234	// backoff. This method will continue trying until a negotiation is successful
235	// before returning the negotiated session to the dispatcher via the succeed
236	// channel.
237	//
238	// NOTE: This method MUST be run as a goroutine.
239	func (n *sessionNegotiator) negotiate() {	83✔
240	defer n.wg.Done()	83✔
241		83✔
242	// On the first pass, initialize the backoff to our configured min	83✔
243	// backoff.	83✔
244	var backoff time.Duration	83✔
245		83✔
246	// Create a closure to update the backoff upon failure such that it	83✔
247	// stays within our min and max backoff parameters.	83✔
248	updateBackoff := func() {	259✔
249	if backoff == 0 {	251✔
250	backoff = n.cfg.MinBackoff	75✔
251	} else {	176✔
252	backoff *= 2	101✔
253	if backoff > n.cfg.MaxBackoff {	159✔
254	backoff = n.cfg.MaxBackoff	58✔
255	}	58✔
256	}
257	}
258
259	retryWithBackoff:
260	// If we are retrying, wait out the delay before continuing.
261	if backoff > 0 {	431✔
262	select {	176✔
263	case <-time.After(backoff):	170✔
264	case <-n.quit:	10✔
265	return	10✔
266	}
267	}
268
269	tryNextCandidate:	249✔
270	for {	498✔
271	select {	249✔
UNCOV 272	case <-n.quit:	×
UNCOV 273	return	×
274	default:	249✔
275	}
276
277	// Pull the next candidate from our list of addresses.
278	tower, err := n.cfg.Candidates.Next()	249✔
279	if err != nil {	385✔
280	// We've run out of addresses, update our backoff.	136✔
281	updateBackoff()	136✔
282		136✔
283	n.log.Debugf("Unable to get new tower candidate, "+	136✔
284	"retrying after %v -- reason: %v", backoff, err)	136✔
285		136✔
286	// Only reset the iterator once we've exhausted all	136✔
287	// candidates. Doing so allows us to load balance	136✔
288	// sessions better amongst all of the tower candidates.	136✔
289	if err == ErrTowerCandidatesExhausted {	272✔
290	n.cfg.Candidates.Reset()	136✔
291	}	136✔
292
293	goto retryWithBackoff	136✔
294	}
295
296	towerPub := tower.IdentityKey.SerializeCompressed()	117✔
297	n.log.Debugf("Attempting session negotiation with tower=%x",	117✔
298	towerPub)	117✔
299		117✔
300	var forceNextKey bool	117✔
301	for {	235✔
302	// Before proceeding, we will reserve a session key	118✔
303	// index to use with this specific tower. If one is	118✔
304	// already reserved, the existing index will be	118✔
305	// returned.	118✔
306	keyIndex, err := n.cfg.DB.NextSessionKeyIndex(	118✔
307	tower.ID, n.cfg.Policy.BlobType, forceNextKey,	118✔
308	)	118✔
309	if err != nil {	118✔
310	n.log.Debugf("Unable to reserve session key "+	×
311	"index for tower=%x: %v", towerPub, err)	×
312		×
313	goto tryNextCandidate	×
314	}
315
316	// We'll now attempt the CreateSession dance with the
317	// tower to get a new session, trying all addresses if
318	// necessary.
319	err = n.createSession(tower, keyIndex)	118✔
320	if err == nil {	195✔
321	return	77✔
322	} else if errors.Is(err, ErrSessionKeyAlreadyUsed) {	119✔
323	forceNextKey = true	1✔
324	continue	1✔
325	}
326
327	// An unexpected error occurred, update our backoff.
328	updateBackoff()	40✔
329		40✔
330	n.log.Debugf("Session negotiation with tower=%x "+	40✔
331	"failed, trying again -- reason: %v", towerPub,	40✔
332	err)	40✔
333		40✔
334	goto retryWithBackoff	40✔
335	}
336	}
337	}
338
339	// createSession takes a tower and attempts to negotiate a session using any of
340	// its stored addresses. This method returns after the first successful
341	// negotiation, or after all addresses have failed with ErrFailedNegotiation.
342	func (n sessionNegotiator) createSession(tower Tower, keyIndex uint32) error {	118✔
343	sessionKeyDesc, err := n.cfg.SecretKeyRing.DeriveKey(	118✔
344	keychain.KeyLocator{	118✔
345	Family: keychain.KeyFamilyTowerSession,	118✔
346	Index: keyIndex,	118✔
347	},	118✔
348	)	118✔
349	if err != nil {	118✔
350	return err	×
351	}	×
352	sessionKey := keychain.NewPubKeyECDH(	118✔
353	sessionKeyDesc, n.cfg.SecretKeyRing,	118✔
354	)	118✔
355		118✔
356	addr := tower.Addresses.PeekAndLock()	118✔
357	for {	236✔
358	lnAddr := &lnwire.NetAddress{	118✔
359	IdentityKey: tower.IdentityKey,	118✔
360	Address: addr,	118✔
361	}	118✔
362		118✔
363	err = n.tryAddress(sessionKey, keyIndex, tower, lnAddr)	118✔
364	tower.Addresses.ReleaseLock(addr)	118✔
365	switch {	118✔
366	case errors.Is(err, ErrSessionKeyAlreadyUsed):	1✔
367	return err	1✔
368
369	case errors.Is(err, ErrPermanentTowerFailure):	×
370	// TODO(conner): report to iterator? can then be reset	×
371	// with restart	×
372	fallthrough	×
373
374	case err != nil:	40✔
375	n.log.Debugf("Request for session negotiation with "+	40✔
376	"tower=%s failed, trying again -- reason: "+	40✔
377	"%v", lnAddr, err)	40✔
378		40✔
379	// Get the next tower address if there is one.	40✔
380	addr, err = tower.Addresses.NextAndLock()	40✔
381	if err == ErrAddressesExhausted {	80✔
382	tower.Addresses.Reset()	40✔
383		40✔
384	return ErrFailedNegotiation	40✔
385	}	40✔
386
387	continue	×
388
389	default:	77✔
390	return nil	77✔
391	}
392	}
393	}
394
395	// tryAddress executes a single create session dance using the given address.
396	// The address should belong to the tower's set of addresses. This method only
397	// returns true if all steps succeed and the new session has been persisted, and
398	// fails otherwise.
399	func (n *sessionNegotiator) tryAddress(sessionKey keychain.SingleKeyECDH,
400	keyIndex uint32, tower Tower, lnAddr lnwire.NetAddress) error {	118✔
401		118✔
402	// Connect to the tower address using our generated session key.	118✔
403	conn, err := n.cfg.Dial(sessionKey, lnAddr)	118✔
404	if err != nil {	123✔
405	return err	5✔
406	}	5✔
407
408	// Send local Init message.
409	err = n.cfg.SendMessage(conn, n.localInit)	113✔
410	if err != nil {	114✔
411	return fmt.Errorf("unable to send Init: %w", err)	1✔
412	}	1✔
413
414	// Receive remote Init message.
415	remoteMsg, err := n.cfg.ReadMessage(conn)	112✔
416	if err != nil {	112✔
417	return fmt.Errorf("unable to read Init: %w", err)	×
418	}	×
419
420	// Check that returned message is wtwire.Init.
421	remoteInit, ok := remoteMsg.(*wtwire.Init)	112✔
422	if !ok {	112✔
423	return fmt.Errorf("expected Init, got %T in reply", remoteMsg)	×
424	}	×
425
426	// Verify the watchtower's remote Init message against our own.
427	err = n.localInit.CheckRemoteInit(remoteInit, wtwire.FeatureNames)	112✔
428	if err != nil {	112✔
429	return err	×
430	}	×
431
432	policy := n.cfg.Policy	112✔
433	createSession := &wtwire.CreateSession{	112✔
434	BlobType: policy.BlobType,	112✔
435	MaxUpdates: policy.MaxUpdates,	112✔
436	RewardBase: policy.RewardBase,	112✔
437	RewardRate: policy.RewardRate,	112✔
438	SweepFeeRate: policy.SweepFeeRate,	112✔
439	}	112✔
440		112✔
441	// Send CreateSession message.	112✔
442	err = n.cfg.SendMessage(conn, createSession)	112✔
443	if err != nil {	112✔
444	return fmt.Errorf("unable to send CreateSession: %w", err)	×
445	}	×
446
447	// Receive CreateSessionReply message.
448	remoteMsg, err = n.cfg.ReadMessage(conn)	112✔
449	if err != nil {	144✔
450	return fmt.Errorf("unable to read CreateSessionReply: %w", err)	32✔
451	}	32✔
452
453	// Check that returned message is wtwire.CreateSessionReply.
454	createSessionReply, ok := remoteMsg.(*wtwire.CreateSessionReply)	80✔
455	if !ok {	80✔
456	return fmt.Errorf("expected CreateSessionReply, got %T in "+	×
457	"reply", remoteMsg)	×
458	}	×
459
460	switch createSessionReply.Code {	80✔
461	case wtwire.CodeOK:	79✔
462	// TODO(conner): validate reward address	79✔
463	rewardPkScript := createSessionReply.Data	79✔
464		79✔
465	sessionID := wtdb.NewSessionIDFromPubKey(sessionKey.PubKey())	79✔
466	dbClientSession := &wtdb.ClientSession{	79✔
467	ClientSessionBody: wtdb.ClientSessionBody{	79✔
468	TowerID: tower.ID,	79✔
469	KeyIndex: keyIndex,	79✔
470	Policy: n.cfg.Policy,	79✔
471	RewardPkScript: rewardPkScript,	79✔
472	},	79✔
473	ID: sessionID,	79✔
474	}	79✔
475		79✔
476	err = n.cfg.DB.CreateClientSession(dbClientSession)	79✔
477	if err != nil {	79✔
478	return fmt.Errorf("unable to persist ClientSession: %w",	×
479	err)	×
480	}	×
481
482	n.log.Debugf("New session negotiated with %s, policy: %s",	79✔
483	lnAddr, dbClientSession.Policy)	79✔
484		79✔
485	clientSession := &ClientSession{	79✔
486	ID: sessionID,	79✔
487	ClientSessionBody: dbClientSession.ClientSessionBody,	79✔
488	Tower: tower,	79✔
489	SessionKeyECDH: sessionKey,	79✔
490	}	79✔
491		79✔
492	// We have a newly negotiated session, return it to the	79✔
493	// dispatcher so that it can update how many outstanding	79✔
494	// negotiation requests we have.	79✔
495	select {	79✔
496	case n.successfulNegotiations <- clientSession:	77✔
497	return nil	77✔
498	case <-n.quit:	2✔
499	return ErrNegotiatorExiting	2✔
500	}
501
502	case wtwire.CreateSessionCodeAlreadyExists:	1✔
503	// TODO(conner): use the last-applied in the create session	1✔
504	// reply to handle case where we lose state, session already	1✔
505	// exists, and we want to possibly resume using the session.	1✔
506	// NOTE that this should not be done until the server code	1✔
507	// has been adapted to first check that the CreateSession	1✔
508	// request is for the same blob-type as the initial session.	1✔
509		1✔
510	return ErrSessionKeyAlreadyUsed	1✔
511
512	// TODO(conner): handle error codes properly
513	case wtwire.CreateSessionCodeRejectBlobType:	×
514	return fmt.Errorf("tower rejected blob type: %v",	×
515	policy.BlobType)	×
516
517	case wtwire.CreateSessionCodeRejectMaxUpdates:	×
518	return fmt.Errorf("tower rejected max updates: %v",	×
519	policy.MaxUpdates)	×
520
521	case wtwire.CreateSessionCodeRejectRewardRate:	×
522	// The tower rejected the session because of the reward rate. If	×
523	// we didn't request a reward session, we'll treat this as a	×
524	// permanent tower failure.	×
525	if !policy.BlobType.Has(blob.FlagReward) {	×
526	return ErrPermanentTowerFailure	×
527	}	×
528
529	return fmt.Errorf("tower rejected reward rate: %v",	×
530	policy.RewardRate)	×
531
532	case wtwire.CreateSessionCodeRejectSweepFeeRate:	×
533	return fmt.Errorf("tower rejected sweep fee rate: %v",	×
534	policy.SweepFeeRate)	×
535
536	default:	×
537	return fmt.Errorf("received unhandled error code: %v",	×
538	createSessionReply.Code)	×
539	}
540	}

lightningnetwork / lnd / 12430728843

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous