p2p/switch.go

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