Merge 065b9bd8c8 into 8610314918

p2p/peer.go

@@ -123,10 +123,8 @@ func (p *Peer) run() DiscReason {
 	p.startProtocols()
 
 	// Wait for an error or disconnect.
-	var (
-		reason    DiscReason
-		requested bool
-	)
+	var reason DiscReason
+
 	select {
 	case err := <-readErr:
 		if r, ok := err.(DiscReason); ok {
@@ -140,15 +138,11 @@ func (p *Peer) run() DiscReason {
 	case err := <-p.protoErr:
 		reason = discReasonForError(err)
 	case reason = <-p.disc:
-		requested = true
 	}
 	close(p.closed)
 	p.rw.close(reason)
 	p.wg.Wait()
 
-	if requested {
-		reason = DiscRequested
-	}
 	glog.V(logger.Debug).Infof("%v: Disconnected: %v\n", p, reason)
 	return reason
 }

p2p/server.go

@@ -19,6 +19,11 @@ const (
 	refreshPeersInterval    = 30 * time.Second
 	staticPeerCheckInterval = 15 * time.Second
 
+	banStartingTimeout       = time.Minute // Amount of time a first offender is banned
+	banMaximumTimeout        = time.Hour   // Maximum time for banning somebody
+	banAggrevationMultiplier = 4           // Multiplier for repeat offenders
+	banAbsolutionMultiplier  = 24          // Multiplier for absolving a node
+
 	// Maximum number of concurrently handshaking inbound connections.
 	maxAcceptConns = 50
 
@@ -122,6 +127,7 @@ type Server struct {
 
 	quit          chan struct{}
 	addstatic     chan *discover.Node
+	addbanned     chan *Peer
 	posthandshake chan *conn
 	addpeer       chan *conn
 	delpeer       chan *Peer
@@ -307,6 +313,7 @@ func (srv *Server) Start() (err error) {
 	srv.delpeer = make(chan *Peer)
 	srv.posthandshake = make(chan *conn)
 	srv.addstatic = make(chan *discover.Node)
+	srv.addbanned = make(chan *Peer)
 	srv.peerOp = make(chan peerOpFunc)
 	srv.peerOpDone = make(chan struct{})
 
@@ -380,6 +387,9 @@ func (srv *Server) run(dialstate dialer) {
 		peers   = make(map[discover.NodeID]*Peer)
 		trusted = make(map[discover.NodeID]bool, len(srv.TrustedNodes))
 
+		banned    = make(map[discover.NodeID]time.Time)
+		penalties = make(map[discover.NodeID]time.Duration)
+
 		tasks        []task
 		pendingTasks []task
 		taskdone     = make(chan task, maxActiveDialTasks)
@@ -436,6 +446,19 @@ running:
 			// it will keep the node connected.
 			glog.V(logger.Detail).Infoln("<-addstatic:", n)
 			dialstate.addStatic(n)
+		case p := <-srv.addbanned:
+			// Add a node to the banned list, doubling any previous bans
+			glog.V(logger.Detail).Infoln("<-addbanned:", p)
+			penalty := banStartingTimeout
+			if prev, ok := penalties[p.ID()]; ok {
+				penalty = prev * banAggrevationMultiplier
+				if penalty > banMaximumTimeout {
+					penalty = banMaximumTimeout
+				}
+			}
+			penalties[p.ID()] = penalty
+			banned[p.ID()] = time.Now().Add(penalty)
+			glog.V(logger.Debug).Infof("Banned %s until %v", p.ID().String()[:16], banned[p.ID()])
 		case op := <-srv.peerOp:
 			// This channel is used by Peers and PeerCount.
 			op(peers)
@@ -450,6 +473,11 @@ running:
 		case c := <-srv.posthandshake:
 			// A connection has passed the encryption handshake so
 			// the remote identity is known (but hasn't been verified yet).
+			if exp, ok := banned[c.id]; ok && time.Now().Before(exp) {
+				glog.V(logger.Detail).Infoln("<-banned:", c)
+				c.cont <- fmt.Errorf("banned until %v", exp)
+				continue
+			}
 			if trusted[c.id] {
 				// Ensure that the trusted flag is set before checking against MaxPeers.
 				c.flags |= trustedConn
@@ -478,6 +506,15 @@ running:
 			// A peer disconnected.
 			glog.V(logger.Detail).Infoln("<-delpeer:", p)
 			delete(peers, p.ID())
+
+			// Lift any previous bans for good behavior
+			if exp, ok := banned[p.ID()]; ok {
+				lift := exp.Add(penalties[p.ID()] * banAbsolutionMultiplier)
+				if time.Now().After(lift) {
+					delete(banned, p.ID())
+					delete(penalties, p.ID())
+				}
+			}
 		}
 	}
 
@@ -494,6 +531,13 @@ running:
 	// is closed.
 	glog.V(logger.Detail).Infof("ignoring %d pending tasks at spindown", len(tasks))
 	for len(peers) > 0 {
+		// Drop any in flight ban requests
+		select {
+		case p := <-srv.addbanned:
+			glog.V(logger.Detail).Infoln("<-addbanned (spindown):", p)
+		default:
+		}
+		// Drop the peer itself
 		p := <-srv.delpeer
 		glog.V(logger.Detail).Infoln("<-delpeer (spindown):", p)
 		delete(peers, p.ID())
@@ -640,6 +684,10 @@ func (srv *Server) runPeer(p *Peer) {
 		srv.newPeerHook(p)
 	}
 	discreason := p.run()
+	if discreason == DiscUselessPeer {
+		// Temporarily disallow the peer to reconnect
+		srv.addbanned <- p
+	}
 	// Note: run waits for existing peers to be sent on srv.delpeer
 	// before returning, so this send should not select on srv.quit.
 	srv.delpeer <- p