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

/watchtower/wtdb/migration7/client_db.go

package migration7

import (
        "bytes"
        "encoding/binary"
        "errors"
        "fmt"

        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/tlv"
)

var (
        // cSessionBkt is a top-level bucket storing:
        //   session-id => cSessionBody -> encoded ClientSessionBody
        //                 => cSessionDBID -> db-assigned-id
        //              => cSessionCommits => seqnum -> encoded CommittedUpdate
        //              => cSessionAckRangeIndex => chan-id => acked-index-range
        cSessionBkt = []byte("client-session-bucket")

        // cChanDetailsBkt is a top-level bucket storing:
        //   channel-id => cChannelSummary -> encoded ClientChanSummary.
        //                  => cChanDBID -> db-assigned-id
        //                 => cChanSessions => db-session-id -> 1
        cChanDetailsBkt = []byte("client-channel-detail-bucket")

        // cChannelSummary is a sub-bucket of cChanDetailsBkt which stores the
        // encoded body of ClientChanSummary.
        cChannelSummary = []byte("client-channel-summary")

        // cChanSessions is a sub-bucket of cChanDetailsBkt which stores:
        //    session-id -> 1
        cChanSessions = []byte("client-channel-sessions")

        // cSessionAckRangeIndex is a sub-bucket of cSessionBkt storing:
        //    chan-id => start -> end
        cSessionAckRangeIndex = []byte("client-session-ack-range-index")

        // cSessionDBID is a key used in the cSessionBkt to store the
        // db-assigned-d of a session.
        cSessionDBID = []byte("client-session-db-id")

        // cChanIDIndexBkt is a top-level bucket storing:
        //    db-assigned-id -> channel-ID
        cChanIDIndexBkt = []byte("client-channel-id-index")

        // ErrUninitializedDB signals that top-level buckets for the database
        // have not been initialized.
        ErrUninitializedDB = errors.New("db not initialized")

        // ErrCorruptClientSession signals that the client session's on-disk
        // structure deviates from what is expected.
        ErrCorruptClientSession = errors.New("client session corrupted")

        // byteOrder is the default endianness used when serializing integers.
        byteOrder = binary.BigEndian
)

// MigrateChannelToSessionIndex migrates the tower client DB to add an index
// from channel-to-session. This will make it easier in future to check which
// sessions have updates for which channels.
func MigrateChannelToSessionIndex(tx kvdb.RwTx) error {
        log.Infof("Migrating the tower client DB to build a new " +
                "channel-to-session index")

        sessionsBkt := tx.ReadBucket(cSessionBkt)
        if sessionsBkt == nil {
                return ErrUninitializedDB
        }

        chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)
        if chanDetailsBkt == nil {
                return ErrUninitializedDB
        }

        chanIDsBkt := tx.ReadBucket(cChanIDIndexBkt)
        if chanIDsBkt == nil {
                return ErrUninitializedDB
        }

        // First gather all the new channel-to-session pairs that we want to
        // add.
        index, err := collectIndex(sessionsBkt)
        if err != nil {
                return err
        }

        // Then persist those pairs to the db.
        return persistIndex(chanDetailsBkt, chanIDsBkt, index)
}

// collectIndex iterates through all the sessions and uses the keys in the
// cSessionAckRangeIndex bucket to collect all the channels that the session
// has updates for. The function returns a map from channel ID to session ID
// (using the db-assigned IDs for both).
func collectIndex(sessionsBkt kvdb.RBucket) (map[uint64]map[uint64]bool,
        error) {

        index := make(map[uint64]map[uint64]bool)
        err := sessionsBkt.ForEach(func(sessID, _ []byte) error {
                sessionBkt := sessionsBkt.NestedReadBucket(sessID)
                if sessionBkt == nil {
                        return ErrCorruptClientSession
                }

                ackedRanges := sessionBkt.NestedReadBucket(
                        cSessionAckRangeIndex,
                )
                if ackedRanges == nil {
                        return nil
                }

                sessDBIDBytes := sessionBkt.Get(cSessionDBID)
                if sessDBIDBytes == nil {
                        return ErrCorruptClientSession
                }

                sessDBID, err := readUint64(sessDBIDBytes)
                if err != nil {
                        return err
                }

                return ackedRanges.ForEach(func(dbChanIDBytes, _ []byte) error {
                        dbChanID, err := readUint64(dbChanIDBytes)
                        if err != nil {
                                return err
                        }

                        if _, ok := index[dbChanID]; !ok {
                                index[dbChanID] = make(map[uint64]bool)
                        }

                        index[dbChanID][sessDBID] = true

                        return nil
                })
        })
        if err != nil {
                return nil, err
        }

        return index, nil
}

// persistIndex adds the channel-to-session mapping in each channel's details
// bucket.
func persistIndex(chanDetailsBkt kvdb.RwBucket, chanIDsBkt kvdb.RBucket,
        index map[uint64]map[uint64]bool) error {

        for dbChanID, sessIDs := range index {
                dbChanIDBytes, err := writeUint64(dbChanID)
                if err != nil {
                        return err
                }

                realChanID := chanIDsBkt.Get(dbChanIDBytes)

                chanBkt := chanDetailsBkt.NestedReadWriteBucket(realChanID)
                if chanBkt == nil {
                        return fmt.Errorf("channel not found")
                }

                sessIDsBkt, err := chanBkt.CreateBucket(cChanSessions)
                if err != nil {
                        return err
                }

                for id := range sessIDs {
                        sessID, err := writeUint64(id)
                        if err != nil {
                                return err
                        }

                        err = sessIDsBkt.Put(sessID, []byte{1})
                        if err != nil {
                                return err
                        }
                }
        }

        return nil
}

func writeUint64(i uint64) ([]byte, error) {
        var b bytes.Buffer
        err := tlv.WriteVarInt(&b, i, &[8]byte{})
        if err != nil {
                return nil, err
        }

        return b.Bytes(), nil
}

func readUint64(b []byte) (uint64, error) {
        r := bytes.NewReader(b)
        i, err := tlv.ReadVarInt(r, &[8]byte{})
        if err != nil {
                return 0, err
        }

        return i, nil
}

1	package migration7
2
3	import (
4	"bytes"
5	"encoding/binary"
6	"errors"
7	"fmt"
8
9	"github.com/lightningnetwork/lnd/kvdb"
10	"github.com/lightningnetwork/lnd/tlv"
11	)
12
13	var (
14	// cSessionBkt is a top-level bucket storing:
15	// session-id => cSessionBody -> encoded ClientSessionBody
16	// => cSessionDBID -> db-assigned-id
17	// => cSessionCommits => seqnum -> encoded CommittedUpdate
18	// => cSessionAckRangeIndex => chan-id => acked-index-range
19	cSessionBkt = []byte("client-session-bucket")
20
21	// cChanDetailsBkt is a top-level bucket storing:
22	// channel-id => cChannelSummary -> encoded ClientChanSummary.
23	// => cChanDBID -> db-assigned-id
24	// => cChanSessions => db-session-id -> 1
25	cChanDetailsBkt = []byte("client-channel-detail-bucket")
26
27	// cChannelSummary is a sub-bucket of cChanDetailsBkt which stores the
28	// encoded body of ClientChanSummary.
29	cChannelSummary = []byte("client-channel-summary")
30
31	// cChanSessions is a sub-bucket of cChanDetailsBkt which stores:
32	// session-id -> 1
33	cChanSessions = []byte("client-channel-sessions")
34
35	// cSessionAckRangeIndex is a sub-bucket of cSessionBkt storing:
36	// chan-id => start -> end
37	cSessionAckRangeIndex = []byte("client-session-ack-range-index")
38
39	// cSessionDBID is a key used in the cSessionBkt to store the
40	// db-assigned-d of a session.
41	cSessionDBID = []byte("client-session-db-id")
42
43	// cChanIDIndexBkt is a top-level bucket storing:
44	// db-assigned-id -> channel-ID
45	cChanIDIndexBkt = []byte("client-channel-id-index")
46
47	// ErrUninitializedDB signals that top-level buckets for the database
48	// have not been initialized.
49	ErrUninitializedDB = errors.New("db not initialized")
50
51	// ErrCorruptClientSession signals that the client session's on-disk
52	// structure deviates from what is expected.
53	ErrCorruptClientSession = errors.New("client session corrupted")
54
55	// byteOrder is the default endianness used when serializing integers.
56	byteOrder = binary.BigEndian
57	)
58
59	// MigrateChannelToSessionIndex migrates the tower client DB to add an index
60	// from channel-to-session. This will make it easier in future to check which
61	// sessions have updates for which channels.
UNCOV 62	func MigrateChannelToSessionIndex(tx kvdb.RwTx) error {	×
UNCOV 63	log.Infof("Migrating the tower client DB to build a new " +	×
UNCOV 64	"channel-to-session index")	×
UNCOV 65		×
UNCOV 66	sessionsBkt := tx.ReadBucket(cSessionBkt)	×
UNCOV 67	if sessionsBkt == nil {	×
68	return ErrUninitializedDB	×
69	}	×
70
UNCOV 71	chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)	×
UNCOV 72	if chanDetailsBkt == nil {	×
73	return ErrUninitializedDB	×
74	}	×
75
UNCOV 76	chanIDsBkt := tx.ReadBucket(cChanIDIndexBkt)	×
UNCOV 77	if chanIDsBkt == nil {	×
78	return ErrUninitializedDB	×
79	}	×
80
81	// First gather all the new channel-to-session pairs that we want to
82	// add.
UNCOV 83	index, err := collectIndex(sessionsBkt)	×
UNCOV 84	if err != nil {	×
85	return err	×
86	}	×
87
88	// Then persist those pairs to the db.
UNCOV 89	return persistIndex(chanDetailsBkt, chanIDsBkt, index)	×
90	}
91
92	// collectIndex iterates through all the sessions and uses the keys in the
93	// cSessionAckRangeIndex bucket to collect all the channels that the session
94	// has updates for. The function returns a map from channel ID to session ID
95	// (using the db-assigned IDs for both).
96	func collectIndex(sessionsBkt kvdb.RBucket) (map[uint64]map[uint64]bool,
UNCOV 97	error) {	×
UNCOV 98		×
UNCOV 99	index := make(map[uint64]map[uint64]bool)	×
UNCOV 100	err := sessionsBkt.ForEach(func(sessID, _ []byte) error {	×
UNCOV 101	sessionBkt := sessionsBkt.NestedReadBucket(sessID)	×
UNCOV 102	if sessionBkt == nil {	×
103	return ErrCorruptClientSession	×
104	}	×
105
UNCOV 106	ackedRanges := sessionBkt.NestedReadBucket(	×
UNCOV 107	cSessionAckRangeIndex,	×
UNCOV 108	)	×
UNCOV 109	if ackedRanges == nil {	×
110	return nil	×
111	}	×
112
UNCOV 113	sessDBIDBytes := sessionBkt.Get(cSessionDBID)	×
UNCOV 114	if sessDBIDBytes == nil {	×
115	return ErrCorruptClientSession	×
116	}	×
117
UNCOV 118	sessDBID, err := readUint64(sessDBIDBytes)	×
UNCOV 119	if err != nil {	×
120	return err	×
121	}	×
122
UNCOV 123	return ackedRanges.ForEach(func(dbChanIDBytes, _ []byte) error {	×
UNCOV 124	dbChanID, err := readUint64(dbChanIDBytes)	×
UNCOV 125	if err != nil {	×
126	return err	×
127	}	×
128
UNCOV 129	if _, ok := index[dbChanID]; !ok {	×
UNCOV 130	index[dbChanID] = make(map[uint64]bool)	×
UNCOV 131	}	×
132
UNCOV 133	index[dbChanID][sessDBID] = true	×
UNCOV 134		×
UNCOV 135	return nil	×
136	})
137	})
UNCOV 138	if err != nil {	×
139	return nil, err	×
140	}	×
141
UNCOV 142	return index, nil	×
143	}
144
145	// persistIndex adds the channel-to-session mapping in each channel's details
146	// bucket.
147	func persistIndex(chanDetailsBkt kvdb.RwBucket, chanIDsBkt kvdb.RBucket,
UNCOV 148	index map[uint64]map[uint64]bool) error {	×
UNCOV 149		×
UNCOV 150	for dbChanID, sessIDs := range index {	×
UNCOV 151	dbChanIDBytes, err := writeUint64(dbChanID)	×
UNCOV 152	if err != nil {	×
153	return err	×
154	}	×
155
UNCOV 156	realChanID := chanIDsBkt.Get(dbChanIDBytes)	×
UNCOV 157		×
UNCOV 158	chanBkt := chanDetailsBkt.NestedReadWriteBucket(realChanID)	×
UNCOV 159	if chanBkt == nil {	×
UNCOV 160	return fmt.Errorf("channel not found")	×
UNCOV 161	}	×
162
UNCOV 163	sessIDsBkt, err := chanBkt.CreateBucket(cChanSessions)	×
UNCOV 164	if err != nil {	×
165	return err	×
166	}	×
167
UNCOV 168	for id := range sessIDs {	×
UNCOV 169	sessID, err := writeUint64(id)	×
UNCOV 170	if err != nil {	×
171	return err	×
172	}	×
173
UNCOV 174	err = sessIDsBkt.Put(sessID, []byte{1})	×
UNCOV 175	if err != nil {	×
176	return err	×
177	}	×
178	}
179	}
180
UNCOV 181	return nil	×
182	}
183
UNCOV 184	func writeUint64(i uint64) ([]byte, error) {	×
UNCOV 185	var b bytes.Buffer	×
UNCOV 186	err := tlv.WriteVarInt(&b, i, &[8]byte{})	×
UNCOV 187	if err != nil {	×
188	return nil, err	×
189	}	×
190
UNCOV 191	return b.Bytes(), nil	×
192	}
193
UNCOV 194	func readUint64(b []byte) (uint64, error) {	×
UNCOV 195	r := bytes.NewReader(b)	×
UNCOV 196	i, err := tlv.ReadVarInt(r, &[8]byte{})	×
UNCOV 197	if err != nil {	×
198	return 0, err	×
199	}	×
200
UNCOV 201	return i, nil	×
202	}

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