package p2p
import (
- "errors"
+ "encoding/json"
"fmt"
"math/rand"
"net"
+ "strings"
+ "sync"
"time"
- cfg "github.com/bytom/config"
log "github.com/sirupsen/logrus"
crypto "github.com/tendermint/go-crypto"
cmn "github.com/tendermint/tmlibs/common"
+ dbm "github.com/tendermint/tmlibs/db"
+
+ cfg "github.com/bytom/config"
+ "github.com/bytom/errors"
+ "github.com/bytom/p2p/trust"
)
const (
- reconnectAttempts = 30
- reconnectInterval = 3 * time.Second
+ reconnectAttempts = 10
+ reconnectInterval = 10 * time.Second
+
+ bannedPeerKey = "BannedPeer"
+ defaultBanDuration = time.Hour * 24
)
+var ErrConnectBannedPeer = errors.New("Connect banned peer")
+
type Reactor interface {
cmn.Service // Start, Stop
SetSwitch(*Switch)
GetChannels() []*ChannelDescriptor
- AddPeer(peer *Peer)
+ AddPeer(peer *Peer) error
RemovePeer(peer *Peer, reason interface{})
Receive(chID byte, peer *Peer, msgBytes []byte)
}
dialing *cmn.CMap
nodeInfo *NodeInfo // our node info
nodePrivKey crypto.PrivKeyEd25519 // our node privkey
+ bannedPeer map[string]time.Time
+ db dbm.DB
+ mtx sync.Mutex
filterConnByAddr func(net.Addr) error
filterConnByPubKey func(crypto.PubKeyEd25519) error
var (
ErrSwitchDuplicatePeer = errors.New("Duplicate peer")
+ ErrConnectSelf = errors.New("Connect self")
+ ErrPeerConnected = errors.New("Peer is connected")
)
-func NewSwitch(config *cfg.P2PConfig) *Switch {
+func NewSwitch(config *cfg.P2PConfig, trustHistoryDB dbm.DB) *Switch {
sw := &Switch{
config: config,
- peerConfig: DefaultPeerConfig(),
+ peerConfig: DefaultPeerConfig(config),
reactors: make(map[string]Reactor),
chDescs: make([]*ChannelDescriptor, 0),
reactorsByCh: make(map[byte]Reactor),
peers: NewPeerSet(),
dialing: cmn.NewCMap(),
nodeInfo: nil,
+ db: trustHistoryDB,
}
sw.BaseService = *cmn.NewBaseService(nil, "P2P Switch", sw)
+
+ sw.bannedPeer = make(map[string]time.Time)
+ if datajson := sw.db.Get([]byte(bannedPeerKey)); datajson != nil {
+ if err := json.Unmarshal(datajson, &sw.bannedPeer); err != nil {
+ return nil
+ }
+ }
+ trust.Init()
return sw
}
return err
}
+ if err := sw.checkBannedPeer(peer.NodeInfo.ListenHost()); err != nil {
+ return err
+ }
+
// Avoid self
if sw.nodeInfo.PubKey.Equals(peer.PubKey().Wrap()) {
return errors.New("Ignoring connection from self")
// Start peer
if sw.IsRunning() {
- sw.startInitPeer(peer)
+ if err := sw.startInitPeer(peer); err != nil {
+ return err
+ }
}
// Add the peer to .peers.
sw.filterConnByPubKey = f
}
-func (sw *Switch) startInitPeer(peer *Peer) {
+func (sw *Switch) startInitPeer(peer *Peer) error {
peer.Start() // spawn send/recv routines
for _, reactor := range sw.reactors {
- reactor.AddPeer(peer)
+ if err := reactor.AddPeer(peer); err != nil {
+ return err
+ }
}
+ return nil
}
// Dial a list of seeds asynchronously in random order
}
addrBook.AddAddress(netAddr, ourAddr)
}
+
addrBook.Save()
}
-
- // permute the list, dial them in random order.
+ //permute the list, dial them in random order.
perm := rand.Perm(len(netAddrs))
- for i := 0; i < len(perm); i++ {
- go func(i int) {
- time.Sleep(time.Duration(rand.Int63n(3000)) * time.Millisecond)
- j := perm[i]
- sw.dialSeed(netAddrs[j])
- }(i)
+ for i := 0; i < len(perm)/2; i++ {
+ j := perm[i]
+ sw.dialSeed(netAddrs[j])
}
+
return nil
}
}
func (sw *Switch) DialPeerWithAddress(addr *NetAddress, persistent bool) (*Peer, error) {
+ if err := sw.checkBannedPeer(addr.IP.String()); err != nil {
+ return nil, err
+ }
+ if strings.Compare(addr.IP.String(), sw.nodeInfo.ListenHost()) == 0 {
+ return nil, ErrConnectSelf
+ }
+ for _, v := range sw.Peers().list {
+ if strings.Compare(v.mconn.RemoteAddress.IP.String(), addr.IP.String()) == 0 {
+ return nil, ErrPeerConnected
+ }
+ }
sw.dialing.Set(addr.IP.String(), addr)
defer sw.dialing.Delete(addr.IP.String())
}
log.WithFields(log.Fields{
"address": addr,
- "error": err,
}).Info("Dialed and added peer")
return peer, nil
}
return
}
-func (sw *Switch) Peers() IPeerSet {
+func (sw *Switch) Peers() *PeerSet {
return sw.peers
}
sw.stopAndRemovePeer(peer, reason)
if peer.IsPersistent() {
- go func() {
- log.WithField("peer", peer).Info("Reconnecting to peer")
- for i := 1; i < reconnectAttempts; i++ {
- if !sw.IsRunning() {
- return
- }
+ log.WithField("peer", peer).Info("Reconnecting to peer")
+ for i := 1; i < reconnectAttempts; i++ {
+ if !sw.IsRunning() {
+ return
+ }
- peer, err := sw.DialPeerWithAddress(addr, true)
- if err != nil {
- if i == reconnectAttempts {
- log.WithFields(log.Fields{
- "retries": i,
- "error": err,
- }).Info("Error reconnecting to peer. Giving up")
- return
- }
+ peer, err := sw.DialPeerWithAddress(addr, true)
+ if err != nil {
+ if i == reconnectAttempts {
log.WithFields(log.Fields{
"retries": i,
"error": err,
- }).Info("Error reconnecting to peer. Trying again")
- time.Sleep(reconnectInterval)
- continue
+ }).Info("Error reconnecting to peer. Giving up")
+ return
}
- log.WithField("peer", peer).Info("Reconnected to peer")
- return
+ if errors.Root(err) == ErrConnectBannedPeer || errors.Root(err) == ErrPeerConnected || errors.Root(err) == ErrSwitchDuplicatePeer || errors.Root(err) == ErrConnectSelf {
+ log.WithField("error", err).Info("Error reconnecting to peer. ")
+ return
+ }
+
+ log.WithFields(log.Fields{
+ "retries": i,
+ "error": err,
+ }).Info("Error reconnecting to peer. Trying again")
+ time.Sleep(reconnectInterval)
+ continue
}
- }()
+
+ log.WithField("peer", peer).Info("Reconnected to peer")
+ return
+ }
}
}
}
func (sw *Switch) stopAndRemovePeer(peer *Peer, reason interface{}) {
- sw.peers.Remove(peer)
- peer.Stop()
for _, reactor := range sw.reactors {
reactor.RemovePeer(peer, reason)
}
+ sw.peers.Remove(peer)
+ peer.Stop()
}
func (sw *Switch) listenerRoutine(l Listener) {
// ignore connection if we already have enough
maxPeers := sw.config.MaxNumPeers
if maxPeers <= sw.peers.Size() {
+ // close inConn
+ inConn.Close()
log.WithFields(log.Fields{
"address": inConn.RemoteAddr().String(),
"numPeers": sw.peers.Size(),
// New inbound connection!
err := sw.addPeerWithConnectionAndConfig(inConn, sw.peerConfig)
if err != nil {
+ // conn close for returing err
+ inConn.Close()
log.WithFields(log.Fields{
"address": inConn.RemoteAddr().String(),
"error": err,
privKey := crypto.GenPrivKeyEd25519()
// new switch, add reactors
// TODO: let the config be passed in?
- s := initSwitch(i, NewSwitch(cfg))
+ s := initSwitch(i, NewSwitch(cfg, nil))
s.SetNodeInfo(&NodeInfo{
PubKey: privKey.PubKey().Unwrap().(crypto.PubKeyEd25519),
Moniker: cmn.Fmt("switch%d", i),
}
func (sw *Switch) addPeerWithConnection(conn net.Conn) error {
- peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey)
+ peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.config)
if err != nil {
conn.Close()
return err
}
func (sw *Switch) addPeerWithConnectionAndConfig(conn net.Conn, config *PeerConfig) error {
+ fullAddr := conn.RemoteAddr().String()
+ host, _, err := net.SplitHostPort(fullAddr)
+ if err != nil {
+ return err
+ }
+
+ if err = sw.checkBannedPeer(host); err != nil {
+ return err
+ }
+
peer, err := newInboundPeerWithConfig(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, config)
if err != nil {
- conn.Close()
return err
}
peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
if err = sw.AddPeer(peer); err != nil {
- conn.Close()
return err
}
return nil
}
+
+func (sw *Switch) AddBannedPeer(peer *Peer) error {
+ sw.mtx.Lock()
+ defer sw.mtx.Unlock()
+
+ key := peer.mconn.RemoteAddress.IP.String()
+ sw.bannedPeer[key] = time.Now().Add(defaultBanDuration)
+ datajson, err := json.Marshal(sw.bannedPeer)
+ if err != nil {
+ return err
+ }
+ sw.db.Set([]byte(bannedPeerKey), datajson)
+ return nil
+}
+
+func (sw *Switch) DelBannedPeer(addr string) error {
+ sw.mtx.Lock()
+ defer sw.mtx.Unlock()
+
+ delete(sw.bannedPeer, addr)
+ datajson, err := json.Marshal(sw.bannedPeer)
+ if err != nil {
+ return err
+ }
+ sw.db.Set([]byte(bannedPeerKey), datajson)
+ return nil
+}
+
+func (sw *Switch) checkBannedPeer(peer string) error {
+ if banEnd, ok := sw.bannedPeer[peer]; ok {
+ if time.Now().Before(banEnd) {
+ return ErrConnectBannedPeer
+ }
+ sw.DelBannedPeer(peer)
+ }
+ return nil
+}
OSDN Git Service
