12986279612

Committed 27 Jan 2025 09:51AM UTC coverage: 57.652% (-1.1%) from 58.788%

Build # 12986279612

Build Type

Pull #9447

github

Committed by

yyforyongyu

Commit Message

sweep: rename methods for clarity

We now rename "third party" to "unknown" as the inputs can be spent via
an older sweeping tx, a third party (anchor), or a remote party (pin).
In fee bumper we don't have the info to distinguish the above cases, and
leave them to be further handled by the sweeper as it has more context.

Pull Request Pull Request #9447: sweep: start tracking input spending status in the fee bumper

Run Details

83 of 87 new or added lines in 2 files covered. (95.4%)

19578 existing lines in 256 files now uncovered.

103448 of 179434 relevant lines covered (57.65%)

24884.58 hits per line

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

64.62

/watchtower/lookout/lookout.go

package lookout

import (
        "errors"
        "fmt"
        "sync"
        "sync/atomic"
        "time"

        "github.com/btcsuite/btcd/chaincfg/chainhash"
        "github.com/btcsuite/btcd/wire"
        "github.com/lightningnetwork/lnd/chainntnfs"
        "github.com/lightningnetwork/lnd/watchtower/blob"
)

// ErrLookoutExiting is an error that is returned when the lookout server is
// in the process of shutting down.
var ErrLookoutExiting = errors.New("lookout server is shutting down")

// Config houses the Lookout's required resources to properly fulfill it's duty,
// including block fetching, querying accepted state updates, and construction
// and publication of justice transactions.
type Config struct {
        // DB provides persistent access to the watchtower's accepted state
        // updates such that they can be queried as new blocks arrive from the
        // network.
        DB DB

        // EpochRegistrar supports the ability to register for events corresponding to
        // newly created blocks.
        EpochRegistrar EpochRegistrar

        // BlockFetcher supports the ability to fetch blocks from the backend or
        // network.
        BlockFetcher BlockFetcher

        // Punisher handles the responsibility of crafting and broadcasting
        // justice transaction for any breached transactions.
        Punisher Punisher

        // MinBackoff is the minimum amount of time to back-off before
        // re-attempting to fetch a block.
        MinBackoff time.Duration

        // MaxBackoff is the maximum amount of time to back-off before
        // re-attempting to fetch a block.
        MaxBackoff time.Duration

        // MaxNumRetries is the maximum number of times that we should
        // re-attempt fetching a block before moving on.
        MaxNumRetries int
}

// Lookout will check any incoming blocks against the transactions found in the
// database, and in case of matches send the information needed to create a
// penalty transaction to the punisher.
type Lookout struct {
        started  int32 // atomic
        shutdown int32 // atomic

        cfg *Config

        wg   sync.WaitGroup
        quit chan struct{}
}

// New constructs a new Lookout from the given LookoutConfig.
func New(cfg *Config) *Lookout {
        return &Lookout{
                cfg:  cfg,
                quit: make(chan struct{}),
        }
}

// Start safely spins up the Lookout and begins monitoring for breaches.
func (l *Lookout) Start() error {
        if !atomic.CompareAndSwapInt32(&l.started, 0, 1) {
                return nil
        }

        log.Infof("Starting lookout")

        startEpoch, err := l.cfg.DB.GetLookoutTip()
        if err != nil {
                return err
        }

        if startEpoch == nil {
                log.Infof("Starting lookout from chain tip")
        } else {
                log.Infof("Starting lookout from epoch(height=%d hash=%v)",
                        startEpoch.Height, startEpoch.Hash)
        }

        events, err := l.cfg.EpochRegistrar.RegisterBlockEpochNtfn(startEpoch)
        if err != nil {
                log.Errorf("Unable to register for block epochs: %v", err)
                return err
        }

        l.wg.Add(1)
        go l.watchBlocks(events)

        log.Infof("Lookout started successfully")

        return nil
}

// Stop safely shuts down the Lookout.
func (l *Lookout) Stop() error {
        if !atomic.CompareAndSwapInt32(&l.shutdown, 0, 1) {
                return nil
        }

        log.Infof("Stopping lookout")

        close(l.quit)
        l.wg.Wait()

        log.Infof("Lookout stopped successfully")

        return nil
}

// fetchBlockWithRetries attempts to fetch a block from the blockchain using
// its hash. If it fails to fetch the block, it will back-off and retry up to
// MaxNumRetries times.
func (l *Lookout) fetchBlockWithRetries(hash *chainhash.Hash) (*wire.MsgBlock,
        error) {

        backoff := l.cfg.MinBackoff

        updateBackoff := func() {
                backoff *= 2
                if backoff > l.cfg.MaxBackoff {
                        backoff = l.cfg.MaxBackoff
                }
        }

        var attempt int
        for {
                attempt++

                block, err := l.cfg.BlockFetcher.GetBlock(hash)
                if err == nil {
                        return block, nil
                }

                if attempt > l.cfg.MaxNumRetries {
                        return nil, fmt.Errorf("failed to fetch block %s "+
                                "after %d attempts: %v", hash, attempt, err)
                }

                log.Errorf("Failed to fetch block %s (attempt %d): %v. "+
                        "Retrying in %v seconds", hash, attempt, err,
                        backoff.Seconds())

                select {
                case <-time.After(backoff):
                case <-l.quit:
                        return nil, ErrLookoutExiting
                }

                updateBackoff()
        }
}

// watchBlocks serially pulls incoming epochs from the epoch source and searches
// our accepted state updates for any breached transactions. If any are found,
// we will attempt to decrypt the state updates' encrypted blobs and exact
// justice for the victim.
//
// This method MUST be run as a goroutine.
func (l *Lookout) watchBlocks(epochs *chainntnfs.BlockEpochEvent) {
        defer l.wg.Done()
        defer epochs.Cancel()

        for {
                select {
                case epoch := <-epochs.Epochs:
                        log.Debugf("Fetching block for (height=%d, hash=%s)",
                                epoch.Height, epoch.Hash)

                        // Fetch the full block corresponding to the newly
                        // arriving epoch from the backend.
                        block, err := l.fetchBlockWithRetries(epoch.Hash)
                        if err != nil {
                                log.Errorf("Unable to fetch block for "+
                                        "(height=%x, hash=%s): %v",
                                        epoch.Height, epoch.Hash, err)
                                continue
                        }

                        // Process the block to see if it contains any breaches
                        // that we are monitoring on behalf of our clients.
                        err = l.processEpoch(epoch, block)
                        if err != nil {
                                log.Errorf("Unable to process %v: %v",
                                        epoch, err)
                        }

                case <-l.quit:
                        return
                }
        }
}

// processEpoch accepts an Epoch and queries the database for any matching state
// updates for the confirmed transactions. If any are found, the lookout
// responds by attempting to decrypt the encrypted blob and publishing the
// justice transaction.
func (l *Lookout) processEpoch(epoch *chainntnfs.BlockEpoch,
        block *wire.MsgBlock) error {

        numTxnsInBlock := len(block.Transactions)

        log.Debugf("Scanning %d transaction in block (height=%d, hash=%s) "+
                "for breaches", numTxnsInBlock, epoch.Height, epoch.Hash)

        // Iterate over the transactions contained in the block, deriving a
        // breach hint for each transaction and constructing an index mapping
        // the hint back to it's original transaction.
        hintToTx := make(map[blob.BreachHint]*wire.MsgTx, numTxnsInBlock)
        txHints := make([]blob.BreachHint, 0, numTxnsInBlock)
        for _, tx := range block.Transactions {
                hash := tx.TxHash()
                hint := blob.NewBreachHintFromHash(&hash)

                txHints = append(txHints, hint)
                hintToTx[hint] = tx.Copy()
        }

        // Query the database to see if any of the breach hints cause a match
        // with any of our accepted state updates.
        matches, err := l.cfg.DB.QueryMatches(txHints)
        if err != nil {
                return err
        }

        // No matches were found, we are done.
        if len(matches) == 0 {
                log.Debugf("No breaches found in (height=%d, hash=%s)",
                        epoch.Height, epoch.Hash)
                return nil
        }

        breachCountStr := "breach"
        if len(matches) > 1 {
                breachCountStr = "breaches"
        }

        log.Infof("Found %d %s in (height=%d, hash=%s)",
                len(matches), breachCountStr, epoch.Height, epoch.Hash)

        // For each match, use our index to retrieve the original transaction,
        // which corresponds to the breaching commitment transaction. If the
        // decryption succeeds, we will accumulate the assembled justice
        // descriptors in a single slice
        var successes []*JusticeDescriptor
        for _, match := range matches {
                commitTx := hintToTx[match.Hint]
                log.Infof("Dispatching punisher for client %s, breach-txid=%s",
                        match.ID, commitTx.TxHash())

                // The decryption key for the state update should be the full
                // txid of the breaching commitment transaction.
                // The decryption key for the state update should be computed as
                //   key = SHA256(txid || txid).
                breachTxID := commitTx.TxHash()
                breachKey := blob.NewBreachKeyFromHash(&breachTxID)

                // Now, decrypt the blob of justice that we received in the
                // state update. This will contain all information required to
                // sweep the breached commitment outputs.
                justiceKit, err := blob.Decrypt(
                        breachKey, match.EncryptedBlob,
                        match.SessionInfo.Policy.BlobType,
                )
                if err != nil {
                        // If the decryption fails, this implies either that the
                        // client sent an invalid blob, or that the breach hint
                        // caused a match on the txid, but this isn't actually
                        // the right transaction.
                        log.Debugf("Unable to decrypt blob for client %s, "+
                                "breach-txid %s: %v", match.ID,
                                commitTx.TxHash(), err)
                        continue
                }

                justiceDesc := &JusticeDescriptor{
                        BreachedCommitTx: commitTx,
                        SessionInfo:      match.SessionInfo,
                        JusticeKit:       justiceKit,
                }
                successes = append(successes, justiceDesc)
        }

        // TODO(conner): mark successfully decrypted blob so that we can
        // reliably rebroadcast on startup

        // Now, we'll dispatch a punishment for each successful match in
        // parallel. This will assemble the justice transaction for each and
        // watch for their confirmation on chain.
        for _, justiceDesc := range successes {
                l.wg.Add(1)
                go l.dispatchPunisher(justiceDesc)
        }

        return l.cfg.DB.SetLookoutTip(epoch)
}

// dispatchPunisher accepts a justice descriptor corresponding to a successfully
// decrypted blob.  The punisher will then construct the witness scripts and
// witness stacks for the breached outputs. If construction of the justice
// transaction is successful, it will be published to the network to retrieve
// the funds and claim the watchtower's reward.
//
// This method MUST be run as a goroutine.
func (l *Lookout) dispatchPunisher(desc *JusticeDescriptor) {
        defer l.wg.Done()

        // Give the justice descriptor to the punisher to construct and publish
        // the justice transaction. The lookout's quit channel is provided so
        // that long-running tasks that watch for on-chain events can be
        // canceled during shutdown since this method is waitgrouped.
        err := l.cfg.Punisher.Punish(desc, l.quit)
        if err != nil {
                log.Errorf("Unable to punish breach-txid %s for %s: %v",
                        desc.BreachedCommitTx.TxHash(), desc.SessionInfo.ID,
                        err)
                return
        }

        log.Infof("Punishment for client %s with breach-txid=%s dispatched",
                desc.SessionInfo.ID, desc.BreachedCommitTx.TxHash())
}

1	package lookout
2
3	import (
4	"errors"
5	"fmt"
6	"sync"
7	"sync/atomic"
8	"time"
9
10	"github.com/btcsuite/btcd/chaincfg/chainhash"
11	"github.com/btcsuite/btcd/wire"
12	"github.com/lightningnetwork/lnd/chainntnfs"
13	"github.com/lightningnetwork/lnd/watchtower/blob"
14	)
15
16	// ErrLookoutExiting is an error that is returned when the lookout server is
17	// in the process of shutting down.
18	var ErrLookoutExiting = errors.New("lookout server is shutting down")
19
20	// Config houses the Lookout's required resources to properly fulfill it's duty,
21	// including block fetching, querying accepted state updates, and construction
22	// and publication of justice transactions.
23	type Config struct {
24	// DB provides persistent access to the watchtower's accepted state
25	// updates such that they can be queried as new blocks arrive from the
26	// network.
27	DB DB
28
29	// EpochRegistrar supports the ability to register for events corresponding to
30	// newly created blocks.
31	EpochRegistrar EpochRegistrar
32
33	// BlockFetcher supports the ability to fetch blocks from the backend or
34	// network.
35	BlockFetcher BlockFetcher
36
37	// Punisher handles the responsibility of crafting and broadcasting
38	// justice transaction for any breached transactions.
39	Punisher Punisher
40
41	// MinBackoff is the minimum amount of time to back-off before
42	// re-attempting to fetch a block.
43	MinBackoff time.Duration
44
45	// MaxBackoff is the maximum amount of time to back-off before
46	// re-attempting to fetch a block.
47	MaxBackoff time.Duration
48
49	// MaxNumRetries is the maximum number of times that we should
50	// re-attempt fetching a block before moving on.
51	MaxNumRetries int
52	}
53
54	// Lookout will check any incoming blocks against the transactions found in the
55	// database, and in case of matches send the information needed to create a
56	// penalty transaction to the punisher.
57	type Lookout struct {
58	started int32 // atomic
59	shutdown int32 // atomic
60
61	cfg *Config
62
63	wg sync.WaitGroup
64	quit chan struct{}
65	}
66
67	// New constructs a new Lookout from the given LookoutConfig.
68	func New(cfg Config) Lookout {	1✔
69	return &Lookout{	1✔
70	cfg: cfg,	1✔
71	quit: make(chan struct{}),	1✔
72	}	1✔
73	}	1✔
74
75	// Start safely spins up the Lookout and begins monitoring for breaches.
76	func (l *Lookout) Start() error {	1✔
77	if !atomic.CompareAndSwapInt32(&l.started, 0, 1) {	1✔
78	return nil	×
79	}	×
80
81	log.Infof("Starting lookout")	1✔
82		1✔
83	startEpoch, err := l.cfg.DB.GetLookoutTip()	1✔
84	if err != nil {	1✔
85	return err	×
86	}	×
87
88	if startEpoch == nil {	2✔
89	log.Infof("Starting lookout from chain tip")	1✔
90	} else {	1✔
91	log.Infof("Starting lookout from epoch(height=%d hash=%v)",	×
92	startEpoch.Height, startEpoch.Hash)	×
93	}	×
94
95	events, err := l.cfg.EpochRegistrar.RegisterBlockEpochNtfn(startEpoch)	1✔
96	if err != nil {	1✔
97	log.Errorf("Unable to register for block epochs: %v", err)	×
98	return err	×
99	}	×
100
101	l.wg.Add(1)	1✔
102	go l.watchBlocks(events)	1✔
103		1✔
104	log.Infof("Lookout started successfully")	1✔
105		1✔
106	return nil	1✔
107	}
108
109	// Stop safely shuts down the Lookout.
UNCOV 110	func (l *Lookout) Stop() error {	×
UNCOV 111	if !atomic.CompareAndSwapInt32(&l.shutdown, 0, 1) {	×
112	return nil	×
113	}	×
114
UNCOV 115	log.Infof("Stopping lookout")	×
UNCOV 116		×
UNCOV 117	close(l.quit)	×
UNCOV 118	l.wg.Wait()	×
UNCOV 119		×
UNCOV 120	log.Infof("Lookout stopped successfully")	×
UNCOV 121		×
UNCOV 122	return nil	×
123	}
124
125	// fetchBlockWithRetries attempts to fetch a block from the blockchain using
126	// its hash. If it fails to fetch the block, it will back-off and retry up to
127	// MaxNumRetries times.
128	func (l Lookout) fetchBlockWithRetries(hash chainhash.Hash) (*wire.MsgBlock,
129	error) {	2✔
130		2✔
131	backoff := l.cfg.MinBackoff	2✔
132		2✔
133	updateBackoff := func() {	2✔
134	backoff *= 2	×
135	if backoff > l.cfg.MaxBackoff {	×
136	backoff = l.cfg.MaxBackoff	×
137	}	×
138	}
139
140	var attempt int	2✔
141	for {	4✔
142	attempt++	2✔
143		2✔
144	block, err := l.cfg.BlockFetcher.GetBlock(hash)	2✔
145	if err == nil {	4✔
146	return block, nil	2✔
147	}	2✔
148
149	if attempt > l.cfg.MaxNumRetries {	×
150	return nil, fmt.Errorf("failed to fetch block %s "+	×
151	"after %d attempts: %v", hash, attempt, err)	×
152	}	×
153
154	log.Errorf("Failed to fetch block %s (attempt %d): %v. "+	×
155	"Retrying in %v seconds", hash, attempt, err,	×
156	backoff.Seconds())	×
157		×
158	select {	×
159	case <-time.After(backoff):	×
160	case <-l.quit:	×
161	return nil, ErrLookoutExiting	×
162	}
163
164	updateBackoff()	×
165	}
166	}
167
168	// watchBlocks serially pulls incoming epochs from the epoch source and searches
169	// our accepted state updates for any breached transactions. If any are found,
170	// we will attempt to decrypt the state updates' encrypted blobs and exact
171	// justice for the victim.
172	//
173	// This method MUST be run as a goroutine.
174	func (l Lookout) watchBlocks(epochs chainntnfs.BlockEpochEvent) {	1✔
175	defer l.wg.Done()	1✔
176	defer epochs.Cancel()	1✔
177		1✔
178	for {	4✔
179	select {	3✔
180	case epoch := <-epochs.Epochs:	2✔
181	log.Debugf("Fetching block for (height=%d, hash=%s)",	2✔
182	epoch.Height, epoch.Hash)	2✔
183		2✔
184	// Fetch the full block corresponding to the newly	2✔
185	// arriving epoch from the backend.	2✔
186	block, err := l.fetchBlockWithRetries(epoch.Hash)	2✔
187	if err != nil {	2✔
188	log.Errorf("Unable to fetch block for "+	×
189	"(height=%x, hash=%s): %v",	×
190	epoch.Height, epoch.Hash, err)	×
191	continue	×
192	}
193
194	// Process the block to see if it contains any breaches
195	// that we are monitoring on behalf of our clients.
196	err = l.processEpoch(epoch, block)	2✔
197	if err != nil {	2✔
198	log.Errorf("Unable to process %v: %v",	×
199	epoch, err)	×
200	}	×
201
UNCOV 202	case <-l.quit:	×
UNCOV 203	return	×
204	}
205	}
206	}
207
208	// processEpoch accepts an Epoch and queries the database for any matching state
209	// updates for the confirmed transactions. If any are found, the lookout
210	// responds by attempting to decrypt the encrypted blob and publishing the
211	// justice transaction.
212	func (l Lookout) processEpoch(epoch chainntnfs.BlockEpoch,
213	block *wire.MsgBlock) error {	2✔
214		2✔
215	numTxnsInBlock := len(block.Transactions)	2✔
216		2✔
217	log.Debugf("Scanning %d transaction in block (height=%d, hash=%s) "+	2✔
218	"for breaches", numTxnsInBlock, epoch.Height, epoch.Hash)	2✔
219		2✔
220	// Iterate over the transactions contained in the block, deriving a	2✔
221	// breach hint for each transaction and constructing an index mapping	2✔
222	// the hint back to it's original transaction.	2✔
223	hintToTx := make(map[blob.BreachHint]*wire.MsgTx, numTxnsInBlock)	2✔
224	txHints := make([]blob.BreachHint, 0, numTxnsInBlock)	2✔
225	for _, tx := range block.Transactions {	4✔
226	hash := tx.TxHash()	2✔
227	hint := blob.NewBreachHintFromHash(&hash)	2✔
228		2✔
229	txHints = append(txHints, hint)	2✔
230	hintToTx[hint] = tx.Copy()	2✔
231	}	2✔
232
233	// Query the database to see if any of the breach hints cause a match
234	// with any of our accepted state updates.
235	matches, err := l.cfg.DB.QueryMatches(txHints)	2✔
236	if err != nil {	2✔
237	return err	×
238	}	×
239
240	// No matches were found, we are done.
241	if len(matches) == 0 {	2✔
UNCOV 242	log.Debugf("No breaches found in (height=%d, hash=%s)",	×
UNCOV 243	epoch.Height, epoch.Hash)	×
UNCOV 244	return nil	×
UNCOV 245	}	×
246
247	breachCountStr := "breach"	2✔
248	if len(matches) > 1 {	2✔
249	breachCountStr = "breaches"	×
250	}	×
251
252	log.Infof("Found %d %s in (height=%d, hash=%s)",	2✔
253	len(matches), breachCountStr, epoch.Height, epoch.Hash)	2✔
254		2✔
255	// For each match, use our index to retrieve the original transaction,	2✔
256	// which corresponds to the breaching commitment transaction. If the	2✔
257	// decryption succeeds, we will accumulate the assembled justice	2✔
258	// descriptors in a single slice	2✔
259	var successes []*JusticeDescriptor	2✔
260	for _, match := range matches {	4✔
261	commitTx := hintToTx[match.Hint]	2✔
262	log.Infof("Dispatching punisher for client %s, breach-txid=%s",	2✔
263	match.ID, commitTx.TxHash())	2✔
264		2✔
265	// The decryption key for the state update should be the full	2✔
266	// txid of the breaching commitment transaction.	2✔
267	// The decryption key for the state update should be computed as	2✔
268	// key = SHA256(txid \|\| txid).	2✔
269	breachTxID := commitTx.TxHash()	2✔
270	breachKey := blob.NewBreachKeyFromHash(&breachTxID)	2✔
271		2✔
272	// Now, decrypt the blob of justice that we received in the	2✔
273	// state update. This will contain all information required to	2✔
274	// sweep the breached commitment outputs.	2✔
275	justiceKit, err := blob.Decrypt(	2✔
276	breachKey, match.EncryptedBlob,	2✔
277	match.SessionInfo.Policy.BlobType,	2✔
278	)	2✔
279	if err != nil {	2✔
280	// If the decryption fails, this implies either that the	×
281	// client sent an invalid blob, or that the breach hint	×
282	// caused a match on the txid, but this isn't actually	×
283	// the right transaction.	×
284	log.Debugf("Unable to decrypt blob for client %s, "+	×
285	"breach-txid %s: %v", match.ID,	×
286	commitTx.TxHash(), err)	×
287	continue	×
288	}
289
290	justiceDesc := &JusticeDescriptor{	2✔
291	BreachedCommitTx: commitTx,	2✔
292	SessionInfo: match.SessionInfo,	2✔
293	JusticeKit: justiceKit,	2✔
294	}	2✔
295	successes = append(successes, justiceDesc)	2✔
296	}
297
298	// TODO(conner): mark successfully decrypted blob so that we can
299	// reliably rebroadcast on startup
300
301	// Now, we'll dispatch a punishment for each successful match in
302	// parallel. This will assemble the justice transaction for each and
303	// watch for their confirmation on chain.
304	for _, justiceDesc := range successes {	4✔
305	l.wg.Add(1)	2✔
306	go l.dispatchPunisher(justiceDesc)	2✔
307	}	2✔
308
309	return l.cfg.DB.SetLookoutTip(epoch)	2✔
310	}
311
312	// dispatchPunisher accepts a justice descriptor corresponding to a successfully
313	// decrypted blob. The punisher will then construct the witness scripts and
314	// witness stacks for the breached outputs. If construction of the justice
315	// transaction is successful, it will be published to the network to retrieve
316	// the funds and claim the watchtower's reward.
317	//
318	// This method MUST be run as a goroutine.
319	func (l Lookout) dispatchPunisher(desc JusticeDescriptor) {	2✔
320	defer l.wg.Done()	2✔
321		2✔
322	// Give the justice descriptor to the punisher to construct and publish	2✔
323	// the justice transaction. The lookout's quit channel is provided so	2✔
324	// that long-running tasks that watch for on-chain events can be	2✔
325	// canceled during shutdown since this method is waitgrouped.	2✔
326	err := l.cfg.Punisher.Punish(desc, l.quit)	2✔
327	if err != nil {	2✔
328	log.Errorf("Unable to punish breach-txid %s for %s: %v",	×
329	desc.BreachedCommitTx.TxHash(), desc.SessionInfo.ID,	×
330	err)	×
331	return	×
332	}	×
333
334	log.Infof("Punishment for client %s with breach-txid=%s dispatched",	2✔
335	desc.SessionInfo.ID, desc.BreachedCommitTx.TxHash())	2✔
336	}

lightningnetwork / lnd / 12986279612

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