14358372723

Committed 09 Apr 2025 01:26PM UTC coverage: 56.696% (-12.3%) from 69.037%

Build # 14358372723

Build Type

Pull #9696

github

Committed by

web-flow

Commit Message

Merge e2837e400 into 867d27d68

Pull Request Pull Request #9696: Add `development_guidelines.md` for both human and machine

Run Details

107055 of 188823 relevant lines covered (56.7%)

22721.56 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 {	36✔
122	// Generate the set of features the negotiator will present to the tower	36✔
123	// upon connection.	36✔
124	features := cfg.Policy.FeatureBits()	36✔
125		36✔
126	localInit := wtwire.NewInitMessage(	36✔
127	lnwire.NewRawFeatureVector(features...),	36✔
128	cfg.ChainHash,	36✔
129	)	36✔
130		36✔
131	return &sessionNegotiator{	36✔
132	cfg: cfg,	36✔
133	log: cfg.Log,	36✔
134	localInit: localInit,	36✔
135	dispatcher: make(chan struct{}, 1),	36✔
136	newSessions: make(chan *ClientSession),	36✔
137	successfulNegotiations: make(chan *ClientSession),	36✔
138	quit: make(chan struct{}),	36✔
139	}	36✔
140	}	36✔
141
142	// Start safely starts up the sessionNegotiator.
143	func (n *sessionNegotiator) Start() error {	36✔
144	n.started.Do(func() {	72✔
145	n.log.Debugf("Starting session negotiator")	36✔
146		36✔
147	n.wg.Add(1)	36✔
148	go n.negotiationDispatcher()	36✔
149	})	36✔
150
151	return nil	36✔
152	}
153
154	// Stop safely shuts down the sessionNegotiator.
155	func (n *sessionNegotiator) Stop() error {	36✔
156	n.stopped.Do(func() {	72✔
157	n.log.Debugf("Stopping session negotiator")	36✔
158		36✔
159	close(n.quit)	36✔
160	n.wg.Wait()	36✔
161	})	36✔
162
163	return nil	36✔
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 {	649✔
169	return n.newSessions	649✔
170	}	649✔
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() {	79✔
176	select {	79✔
177	case n.dispatcher <- struct{}{}:	79✔
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() {	36✔
186	defer n.wg.Done()	36✔
187		36✔
188	var pendingNegotiations int	36✔
189	for {	224✔
190	select {	188✔
191	case <-n.dispatcher:	79✔
192	pendingNegotiations++	79✔
193		79✔
194	if pendingNegotiations > 1 {	79✔
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")	79✔
204		79✔
205	n.wg.Add(1)	79✔
206	go n.negotiate()	79✔
207
208	case session := <-n.successfulNegotiations:	73✔
209	select {	73✔
210	case n.newSessions <- session:	73✔
211	pendingNegotiations--	73✔
212	case <-n.quit:	×
213	return	×
214	}
215
216	if pendingNegotiations > 0 {	73✔
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:	36✔
225	return	36✔
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() {	79✔
240	defer n.wg.Done()	79✔
241		79✔
242	// On the first pass, initialize the backoff to our configured min	79✔
243	// backoff.	79✔
244	var backoff time.Duration	79✔
245		79✔
246	// Create a closure to update the backoff upon failure such that it	79✔
247	// stays within our min and max backoff parameters.	79✔
248	updateBackoff := func() {	265✔
249	if backoff == 0 {	259✔
250	backoff = n.cfg.MinBackoff	73✔
251	} else {	186✔
252	backoff *= 2	113✔
253	if backoff > n.cfg.MaxBackoff {	171✔
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 {	451✔
262	select {	186✔
263	case <-time.After(backoff):	180✔
264	case <-n.quit:	6✔
265	return	6✔
266	}
267	}
268
269	tryNextCandidate:	259✔
270	for {	518✔
271	select {	259✔
272	case <-n.quit:	×
273	return	×
274	default:	259✔
275	}
276
277	// Pull the next candidate from our list of addresses.
278	tower, err := n.cfg.Candidates.Next()	259✔
279	if err != nil {	405✔
280	// We've run out of addresses, update our backoff.	146✔
281	updateBackoff()	146✔
282		146✔
283	n.log.Debugf("Unable to get new tower candidate, "+	146✔
284	"retrying after %v -- reason: %v", backoff, err)	146✔
285		146✔
286	// Only reset the iterator once we've exhausted all	146✔
287	// candidates. Doing so allows us to load balance	146✔
288	// sessions better amongst all of the tower candidates.	146✔
289	if err == ErrTowerCandidatesExhausted {	292✔
290	n.cfg.Candidates.Reset()	146✔
291	}	146✔
292
293	goto retryWithBackoff	146✔
294	}
295
296	towerPub := tower.IdentityKey.SerializeCompressed()	113✔
297	n.log.Debugf("Attempting session negotiation with tower=%x",	113✔
298	towerPub)	113✔
299		113✔
300	var forceNextKey bool	113✔
301	for {	227✔
302	// Before proceeding, we will reserve a session key	114✔
303	// index to use with this specific tower. If one is	114✔
304	// already reserved, the existing index will be	114✔
305	// returned.	114✔
306	keyIndex, err := n.cfg.DB.NextSessionKeyIndex(	114✔
307	tower.ID, n.cfg.Policy.BlobType, forceNextKey,	114✔
308	)	114✔
309	if err != nil {	114✔
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)	114✔
320	if err == nil {	187✔
321	return	73✔
322	} else if errors.Is(err, ErrSessionKeyAlreadyUsed) {	115✔
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 {	114✔
343	sessionKeyDesc, err := n.cfg.SecretKeyRing.DeriveKey(	114✔
344	keychain.KeyLocator{	114✔
345	Family: keychain.KeyFamilyTowerSession,	114✔
346	Index: keyIndex,	114✔
347	},	114✔
348	)	114✔
349	if err != nil {	114✔
350	return err	×
351	}	×
352	sessionKey := keychain.NewPubKeyECDH(	114✔
353	sessionKeyDesc, n.cfg.SecretKeyRing,	114✔
354	)	114✔
355		114✔
356	addr := tower.Addresses.PeekAndLock()	114✔
357	for {	228✔
358	lnAddr := &lnwire.NetAddress{	114✔
359	IdentityKey: tower.IdentityKey,	114✔
360	Address: addr,	114✔
361	}	114✔
362		114✔
363	err = n.tryAddress(sessionKey, keyIndex, tower, lnAddr)	114✔
364	tower.Addresses.ReleaseLock(addr)	114✔
365	switch {	114✔
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:	73✔
390	return nil	73✔
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 {	114✔
401		114✔
402	// Connect to the tower address using our generated session key.	114✔
403	conn, err := n.cfg.Dial(sessionKey, lnAddr)	114✔
404	if err != nil {	119✔
405	return err	5✔
406	}	5✔
407
408	// Send local Init message.
409	err = n.cfg.SendMessage(conn, n.localInit)	109✔
410	if err != nil {	110✔
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)	108✔
416	if err != nil {	108✔
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)	108✔
422	if !ok {	108✔
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)	108✔
428	if err != nil {	108✔
429	return err	×
430	}	×
431
432	policy := n.cfg.Policy	108✔
433	createSession := &wtwire.CreateSession{	108✔
434	BlobType: policy.BlobType,	108✔
435	MaxUpdates: policy.MaxUpdates,	108✔
436	RewardBase: policy.RewardBase,	108✔
437	RewardRate: policy.RewardRate,	108✔
438	SweepFeeRate: policy.SweepFeeRate,	108✔
439	}	108✔
440		108✔
441	// Send CreateSession message.	108✔
442	err = n.cfg.SendMessage(conn, createSession)	108✔
443	if err != nil {	108✔
444	return fmt.Errorf("unable to send CreateSession: %w", err)	×
445	}	×
446
447	// Receive CreateSessionReply message.
448	remoteMsg, err = n.cfg.ReadMessage(conn)	108✔
449	if err != nil {	140✔
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)	76✔
455	if !ok {	76✔
456	return fmt.Errorf("expected CreateSessionReply, got %T in "+	×
457	"reply", remoteMsg)	×
458	}	×
459
460	switch createSessionReply.Code {	76✔
461	case wtwire.CodeOK:	75✔
462	// TODO(conner): validate reward address	75✔
463	rewardPkScript := createSessionReply.Data	75✔
464		75✔
465	sessionID := wtdb.NewSessionIDFromPubKey(sessionKey.PubKey())	75✔
466	dbClientSession := &wtdb.ClientSession{	75✔
467	ClientSessionBody: wtdb.ClientSessionBody{	75✔
468	TowerID: tower.ID,	75✔
469	KeyIndex: keyIndex,	75✔
470	Policy: n.cfg.Policy,	75✔
471	RewardPkScript: rewardPkScript,	75✔
472	},	75✔
473	ID: sessionID,	75✔
474	}	75✔
475		75✔
476	err = n.cfg.DB.CreateClientSession(dbClientSession)	75✔
477	if err != nil {	75✔
478	return fmt.Errorf("unable to persist ClientSession: %w",	×
479	err)	×
480	}	×
481
482	n.log.Debugf("New session negotiated with %s, policy: %s",	75✔
483	lnAddr, dbClientSession.Policy)	75✔
484		75✔
485	clientSession := &ClientSession{	75✔
486	ID: sessionID,	75✔
487	ClientSessionBody: dbClientSession.ClientSessionBody,	75✔
488	Tower: tower,	75✔
489	SessionKeyECDH: sessionKey,	75✔
490	}	75✔
491		75✔
492	// We have a newly negotiated session, return it to the	75✔
493	// dispatcher so that it can update how many outstanding	75✔
494	// negotiation requests we have.	75✔
495	select {	75✔
496	case n.successfulNegotiations <- clientSession:	73✔
497	return nil	73✔
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 / 14358372723

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