channeldb: FetchPermAndTempPeers to load access perms on startup

We introduce a new func FetchPermAndTempPeers that returns two maps.
The first map indicates the nodes that will have "protected" access
to the server. The second map indicates the nodes that have
"temporary" access to the server. This will be used in a future
commit in the server.go code.

channeldb/channel_test.go

@@ -107,6 +107,10 @@ type testChannelParams struct {
 	// openChannel is set to true if the channel should be fully marked as
 	// open if this is false, the channel will be left in pending state.
 	openChannel bool
+
+	// closedChannel is set to true if the channel should be marked as
+	// closed after opening it.
+	closedChannel bool
 }
 
 // testChannelOption is a functional option which can be used to alter the
@@ -129,6 +133,21 @@ func openChannelOption() testChannelOption {
 	}
 }
 
+// closedChannelOption is an option which can be used to create a test channel
+// that is closed.
+func closedChannelOption() testChannelOption {
+	return func(params *testChannelParams) {
+		params.closedChannel = true
+	}
+}
+
+// pubKeyOption is an option which can be used to set the remote's pubkey.
+func pubKeyOption(pubKey *btcec.PublicKey) testChannelOption {
+	return func(params *testChannelParams) {
+		params.channel.IdentityPub = pubKey
+	}
+}
+
 // localHtlcsOption is an option which allows setting of htlcs on the local
 // commitment.
 func localHtlcsOption(htlcs []HTLC) testChannelOption {
@@ -231,6 +250,17 @@ func createTestChannel(t *testing.T, cdb *ChannelStateDB,
 	err = params.channel.MarkAsOpen(params.channel.ShortChannelID)
 	require.NoError(t, err, "unable to mark channel open")
 
+	if params.closedChannel {
+		// Set the other public keys so that serialization doesn't
+		// panic.
+		err = params.channel.CloseChannel(&ChannelCloseSummary{
+			RemotePub:               params.channel.IdentityPub,
+			RemoteCurrentRevocation: params.channel.IdentityPub,
+			RemoteNextRevocation:    params.channel.IdentityPub,
+		})
+		require.NoError(t, err, "unable to close channel")
+	}
+
 	return params.channel
 }
 

channeldb/db.go

@@ -730,6 +730,194 @@ func (c *ChannelStateDB) FetchChannelByID(tx kvdb.RTx, id lnwire.ChannelID) (
 	return c.channelScanner(tx, selector)
 }
 
+// ChanCount is used by the server in determining access control.
+type ChanCount struct {
+	HasOpenOrClosedChan bool
+	PendingOpenCount    uint64
+}
+
+// FetchPermAndTempPeers returns a map where the key is the remote node's
+// public key and the value is a struct that has a tally of the pending-open
+// channels and whether the peer has an open or closed channel with us.
+func (c *ChannelStateDB) FetchPermAndTempPeers(
+	chainHash []byte) (map[string]ChanCount, error) {
+
+	peerCounts := make(map[string]ChanCount)
+
+	err := kvdb.View(c.backend, func(tx kvdb.RTx) error {
+		openChanBucket := tx.ReadBucket(openChannelBucket)
+		if openChanBucket == nil {
+			return ErrNoChanDBExists
+		}
+
+		openChanErr := openChanBucket.ForEach(func(nodePub,
+			v []byte) error {
+
+			// If there is a value, this is not a bucket.
+			if v != nil {
+				return nil
+			}
+
+			nodeChanBucket := openChanBucket.NestedReadBucket(
+				nodePub,
+			)
+			if nodeChanBucket == nil {
+				return nil
+			}
+
+			chainBucket := nodeChanBucket.NestedReadBucket(
+				chainHash,
+			)
+			if chainBucket == nil {
+				return fmt.Errorf("no chain bucket exists")
+			}
+
+			var isPermPeer bool
+			var pendingOpenCount uint64
+
+			internalErr := chainBucket.ForEach(func(chanPoint,
+				val []byte) error {
+
+				// If there is a value, this is not a bucket.
+				if val != nil {
+					return nil
+				}
+
+				chanBucket := chainBucket.NestedReadBucket(
+					chanPoint,
+				)
+				if chanBucket == nil {
+					return nil
+				}
+
+				var op wire.OutPoint
+				readErr := graphdb.ReadOutpoint(
+					bytes.NewReader(chanPoint), &op,
+				)
+				if readErr != nil {
+					return readErr
+				}
+
+				// We need to go through each channel and look
+				// at the IsPending status.
+				openChan, err := fetchOpenChannel(
+					chanBucket, &op,
+				)
+				if err != nil {
+					return err
+				}
+
+				if openChan.IsPending {
+					// Add to the pending-open count since
+					// this is a temp peer.
+					pendingOpenCount++
+					return nil
+				}
+
+				// Since IsPending is false, this is a perm
+				// peer.
+				isPermPeer = true
+
+				return nil
+			})
+			if internalErr != nil {
+				return internalErr
+			}
+
+			peerCount := ChanCount{
+				HasOpenOrClosedChan: isPermPeer,
+				PendingOpenCount:    pendingOpenCount,
+			}
+			peerCounts[string(nodePub)] = peerCount
+
+			return nil
+		})
+		if openChanErr != nil {
+			return openChanErr
+		}
+
+		// Now check the closed channel bucket.
+		historicalChanBucket := tx.ReadBucket(historicalChannelBucket)
+		if historicalChanBucket == nil {
+			return ErrNoHistoricalBucket
+		}
+
+		historicalErr := historicalChanBucket.ForEach(func(chanPoint,
+			v []byte) error {
+			// Parse each nested bucket and the chanInfoKey to get
+			// the IsPending bool. This determines whether the
+			// peer is protected or not.
+			if v != nil {
+				// This is not a bucket. This is currently not
+				// possible.
+				return nil
+			}
+
+			chanBucket := historicalChanBucket.NestedReadBucket(
+				chanPoint,
+			)
+			if chanBucket == nil {
+				// This is not possible.
+				return fmt.Errorf("no historical channel " +
+					"bucket exists")
+			}
+
+			var op wire.OutPoint
+			readErr := graphdb.ReadOutpoint(
+				bytes.NewReader(chanPoint), &op,
+			)
+			if readErr != nil {
+				return readErr
+			}
+
+			// This channel is closed, but the structure of the
+			// historical bucket is the same. This is by design,
+			// which means we can call fetchOpenChannel.
+			channel, fetchErr := fetchOpenChannel(chanBucket, &op)
+			if fetchErr != nil {
+				return fetchErr
+			}
+
+			// Only include this peer in the protected class if
+			// the closing transaction confirmed. Note that
+			// CloseChannel can be called in the funding manager
+			// while IsPending is true which is why we need this
+			// special-casing to not count premature funding
+			// manager calls to CloseChannel.
+			if !channel.IsPending {
+				// Fetch the public key of the remote node. We
+				// need to use the string-ified serialized,
+				// compressed bytes as the key.
+				remotePub := channel.IdentityPub
+				remoteSer := remotePub.SerializeCompressed()
+				remoteKey := string(remoteSer)
+
+				count, exists := peerCounts[remoteKey]
+				if exists {
+					count.HasOpenOrClosedChan = true
+					peerCounts[remoteKey] = count
+				} else {
+					peerCount := ChanCount{
+						HasOpenOrClosedChan: true,
+					}
+					peerCounts[remoteKey] = peerCount
+				}
+			}
+
+			return nil
+		})
+		if historicalErr != nil {
+			return historicalErr
+		}
+
+		return nil
+	}, func() {
+		clear(peerCounts)
+	})
+
+	return peerCounts, err
+}
+
 // channelSelector describes a function that takes a chain-hash bucket from
 // within the open-channel DB and returns the wanted channel point bytes, and
 // channel point. It must return the ErrChannelNotFound error if the wanted

channeldb/db_test.go

@@ -721,6 +721,91 @@ func TestFetchHistoricalChannel(t *testing.T) {
 	}
 }
 
+// TestFetchPermTempPeer tests that we're able to call FetchPermAndTempPeers
+// successfully.
+func TestFetchPermTempPeer(t *testing.T) {
+	t.Parallel()
+
+	fullDB, err := MakeTestDB(t)
+	require.NoError(t, err, "unable to make test database")
+
+	cdb := fullDB.ChannelStateDB()
+
+	// Create an open channel.
+	privKey1, err := btcec.NewPrivateKey()
+	require.NoError(t, err, "unable to generate new private key")
+
+	pubKey1 := privKey1.PubKey()
+
+	channelState1 := createTestChannel(
+		t, cdb, openChannelOption(), pubKeyOption(pubKey1),
+	)
+
+	// Next, assert that the channel exists in the database.
+	_, err = cdb.FetchChannel(channelState1.FundingOutpoint)
+	require.NoError(t, err, "unable to fetch channel")
+
+	// Create a pending channel.
+	privKey2, err := btcec.NewPrivateKey()
+	require.NoError(t, err, "unable to generate private key")
+
+	pubKey2 := privKey2.PubKey()
+	channelState2 := createTestChannel(t, cdb, pubKeyOption(pubKey2))
+
+	// Assert that the channel exists in the database.
+	_, err = cdb.FetchChannel(channelState2.FundingOutpoint)
+	require.NoError(t, err, "unable to fetch channel")
+
+	// Create a closed channel.
+	privKey3, err := btcec.NewPrivateKey()
+	require.NoError(t, err, "unable to generate new private key")
+
+	pubKey3 := privKey3.PubKey()
+
+	_ = createTestChannel(
+		t, cdb, pubKeyOption(pubKey3), openChannelOption(),
+		closedChannelOption(),
+	)
+
+	// Fetch the ChanCount for our peers.
+	peerCounts, err := cdb.FetchPermAndTempPeers(key[:])
+	require.NoError(t, err, "unable to fetch perm and temp peers")
+
+	// There should only be three entries.
+	require.Len(t, peerCounts, 3)
+
+	// The first entry should have OpenClosed set to true and Pending set
+	// to 0.
+	count1, found := peerCounts[string(pubKey1.SerializeCompressed())]
+	require.True(t, found, "unable to find peer 1 in peerCounts")
+	require.True(
+		t, count1.HasOpenOrClosedChan,
+		"couldn't find peer 1's channels",
+	)
+	require.Zero(
+		t, count1.PendingOpenCount,
+		"peer 1 doesn't have 0 pending-open",
+	)
+
+	count2, found := peerCounts[string(pubKey2.SerializeCompressed())]
+	require.True(t, found, "unable to find peer 2 in peerCounts")
+	require.False(
+		t, count2.HasOpenOrClosedChan, "found erroneous channels",
+	)
+	require.Equal(t, uint64(1), count2.PendingOpenCount)
+
+	count3, found := peerCounts[string(pubKey3.SerializeCompressed())]
+	require.True(t, found, "unable to find peer 3 in peerCounts")
+	require.True(
+		t, count3.HasOpenOrClosedChan,
+		"couldn't find peer 3's channels",
+	)
+	require.Zero(
+		t, count3.PendingOpenCount,
+		"peer 3 doesn't have 0 pending-open",
+	)
+}
+
 func createLightningNode(priv *btcec.PrivateKey) *models.LightningNode {
 	updateTime := rand.Int63()