13586005509

Committed 28 Feb 2025 10:14AM UTC coverage: 68.629% (+9.9%) from 58.77%

Build # 13586005509

Build Type

Pull #9521

github

Committed by

web-flow

Commit Message

Merge 37d3a70a5 into 8532955b3

Pull Request Pull Request #9521: unit: remove GOACC, use Go 1.20 native coverage functionality

Run Details

129950 of 189351 relevant lines covered (68.63%)

23726.46 hits per line

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

75.33

/watchtower/wtdb/tower_db.go

package wtdb

import (
        "bytes"
        "errors"

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

var (
        // sessionsBkt is a bucket containing all negotiated client sessions.
        //  session id -> session
        sessionsBkt = []byte("sessions-bucket")

        // updatesBkt is a bucket containing all state updates sent by clients.
        // The updates are further bucketed by session id to prevent clients
        // from overwrite each other.
        //   hint => session id -> update
        updatesBkt = []byte("updates-bucket")

        // updateIndexBkt is a bucket that indexes all state updates by their
        // overarching session id. This allows for efficient lookup of updates
        // by their session id, which is currently used to aide deletion
        // performance.
        //  session id => hint1 -> []byte{}
        //             => hint2 -> []byte{}
        updateIndexBkt = []byte("update-index-bucket")

        // lookoutTipBkt is a bucket containing the last block epoch processed
        // by the lookout subsystem. It has one key, lookoutTipKey.
        //   lookoutTipKey -> block epoch
        lookoutTipBkt = []byte("lookout-tip-bucket")

        // lookoutTipKey is a static key used to retrieve lookout tip's block
        // epoch from the lookoutTipBkt.
        lookoutTipKey = []byte("lookout-tip")

        // ErrNoSessionHintIndex signals that an active session does not have an
        // initialized index for tracking its own state updates.
        ErrNoSessionHintIndex = errors.New("session hint index missing")

        // ErrInvalidBlobSize indicates that the encrypted blob provided by the
        // client is not valid according to the blob type of the session.
        ErrInvalidBlobSize = errors.New("invalid blob size")
)

// TowerDB is single database providing a persistent storage engine for the
// wtserver and lookout subsystems.
type TowerDB struct {
        db kvdb.Backend
}

// OpenTowerDB opens the tower database given the path to the database's
// directory. If no such database exists, this method will initialize a fresh
// one using the latest version number and bucket structure. If a database
// exists but has a lower version number than the current version, any necessary
// migrations will be applied before returning. Any attempt to open a database
// with a version number higher that the latest version will fail to prevent
// accidental reversion.
func OpenTowerDB(db kvdb.Backend) (*TowerDB, error) {
        firstInit, err := isFirstInit(db)
        if err != nil {
                return nil, err
        }

        towerDB := &TowerDB{
                db: db,
        }

        err = initOrSyncVersions(towerDB, firstInit, towerDBVersions)
        if err != nil {
                db.Close()
                return nil, err
        }

        // Now that the database version fully consistent with our latest known
        // version, ensure that all top-level buckets known to this version are
        // initialized. This allows us to assume their presence throughout all
        // operations. If an known top-level bucket is expected to exist but is
        // missing, this will trigger a ErrUninitializedDB error.
        err = kvdb.Update(towerDB.db, initTowerDBBuckets, func() {})
        if err != nil {
                db.Close()
                return nil, err
        }

        return towerDB, nil
}

// initTowerDBBuckets creates all top-level buckets required to handle database
// operations required by the latest version.
func initTowerDBBuckets(tx kvdb.RwTx) error {
        buckets := [][]byte{
                sessionsBkt,
                updateIndexBkt,
                updatesBkt,
                lookoutTipBkt,
        }

        for _, bucket := range buckets {
                _, err := tx.CreateTopLevelBucket(bucket)
                if err != nil {
                        return err
                }
        }

        return nil
}

// bdb returns the backing bbolt.DB instance.
//
// NOTE: Part of the versionedDB interface.
func (t *TowerDB) bdb() kvdb.Backend {
        return t.db
}

// Version returns the database's current version number.
//
// NOTE: Part of the versionedDB interface.
func (t *TowerDB) Version() (uint32, error) {
        var version uint32
        err := kvdb.View(t.db, func(tx kvdb.RTx) error {
                var err error
                version, err = getDBVersion(tx)
                return err
        }, func() {
                version = 0
        })
        if err != nil {
                return 0, err
        }

        return version, nil
}

// Close closes the underlying database.
func (t *TowerDB) Close() error {
        return t.db.Close()
}

// GetSessionInfo retrieves the session for the passed session id. An error is
// returned if the session could not be found.
func (t *TowerDB) GetSessionInfo(id *SessionID) (*SessionInfo, error) {
        var session *SessionInfo
        err := kvdb.View(t.db, func(tx kvdb.RTx) error {
                sessions := tx.ReadBucket(sessionsBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                var err error
                session, err = getSession(sessions, id[:])
                return err
        }, func() {
                session = nil
        })
        if err != nil {
                return nil, err
        }

        return session, nil
}

// InsertSessionInfo records a negotiated session in the tower database. An
// error is returned if the session already exists.
func (t *TowerDB) InsertSessionInfo(session *SessionInfo) error {
        return kvdb.Update(t.db, func(tx kvdb.RwTx) error {
                sessions := tx.ReadWriteBucket(sessionsBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                updateIndex := tx.ReadWriteBucket(updateIndexBkt)
                if updateIndex == nil {
                        return ErrUninitializedDB
                }

                dbSession, err := getSession(sessions, session.ID[:])
                switch {
                case err == ErrSessionNotFound:
                        // proceed.

                case err != nil:
                        return err

                case dbSession.LastApplied > 0:
                        return ErrSessionAlreadyExists
                }

                // Perform a quick sanity check on the session policy before
                // accepting.
                if err := session.Policy.Validate(); err != nil {
                        return err
                }

                err = putSession(sessions, session)
                if err != nil {
                        return err
                }

                // Initialize the session-hint index which will be used to track
                // all updates added for this session. Upon deletion, we will
                // consult the index to determine exactly which updates should
                // be deleted without needing to iterate over the entire
                // database.
                return touchSessionHintBkt(updateIndex, &session.ID)
        }, func() {})
}

// InsertStateUpdate stores an update sent by the client after validating that
// the update is well-formed in the context of other updates sent for the same
// session. This include verifying that the sequence number is incremented
// properly and the last applied values echoed by the client are sane.
func (t *TowerDB) InsertStateUpdate(update *SessionStateUpdate) (uint16, error) {
        var lastApplied uint16
        err := kvdb.Update(t.db, func(tx kvdb.RwTx) error {
                sessions := tx.ReadWriteBucket(sessionsBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                updates := tx.ReadWriteBucket(updatesBkt)
                if updates == nil {
                        return ErrUninitializedDB
                }

                updateIndex := tx.ReadWriteBucket(updateIndexBkt)
                if updateIndex == nil {
                        return ErrUninitializedDB
                }

                // Fetch the session corresponding to the update's session id.
                // This will be used to validate that the update's sequence
                // number and last applied values are sane.
                session, err := getSession(sessions, update.ID[:])
                if err != nil {
                        return err
                }

                commitType, err := session.Policy.BlobType.CommitmentType(nil)
                if err != nil {
                        return err
                }

                kit, err := commitType.EmptyJusticeKit()
                if err != nil {
                        return err
                }

                // Assert that the blob is the correct size for the session's
                // blob type.
                expBlobSize := blob.Size(kit)
                if len(update.EncryptedBlob) != expBlobSize {
                        return ErrInvalidBlobSize
                }

                // Validate the update against the current state of the session.
                err = session.AcceptUpdateSequence(
                        update.SeqNum, update.LastApplied,
                )
                if err != nil {
                        return err
                }

                // Validation succeeded, therefore the update is committed and
                // the session's last applied value is equal to the update's
                // sequence number.
                lastApplied = session.LastApplied

                // Store the updated session to persist the updated last applied
                // values.
                err = putSession(sessions, session)
                if err != nil {
                        return err
                }

                // Create or load the hint bucket for this state update's hint
                // and write the given update.
                hints, err := updates.CreateBucketIfNotExists(update.Hint[:])
                if err != nil {
                        return err
                }

                var b bytes.Buffer
                err = update.Encode(&b)
                if err != nil {
                        return err
                }

                err = hints.Put(update.ID[:], b.Bytes())
                if err != nil {
                        return err
                }

                // Finally, create an entry in the update index to track this
                // hint under its session id. This will allow us to delete the
                // entries efficiently if the session is ever removed.
                return putHintForSession(updateIndex, &update.ID, update.Hint)
        }, func() {
                lastApplied = 0
        })
        if err != nil {
                return 0, err
        }

        return lastApplied, nil
}

// DeleteSession removes all data associated with a particular session id from
// the tower's database.
func (t *TowerDB) DeleteSession(target SessionID) error {
        return kvdb.Update(t.db, func(tx kvdb.RwTx) error {
                sessions := tx.ReadWriteBucket(sessionsBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                updates := tx.ReadWriteBucket(updatesBkt)
                if updates == nil {
                        return ErrUninitializedDB
                }

                updateIndex := tx.ReadWriteBucket(updateIndexBkt)
                if updateIndex == nil {
                        return ErrUninitializedDB
                }

                // Fail if the session doesn't exit.
                _, err := getSession(sessions, target[:])
                if err != nil {
                        return err
                }

                // Remove the target session.
                err = sessions.Delete(target[:])
                if err != nil {
                        return err
                }

                // Next, check the update index for any hints that were added
                // under this session.
                hints, err := getHintsForSession(updateIndex, &target)
                if err != nil {
                        return err
                }

                for _, hint := range hints {
                        // Remove the state updates for any blobs stored under
                        // the target session identifier.
                        updatesForHint := updates.NestedReadWriteBucket(hint[:])
                        if updatesForHint == nil {
                                continue
                        }

                        update := updatesForHint.Get(target[:])
                        if update == nil {
                                continue
                        }

                        err := updatesForHint.Delete(target[:])
                        if err != nil {
                                return err
                        }

                        // If this was the last state update, we can also remove
                        // the hint that would map to an empty set.
                        err = isBucketEmpty(updatesForHint)
                        switch {

                        // Other updates exist for this hint, keep the bucket.
                        case err == errBucketNotEmpty:
                                continue

                        // Unexpected error.
                        case err != nil:
                                return err

                        // No more updates for this hint, prune hint bucket.
                        default:
                                err = updates.DeleteNestedBucket(hint[:])
                                if err != nil {
                                        return err
                                }
                        }
                }

                // Finally, remove this session from the update index, which
                // also removes any of the indexed hints beneath it.
                return removeSessionHintBkt(updateIndex, &target)
        }, func() {})
}

// QueryMatches searches against all known state updates for any that match the
// passed breachHints. More than one Match will be returned for a given hint if
// they exist in the database.
func (t *TowerDB) QueryMatches(breachHints []blob.BreachHint) ([]Match, error) {
        var matches []Match
        err := kvdb.View(t.db, func(tx kvdb.RTx) error {
                sessions := tx.ReadBucket(sessionsBkt)
                if sessions == nil {
                        return ErrUninitializedDB
                }

                updates := tx.ReadBucket(updatesBkt)
                if updates == nil {
                        return ErrUninitializedDB
                }

                // Iterate through the target breach hints, appending any
                // matching updates to the set of matches.
                for _, hint := range breachHints {
                        // If a bucket does not exist for this hint, no matches
                        // are known.
                        updatesForHint := updates.NestedReadBucket(hint[:])
                        if updatesForHint == nil {
                                continue
                        }

                        // Otherwise, iterate through all (session id, update)
                        // pairs, creating a Match for each.
                        err := updatesForHint.ForEach(func(k, v []byte) error {
                                // Load the session via the session id for this
                                // update. The session info contains further
                                // instructions for how to process the state
                                // update.
                                session, err := getSession(sessions, k)
                                switch {
                                case err == ErrSessionNotFound:
                                        log.Warnf("Missing session=%x for "+
                                                "matched state update hint=%x",
                                                k, hint)
                                        return nil

                                case err != nil:
                                        return err
                                }

                                // Decode the state update containing the
                                // encrypted blob.
                                update := &SessionStateUpdate{}
                                err = update.Decode(bytes.NewReader(v))
                                if err != nil {
                                        return err
                                }

                                var id SessionID
                                copy(id[:], k)

                                // Construct the final match using the found
                                // update and its session info.
                                match := Match{
                                        ID:            id,
                                        SeqNum:        update.SeqNum,
                                        Hint:          hint,
                                        EncryptedBlob: update.EncryptedBlob,
                                        SessionInfo:   session,
                                }

                                matches = append(matches, match)

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

                return nil
        }, func() {
                matches = nil
        })
        if err != nil {
                return nil, err
        }

        return matches, nil
}

// SetLookoutTip stores the provided epoch as the latest lookout tip epoch in
// the tower database.
func (t *TowerDB) SetLookoutTip(epoch *chainntnfs.BlockEpoch) error {
        return kvdb.Update(t.db, func(tx kvdb.RwTx) error {
                lookoutTip := tx.ReadWriteBucket(lookoutTipBkt)
                if lookoutTip == nil {
                        return ErrUninitializedDB
                }

                return putLookoutEpoch(lookoutTip, epoch)
        }, func() {})
}

// GetLookoutTip retrieves the current lookout tip block epoch from the tower
// database.
func (t *TowerDB) GetLookoutTip() (*chainntnfs.BlockEpoch, error) {
        var epoch *chainntnfs.BlockEpoch
        err := kvdb.View(t.db, func(tx kvdb.RTx) error {
                lookoutTip := tx.ReadBucket(lookoutTipBkt)
                if lookoutTip == nil {
                        return ErrUninitializedDB
                }

                epoch = getLookoutEpoch(lookoutTip)

                return nil
        }, func() {
                epoch = nil
        })
        if err != nil {
                return nil, err
        }

        return epoch, nil
}

// getSession retrieves the session info from the sessions bucket identified by
// its session id. An error is returned if the session is not found or a
// deserialization error occurs.
func getSession(sessions kvdb.RBucket, id []byte) (*SessionInfo, error) {
        sessionBytes := sessions.Get(id)
        if sessionBytes == nil {
                return nil, ErrSessionNotFound
        }

        var session SessionInfo
        err := session.Decode(bytes.NewReader(sessionBytes))
        if err != nil {
                return nil, err
        }

        return &session, nil
}

// putSession stores the session info in the sessions bucket identified by its
// session id. An error is returned if a serialization error occurs.
func putSession(sessions kvdb.RwBucket, session *SessionInfo) error {
        var b bytes.Buffer
        err := session.Encode(&b)
        if err != nil {
                return err
        }

        return sessions.Put(session.ID[:], b.Bytes())
}

// touchSessionHintBkt initializes the session-hint bucket for a particular
// session id. This ensures that future calls to getHintsForSession or
// putHintForSession can rely on the bucket already being created, and fail if
// index has not been initialized as this points to improper usage.
func touchSessionHintBkt(updateIndex kvdb.RwBucket, id *SessionID) error {
        _, err := updateIndex.CreateBucketIfNotExists(id[:])
        return err
}

// removeSessionHintBkt prunes the session-hint bucket for the given session id
// and all of the hints contained inside. This should be used to clean up the
// index upon session deletion.
func removeSessionHintBkt(updateIndex kvdb.RwBucket, id *SessionID) error {
        return updateIndex.DeleteNestedBucket(id[:])
}

// getHintsForSession returns all known hints belonging to the given session id.
// If the index for the session has not been initialized, this method returns
// ErrNoSessionHintIndex.
func getHintsForSession(updateIndex kvdb.RBucket,
        id *SessionID) ([]blob.BreachHint, error) {

        sessionHints := updateIndex.NestedReadBucket(id[:])
        if sessionHints == nil {
                return nil, ErrNoSessionHintIndex
        }

        var hints []blob.BreachHint
        err := sessionHints.ForEach(func(k, _ []byte) error {
                if len(k) != blob.BreachHintSize {
                        return nil
                }

                var hint blob.BreachHint
                copy(hint[:], k)
                hints = append(hints, hint)
                return nil
        })
        if err != nil {
                return nil, err
        }

        return hints, nil
}

// putHintForSession inserts a record into the update index for a given
// (session, hint) pair. The hints are coalesced under a bucket for the target
// session id, and used to perform efficient removal of updates. If the index
// for the session has not been initialized, this method returns
// ErrNoSessionHintIndex.
func putHintForSession(updateIndex kvdb.RwBucket, id *SessionID,
        hint blob.BreachHint) error {

        sessionHints := updateIndex.NestedReadWriteBucket(id[:])
        if sessionHints == nil {
                return ErrNoSessionHintIndex
        }

        return sessionHints.Put(hint[:], []byte{})
}

// putLookoutEpoch stores the given lookout tip block epoch in provided bucket.
func putLookoutEpoch(bkt kvdb.RwBucket, epoch *chainntnfs.BlockEpoch) error {
        epochBytes := make([]byte, 36)
        copy(epochBytes, epoch.Hash[:])
        byteOrder.PutUint32(epochBytes[32:], uint32(epoch.Height))

        return bkt.Put(lookoutTipKey, epochBytes)
}

// getLookoutEpoch retrieves the lookout tip block epoch from the given bucket.
// A nil epoch is returned if no update exists.
func getLookoutEpoch(bkt kvdb.RBucket) *chainntnfs.BlockEpoch {
        epochBytes := bkt.Get(lookoutTipKey)
        if len(epochBytes) != 36 {
                return nil
        }

        var hash chainhash.Hash
        copy(hash[:], epochBytes[:32])
        height := byteOrder.Uint32(epochBytes[32:])

        return &chainntnfs.BlockEpoch{
                Hash:   &hash,
                Height: int32(height),
        }
}

// errBucketNotEmpty is a helper error returned when testing whether a bucket is
// empty or not.
var errBucketNotEmpty = errors.New("bucket not empty")

// isBucketEmpty returns errBucketNotEmpty if the bucket is not empty.
func isBucketEmpty(bkt kvdb.RBucket) error {
        return bkt.ForEach(func(_, _ []byte) error {
                return errBucketNotEmpty
        })
}

1	package wtdb
2
3	import (
4	"bytes"
5	"errors"
6
7	"github.com/btcsuite/btcd/chaincfg/chainhash"
8	"github.com/lightningnetwork/lnd/chainntnfs"
9	"github.com/lightningnetwork/lnd/kvdb"
10	"github.com/lightningnetwork/lnd/watchtower/blob"
11	)
12
13	var (
14	// sessionsBkt is a bucket containing all negotiated client sessions.
15	// session id -> session
16	sessionsBkt = []byte("sessions-bucket")
17
18	// updatesBkt is a bucket containing all state updates sent by clients.
19	// The updates are further bucketed by session id to prevent clients
20	// from overwrite each other.
21	// hint => session id -> update
22	updatesBkt = []byte("updates-bucket")
23
24	// updateIndexBkt is a bucket that indexes all state updates by their
25	// overarching session id. This allows for efficient lookup of updates
26	// by their session id, which is currently used to aide deletion
27	// performance.
28	// session id => hint1 -> []byte{}
29	// => hint2 -> []byte{}
30	updateIndexBkt = []byte("update-index-bucket")
31
32	// lookoutTipBkt is a bucket containing the last block epoch processed
33	// by the lookout subsystem. It has one key, lookoutTipKey.
34	// lookoutTipKey -> block epoch
35	lookoutTipBkt = []byte("lookout-tip-bucket")
36
37	// lookoutTipKey is a static key used to retrieve lookout tip's block
38	// epoch from the lookoutTipBkt.
39	lookoutTipKey = []byte("lookout-tip")
40
41	// ErrNoSessionHintIndex signals that an active session does not have an
42	// initialized index for tracking its own state updates.
43	ErrNoSessionHintIndex = errors.New("session hint index missing")
44
45	// ErrInvalidBlobSize indicates that the encrypted blob provided by the
46	// client is not valid according to the blob type of the session.
47	ErrInvalidBlobSize = errors.New("invalid blob size")
48	)
49
50	// TowerDB is single database providing a persistent storage engine for the
51	// wtserver and lookout subsystems.
52	type TowerDB struct {
53	db kvdb.Backend
54	}
55
56	// OpenTowerDB opens the tower database given the path to the database's
57	// directory. If no such database exists, this method will initialize a fresh
58	// one using the latest version number and bucket structure. If a database
59	// exists but has a lower version number than the current version, any necessary
60	// migrations will be applied before returning. Any attempt to open a database
61	// with a version number higher that the latest version will fail to prevent
62	// accidental reversion.
63	func OpenTowerDB(db kvdb.Backend) (*TowerDB, error) {	42✔
64	firstInit, err := isFirstInit(db)	42✔
65	if err != nil {	42✔
66	return nil, err	×
67	}	×
68
69	towerDB := &TowerDB{	42✔
70	db: db,	42✔
71	}	42✔
72		42✔
73	err = initOrSyncVersions(towerDB, firstInit, towerDBVersions)	42✔
74	if err != nil {	42✔
75	db.Close()	×
76	return nil, err	×
77	}	×
78
79	// Now that the database version fully consistent with our latest known
80	// version, ensure that all top-level buckets known to this version are
81	// initialized. This allows us to assume their presence throughout all
82	// operations. If an known top-level bucket is expected to exist but is
83	// missing, this will trigger a ErrUninitializedDB error.
84	err = kvdb.Update(towerDB.db, initTowerDBBuckets, func() {})	84✔
85	if err != nil {	42✔
86	db.Close()	×
87	return nil, err	×
88	}	×
89
90	return towerDB, nil	42✔
91	}
92
93	// initTowerDBBuckets creates all top-level buckets required to handle database
94	// operations required by the latest version.
95	func initTowerDBBuckets(tx kvdb.RwTx) error {	42✔
96	buckets := [][]byte{	42✔
97	sessionsBkt,	42✔
98	updateIndexBkt,	42✔
99	updatesBkt,	42✔
100	lookoutTipBkt,	42✔
101	}	42✔
102		42✔
103	for _, bucket := range buckets {	201✔
104	_, err := tx.CreateTopLevelBucket(bucket)	159✔
105	if err != nil {	159✔
106	return err	×
107	}	×
108	}
109
110	return nil	42✔
111	}
112
113	// bdb returns the backing bbolt.DB instance.
114	//
115	// NOTE: Part of the versionedDB interface.
116	func (t *TowerDB) bdb() kvdb.Backend {	29✔
117	return t.db	29✔
118	}	29✔
119
120	// Version returns the database's current version number.
121	//
122	// NOTE: Part of the versionedDB interface.
123	func (t *TowerDB) Version() (uint32, error) {	13✔
124	var version uint32	13✔
125	err := kvdb.View(t.db, func(tx kvdb.RTx) error {	26✔
126	var err error	13✔
127	version, err = getDBVersion(tx)	13✔
128	return err	13✔
129	}, func() {	26✔
130	version = 0	13✔
131	})	13✔
132	if err != nil {	13✔
133	return 0, err	×
134	}	×
135
136	return version, nil	13✔
137	}
138
139	// Close closes the underlying database.
140	func (t *TowerDB) Close() error {	39✔
141	return t.db.Close()	39✔
142	}	39✔
143
144	// GetSessionInfo retrieves the session for the passed session id. An error is
145	// returned if the session could not be found.
146	func (t TowerDB) GetSessionInfo(id SessionID) (*SessionInfo, error) {	15✔
147	var session *SessionInfo	15✔
148	err := kvdb.View(t.db, func(tx kvdb.RTx) error {	30✔
149	sessions := tx.ReadBucket(sessionsBkt)	15✔
150	if sessions == nil {	15✔
151	return ErrUninitializedDB	×
152	}	×
153
154	var err error	15✔
155	session, err = getSession(sessions, id[:])	15✔
156	return err	15✔
157	}, func() {	15✔
158	session = nil	15✔
159	})	15✔
160	if err != nil {	24✔
161	return nil, err	9✔
162	}	9✔
163
164	return session, nil	6✔
165	}
166
167	// InsertSessionInfo records a negotiated session in the tower database. An
168	// error is returned if the session already exists.
169	func (t TowerDB) InsertSessionInfo(session SessionInfo) error {	37✔
170	return kvdb.Update(t.db, func(tx kvdb.RwTx) error {	74✔
171	sessions := tx.ReadWriteBucket(sessionsBkt)	37✔
172	if sessions == nil {	37✔
173	return ErrUninitializedDB	×
174	}	×
175
176	updateIndex := tx.ReadWriteBucket(updateIndexBkt)	37✔
177	if updateIndex == nil {	37✔
178	return ErrUninitializedDB	×
179	}	×
180
181	dbSession, err := getSession(sessions, session.ID[:])	37✔
182	switch {	37✔
183	case err == ErrSessionNotFound:	33✔
184	// proceed.
185
186	case err != nil:	×
187	return err	×
188
189	case dbSession.LastApplied > 0:	2✔
190	return ErrSessionAlreadyExists	2✔
191	}
192
193	// Perform a quick sanity check on the session policy before
194	// accepting.
195	if err := session.Policy.Validate(); err != nil {	37✔
196	return err	2✔
197	}	2✔
198
199	err = putSession(sessions, session)	33✔
200	if err != nil {	33✔
201	return err	×
202	}	×
203
204	// Initialize the session-hint index which will be used to track
205	// all updates added for this session. Upon deletion, we will
206	// consult the index to determine exactly which updates should
207	// be deleted without needing to iterate over the entire
208	// database.
209	return touchSessionHintBkt(updateIndex, &session.ID)	33✔
210	}, func() {})	37✔
211	}
212
213	// InsertStateUpdate stores an update sent by the client after validating that
214	// the update is well-formed in the context of other updates sent for the same
215	// session. This include verifying that the sequence number is incremented
216	// properly and the last applied values echoed by the client are sane.
217	func (t TowerDB) InsertStateUpdate(update SessionStateUpdate) (uint16, error) {	59✔
218	var lastApplied uint16	59✔
219	err := kvdb.Update(t.db, func(tx kvdb.RwTx) error {	118✔
220	sessions := tx.ReadWriteBucket(sessionsBkt)	59✔
221	if sessions == nil {	59✔
222	return ErrUninitializedDB	×
223	}	×
224
225	updates := tx.ReadWriteBucket(updatesBkt)	59✔
226	if updates == nil {	59✔
227	return ErrUninitializedDB	×
228	}	×
229
230	updateIndex := tx.ReadWriteBucket(updateIndexBkt)	59✔
231	if updateIndex == nil {	59✔
232	return ErrUninitializedDB	×
233	}	×
234
235	// Fetch the session corresponding to the update's session id.
236	// This will be used to validate that the update's sequence
237	// number and last applied values are sane.
238	session, err := getSession(sessions, update.ID[:])	59✔
239	if err != nil {	61✔
240	return err	2✔
241	}	2✔
242
243	commitType, err := session.Policy.BlobType.CommitmentType(nil)	57✔
244	if err != nil {	57✔
245	return err	×
246	}	×
247
248	kit, err := commitType.EmptyJusticeKit()	57✔
249	if err != nil {	57✔
250	return err	×
251	}	×
252
253	// Assert that the blob is the correct size for the session's
254	// blob type.
255	expBlobSize := blob.Size(kit)	57✔
256	if len(update.EncryptedBlob) != expBlobSize {	59✔
257	return ErrInvalidBlobSize	2✔
258	}	2✔
259
260	// Validate the update against the current state of the session.
261	err = session.AcceptUpdateSequence(	55✔
262	update.SeqNum, update.LastApplied,	55✔
263	)	55✔
264	if err != nil {	69✔
265	return err	14✔
266	}	14✔
267
268	// Validation succeeded, therefore the update is committed and
269	// the session's last applied value is equal to the update's
270	// sequence number.
271	lastApplied = session.LastApplied	41✔
272		41✔
273	// Store the updated session to persist the updated last applied	41✔
274	// values.	41✔
275	err = putSession(sessions, session)	41✔
276	if err != nil {	41✔
277	return err	×
278	}	×
279
280	// Create or load the hint bucket for this state update's hint
281	// and write the given update.
282	hints, err := updates.CreateBucketIfNotExists(update.Hint[:])	41✔
283	if err != nil {	41✔
284	return err	×
285	}	×
286
287	var b bytes.Buffer	41✔
288	err = update.Encode(&b)	41✔
289	if err != nil {	41✔
290	return err	×
291	}	×
292
293	err = hints.Put(update.ID[:], b.Bytes())	41✔
294	if err != nil {	41✔
295	return err	×
296	}	×
297
298	// Finally, create an entry in the update index to track this
299	// hint under its session id. This will allow us to delete the
300	// entries efficiently if the session is ever removed.
301	return putHintForSession(updateIndex, &update.ID, update.Hint)	41✔
302	}, func() {	59✔
303	lastApplied = 0	59✔
304	})	59✔
305	if err != nil {	77✔
306	return 0, err	18✔
307	}	18✔
308
309	return lastApplied, nil	41✔
310	}
311
312	// DeleteSession removes all data associated with a particular session id from
313	// the tower's database.
314	func (t *TowerDB) DeleteSession(target SessionID) error {	9✔
315	return kvdb.Update(t.db, func(tx kvdb.RwTx) error {	18✔
316	sessions := tx.ReadWriteBucket(sessionsBkt)	9✔
317	if sessions == nil {	9✔
318	return ErrUninitializedDB	×
319	}	×
320
321	updates := tx.ReadWriteBucket(updatesBkt)	9✔
322	if updates == nil {	9✔
323	return ErrUninitializedDB	×
324	}	×
325
326	updateIndex := tx.ReadWriteBucket(updateIndexBkt)	9✔
327	if updateIndex == nil {	9✔
328	return ErrUninitializedDB	×
329	}	×
330
331	// Fail if the session doesn't exit.
332	_, err := getSession(sessions, target[:])	9✔
333	if err != nil {	11✔
334	return err	2✔
335	}	2✔
336
337	// Remove the target session.
338	err = sessions.Delete(target[:])	7✔
339	if err != nil {	7✔
340	return err	×
341	}	×
342
343	// Next, check the update index for any hints that were added
344	// under this session.
345	hints, err := getHintsForSession(updateIndex, &target)	7✔
346	if err != nil {	7✔
347	return err	×
348	}	×
349
350	for _, hint := range hints {	14✔
351	// Remove the state updates for any blobs stored under	7✔
352	// the target session identifier.	7✔
353	updatesForHint := updates.NestedReadWriteBucket(hint[:])	7✔
354	if updatesForHint == nil {	7✔
355	continue	×
356	}
357
358	update := updatesForHint.Get(target[:])	7✔
359	if update == nil {	7✔
360	continue	×
361	}
362
363	err := updatesForHint.Delete(target[:])	7✔
364	if err != nil {	7✔
365	return err	×
366	}	×
367
368	// If this was the last state update, we can also remove
369	// the hint that would map to an empty set.
370	err = isBucketEmpty(updatesForHint)	7✔
371	switch {	7✔
372
373	// Other updates exist for this hint, keep the bucket.
374	case err == errBucketNotEmpty:	2✔
375	continue	2✔
376
377	// Unexpected error.
378	case err != nil:	×
379	return err	×
380
381	// No more updates for this hint, prune hint bucket.
382	default:	5✔
383	err = updates.DeleteNestedBucket(hint[:])	5✔
384	if err != nil {	5✔
385	return err	×
386	}	×
387	}
388	}
389
390	// Finally, remove this session from the update index, which
391	// also removes any of the indexed hints beneath it.
392	return removeSessionHintBkt(updateIndex, &target)	7✔
393	}, func() {})	9✔
394	}
395
396	// QueryMatches searches against all known state updates for any that match the
397	// passed breachHints. More than one Match will be returned for a given hint if
398	// they exist in the database.
399	func (t *TowerDB) QueryMatches(breachHints []blob.BreachHint) ([]Match, error) {	35✔
400	var matches []Match	35✔
401	err := kvdb.View(t.db, func(tx kvdb.RTx) error {	70✔
402	sessions := tx.ReadBucket(sessionsBkt)	35✔
403	if sessions == nil {	35✔
404	return ErrUninitializedDB	×
405	}	×
406
407	updates := tx.ReadBucket(updatesBkt)	35✔
408	if updates == nil {	35✔
409	return ErrUninitializedDB	×
410	}	×
411
412	// Iterate through the target breach hints, appending any
413	// matching updates to the set of matches.
414	for _, hint := range breachHints {	70✔
415	// If a bucket does not exist for this hint, no matches	35✔
416	// are known.	35✔
417	updatesForHint := updates.NestedReadBucket(hint[:])	35✔
418	if updatesForHint == nil {	40✔
419	continue	5✔
420	}
421
422	// Otherwise, iterate through all (session id, update)
423	// pairs, creating a Match for each.
424	err := updatesForHint.ForEach(func(k, v []byte) error {	70✔
425	// Load the session via the session id for this	37✔
426	// update. The session info contains further	37✔
427	// instructions for how to process the state	37✔
428	// update.	37✔
429	session, err := getSession(sessions, k)	37✔
430	switch {	37✔
431	case err == ErrSessionNotFound:	×
432	log.Warnf("Missing session=%x for "+	×
433	"matched state update hint=%x",	×
434	k, hint)	×
435	return nil	×
436
437	case err != nil:	×
438	return err	×
439	}
440
441	// Decode the state update containing the
442	// encrypted blob.
443	update := &SessionStateUpdate{}	37✔
444	err = update.Decode(bytes.NewReader(v))	37✔
445	if err != nil {	37✔
446	return err	×
447	}	×
448
449	var id SessionID	37✔
450	copy(id[:], k)	37✔
451		37✔
452	// Construct the final match using the found	37✔
453	// update and its session info.	37✔
454	match := Match{	37✔
455	ID: id,	37✔
456	SeqNum: update.SeqNum,	37✔
457	Hint: hint,	37✔
458	EncryptedBlob: update.EncryptedBlob,	37✔
459	SessionInfo: session,	37✔
460	}	37✔
461		37✔
462	matches = append(matches, match)	37✔
463		37✔
464	return nil	37✔
465	})
466	if err != nil {	33✔
467	return err	×
468	}	×
469	}
470
471	return nil	35✔
472	}, func() {	35✔
473	matches = nil	35✔
474	})	35✔
475	if err != nil {	35✔
476	return nil, err	×
477	}	×
478
479	return matches, nil	35✔
480	}
481
482	// SetLookoutTip stores the provided epoch as the latest lookout tip epoch in
483	// the tower database.
484	func (t TowerDB) SetLookoutTip(epoch chainntnfs.BlockEpoch) error {	13✔
485	return kvdb.Update(t.db, func(tx kvdb.RwTx) error {	26✔
486	lookoutTip := tx.ReadWriteBucket(lookoutTipBkt)	13✔
487	if lookoutTip == nil {	13✔
488	return ErrUninitializedDB	×
489	}	×
490
491	return putLookoutEpoch(lookoutTip, epoch)	13✔
492	}, func() {})	13✔
493	}
494
495	// GetLookoutTip retrieves the current lookout tip block epoch from the tower
496	// database.
497	func (t TowerDB) GetLookoutTip() (chainntnfs.BlockEpoch, error) {	15✔
498	var epoch *chainntnfs.BlockEpoch	15✔
499	err := kvdb.View(t.db, func(tx kvdb.RTx) error {	30✔
500	lookoutTip := tx.ReadBucket(lookoutTipBkt)	15✔
501	if lookoutTip == nil {	15✔
502	return ErrUninitializedDB	×
503	}	×
504
505	epoch = getLookoutEpoch(lookoutTip)	15✔
506		15✔
507	return nil	15✔
508	}, func() {	15✔
509	epoch = nil	15✔
510	})	15✔
511	if err != nil {	15✔
512	return nil, err	×
513	}	×
514
515	return epoch, nil	15✔
516	}
517
518	// getSession retrieves the session info from the sessions bucket identified by
519	// its session id. An error is returned if the session is not found or a
520	// deserialization error occurs.
521	func getSession(sessions kvdb.RBucket, id []byte) (*SessionInfo, error) {	145✔
522	sessionBytes := sessions.Get(id)	145✔
523	if sessionBytes == nil {	188✔
524	return nil, ErrSessionNotFound	43✔
525	}	43✔
526
527	var session SessionInfo	105✔
528	err := session.Decode(bytes.NewReader(sessionBytes))	105✔
529	if err != nil {	105✔
530	return nil, err	×
531	}	×
532
533	return &session, nil	105✔
534	}
535
536	// putSession stores the session info in the sessions bucket identified by its
537	// session id. An error is returned if a serialization error occurs.
538	func putSession(sessions kvdb.RwBucket, session *SessionInfo) error {	71✔
539	var b bytes.Buffer	71✔
540	err := session.Encode(&b)	71✔
541	if err != nil {	71✔
542	return err	×
543	}	×
544
545	return sessions.Put(session.ID[:], b.Bytes())	71✔
546	}
547
548	// touchSessionHintBkt initializes the session-hint bucket for a particular
549	// session id. This ensures that future calls to getHintsForSession or
550	// putHintForSession can rely on the bucket already being created, and fail if
551	// index has not been initialized as this points to improper usage.
552	func touchSessionHintBkt(updateIndex kvdb.RwBucket, id *SessionID) error {	33✔
553	_, err := updateIndex.CreateBucketIfNotExists(id[:])	33✔
554	return err	33✔
555	}	33✔
556
557	// removeSessionHintBkt prunes the session-hint bucket for the given session id
558	// and all of the hints contained inside. This should be used to clean up the
559	// index upon session deletion.
560	func removeSessionHintBkt(updateIndex kvdb.RwBucket, id *SessionID) error {	7✔
561	return updateIndex.DeleteNestedBucket(id[:])	7✔
562	}	7✔
563
564	// getHintsForSession returns all known hints belonging to the given session id.
565	// If the index for the session has not been initialized, this method returns
566	// ErrNoSessionHintIndex.
567	func getHintsForSession(updateIndex kvdb.RBucket,
568	id *SessionID) ([]blob.BreachHint, error) {	7✔
569		7✔
570	sessionHints := updateIndex.NestedReadBucket(id[:])	7✔
571	if sessionHints == nil {	7✔
572	return nil, ErrNoSessionHintIndex	×
573	}	×
574
575	var hints []blob.BreachHint	7✔
576	err := sessionHints.ForEach(func(k, _ []byte) error {	14✔
577	if len(k) != blob.BreachHintSize {	7✔
578	return nil	×
579	}	×
580
581	var hint blob.BreachHint	7✔
582	copy(hint[:], k)	7✔
583	hints = append(hints, hint)	7✔
584	return nil	7✔
585	})
586	if err != nil {	7✔
587	return nil, err	×
588	}	×
589
590	return hints, nil	7✔
591	}
592
593	// putHintForSession inserts a record into the update index for a given
594	// (session, hint) pair. The hints are coalesced under a bucket for the target
595	// session id, and used to perform efficient removal of updates. If the index
596	// for the session has not been initialized, this method returns
597	// ErrNoSessionHintIndex.
598	func putHintForSession(updateIndex kvdb.RwBucket, id *SessionID,
599	hint blob.BreachHint) error {	41✔
600		41✔
601	sessionHints := updateIndex.NestedReadWriteBucket(id[:])	41✔
602	if sessionHints == nil {	41✔
603	return ErrNoSessionHintIndex	×
604	}	×
605
606	return sessionHints.Put(hint[:], []byte{})	41✔
607	}
608
609	// putLookoutEpoch stores the given lookout tip block epoch in provided bucket.
610	func putLookoutEpoch(bkt kvdb.RwBucket, epoch *chainntnfs.BlockEpoch) error {	13✔
611	epochBytes := make([]byte, 36)	13✔
612	copy(epochBytes, epoch.Hash[:])	13✔
613	byteOrder.PutUint32(epochBytes[32:], uint32(epoch.Height))	13✔
614		13✔
615	return bkt.Put(lookoutTipKey, epochBytes)	13✔
616	}	13✔
617
618	// getLookoutEpoch retrieves the lookout tip block epoch from the given bucket.
619	// A nil epoch is returned if no update exists.
620	func getLookoutEpoch(bkt kvdb.RBucket) *chainntnfs.BlockEpoch {	15✔
621	epochBytes := bkt.Get(lookoutTipKey)	15✔
622	if len(epochBytes) != 36 {	20✔
623	return nil	5✔
624	}	5✔
625
626	var hash chainhash.Hash	10✔
627	copy(hash[:], epochBytes[:32])	10✔
628	height := byteOrder.Uint32(epochBytes[32:])	10✔
629		10✔
630	return &chainntnfs.BlockEpoch{	10✔
631	Hash: &hash,	10✔
632	Height: int32(height),	10✔
633	}	10✔
634	}
635
636	// errBucketNotEmpty is a helper error returned when testing whether a bucket is
637	// empty or not.
638	var errBucketNotEmpty = errors.New("bucket not empty")
639
640	// isBucketEmpty returns errBucketNotEmpty if the bucket is not empty.
641	func isBucketEmpty(bkt kvdb.RBucket) error {	7✔
642	return bkt.ForEach(func(_, _ []byte) error {	9✔
643	return errBucketNotEmpty	2✔
644	})	2✔
645	}

lightningnetwork / lnd / 13586005509

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