From 13712f52d8ccb36daf589863d147cb5b059c3992 Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Thu, 12 Mar 2020 16:19:20 +0800 Subject: [PATCH] p2p: add ban peer feature --- p2p/dial.go | 26 ++++++++++++++++++++++++++ p2p/peer.go | 28 +++++++++++++++++++++++++--- p2p/peer_test.go | 2 +- p2p/server.go | 29 ++++++++++++++++++++++++++--- 4 files changed, 78 insertions(+), 7 deletions(-) diff --git a/p2p/dial.go b/p2p/dial.go index d190e866af..1bc82e9d8b 100644 --- a/p2p/dial.go +++ b/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 } diff --git a/p2p/peer.go b/p2p/peer.go index 4398ad0f23..faf22fc90b 100644 --- a/p2p/peer.go +++ b/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() { diff --git a/p2p/peer_test.go b/p2p/peer_test.go index e40deb98f0..af31bf6170 100644 --- a/p2p/peer_test.go +++ b/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 }() diff --git a/p2p/server.go b/p2p/server.go index c87b7758df..2b33b869cf 100644 --- a/p2p/server.go +++ b/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