13211764208

Committed 08 Feb 2025 03:08AM UTC coverage: 49.288% (-9.5%) from 58.815%

Build # 13211764208

Build Type

Pull #9489

github

Committed by

calvinrzachman

Commit Message

itest: verify switchrpc server enforces send then track

We prevent the rpc server from allowing onion dispatches for
attempt IDs which have already been tracked by rpc clients.

This helps protect the client from leaking a duplicate onion
attempt. NOTE: This is not the only method for solving this
issue! The issue could be addressed via careful client side
programming which accounts for the uncertainty and async
nature of dispatching onions to a remote process via RPC.
This would require some lnd ChannelRouter changes for how
we intend to use these RPCs though.

Pull Request Pull Request #9489: multi: add BuildOnion, SendOnion, and TrackOnion RPCs

Run Details

474 of 990 new or added lines in 11 files covered. (47.88%)

27321 existing lines in 435 files now uncovered.

101192 of 205306 relevant lines covered (49.29%)

1.54 hits per line

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

0.0

/channeldb/migration25/migration.go

package migration25

import (
        "bytes"
        "fmt"

        mig "github.com/lightningnetwork/lnd/channeldb/migration_01_to_11"
        "github.com/lightningnetwork/lnd/kvdb"
)

var (
        // openChanBucket stores all the currently open channels. This bucket
        // has a second, nested bucket which is keyed by a node's ID. Within
        // that node ID bucket, all attributes required to track, update, and
        // close a channel are stored.
        openChannelBucket = []byte("open-chan-bucket")

        // chanInfoKey can be accessed within the bucket for a channel
        // (identified by its chanPoint). This key stores all the static
        // information for a channel which is decided at the end of  the
        // funding flow.
        chanInfoKey = []byte("chan-info-key")

        // ErrNoChanDBExists is returned when a channel bucket hasn't been
        // created.
        ErrNoChanDBExists = fmt.Errorf("channel db has not yet been created")

        // ErrNoActiveChannels  is returned when there is no active (open)
        // channels within the database.
        ErrNoActiveChannels = fmt.Errorf("no active channels exist")

        // ErrChannelNotFound is returned when we attempt to locate a channel
        // for a specific chain, but it is not found.
        ErrChannelNotFound = fmt.Errorf("channel not found")
)

// MigrateInitialBalances patches the two new fields, InitialLocalBalance and
// InitialRemoteBalance, for all the open channels. It does so by reading the
// revocation log at height 0 to learn the initial balances and then updates
// the channel's info.
// The channel info is saved in the nested bucket which is accessible via
// nodePub:chainHash:chanPoint. If any of the sub-buckets turns out to be nil,
// we will log the error and continue to process the rest.
func MigrateInitialBalances(tx kvdb.RwTx) error {
        log.Infof("Migrating initial local and remote balances...")

        openChanBucket := tx.ReadWriteBucket(openChannelBucket)

        // If no bucket is found, we can exit early.
        if openChanBucket == nil {
                return nil
        }

        // Read a list of open channels.
        channels, err := findOpenChannels(openChanBucket)
        if err != nil {
                return err
        }

        // Migrate the balances.
        for _, c := range channels {
                if err := migrateBalances(tx, c); err != nil {
                        return err
                }
        }

        return err
}

// findOpenChannels finds all open channels.
func findOpenChannels(openChanBucket kvdb.RBucket) ([]*OpenChannel, error) {
        channels := []*OpenChannel{}

        // readChannel is a helper closure that reads the channel info from the
        // channel bucket.
        readChannel := func(chainBucket kvdb.RBucket, cp []byte) error {
                c := &OpenChannel{}

                // Read the sub-bucket level 3.
                chanBucket := chainBucket.NestedReadBucket(
                        cp,
                )
                if chanBucket == nil {
                        log.Errorf("unable to read bucket for chanPoint=%x", cp)
                        return nil
                }
                // Get the old channel info.
                if err := fetchChanInfo(chanBucket, c, true); err != nil {
                        return fmt.Errorf("unable to fetch chan info: %w", err)
                }

                // Fetch the channel commitments, which are useful for freshly
                // open channels as they don't have any revocation logs and
                // their current commitments reflect the initial balances.
                if err := FetchChanCommitments(chanBucket, c); err != nil {
                        return fmt.Errorf("unable to fetch chan commits: %w",
                                err)
                }

                channels = append(channels, c)

                return nil
        }

        // Iterate the root bucket.
        err := openChanBucket.ForEach(func(nodePub, v []byte) error {
                // Ensure that this is a key the same size as a pubkey, and
                // also that it leads directly to a bucket.
                if len(nodePub) != 33 || v != nil {
                        return nil
                }

                // Read the sub-bucket level 1.
                nodeChanBucket := openChanBucket.NestedReadBucket(nodePub)
                if nodeChanBucket == nil {
                        log.Errorf("no bucket for node %x", nodePub)
                        return nil
                }

                // Iterate the bucket.
                return nodeChanBucket.ForEach(func(chainHash, _ []byte) error {
                        // Read the sub-bucket level 2.
                        chainBucket := nodeChanBucket.NestedReadBucket(
                                chainHash,
                        )
                        if chainBucket == nil {
                                log.Errorf("unable to read bucket for chain=%x",
                                        chainHash)
                                return nil
                        }

                        // Iterate the bucket.
                        return chainBucket.ForEach(func(cp, _ []byte) error {
                                return readChannel(chainBucket, cp)
                        })
                })
        })

        if err != nil {
                return nil, err
        }

        return channels, nil
}

// migrateBalances queries the revocation log at height 0 to find the initial
// balances and save them to the channel info.
func migrateBalances(tx kvdb.RwTx, c *OpenChannel) error {
        // Get the bucket.
        chanBucket, err := FetchChanBucket(tx, c)
        if err != nil {
                return err
        }
        // Get the initial balances.
        localAmt, remoteAmt, err := c.balancesAtHeight(chanBucket, 0)
        if err != nil {
                return fmt.Errorf("unable to get initial balances: %w", err)
        }

        c.InitialLocalBalance = localAmt
        c.InitialRemoteBalance = remoteAmt

        // Update the channel info.
        if err := putChanInfo(chanBucket, c, false); err != nil {
                return fmt.Errorf("unable to put chan info: %w", err)
        }

        return nil
}

// FetchChanBucket is a helper function that returns the bucket where a
// channel's data resides in given: the public key for the node, the outpoint,
// and the chainhash that the channel resides on.
func FetchChanBucket(tx kvdb.RwTx, c *OpenChannel) (kvdb.RwBucket, error) {
        // First fetch the top level bucket which stores all data related to
        // current, active channels.
        openChanBucket := tx.ReadWriteBucket(openChannelBucket)
        if openChanBucket == nil {
                return nil, ErrNoChanDBExists
        }

        // Within this top level bucket, fetch the bucket dedicated to storing
        // open channel data specific to the remote node.
        nodePub := c.IdentityPub.SerializeCompressed()
        nodeChanBucket := openChanBucket.NestedReadWriteBucket(nodePub)
        if nodeChanBucket == nil {
                return nil, ErrNoActiveChannels
        }

        // We'll then recurse down an additional layer in order to fetch the
        // bucket for this particular chain.
        chainBucket := nodeChanBucket.NestedReadWriteBucket(c.ChainHash[:])
        if chainBucket == nil {
                return nil, ErrNoActiveChannels
        }

        // With the bucket for the node and chain fetched, we can now go down
        // another level, for this channel itself.
        var chanPointBuf bytes.Buffer
        err := mig.WriteOutpoint(&chanPointBuf, &c.FundingOutpoint)
        if err != nil {
                return nil, err
        }

        chanBucket := chainBucket.NestedReadWriteBucket(chanPointBuf.Bytes())
        if chanBucket == nil {
                return nil, ErrChannelNotFound
        }

        return chanBucket, nil
}

1	package migration25
2
3	import (
4	"bytes"
5	"fmt"
6
7	mig "github.com/lightningnetwork/lnd/channeldb/migration_01_to_11"
8	"github.com/lightningnetwork/lnd/kvdb"
9	)
10
11	var (
12	// openChanBucket stores all the currently open channels. This bucket
13	// has a second, nested bucket which is keyed by a node's ID. Within
14	// that node ID bucket, all attributes required to track, update, and
15	// close a channel are stored.
16	openChannelBucket = []byte("open-chan-bucket")
17
18	// chanInfoKey can be accessed within the bucket for a channel
19	// (identified by its chanPoint). This key stores all the static
20	// information for a channel which is decided at the end of the
21	// funding flow.
22	chanInfoKey = []byte("chan-info-key")
23
24	// ErrNoChanDBExists is returned when a channel bucket hasn't been
25	// created.
26	ErrNoChanDBExists = fmt.Errorf("channel db has not yet been created")
27
28	// ErrNoActiveChannels is returned when there is no active (open)
29	// channels within the database.
30	ErrNoActiveChannels = fmt.Errorf("no active channels exist")
31
32	// ErrChannelNotFound is returned when we attempt to locate a channel
33	// for a specific chain, but it is not found.
34	ErrChannelNotFound = fmt.Errorf("channel not found")
35	)
36
37	// MigrateInitialBalances patches the two new fields, InitialLocalBalance and
38	// InitialRemoteBalance, for all the open channels. It does so by reading the
39	// revocation log at height 0 to learn the initial balances and then updates
40	// the channel's info.
41	// The channel info is saved in the nested bucket which is accessible via
42	// nodePub:chainHash:chanPoint. If any of the sub-buckets turns out to be nil,
43	// we will log the error and continue to process the rest.
UNCOV 44	func MigrateInitialBalances(tx kvdb.RwTx) error {	×
UNCOV 45	log.Infof("Migrating initial local and remote balances...")	×
UNCOV 46		×
UNCOV 47	openChanBucket := tx.ReadWriteBucket(openChannelBucket)	×
UNCOV 48		×
UNCOV 49	// If no bucket is found, we can exit early.	×
UNCOV 50	if openChanBucket == nil {	×
51	return nil	×
52	}	×
53
54	// Read a list of open channels.
UNCOV 55	channels, err := findOpenChannels(openChanBucket)	×
UNCOV 56	if err != nil {	×
57	return err	×
58	}	×
59
60	// Migrate the balances.
UNCOV 61	for _, c := range channels {	×
UNCOV 62	if err := migrateBalances(tx, c); err != nil {	×
UNCOV 63	return err	×
UNCOV 64	}	×
65	}
66
UNCOV 67	return err	×
68	}
69
70	// findOpenChannels finds all open channels.
UNCOV 71	func findOpenChannels(openChanBucket kvdb.RBucket) ([]*OpenChannel, error) {	×
UNCOV 72	channels := []*OpenChannel{}	×
UNCOV 73		×
UNCOV 74	// readChannel is a helper closure that reads the channel info from the	×
UNCOV 75	// channel bucket.	×
UNCOV 76	readChannel := func(chainBucket kvdb.RBucket, cp []byte) error {	×
UNCOV 77	c := &OpenChannel{}	×
UNCOV 78		×
UNCOV 79	// Read the sub-bucket level 3.	×
UNCOV 80	chanBucket := chainBucket.NestedReadBucket(	×
UNCOV 81	cp,	×
UNCOV 82	)	×
UNCOV 83	if chanBucket == nil {	×
84	log.Errorf("unable to read bucket for chanPoint=%x", cp)	×
85	return nil	×
86	}	×
87	// Get the old channel info.
UNCOV 88	if err := fetchChanInfo(chanBucket, c, true); err != nil {	×
89	return fmt.Errorf("unable to fetch chan info: %w", err)	×
90	}	×
91
92	// Fetch the channel commitments, which are useful for freshly
93	// open channels as they don't have any revocation logs and
94	// their current commitments reflect the initial balances.
UNCOV 95	if err := FetchChanCommitments(chanBucket, c); err != nil {	×
96	return fmt.Errorf("unable to fetch chan commits: %w",	×
97	err)	×
98	}	×
99
UNCOV 100	channels = append(channels, c)	×
UNCOV 101		×
UNCOV 102	return nil	×
103	}
104
105	// Iterate the root bucket.
UNCOV 106	err := openChanBucket.ForEach(func(nodePub, v []byte) error {	×
UNCOV 107	// Ensure that this is a key the same size as a pubkey, and	×
UNCOV 108	// also that it leads directly to a bucket.	×
UNCOV 109	if len(nodePub) != 33 \|\| v != nil {	×
110	return nil	×
111	}	×
112
113	// Read the sub-bucket level 1.
UNCOV 114	nodeChanBucket := openChanBucket.NestedReadBucket(nodePub)	×
UNCOV 115	if nodeChanBucket == nil {	×
116	log.Errorf("no bucket for node %x", nodePub)	×
117	return nil	×
118	}	×
119
120	// Iterate the bucket.
UNCOV 121	return nodeChanBucket.ForEach(func(chainHash, _ []byte) error {	×
UNCOV 122	// Read the sub-bucket level 2.	×
UNCOV 123	chainBucket := nodeChanBucket.NestedReadBucket(	×
UNCOV 124	chainHash,	×
UNCOV 125	)	×
UNCOV 126	if chainBucket == nil {	×
127	log.Errorf("unable to read bucket for chain=%x",	×
128	chainHash)	×
129	return nil	×
130	}	×
131
132	// Iterate the bucket.
UNCOV 133	return chainBucket.ForEach(func(cp, _ []byte) error {	×
UNCOV 134	return readChannel(chainBucket, cp)	×
UNCOV 135	})	×
136	})
137	})
138
UNCOV 139	if err != nil {	×
140	return nil, err	×
141	}	×
142
UNCOV 143	return channels, nil	×
144	}
145
146	// migrateBalances queries the revocation log at height 0 to find the initial
147	// balances and save them to the channel info.
UNCOV 148	func migrateBalances(tx kvdb.RwTx, c *OpenChannel) error {	×
UNCOV 149	// Get the bucket.	×
UNCOV 150	chanBucket, err := FetchChanBucket(tx, c)	×
UNCOV 151	if err != nil {	×
152	return err	×
153	}	×
154	// Get the initial balances.
UNCOV 155	localAmt, remoteAmt, err := c.balancesAtHeight(chanBucket, 0)	×
UNCOV 156	if err != nil {	×
UNCOV 157	return fmt.Errorf("unable to get initial balances: %w", err)	×
UNCOV 158	}	×
159
UNCOV 160	c.InitialLocalBalance = localAmt	×
UNCOV 161	c.InitialRemoteBalance = remoteAmt	×
UNCOV 162		×
UNCOV 163	// Update the channel info.	×
UNCOV 164	if err := putChanInfo(chanBucket, c, false); err != nil {	×
165	return fmt.Errorf("unable to put chan info: %w", err)	×
166	}	×
167
UNCOV 168	return nil	×
169	}
170
171	// FetchChanBucket is a helper function that returns the bucket where a
172	// channel's data resides in given: the public key for the node, the outpoint,
173	// and the chainhash that the channel resides on.
UNCOV 174	func FetchChanBucket(tx kvdb.RwTx, c *OpenChannel) (kvdb.RwBucket, error) {	×
UNCOV 175	// First fetch the top level bucket which stores all data related to	×
UNCOV 176	// current, active channels.	×
UNCOV 177	openChanBucket := tx.ReadWriteBucket(openChannelBucket)	×
UNCOV 178	if openChanBucket == nil {	×
179	return nil, ErrNoChanDBExists	×
180	}	×
181
182	// Within this top level bucket, fetch the bucket dedicated to storing
183	// open channel data specific to the remote node.
UNCOV 184	nodePub := c.IdentityPub.SerializeCompressed()	×
UNCOV 185	nodeChanBucket := openChanBucket.NestedReadWriteBucket(nodePub)	×
UNCOV 186	if nodeChanBucket == nil {	×
187	return nil, ErrNoActiveChannels	×
188	}	×
189
190	// We'll then recurse down an additional layer in order to fetch the
191	// bucket for this particular chain.
UNCOV 192	chainBucket := nodeChanBucket.NestedReadWriteBucket(c.ChainHash[:])	×
UNCOV 193	if chainBucket == nil {	×
194	return nil, ErrNoActiveChannels	×
195	}	×
196
197	// With the bucket for the node and chain fetched, we can now go down
198	// another level, for this channel itself.
UNCOV 199	var chanPointBuf bytes.Buffer	×
UNCOV 200	err := mig.WriteOutpoint(&chanPointBuf, &c.FundingOutpoint)	×
UNCOV 201	if err != nil {	×
202	return nil, err	×
203	}	×
204
UNCOV 205	chanBucket := chainBucket.NestedReadWriteBucket(chanPointBuf.Bytes())	×
UNCOV 206	if chanBucket == nil {	×
207	return nil, ErrChannelNotFound	×
208	}	×
209
UNCOV 210	return chanBucket, nil	×
211	}

lightningnetwork / lnd / 13211764208

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