p2p/switch.go

   1 package p2p
   2
   3 import (
   4         "encoding/json"
   5         "fmt"
   6         "math/rand"
   7         "net"
   8         "sync"
   9         "time"
  10         "strings"
  11
  12         log "github.com/sirupsen/logrus"
  13         crypto "github.com/tendermint/go-crypto"
  14         cmn "github.com/tendermint/tmlibs/common"
  15         dbm "github.com/tendermint/tmlibs/db"
  16
  17         cfg "github.com/bytom/config"
  18         "github.com/bytom/errors"
  19         "github.com/bytom/p2p/trust"
  20 )
  21
  22 const (
  23         reconnectAttempts = 10
  24         reconnectInterval = 10 * time.Second
  25
  26         bannedPeerKey      = "BannedPeer"
  27         defaultBanDuration = time.Hour * 24
  28         peerBannedTM       = 20
  29 )
  30
  31 var ErrConnectBannedPeer = errors.New("Connect banned peer")
  32
  33 type Reactor interface {
  34         cmn.Service // Start, Stop
  35
  36         SetSwitch(*Switch)
  37         GetChannels() []*ChannelDescriptor
  38         AddPeer(peer *Peer) error
  39         RemovePeer(peer *Peer, reason interface{})
  40         Receive(chID byte, peer *Peer, msgBytes []byte)
  41 }
  42
  43 //--------------------------------------
  44
  45 type BaseReactor struct {
  46         cmn.BaseService // Provides Start, Stop, .Quit
  47         Switch          *Switch
  48 }
  49
  50 func NewBaseReactor(name string, impl Reactor) *BaseReactor {
  51         return &BaseReactor{
  52                 BaseService: *cmn.NewBaseService(nil, name, impl),
  53                 Switch:      nil,
  54         }
  55 }
  56
  57 func (br *BaseReactor) SetSwitch(sw *Switch) {
  58         br.Switch = sw
  59 }
  60 func (_ *BaseReactor) GetChannels() []*ChannelDescriptor              { return nil }
  61 func (_ *BaseReactor) AddPeer(peer *Peer)                             {}
  62 func (_ *BaseReactor) RemovePeer(peer *Peer, reason interface{})      {}
  63 func (_ *BaseReactor) Receive(chID byte, peer *Peer, msgBytes []byte) {}
  64
  65 //-----------------------------------------------------------------------------
  66
  67 /*
  68 The `Switch` handles peer connections and exposes an API to receive incoming messages
  69 on `Reactors`.  Each `Reactor` is responsible for handling incoming messages of one
  70 or more `Channels`.  So while sending outgoing messages is typically performed on the peer,
  71 incoming messages are received on the reactor.
  72 */
  73 type Switch struct {
  74         cmn.BaseService
  75
  76         config           *cfg.P2PConfig
  77         peerConfig       *PeerConfig
  78         listeners        []Listener
  79         reactors         map[string]Reactor
  80         chDescs          []*ChannelDescriptor
  81         reactorsByCh     map[byte]Reactor
  82         peers            *PeerSet
  83         dialing          *cmn.CMap
  84         nodeInfo         *NodeInfo             // our node info
  85         nodePrivKey      crypto.PrivKeyEd25519 // our node privkey
  86         bannedPeer       map[string]time.Time
  87         db               dbm.DB
  88         TrustMetricStore *trust.TrustMetricStore
  89         ScamPeerCh       chan *Peer
  90         mtx              sync.Mutex
  91
  92         filterConnByAddr   func(net.Addr) error
  93         filterConnByPubKey func(crypto.PubKeyEd25519) error
  94 }
  95
  96 var (
  97         ErrSwitchDuplicatePeer = errors.New("Duplicate peer")
  98         ErrConnectSelf         = errors.New("Connect self")
  99         ErrPeerConnected       = errors.New("Peer is connected")
 100 )
 101
 102 func NewSwitch(config *cfg.P2PConfig, trustHistoryDB dbm.DB) *Switch {
 103         sw := &Switch{
 104                 config:       config,
 105                 peerConfig:   DefaultPeerConfig(config),
 106                 reactors:     make(map[string]Reactor),
 107                 chDescs:      make([]*ChannelDescriptor, 0),
 108                 reactorsByCh: make(map[byte]Reactor),
 109                 peers:        NewPeerSet(),
 110                 dialing:      cmn.NewCMap(),
 111                 nodeInfo:     nil,
 112                 db:           trustHistoryDB,
 113                 ScamPeerCh:   make(chan *Peer),
 114         }
 115         sw.BaseService = *cmn.NewBaseService(nil, "P2P Switch", sw)
 116         sw.TrustMetricStore = trust.NewTrustMetricStore(trustHistoryDB, trust.DefaultConfig())
 117         sw.TrustMetricStore.Start()
 118
 119         sw.bannedPeer = make(map[string]time.Time)
 120         if datajson := sw.db.Get([]byte(bannedPeerKey)); datajson != nil {
 121                 if err := json.Unmarshal(datajson, &sw.bannedPeer); err != nil {
 122                         return nil
 123                 }
 124         }
 125         go sw.scamPeerHandler()
 126         return sw
 127 }
 128
 129 // Not goroutine safe.
 130 func (sw *Switch) AddReactor(name string, reactor Reactor) Reactor {
 131         // Validate the reactor.
 132         // No two reactors can share the same channel.
 133         reactorChannels := reactor.GetChannels()
 134         for _, chDesc := range reactorChannels {
 135                 chID := chDesc.ID
 136                 if sw.reactorsByCh[chID] != nil {
 137                         cmn.PanicSanity(fmt.Sprintf("Channel %X has multiple reactors %v & %v", chID, sw.reactorsByCh[chID], reactor))
 138                 }
 139                 sw.chDescs = append(sw.chDescs, chDesc)
 140                 sw.reactorsByCh[chID] = reactor
 141         }
 142         sw.reactors[name] = reactor
 143         reactor.SetSwitch(sw)
 144         return reactor
 145 }
 146
 147 // Not goroutine safe.
 148 func (sw *Switch) Reactors() map[string]Reactor {
 149         return sw.reactors
 150 }
 151
 152 // Not goroutine safe.
 153 func (sw *Switch) Reactor(name string) Reactor {
 154         return sw.reactors[name]
 155 }
 156
 157 // Not goroutine safe.
 158 func (sw *Switch) AddListener(l Listener) {
 159         sw.listeners = append(sw.listeners, l)
 160 }
 161
 162 // Not goroutine safe.
 163 func (sw *Switch) Listeners() []Listener {
 164         return sw.listeners
 165 }
 166
 167 // Not goroutine safe.
 168 func (sw *Switch) IsListening() bool {
 169         return len(sw.listeners) > 0
 170 }
 171
 172 // Not goroutine safe.
 173 func (sw *Switch) SetNodeInfo(nodeInfo *NodeInfo) {
 174         sw.nodeInfo = nodeInfo
 175 }
 176
 177 // Not goroutine safe.
 178 func (sw *Switch) NodeInfo() *NodeInfo {
 179         return sw.nodeInfo
 180 }
 181
 182 // Not goroutine safe.
 183 // NOTE: Overwrites sw.nodeInfo.PubKey
 184 func (sw *Switch) SetNodePrivKey(nodePrivKey crypto.PrivKeyEd25519) {
 185         sw.nodePrivKey = nodePrivKey
 186         if sw.nodeInfo != nil {
 187                 sw.nodeInfo.PubKey = nodePrivKey.PubKey().Unwrap().(crypto.PubKeyEd25519)
 188         }
 189 }
 190
 191 // Switch.Start() starts all the reactors, peers, and listeners.
 192 func (sw *Switch) OnStart() error {
 193         sw.BaseService.OnStart()
 194         // Start reactors
 195         for _, reactor := range sw.reactors {
 196                 _, err := reactor.Start()
 197                 if err != nil {
 198                         return err
 199                 }
 200         }
 201         // Start peers
 202         for _, peer := range sw.peers.List() {
 203                 sw.startInitPeer(peer)
 204         }
 205         // Start listeners
 206         for _, listener := range sw.listeners {
 207                 go sw.listenerRoutine(listener)
 208         }
 209         return nil
 210 }
 211
 212 func (sw *Switch) OnStop() {
 213         sw.BaseService.OnStop()
 214         // Stop listeners
 215         for _, listener := range sw.listeners {
 216                 listener.Stop()
 217         }
 218         sw.listeners = nil
 219         // Stop peers
 220         for _, peer := range sw.peers.List() {
 221                 peer.Stop()
 222                 sw.peers.Remove(peer)
 223         }
 224         // Stop reactors
 225         for _, reactor := range sw.reactors {
 226                 reactor.Stop()
 227         }
 228 }
 229
 230 // NOTE: This performs a blocking handshake before the peer is added.
 231 // CONTRACT: If error is returned, peer is nil, and conn is immediately closed.
 232 func (sw *Switch) AddPeer(peer *Peer) error {
 233         if err := sw.FilterConnByAddr(peer.Addr()); err != nil {
 234                 return err
 235         }
 236
 237         if err := sw.FilterConnByPubKey(peer.PubKey()); err != nil {
 238                 return err
 239         }
 240
 241         if err := peer.HandshakeTimeout(sw.nodeInfo, time.Duration(sw.peerConfig.HandshakeTimeout*time.Second)); err != nil {
 242                 return err
 243         }
 244
 245         if err := sw.checkBannedPeer(peer.NodeInfo.ListenHost()); err != nil {
 246                 return err
 247         }
 248
 249         // Avoid self
 250         if sw.nodeInfo.PubKey.Equals(peer.PubKey().Wrap()) {
 251                 return errors.New("Ignoring connection from self")
 252         }
 253
 254         // Check version, chain id
 255         if err := sw.nodeInfo.CompatibleWith(peer.NodeInfo); err != nil {
 256                 return err
 257         }
 258
 259         // Check for duplicate peer
 260         if sw.peers.Has(peer.Key) {
 261                 return ErrSwitchDuplicatePeer
 262
 263         }
 264
 265         // Start peer
 266         if sw.IsRunning() {
 267                 if err := sw.startInitPeer(peer); err != nil {
 268                         return err
 269                 }
 270         }
 271
 272         // Add the peer to .peers.
 273         // We start it first so that a peer in the list is safe to Stop.
 274         // It should not err since we already checked peers.Has()
 275         if err := sw.peers.Add(peer); err != nil {
 276                 return err
 277         }
 278
 279         tm := trust.NewMetric()
 280
 281         tm.Start()
 282         sw.TrustMetricStore.AddPeerTrustMetric(peer.mconn.RemoteAddress.IP.String(), tm)
 283
 284         log.WithField("peer", peer).Info("Added peer")
 285         return nil
 286 }
 287
 288 func (sw *Switch) FilterConnByAddr(addr net.Addr) error {
 289         if sw.filterConnByAddr != nil {
 290                 return sw.filterConnByAddr(addr)
 291         }
 292         return nil
 293 }
 294
 295 func (sw *Switch) FilterConnByPubKey(pubkey crypto.PubKeyEd25519) error {
 296         if sw.filterConnByPubKey != nil {
 297                 return sw.filterConnByPubKey(pubkey)
 298         }
 299         return nil
 300
 301 }
 302
 303 func (sw *Switch) SetAddrFilter(f func(net.Addr) error) {
 304         sw.filterConnByAddr = f
 305 }
 306
 307 func (sw *Switch) SetPubKeyFilter(f func(crypto.PubKeyEd25519) error) {
 308         sw.filterConnByPubKey = f
 309 }
 310
 311 func (sw *Switch) startInitPeer(peer *Peer) error {
 312         peer.Start() // spawn send/recv routines
 313         for _, reactor := range sw.reactors {
 314                 if err := reactor.AddPeer(peer); err != nil {
 315                         return err
 316                 }
 317         }
 318         return nil
 319 }
 320
 321 // Dial a list of seeds asynchronously in random order
 322 func (sw *Switch) DialSeeds(addrBook *AddrBook, seeds []string) error {
 323
 324         netAddrs, err := NewNetAddressStrings(seeds)
 325         if err != nil {
 326                 return err
 327         }
 328
 329         if addrBook != nil {
 330                 // add seeds to `addrBook`
 331                 ourAddrS := sw.nodeInfo.ListenAddr
 332                 ourAddr, _ := NewNetAddressString(ourAddrS)
 333                 for _, netAddr := range netAddrs {
 334                         // do not add ourselves
 335                         if netAddr.Equals(ourAddr) {
 336                                 continue
 337                         }
 338                         addrBook.AddAddress(netAddr, ourAddr)
 339                 }
 340                 addrBook.Save()
 341         }
 342
 343         // permute the list, dial them in random order.
 344         perm := rand.Perm(len(netAddrs))
 345         for i := 0; i < len(perm); i++ {
 346                 j := perm[i]
 347                 sw.dialSeed(netAddrs[j])
 348         }
 349         return nil
 350 }
 351
 352 func (sw *Switch) dialSeed(addr *NetAddress) {
 353         peer, err := sw.DialPeerWithAddress(addr, true)
 354         if err != nil {
 355                 log.WithField("error", err).Error("Error dialing seed")
 356         } else {
 357                 log.WithField("peer", peer).Info("Connected to seed")
 358         }
 359 }
 360
 361 func (sw *Switch) DialPeerWithAddress(addr *NetAddress, persistent bool) (*Peer, error) {
 362         if err := sw.checkBannedPeer(addr.IP.String()); err != nil {
 363                 return nil, err
 364         }
 365         if strings.Compare(addr.IP.String(), sw.nodeInfo.ListenHost()) == 0 {
 366                 return nil, ErrConnectSelf
 367         }
 368         for _, v := range sw.Peers().list {
 369                 if strings.Compare(v.mconn.RemoteAddress.IP.String(), addr.IP.String()) == 0 {
 370                         return nil, ErrPeerConnected
 371                 }
 372         }
 373         sw.dialing.Set(addr.IP.String(), addr)
 374         defer sw.dialing.Delete(addr.IP.String())
 375
 376         log.WithField("address", addr).Info("Dialing peer")
 377         peer, err := newOutboundPeerWithConfig(addr, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.peerConfig)
 378         if err != nil {
 379                 log.WithFields(log.Fields{
 380                         "address": addr,
 381                         "error":   err,
 382                 }).Info("Failed to dial peer")
 383                 return nil, err
 384         }
 385         peer.SetLogger(sw.Logger.With("peer", addr))
 386         if persistent {
 387                 peer.makePersistent()
 388         }
 389         err = sw.AddPeer(peer)
 390         if err != nil {
 391                 log.WithFields(log.Fields{
 392                         "address": addr,
 393                         "error":   err,
 394                 }).Info("Failed to add peer")
 395                 peer.CloseConn()
 396                 return nil, err
 397         }
 398         log.WithFields(log.Fields{
 399                 "address": addr,
 400         }).Info("Dialed and added peer")
 401         return peer, nil
 402 }
 403
 404 func (sw *Switch) IsDialing(addr *NetAddress) bool {
 405         return sw.dialing.Has(addr.IP.String())
 406 }
 407
 408 // Broadcast runs a go routine for each attempted send, which will block
 409 // trying to send for defaultSendTimeoutSeconds. Returns a channel
 410 // which receives success values for each attempted send (false if times out)
 411 // NOTE: Broadcast uses goroutines, so order of broadcast may not be preserved.
 412 func (sw *Switch) Broadcast(chID byte, msg interface{}) chan bool {
 413         successChan := make(chan bool, len(sw.peers.List()))
 414         log.WithFields(log.Fields{
 415                 "chID": chID,
 416                 "msg":  msg,
 417         }).Debug("Broadcast")
 418         for _, peer := range sw.peers.List() {
 419                 go func(peer *Peer) {
 420                         success := peer.Send(chID, msg)
 421                         successChan <- success
 422                 }(peer)
 423         }
 424         return successChan
 425 }
 426
 427 // Returns the count of outbound/inbound and outbound-dialing peers.
 428 func (sw *Switch) NumPeers() (outbound, inbound, dialing int) {
 429         peers := sw.peers.List()
 430         for _, peer := range peers {
 431                 if peer.outbound {
 432                         outbound++
 433                 } else {
 434                         inbound++
 435                 }
 436         }
 437         dialing = sw.dialing.Size()
 438         return
 439 }
 440
 441 func (sw *Switch) Peers() *PeerSet {
 442         return sw.peers
 443 }
 444
 445 // Disconnect from a peer due to external error, retry if it is a persistent peer.
 446 // TODO: make record depending on reason.
 447 func (sw *Switch) StopPeerForError(peer *Peer, reason interface{}) {
 448         addr := NewNetAddress(peer.Addr())
 449         log.WithFields(log.Fields{
 450                 "peer":  peer,
 451                 "error": reason,
 452         }).Info("Stopping peer due to error")
 453         sw.stopAndRemovePeer(peer, reason)
 454
 455         if peer.IsPersistent() {
 456                 log.WithField("peer", peer).Info("Reconnecting to peer")
 457                 for i := 1; i < reconnectAttempts; i++ {
 458                         if !sw.IsRunning() {
 459                                 return
 460                         }
 461
 462                         peer, err := sw.DialPeerWithAddress(addr, true)
 463                         if err != nil {
 464                                 if i == reconnectAttempts {
 465                                         log.WithFields(log.Fields{
 466                                                 "retries": i,
 467                                                 "error":   err,
 468                                         }).Info("Error reconnecting to peer. Giving up")
 469                                         return
 470                                 }
 471
 472                                 if errors.Root(err) == ErrConnectBannedPeer || errors.Root(err) == ErrPeerConnected || errors.Root(err) == ErrSwitchDuplicatePeer || errors.Root(err) == ErrConnectSelf {
 473                                         log.WithField("error", err).Info("Error reconnecting to peer. ")
 474                                         return
 475                                 }
 476
 477                                 log.WithFields(log.Fields{
 478                                         "retries": i,
 479                                         "error":   err,
 480                                 }).Info("Error reconnecting to peer. Trying again")
 481                                 time.Sleep(reconnectInterval)
 482                                 continue
 483                         }
 484
 485                         log.WithField("peer", peer).Info("Reconnected to peer")
 486                         return
 487                 }
 488         }
 489 }
 490
 491 // Disconnect from a peer gracefully.
 492 // TODO: handle graceful disconnects.
 493 func (sw *Switch) StopPeerGracefully(peer *Peer) {
 494         log.Info("Stopping peer gracefully")
 495         sw.stopAndRemovePeer(peer, nil)
 496 }
 497
 498 func (sw *Switch) stopAndRemovePeer(peer *Peer, reason interface{}) {
 499         sw.peers.Remove(peer)
 500         peer.Stop()
 501         for _, reactor := range sw.reactors {
 502                 reactor.RemovePeer(peer, reason)
 503         }
 504 }
 505
 506 func (sw *Switch) listenerRoutine(l Listener) {
 507         for {
 508                 inConn, ok := <-l.Connections()
 509                 if !ok {
 510                         break
 511                 }
 512
 513                 // ignore connection if we already have enough
 514                 maxPeers := sw.config.MaxNumPeers
 515                 if maxPeers <= sw.peers.Size() {
 516                         // close inConn
 517                         inConn.Close()
 518                         log.WithFields(log.Fields{
 519                                 "address":  inConn.RemoteAddr().String(),
 520                                 "numPeers": sw.peers.Size(),
 521                                 "max":      maxPeers,
 522                         }).Info("Ignoring inbound connection: already have enough peers")
 523                         continue
 524                 }
 525
 526                 // New inbound connection!
 527                 err := sw.addPeerWithConnectionAndConfig(inConn, sw.peerConfig)
 528                 if err != nil {
 529                         // conn close for returing err
 530                         inConn.Close()
 531                         log.WithFields(log.Fields{
 532                                 "address": inConn.RemoteAddr().String(),
 533                                 "error":   err,
 534                         }).Info("Ignoring inbound connection: error while adding peer")
 535                         continue
 536                 }
 537
 538                 // NOTE: We don't yet have the listening port of the
 539                 // remote (if they have a listener at all).
 540                 // The peerHandshake will handle that
 541         }
 542
 543         // cleanup
 544 }
 545
 546 //-----------------------------------------------------------------------------
 547
 548 type SwitchEventNewPeer struct {
 549         Peer *Peer
 550 }
 551
 552 type SwitchEventDonePeer struct {
 553         Peer  *Peer
 554         Error interface{}
 555 }
 556
 557 //------------------------------------------------------------------
 558 // Switches connected via arbitrary net.Conn; useful for testing
 559
 560 // Returns n switches, connected according to the connect func.
 561 // If connect==Connect2Switches, the switches will be fully connected.
 562 // initSwitch defines how the ith switch should be initialized (ie. with what reactors).
 563 // NOTE: panics if any switch fails to start.
 564 func MakeConnectedSwitches(cfg *cfg.P2PConfig, n int, initSwitch func(int, *Switch) *Switch, connect func([]*Switch, int, int)) []*Switch {
 565         switches := make([]*Switch, n)
 566         for i := 0; i < n; i++ {
 567                 switches[i] = makeSwitch(cfg, i, "testing", "123.123.123", initSwitch)
 568         }
 569
 570         if err := StartSwitches(switches); err != nil {
 571                 panic(err)
 572         }
 573
 574         for i := 0; i < n; i++ {
 575                 for j := i; j < n; j++ {
 576                         connect(switches, i, j)
 577                 }
 578         }
 579
 580         return switches
 581 }
 582
 583 var PanicOnAddPeerErr = false
 584
 585 // Will connect switches i and j via net.Pipe()
 586 // Blocks until a conection is established.
 587 // NOTE: caller ensures i and j are within bounds
 588 func Connect2Switches(switches []*Switch, i, j int) {
 589         switchI := switches[i]
 590         switchJ := switches[j]
 591         c1, c2 := net.Pipe()
 592         doneCh := make(chan struct{})
 593         go func() {
 594                 err := switchI.addPeerWithConnection(c1)
 595                 if PanicOnAddPeerErr && err != nil {
 596                         panic(err)
 597                 }
 598                 doneCh <- struct{}{}
 599         }()
 600         go func() {
 601                 err := switchJ.addPeerWithConnection(c2)
 602                 if PanicOnAddPeerErr && err != nil {
 603                         panic(err)
 604                 }
 605                 doneCh <- struct{}{}
 606         }()
 607         <-doneCh
 608         <-doneCh
 609 }
 610
 611 func StartSwitches(switches []*Switch) error {
 612         for _, s := range switches {
 613                 _, err := s.Start() // start switch and reactors
 614                 if err != nil {
 615                         return err
 616                 }
 617         }
 618         return nil
 619 }
 620
 621 func makeSwitch(cfg *cfg.P2PConfig, i int, network, version string, initSwitch func(int, *Switch) *Switch) *Switch {
 622         privKey := crypto.GenPrivKeyEd25519()
 623         // new switch, add reactors
 624         // TODO: let the config be passed in?
 625         s := initSwitch(i, NewSwitch(cfg, nil))
 626         s.SetNodeInfo(&NodeInfo{
 627                 PubKey:     privKey.PubKey().Unwrap().(crypto.PubKeyEd25519),
 628                 Moniker:    cmn.Fmt("switch%d", i),
 629                 Network:    network,
 630                 Version:    version,
 631                 RemoteAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
 632                 ListenAddr: cmn.Fmt("%v:%v", network, rand.Intn(64512)+1023),
 633         })
 634         s.SetNodePrivKey(privKey)
 635         return s
 636 }
 637
 638 func (sw *Switch) addPeerWithConnection(conn net.Conn) error {
 639         peer, err := newInboundPeer(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, sw.config)
 640         if err != nil {
 641                 conn.Close()
 642                 return err
 643         }
 644         peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
 645         if err = sw.AddPeer(peer); err != nil {
 646                 conn.Close()
 647                 return err
 648         }
 649
 650         return nil
 651 }
 652
 653 func (sw *Switch) addPeerWithConnectionAndConfig(conn net.Conn, config *PeerConfig) error {
 654         fullAddr := conn.RemoteAddr().String()
 655         host, _, err := net.SplitHostPort(fullAddr)
 656         if err != nil {
 657                 return err
 658         }
 659
 660         if err = sw.checkBannedPeer(host); err != nil {
 661                 return err
 662         }
 663
 664         peer, err := newInboundPeerWithConfig(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, config)
 665         if err != nil {
 666                 return err
 667         }
 668         peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
 669         if err = sw.AddPeer(peer); err != nil {
 670                 return err
 671         }
 672
 673         return nil
 674 }
 675
 676 func (sw *Switch) AddBannedPeer(peer *Peer) error {
 677         sw.mtx.Lock()
 678         defer sw.mtx.Unlock()
 679
 680         key := peer.mconn.RemoteAddress.IP.String()
 681         sw.bannedPeer[key] = time.Now().Add(defaultBanDuration)
 682         datajson, err := json.Marshal(sw.bannedPeer)
 683         if err != nil {
 684                 return err
 685         }
 686         sw.db.Set([]byte(bannedPeerKey), datajson)
 687         return nil
 688 }
 689
 690 func (sw *Switch) DelBannedPeer(addr string) error {
 691         sw.mtx.Lock()
 692         defer sw.mtx.Unlock()
 693
 694         delete(sw.bannedPeer, addr)
 695         datajson, err := json.Marshal(sw.bannedPeer)
 696         if err != nil {
 697                 return err
 698         }
 699         sw.db.Set([]byte(bannedPeerKey), datajson)
 700         return nil
 701 }
 702
 703 func (sw *Switch) scamPeerHandler() {
 704         for src := range sw.ScamPeerCh {
 705                 var tm *trust.TrustMetric
 706                 key := src.Connection().RemoteAddress.IP.String()
 707                 if tm = sw.TrustMetricStore.GetPeerTrustMetric(key); tm == nil {
 708                         log.Errorf("Can't get peer trust metric")
 709                         continue
 710                 }
 711                 sw.delTrustMetric(tm, src)
 712         }
 713 }
 714
 715 func (sw *Switch) AddScamPeer(src *Peer) {
 716         sw.ScamPeerCh <- src
 717 }
 718
 719 func (sw *Switch) delTrustMetric(tm *trust.TrustMetric, src *Peer) {
 720         key := src.Connection().RemoteAddress.IP.String()
 721         tm.BadEvents(1)
 722         if tm.TrustScore() < peerBannedTM {
 723                 sw.AddBannedPeer(src)
 724                 sw.TrustMetricStore.PeerDisconnected(key)
 725                 sw.StopPeerGracefully(src)
 726         }
 727 }
 728
 729 func (sw *Switch) checkBannedPeer(peer string) error {
 730         if banEnd, ok := sw.bannedPeer[peer]; ok {
 731                 if time.Now().Before(banEnd) {
 732                         return ErrConnectBannedPeer
 733                 }
 734                 sw.DelBannedPeer(peer)
 735         }
 736         return nil
 737 }