p2p: add ban peer feature

p2p/dial.go

@@ -77,6 +77,7 @@ var (
 	errAlreadyDialing   = errors.New("already dialing")
 	errAlreadyConnected = errors.New("already connected")
 	errRecentlyDialed   = errors.New("recently dialed")
+	errRecentlyBanned   = errors.New("recently banned")
 	errNotWhitelisted   = errors.New("not contained in netrestrict whitelist")
 )
 
@@ -102,6 +103,7 @@ type dialScheduler struct {
 	remStaticCh chan *enode.Node
 	addPeerCh   chan *conn
 	remPeerCh   chan *conn
+	banPeerCh   chan *conn
 
 	// Everything below here belongs to loop and
 	// should only be accessed by code on the loop goroutine.
@@ -121,6 +123,9 @@ type dialScheduler struct {
 	historyTimer     mclock.Timer
 	historyTimerTime mclock.AbsTime
 
+	// The bannedHistory keeps all recently banned nodes which won't be dialed for a short time
+	bannedHistory expHeap
+
 	// for logStats
 	lastStatsLog     mclock.AbsTime
 	doneSinceLastLog int
@@ -172,6 +177,7 @@ func newDialScheduler(config dialConfig, it enode.Iterator, setupFunc dialSetupF
 		remStaticCh: make(chan *enode.Node),
 		addPeerCh:   make(chan *conn),
 		remPeerCh:   make(chan *conn),
+		banPeerCh:   make(chan *conn),
 	}
 	d.lastStatsLog = d.clock.Now()
 	d.ctx, d.cancel = context.WithCancel(context.Background())
@@ -219,6 +225,14 @@ func (d *dialScheduler) peerRemoved(c *conn) {
 	}
 }
 
+// peerBanned notifies the dialer about banned peer
+func (d *dialScheduler) peerBanned(c *conn) {
+	select {
+	case d.banPeerCh <- c:
+	case <-d.ctx.Done():
+	}
+}
+
 // loop is the main loop of the dialer.
 func (d *dialScheduler) loop(it enode.Iterator) {
 	var (
@@ -273,6 +287,14 @@ loop:
 			delete(d.peers, c.node.ID())
 			d.updateStaticPool(c.node.ID())
 
+		case c := <-d.banPeerCh:
+			// If it's not a static node and been banned,
+			// add to the blacklist for dialing.
+			if d.static[c.node.ID()] == nil {
+				remoteIP := netutil.AddrIP(c.fd.RemoteAddr())
+				d.bannedHistory.add(remoteIP.String(), d.clock.Now().Add(banIPThrottleTime))
+			}
+
 		case node := <-d.addStaticCh:
 			id := node.ID()
 			_, exists := d.static[id]
@@ -400,6 +422,10 @@ func (d *dialScheduler) checkDial(n *enode.Node) error {
 	if d.history.contains(string(n.ID().Bytes())) {
 		return errRecentlyDialed
 	}
+	d.bannedHistory.expire(d.clock.Now(), nil)
+	if !n.Incomplete() && !netutil.IsLAN(n.IP()) && d.bannedHistory.contains(n.IP().String()) {
+		return errRecentlyBanned
+	}
 	return nil
 }
 

p2p/peer.go

@@ -113,6 +113,8 @@ type Peer struct {
 	protoErr chan error
 	closed   chan struct{}
 	disc     chan DiscReason
+	banOnce  sync.Once
+	ban      chan struct{}
 
 	// events receives message send / receive events if set
 	events *event.Feed
@@ -168,6 +170,14 @@ func (p *Peer) Disconnect(reason DiscReason) {
 	}
 }
 
+// Ban terminates the peer connection if the peer acts really bad.
+// The banned peer will be forbidden for re-connection in a short time(5 mins).
+func (p *Peer) Ban() {
+	p.banOnce.Do(func() {
+		close(p.ban)
+	})
+}
+
 // String implements fmt.Stringer.
 func (p *Peer) String() string {
 	id := p.ID()
@@ -186,6 +196,7 @@ func newPeer(log log.Logger, conn *conn, protocols []Protocol) *Peer {
 		running:  protomap,
 		created:  mclock.Now(),
 		disc:     make(chan DiscReason),
+		ban:      make(chan struct{}),
 		protoErr: make(chan error, len(protomap)+1), // protocols + pingLoop
 		closed:   make(chan struct{}),
 		log:      log.New("id", conn.node.ID(), "conn", conn.flags),
@@ -197,7 +208,7 @@ func (p *Peer) Log() log.Logger {
 	return p.log
 }
 
-func (p *Peer) run() (remoteRequested bool, err error) {
+func (p *Peer) run() (remoteRequested bool, banned bool, err error) {
 	var (
 		writeStart = make(chan struct{}, 1)
 		writeErr   = make(chan error, 1)
@@ -238,13 +249,24 @@ loop:
 		case err = <-p.disc:
 			reason = discReasonForError(err)
 			break loop
+		case <-p.ban:
+			banned = true
+			break loop
 		}
 	}
-
 	close(p.closed)
+
+	// If the peer is banned by up upper level protocol, just send the
+	// DiscRequested to peer. The reason is:
+	// - The banned disconnection reason is not yet defined
+	// - It's might unnecessary to tell the adversary the specified reason for
+	//   disconnection.
+	if banned {
+		reason = DiscRequested
+	}
 	p.rw.close(reason)
 	p.wg.Wait()
-	return remoteRequested, err
+	return remoteRequested, banned, err
 }
 
 func (p *Peer) pingLoop() {

p2p/peer_test.go

@@ -97,7 +97,7 @@ func testPeer(protos []Protocol) (func(), *conn, *Peer, <-chan error) {
 	peer := newPeer(log.Root(), c1, protos)
 	errc := make(chan error, 1)
 	go func() {
-		_, err := peer.run()
+		_, _, err := peer.run()
 		errc <- err
 	}()
 

p2p/server.go

@@ -57,6 +57,10 @@ const (
 	// This time limits inbound connection attempts per source IP.
 	inboundThrottleTime = 30 * time.Second
 
+	// This time limits inbound connection attempts per source IP
+	// when the IP is banned.
+	banIPThrottleTime = 5 * time.Minute
+
 	// Maximum time allowed for reading a complete message.
 	// This is effectively the amount of time a connection can be idle.
 	frameReadTimeout = 30 * time.Second
@@ -198,6 +202,9 @@ type Server struct {
 
 	// State of run loop and listenLoop.
 	inboundHistory expHeap
+
+	bannedLock    sync.RWMutex // protects bannedHistory
+	bannedHistory expHeap
 }
 
 type peerOpFunc func(map[enode.ID]*Peer)
@@ -206,6 +213,7 @@ type peerDrop struct {
 	*Peer
 	err       error
 	requested bool // true if signaled by the peer
+	banned    bool // true if the peer is banned
 }
 
 type connFlag int32
@@ -770,6 +778,14 @@ running:
 			if pd.Inbound() {
 				inboundCount--
 			}
+			// Ban the IP if a peer is marked as bad peer.
+			if pd.banned {
+				remoteIP := netutil.AddrIP(pd.rw.fd.RemoteAddr())
+				srv.bannedLock.Lock()
+				srv.bannedHistory.add(remoteIP.String(), srv.clock.Now().Add(banIPThrottleTime))
+				srv.bannedLock.Unlock()
+				srv.dialsched.peerBanned(pd.rw)
+			}
 		}
 	}
 
@@ -896,8 +912,15 @@ func (srv *Server) checkInboundConn(fd net.Conn, remoteIP net.IP) error {
 	if srv.NetRestrict != nil && !srv.NetRestrict.Contains(remoteIP) {
 		return fmt.Errorf("not whitelisted in NetRestrict")
 	}
-	// Reject Internet peers that try too often.
+	// Reject Internet peers if they are marked as bad.
 	now := srv.clock.Now()
+	srv.bannedLock.Lock()
+	defer srv.bannedLock.Unlock()
+	srv.bannedHistory.expire(now, nil)
+	if !netutil.IsLAN(remoteIP) && srv.bannedHistory.contains(remoteIP.String()) {
+		return fmt.Errorf("bad peer")
+	}
+	// Reject Internet peers that try too often.
 	srv.inboundHistory.expire(now, nil)
 	if !netutil.IsLAN(remoteIP) && srv.inboundHistory.contains(remoteIP.String()) {
 		return fmt.Errorf("too many attempts")
@@ -1032,12 +1055,12 @@ func (srv *Server) runPeer(p *Peer) {
 	})
 
 	// Run the per-peer main loop.
-	remoteRequested, err := p.run()
+	remoteRequested, banned, err := p.run()
 
 	// Announce disconnect on the main loop to update the peer set.
 	// The main loop waits for existing peers to be sent on srv.delpeer
 	// before returning, so this send should not select on srv.quit.
-	srv.delpeer <- peerDrop{p, err, remoteRequested}
+	srv.delpeer <- peerDrop{p, err, remoteRequested, banned}
 
 	// Broadcast peer drop to external subscribers. This needs to be
 	// after the send to delpeer so subscribers have a consistent view of