brontide+server.go: Listener ban-callback, introduce server caches

Here we introduce the server caches that will determine the access
control status of our peers. Peers that have had their funding
transaction confirm with us are protected. Peers that only have
pending-open channels with us are temporary access and can have
their access revoked. The rest of the peers are granted restricted
access.
The channelPeers map contains protected-access peers and the
pendingChannelPeers map contains temporary-access peers.

Author: Crypt-iQ

brontide/listener.go

@@ -1,6 +1,7 @@
 package brontide
 
 import (
+	"encoding/hex"
 	"errors"
 	"fmt"
 	"io"
@@ -12,7 +13,7 @@ import (
 
 // defaultHandshakes is the maximum number of handshakes that can be done in
 // parallel.
-const defaultHandshakes = 1000
+const defaultHandshakes = 50
 
 // Listener is an implementation of a net.Conn which executes an authenticated
 // key exchange and message encryption protocol dubbed "Machine" after
@@ -24,6 +25,10 @@ type Listener struct {
 
 	tcp *net.TCPListener
 
+	// shouldAccept is a closure that determines if we should accept the
+	// incoming connection or not based on its hex-encoded public key.
+	shouldAccept func(string) (bool, error)
+
 	handshakeSema chan struct{}
 	conns         chan maybeConn
 	quit          chan struct{}
@@ -34,8 +39,8 @@ var _ net.Listener = (*Listener)(nil)
 
 // NewListener returns a new net.Listener which enforces the Brontide scheme
 // during both initial connection establishment and data transfer.
-func NewListener(localStatic keychain.SingleKeyECDH,
-	listenAddr string) (*Listener, error) {
+func NewListener(localStatic keychain.SingleKeyECDH, listenAddr string,
+	shouldAccept func(string) (bool, error)) (*Listener, error) {
 
 	addr, err := net.ResolveTCPAddr("tcp", listenAddr)
 	if err != nil {
@@ -50,6 +55,7 @@ func NewListener(localStatic keychain.SingleKeyECDH,
 	brontideListener := &Listener{
 		localStatic:   localStatic,
 		tcp:           l,
+		shouldAccept:  shouldAccept,
 		handshakeSema: make(chan struct{}, defaultHandshakes),
 		conns:         make(chan maybeConn),
 		quit:          make(chan struct{}),
@@ -193,6 +199,29 @@ func (l *Listener) doHandshake(conn net.Conn) {
 		return
 	}
 
+	// Call the shouldAccept closure to see if the remote node's public key
+	// is allowed according to our banning heuristic. This is here because
+	// we do not learn the remote node's public static key until we've
+	// received and validated Act 3.
+	remoteKey := brontideConn.RemotePub()
+	if remoteKey == nil {
+		connErr := fmt.Errorf("no remote pubkey")
+		brontideConn.conn.Close()
+		l.rejectConn(rejectedConnErr(connErr, remoteAddr))
+
+		return
+	}
+
+	remoteHex := hex.EncodeToString(remoteKey.SerializeCompressed())
+	accepted, acceptErr := l.shouldAccept(remoteHex)
+	if !accepted {
+		// Reject the connection.
+		brontideConn.conn.Close()
+		l.rejectConn(rejectedConnErr(acceptErr, remoteAddr))
+
+		return
+	}
+
 	l.acceptConn(brontideConn)
 }
 
@@ -255,3 +284,9 @@ func (l *Listener) Close() error {
 func (l *Listener) Addr() net.Addr {
 	return l.tcp.Addr()
 }
+
+// DisabledBanClosure is used in places that NewListener is invoked to bypass
+// the ban-scoring.
+func DisabledBanClosure(s string) (bool, error) {
+	return true, nil
+}

brontide/noise_test.go

@@ -35,7 +35,7 @@ func makeListener() (*Listener, *lnwire.NetAddress, error) {
 	addr := "localhost:0"
 
 	// Our listener will be local, and the connection remote.
-	listener, err := NewListener(localKeyECDH, addr)
+	listener, err := NewListener(localKeyECDH, addr, DisabledBanClosure)
 	if err != nil {
 		return nil, nil, err
 	}

discovery/syncer.go

@@ -181,6 +181,9 @@ const (
 	// requestBatchSize is the maximum number of channels we will query the
 	// remote peer for in a QueryShortChanIDs message.
 	requestBatchSize = 500
+
+	// syncerBufferSize is the size of the syncer's buffers.
+	syncerBufferSize = 5
 )
 
 var (
@@ -436,8 +439,8 @@ func newGossipSyncer(cfg gossipSyncerCfg, sema chan struct{}) *GossipSyncer {
 		rateLimiter:        rateLimiter,
 		syncTransitionReqs: make(chan *syncTransitionReq),
 		historicalSyncReqs: make(chan *historicalSyncReq),
-		gossipMsgs:         make(chan lnwire.Message, 100),
-		queryMsgs:          make(chan lnwire.Message, 100),
+		gossipMsgs:         make(chan lnwire.Message, syncerBufferSize),
+		queryMsgs:          make(chan lnwire.Message, syncerBufferSize),
 		syncerSema:         sema,
 		quit:               make(chan struct{}),
 	}

funding/manager.go

@@ -109,7 +109,7 @@ const (
 	// pendingChansLimit is the maximum number of pending channels that we
 	// can have. After this point, pending channel opens will start to be
 	// rejected.
-	pendingChansLimit = 1_000
+	pendingChansLimit = 50
 )
 
 var (

peer/brontide.go

@@ -90,6 +90,9 @@ const (
 	// torTimeoutMultiplier is the scaling factor we use on network timeouts
 	// for Tor peers.
 	torTimeoutMultiplier = 3
+
+	// msgStreamSize is the size of the message streams.
+	msgStreamSize = 5
 )
 
 var (
@@ -1856,7 +1859,7 @@ func newChanMsgStream(p *Brontide, cid lnwire.ChannelID) *msgStream {
 	return newMsgStream(p,
 		fmt.Sprintf("Update stream for ChannelID(%x) created", cid[:]),
 		fmt.Sprintf("Update stream for ChannelID(%x) exiting", cid[:]),
-		1000,
+		msgStreamSize,
 		apply,
 	)
 }
@@ -1875,7 +1878,7 @@ func newDiscMsgStream(p *Brontide) *msgStream {
 		p,
 		"Update stream for gossiper created",
 		"Update stream for gossiper exited",
-		1000,
+		msgStreamSize,
 		apply,
 	)
 }