client.go

   1 package torrent
   2
   3 import (
   4         "bufio"
   5         "bytes"
   6         "context"
   7         "crypto/rand"
   8         "encoding/binary"
   9         "errors"
  10         "fmt"
  11         "io"
  12         "net"
  13         "net/http"
  14         "net/url"
  15         "strconv"
  16         "strings"
  17         "time"
  18
  19         "github.com/anacrolix/dht/v2"
  20         "github.com/anacrolix/dht/v2/krpc"
  21         "github.com/anacrolix/log"
  22         "github.com/anacrolix/missinggo"
  23         "github.com/anacrolix/missinggo/bitmap"
  24         "github.com/anacrolix/missinggo/conntrack"
  25         "github.com/anacrolix/missinggo/perf"
  26         "github.com/anacrolix/missinggo/pproffd"
  27         "github.com/anacrolix/missinggo/pubsub"
  28         "github.com/anacrolix/missinggo/slices"
  29         "github.com/anacrolix/sync"
  30         "github.com/anacrolix/torrent/bencode"
  31         "github.com/anacrolix/torrent/iplist"
  32         "github.com/anacrolix/torrent/metainfo"
  33         "github.com/anacrolix/torrent/mse"
  34         pp "github.com/anacrolix/torrent/peer_protocol"
  35         "github.com/anacrolix/torrent/storage"
  36         "github.com/davecgh/go-spew/spew"
  37         "github.com/dustin/go-humanize"
  38         "github.com/google/btree"
  39         "golang.org/x/time/rate"
  40         "golang.org/x/xerrors"
  41 )
  42
  43 // Clients contain zero or more Torrents. A Client manages a blocklist, the
  44 // TCP/UDP protocol ports, and DHT as desired.
  45 type Client struct {
  46         // An aggregate of stats over all connections. First in struct to ensure
  47         // 64-bit alignment of fields. See #262.
  48         stats ConnStats
  49
  50         _mu    sync.RWMutex
  51         event  sync.Cond
  52         closed missinggo.Event
  53
  54         config *ClientConfig
  55         logger log.Logger
  56
  57         peerID         PeerID
  58         defaultStorage *storage.Client
  59         onClose        []func()
  60         conns          []socket
  61         dhtServers     []*dht.Server
  62         ipBlockList    iplist.Ranger
  63         // Our BitTorrent protocol extension bytes, sent in our BT handshakes.
  64         extensionBytes pp.PeerExtensionBits
  65
  66         // Set of addresses that have our client ID. This intentionally will
  67         // include ourselves if we end up trying to connect to our own address
  68         // through legitimate channels.
  69         dopplegangerAddrs map[string]struct{}
  70         badPeerIPs        map[string]struct{}
  71         torrents          map[InfoHash]*Torrent
  72
  73         acceptLimiter   map[ipStr]int
  74         dialRateLimiter *rate.Limiter
  75 }
  76
  77 type ipStr string
  78
  79 func (cl *Client) BadPeerIPs() []string {
  80         cl.rLock()
  81         defer cl.rUnlock()
  82         return cl.badPeerIPsLocked()
  83 }
  84
  85 func (cl *Client) badPeerIPsLocked() []string {
  86         return slices.FromMapKeys(cl.badPeerIPs).([]string)
  87 }
  88
  89 func (cl *Client) PeerID() PeerID {
  90         return cl.peerID
  91 }
  92
  93 func (cl *Client) LocalPort() (port int) {
  94         cl.eachListener(func(l socket) bool {
  95                 _port := missinggo.AddrPort(l.Addr())
  96                 if _port == 0 {
  97                         panic(l)
  98                 }
  99                 if port == 0 {
 100                         port = _port
 101                 } else if port != _port {
 102                         panic("mismatched ports")
 103                 }
 104                 return true
 105         })
 106         return
 107 }
 108
 109 func writeDhtServerStatus(w io.Writer, s *dht.Server) {
 110         dhtStats := s.Stats()
 111         fmt.Fprintf(w, "\t# Nodes: %d (%d good, %d banned)\n", dhtStats.Nodes, dhtStats.GoodNodes, dhtStats.BadNodes)
 112         fmt.Fprintf(w, "\tServer ID: %x\n", s.ID())
 113         fmt.Fprintf(w, "\tAnnounces: %d\n", dhtStats.SuccessfulOutboundAnnouncePeerQueries)
 114         fmt.Fprintf(w, "\tOutstanding transactions: %d\n", dhtStats.OutstandingTransactions)
 115 }
 116
 117 // Writes out a human readable status of the client, such as for writing to a
 118 // HTTP status page.
 119 func (cl *Client) WriteStatus(_w io.Writer) {
 120         cl.rLock()
 121         defer cl.rUnlock()
 122         w := bufio.NewWriter(_w)
 123         defer w.Flush()
 124         fmt.Fprintf(w, "Listen port: %d\n", cl.LocalPort())
 125         fmt.Fprintf(w, "Peer ID: %+q\n", cl.PeerID())
 126         fmt.Fprintf(w, "Announce key: %x\n", cl.announceKey())
 127         fmt.Fprintf(w, "Banned IPs: %d\n", len(cl.badPeerIPsLocked()))
 128         cl.eachDhtServer(func(s *dht.Server) {
 129                 fmt.Fprintf(w, "%s DHT server at %s:\n", s.Addr().Network(), s.Addr().String())
 130                 writeDhtServerStatus(w, s)
 131         })
 132         spew.Fdump(w, &cl.stats)
 133         fmt.Fprintf(w, "# Torrents: %d\n", len(cl.torrentsAsSlice()))
 134         fmt.Fprintln(w)
 135         for _, t := range slices.Sort(cl.torrentsAsSlice(), func(l, r *Torrent) bool {
 136                 return l.InfoHash().AsString() < r.InfoHash().AsString()
 137         }).([]*Torrent) {
 138                 if t.name() == "" {
 139                         fmt.Fprint(w, "<unknown name>")
 140                 } else {
 141                         fmt.Fprint(w, t.name())
 142                 }
 143                 fmt.Fprint(w, "\n")
 144                 if t.info != nil {
 145                         fmt.Fprintf(w, "%f%% of %d bytes (%s)", 100*(1-float64(t.bytesMissingLocked())/float64(t.info.TotalLength())), t.length, humanize.Bytes(uint64(t.info.TotalLength())))
 146                 } else {
 147                         w.WriteString("<missing metainfo>")
 148                 }
 149                 fmt.Fprint(w, "\n")
 150                 t.writeStatus(w)
 151                 fmt.Fprintln(w)
 152         }
 153 }
 154
 155 const debugLogValue = log.Debug
 156
 157 func (cl *Client) debugLogFilter(m log.Msg) bool {
 158         if cl.config.Debug {
 159                 return true
 160         }
 161         return !m.HasValue(debugLogValue)
 162 }
 163
 164 func (cl *Client) initLogger() {
 165         cl.logger = cl.config.Logger.WithValues(cl).WithFilter(cl.debugLogFilter)
 166 }
 167
 168 func (cl *Client) announceKey() int32 {
 169         return int32(binary.BigEndian.Uint32(cl.peerID[16:20]))
 170 }
 171
 172 func NewClient(cfg *ClientConfig) (cl *Client, err error) {
 173         if cfg == nil {
 174                 cfg = NewDefaultClientConfig()
 175                 cfg.ListenPort = 0
 176         }
 177         defer func() {
 178                 if err != nil {
 179                         cl = nil
 180                 }
 181         }()
 182         cl = &Client{
 183                 config:            cfg,
 184                 dopplegangerAddrs: make(map[string]struct{}),
 185                 torrents:          make(map[metainfo.Hash]*Torrent),
 186                 dialRateLimiter:   rate.NewLimiter(10, 10),
 187         }
 188         go cl.acceptLimitClearer()
 189         cl.initLogger()
 190         defer func() {
 191                 if err == nil {
 192                         return
 193                 }
 194                 cl.Close()
 195         }()
 196         cl.extensionBytes = defaultPeerExtensionBytes()
 197         cl.event.L = cl.locker()
 198         storageImpl := cfg.DefaultStorage
 199         if storageImpl == nil {
 200                 // We'd use mmap but HFS+ doesn't support sparse files.
 201                 storageImpl = storage.NewFile(cfg.DataDir)
 202                 cl.onClose = append(cl.onClose, func() {
 203                         if err := storageImpl.Close(); err != nil {
 204                                 cl.logger.Printf("error closing default storage: %s", err)
 205                         }
 206                 })
 207         }
 208         cl.defaultStorage = storage.NewClient(storageImpl)
 209         if cfg.IPBlocklist != nil {
 210                 cl.ipBlockList = cfg.IPBlocklist
 211         }
 212
 213         if cfg.PeerID != "" {
 214                 missinggo.CopyExact(&cl.peerID, cfg.PeerID)
 215         } else {
 216                 o := copy(cl.peerID[:], cfg.Bep20)
 217                 _, err = rand.Read(cl.peerID[o:])
 218                 if err != nil {
 219                         panic("error generating peer id")
 220                 }
 221         }
 222
 223         if cl.config.HTTPProxy == nil && cl.config.ProxyURL != "" {
 224                 if fixedURL, err := url.Parse(cl.config.ProxyURL); err == nil {
 225                         cl.config.HTTPProxy = http.ProxyURL(fixedURL)
 226                 }
 227         }
 228
 229         cl.conns, err = listenAll(cl.listenNetworks(), cl.config.ListenHost, cl.config.ListenPort, cl.config.ProxyURL, cl.firewallCallback)
 230         if err != nil {
 231                 return
 232         }
 233         // Check for panics.
 234         cl.LocalPort()
 235
 236         for _, s := range cl.conns {
 237                 if peerNetworkEnabled(parseNetworkString(s.Addr().Network()), cl.config) {
 238                         go cl.acceptConnections(s)
 239                 }
 240         }
 241
 242         go cl.forwardPort()
 243         if !cfg.NoDHT {
 244                 for _, s := range cl.conns {
 245                         if pc, ok := s.(net.PacketConn); ok {
 246                                 ds, err := cl.newDhtServer(pc)
 247                                 if err != nil {
 248                                         panic(err)
 249                                 }
 250                                 cl.dhtServers = append(cl.dhtServers, ds)
 251                         }
 252                 }
 253         }
 254
 255         return
 256 }
 257
 258 func (cl *Client) firewallCallback(net.Addr) bool {
 259         cl.rLock()
 260         block := !cl.wantConns()
 261         cl.rUnlock()
 262         if block {
 263                 torrent.Add("connections firewalled", 1)
 264         } else {
 265                 torrent.Add("connections not firewalled", 1)
 266         }
 267         return block
 268 }
 269
 270 func (cl *Client) enabledPeerNetworks() (ns []network) {
 271         for _, n := range allPeerNetworks {
 272                 if peerNetworkEnabled(n, cl.config) {
 273                         ns = append(ns, n)
 274                 }
 275         }
 276         return
 277 }
 278
 279 func (cl *Client) listenOnNetwork(n network) bool {
 280         if n.Ipv4 && cl.config.DisableIPv4 {
 281                 return false
 282         }
 283         if n.Ipv6 && cl.config.DisableIPv6 {
 284                 return false
 285         }
 286         if n.Tcp && cl.config.DisableTCP {
 287                 return false
 288         }
 289         if n.Udp && cl.config.DisableUTP && cl.config.NoDHT {
 290                 return false
 291         }
 292         return true
 293 }
 294
 295 func (cl *Client) listenNetworks() (ns []network) {
 296         for _, n := range allPeerNetworks {
 297                 if cl.listenOnNetwork(n) {
 298                         ns = append(ns, n)
 299                 }
 300         }
 301         return
 302 }
 303
 304 func (cl *Client) newDhtServer(conn net.PacketConn) (s *dht.Server, err error) {
 305         cfg := dht.ServerConfig{
 306                 IPBlocklist:    cl.ipBlockList,
 307                 Conn:           conn,
 308                 OnAnnouncePeer: cl.onDHTAnnouncePeer,
 309                 PublicIP: func() net.IP {
 310                         if connIsIpv6(conn) && cl.config.PublicIp6 != nil {
 311                                 return cl.config.PublicIp6
 312                         }
 313                         return cl.config.PublicIp4
 314                 }(),
 315                 StartingNodes:      cl.config.DhtStartingNodes,
 316                 ConnectionTracking: cl.config.ConnTracker,
 317                 OnQuery:            cl.config.DHTOnQuery,
 318         }
 319         s, err = dht.NewServer(&cfg)
 320         if err == nil {
 321                 go func() {
 322                         ts, err := s.Bootstrap()
 323                         if err != nil {
 324                                 cl.logger.Printf("error bootstrapping dht: %s", err)
 325                         }
 326                         log.Str("completed bootstrap").AddValues(s, ts).Log(cl.logger)
 327                 }()
 328         }
 329         return
 330 }
 331
 332 func (cl *Client) Closed() <-chan struct{} {
 333         cl.lock()
 334         defer cl.unlock()
 335         return cl.closed.C()
 336 }
 337
 338 func (cl *Client) eachDhtServer(f func(*dht.Server)) {
 339         for _, ds := range cl.dhtServers {
 340                 f(ds)
 341         }
 342 }
 343
 344 func (cl *Client) closeSockets() {
 345         cl.eachListener(func(l socket) bool {
 346                 l.Close()
 347                 return true
 348         })
 349         cl.conns = nil
 350 }
 351
 352 // Stops the client. All connections to peers are closed and all activity will
 353 // come to a halt.
 354 func (cl *Client) Close() {
 355         cl.lock()
 356         defer cl.unlock()
 357         cl.closed.Set()
 358         cl.eachDhtServer(func(s *dht.Server) { s.Close() })
 359         cl.closeSockets()
 360         for _, t := range cl.torrents {
 361                 t.close()
 362         }
 363         for _, f := range cl.onClose {
 364                 f()
 365         }
 366         cl.event.Broadcast()
 367 }
 368
 369 func (cl *Client) ipBlockRange(ip net.IP) (r iplist.Range, blocked bool) {
 370         if cl.ipBlockList == nil {
 371                 return
 372         }
 373         return cl.ipBlockList.Lookup(ip)
 374 }
 375
 376 func (cl *Client) ipIsBlocked(ip net.IP) bool {
 377         _, blocked := cl.ipBlockRange(ip)
 378         return blocked
 379 }
 380
 381 func (cl *Client) wantConns() bool {
 382         for _, t := range cl.torrents {
 383                 if t.wantConns() {
 384                         return true
 385                 }
 386         }
 387         return false
 388 }
 389
 390 func (cl *Client) waitAccept() {
 391         for {
 392                 if cl.closed.IsSet() {
 393                         return
 394                 }
 395                 if cl.wantConns() {
 396                         return
 397                 }
 398                 cl.event.Wait()
 399         }
 400 }
 401
 402 func (cl *Client) rejectAccepted(conn net.Conn) bool {
 403         ra := conn.RemoteAddr()
 404         rip := missinggo.AddrIP(ra)
 405         if cl.config.DisableIPv4Peers && rip.To4() != nil {
 406                 return true
 407         }
 408         if cl.config.DisableIPv4 && len(rip) == net.IPv4len {
 409                 return true
 410         }
 411         if cl.config.DisableIPv6 && len(rip) == net.IPv6len && rip.To4() == nil {
 412                 return true
 413         }
 414         if cl.rateLimitAccept(rip) {
 415                 return true
 416         }
 417         return cl.badPeerIPPort(rip, missinggo.AddrPort(ra))
 418 }
 419
 420 func (cl *Client) acceptConnections(l net.Listener) {
 421         for {
 422                 conn, err := l.Accept()
 423                 torrent.Add("client listener accepts", 1)
 424                 conn = pproffd.WrapNetConn(conn)
 425                 cl.rLock()
 426                 closed := cl.closed.IsSet()
 427                 reject := false
 428                 if conn != nil {
 429                         reject = cl.rejectAccepted(conn)
 430                 }
 431                 cl.rUnlock()
 432                 if closed {
 433                         if conn != nil {
 434                                 conn.Close()
 435                         }
 436                         return
 437                 }
 438                 if err != nil {
 439                         cl.logger.Printf("error accepting connection: %s", err)
 440                         continue
 441                 }
 442                 go func() {
 443                         if reject {
 444                                 torrent.Add("rejected accepted connections", 1)
 445                                 conn.Close()
 446                         } else {
 447                                 go cl.incomingConnection(conn)
 448                         }
 449                         log.Fmsg("accepted %s connection from %s", conn.RemoteAddr().Network(), conn.RemoteAddr()).AddValue(debugLogValue).Log(cl.logger)
 450                         torrent.Add(fmt.Sprintf("accepted conn remote IP len=%d", len(missinggo.AddrIP(conn.RemoteAddr()))), 1)
 451                         torrent.Add(fmt.Sprintf("accepted conn network=%s", conn.RemoteAddr().Network()), 1)
 452                         torrent.Add(fmt.Sprintf("accepted on %s listener", l.Addr().Network()), 1)
 453                 }()
 454         }
 455 }
 456
 457 func (cl *Client) incomingConnection(nc net.Conn) {
 458         defer nc.Close()
 459         if tc, ok := nc.(*net.TCPConn); ok {
 460                 tc.SetLinger(0)
 461         }
 462         c := cl.newConnection(nc, false, missinggo.IpPortFromNetAddr(nc.RemoteAddr()), nc.RemoteAddr().Network())
 463         c.Discovery = peerSourceIncoming
 464         cl.runReceivedConn(c)
 465 }
 466
 467 // Returns a handle to the given torrent, if it's present in the client.
 468 func (cl *Client) Torrent(ih metainfo.Hash) (t *Torrent, ok bool) {
 469         cl.lock()
 470         defer cl.unlock()
 471         t, ok = cl.torrents[ih]
 472         return
 473 }
 474
 475 func (cl *Client) torrent(ih metainfo.Hash) *Torrent {
 476         return cl.torrents[ih]
 477 }
 478
 479 type dialResult struct {
 480         Conn    net.Conn
 481         Network string
 482 }
 483
 484 func countDialResult(err error) {
 485         if err == nil {
 486                 torrent.Add("successful dials", 1)
 487         } else {
 488                 torrent.Add("unsuccessful dials", 1)
 489         }
 490 }
 491
 492 func reducedDialTimeout(minDialTimeout, max time.Duration, halfOpenLimit int, pendingPeers int) (ret time.Duration) {
 493         ret = max / time.Duration((pendingPeers+halfOpenLimit)/halfOpenLimit)
 494         if ret < minDialTimeout {
 495                 ret = minDialTimeout
 496         }
 497         return
 498 }
 499
 500 // Returns whether an address is known to connect to a client with our own ID.
 501 func (cl *Client) dopplegangerAddr(addr string) bool {
 502         _, ok := cl.dopplegangerAddrs[addr]
 503         return ok
 504 }
 505
 506 // Returns a connection over UTP or TCP, whichever is first to connect.
 507 func (cl *Client) dialFirst(ctx context.Context, addr string) (res dialResult) {
 508         {
 509                 t := perf.NewTimer(perf.CallerName(0))
 510                 defer func() {
 511                         if res.Conn == nil {
 512                                 t.Mark(fmt.Sprintf("returned no conn (context: %v)", ctx.Err()))
 513                         } else {
 514                                 t.Mark("returned conn over " + res.Network)
 515                         }
 516                 }()
 517         }
 518         ctx, cancel := context.WithCancel(ctx)
 519         // As soon as we return one connection, cancel the others.
 520         defer cancel()
 521         left := 0
 522         resCh := make(chan dialResult, left)
 523         func() {
 524                 cl.lock()
 525                 defer cl.unlock()
 526                 cl.eachListener(func(s socket) bool {
 527                         func() {
 528                                 network := s.Addr().Network()
 529                                 if !peerNetworkEnabled(parseNetworkString(network), cl.config) {
 530                                         return
 531                                 }
 532                                 left++
 533                                 //cl.logger.Printf("dialing %s on %s/%s", addr, s.Addr().Network(), s.Addr())
 534                                 go func() {
 535                                         resCh <- dialResult{
 536                                                 cl.dialFromSocket(ctx, s, addr),
 537                                                 network,
 538                                         }
 539                                 }()
 540                         }()
 541                         return true
 542                 })
 543         }()
 544         // Wait for a successful connection.
 545         func() {
 546                 defer perf.ScopeTimer()()
 547                 for ; left > 0 && res.Conn == nil; left-- {
 548                         res = <-resCh
 549                 }
 550         }()
 551         // There are still incompleted dials.
 552         go func() {
 553                 for ; left > 0; left-- {
 554                         conn := (<-resCh).Conn
 555                         if conn != nil {
 556                                 conn.Close()
 557                         }
 558                 }
 559         }()
 560         if res.Conn != nil {
 561                 go torrent.Add(fmt.Sprintf("network dialed first: %s", res.Conn.RemoteAddr().Network()), 1)
 562         }
 563         //if res.Conn != nil {
 564         //      cl.logger.Printf("first connection for %s from %s/%s", addr, res.Conn.LocalAddr().Network(), res.Conn.LocalAddr().String())
 565         //} else {
 566         //      cl.logger.Printf("failed to dial %s", addr)
 567         //}
 568         return res
 569 }
 570
 571 func (cl *Client) dialFromSocket(ctx context.Context, s socket, addr string) net.Conn {
 572         network := s.Addr().Network()
 573         cte := cl.config.ConnTracker.Wait(
 574                 ctx,
 575                 conntrack.Entry{network, s.Addr().String(), addr},
 576                 "dial torrent client",
 577                 0,
 578         )
 579         // Try to avoid committing to a dial if the context is complete as it's difficult to determine
 580         // which dial errors allow us to forget the connection tracking entry handle.
 581         if ctx.Err() != nil {
 582                 if cte != nil {
 583                         cte.Forget()
 584                 }
 585                 return nil
 586         }
 587         c, err := s.dial(ctx, addr)
 588         // This is a bit optimistic, but it looks non-trivial to thread this through the proxy code. Set
 589         // it now in case we close the connection forthwith.
 590         if tc, ok := c.(*net.TCPConn); ok {
 591                 tc.SetLinger(0)
 592         }
 593         countDialResult(err)
 594         if c == nil {
 595                 if err != nil && forgettableDialError(err) {
 596                         cte.Forget()
 597                 } else {
 598                         cte.Done()
 599                 }
 600                 return nil
 601         }
 602         return closeWrapper{c, func() error {
 603                 err := c.Close()
 604                 cte.Done()
 605                 return err
 606         }}
 607 }
 608
 609 func forgettableDialError(err error) bool {
 610         return strings.Contains(err.Error(), "no suitable address found")
 611 }
 612
 613 func (cl *Client) noLongerHalfOpen(t *Torrent, addr string) {
 614         if _, ok := t.halfOpen[addr]; !ok {
 615                 panic("invariant broken")
 616         }
 617         delete(t.halfOpen, addr)
 618         t.openNewConns()
 619 }
 620
 621 // Performs initiator handshakes and returns a connection. Returns nil
 622 // *connection if no connection for valid reasons.
 623 func (cl *Client) handshakesConnection(ctx context.Context, nc net.Conn, t *Torrent, encryptHeader bool, remoteAddr IpPort, network string) (c *connection, err error) {
 624         c = cl.newConnection(nc, true, remoteAddr, network)
 625         c.headerEncrypted = encryptHeader
 626         ctx, cancel := context.WithTimeout(ctx, cl.config.HandshakesTimeout)
 627         defer cancel()
 628         dl, ok := ctx.Deadline()
 629         if !ok {
 630                 panic(ctx)
 631         }
 632         err = nc.SetDeadline(dl)
 633         if err != nil {
 634                 panic(err)
 635         }
 636         err = cl.initiateHandshakes(c, t)
 637         return
 638 }
 639
 640 // Returns nil connection and nil error if no connection could be established
 641 // for valid reasons.
 642 func (cl *Client) establishOutgoingConnEx(t *Torrent, addr IpPort, obfuscatedHeader bool) (*connection, error) {
 643         dialCtx, cancel := context.WithTimeout(context.Background(), func() time.Duration {
 644                 cl.rLock()
 645                 defer cl.rUnlock()
 646                 return t.dialTimeout()
 647         }())
 648         defer cancel()
 649         dr := cl.dialFirst(dialCtx, addr.String())
 650         nc := dr.Conn
 651         if nc == nil {
 652                 if dialCtx.Err() != nil {
 653                         return nil, xerrors.Errorf("dialing: %w", dialCtx.Err())
 654                 }
 655                 return nil, errors.New("dial failed")
 656         }
 657         c, err := cl.handshakesConnection(context.Background(), nc, t, obfuscatedHeader, addr, dr.Network)
 658         if err != nil {
 659                 nc.Close()
 660         }
 661         return c, err
 662 }
 663
 664 // Returns nil connection and nil error if no connection could be established
 665 // for valid reasons.
 666 func (cl *Client) establishOutgoingConn(t *Torrent, addr IpPort) (c *connection, err error) {
 667         torrent.Add("establish outgoing connection", 1)
 668         obfuscatedHeaderFirst := cl.config.HeaderObfuscationPolicy.Preferred
 669         c, err = cl.establishOutgoingConnEx(t, addr, obfuscatedHeaderFirst)
 670         if err == nil {
 671                 torrent.Add("initiated conn with preferred header obfuscation", 1)
 672                 return
 673         }
 674         //cl.logger.Printf("error establishing connection to %s (obfuscatedHeader=%t): %v", addr, obfuscatedHeaderFirst, err)
 675         if cl.config.HeaderObfuscationPolicy.RequirePreferred {
 676                 // We should have just tried with the preferred header obfuscation. If it was required,
 677                 // there's nothing else to try.
 678                 return
 679         }
 680         // Try again with encryption if we didn't earlier, or without if we did.
 681         c, err = cl.establishOutgoingConnEx(t, addr, !obfuscatedHeaderFirst)
 682         if err == nil {
 683                 torrent.Add("initiated conn with fallback header obfuscation", 1)
 684         }
 685         //cl.logger.Printf("error establishing fallback connection to %v: %v", addr, err)
 686         return
 687 }
 688
 689 // Called to dial out and run a connection. The addr we're given is already
 690 // considered half-open.
 691 func (cl *Client) outgoingConnection(t *Torrent, addr IpPort, ps peerSource) {
 692         cl.dialRateLimiter.Wait(context.Background())
 693         c, err := cl.establishOutgoingConn(t, addr)
 694         cl.lock()
 695         defer cl.unlock()
 696         // Don't release lock between here and addConnection, unless it's for
 697         // failure.
 698         cl.noLongerHalfOpen(t, addr.String())
 699         if err != nil {
 700                 if cl.config.Debug {
 701                         cl.logger.Printf("error establishing outgoing connection to %v: %v", addr, err)
 702                 }
 703                 return
 704         }
 705         defer c.Close()
 706         c.Discovery = ps
 707         cl.runHandshookConn(c, t)
 708 }
 709
 710 // The port number for incoming peer connections. 0 if the client isn't
 711 // listening.
 712 func (cl *Client) incomingPeerPort() int {
 713         return cl.LocalPort()
 714 }
 715
 716 func (cl *Client) initiateHandshakes(c *connection, t *Torrent) error {
 717         if c.headerEncrypted {
 718                 var rw io.ReadWriter
 719                 var err error
 720                 rw, c.cryptoMethod, err = mse.InitiateHandshake(
 721                         struct {
 722                                 io.Reader
 723                                 io.Writer
 724                         }{c.r, c.w},
 725                         t.infoHash[:],
 726                         nil,
 727                         cl.config.CryptoProvides,
 728                 )
 729                 c.setRW(rw)
 730                 if err != nil {
 731                         return xerrors.Errorf("header obfuscation handshake: %w", err)
 732                 }
 733         }
 734         ih, err := cl.connBtHandshake(c, &t.infoHash)
 735         if err != nil {
 736                 return xerrors.Errorf("bittorrent protocol handshake: %w", err)
 737         }
 738         if ih != t.infoHash {
 739                 return errors.New("bittorrent protocol handshake: peer infohash didn't match")
 740         }
 741         return nil
 742 }
 743
 744 // Calls f with any secret keys.
 745 func (cl *Client) forSkeys(f func([]byte) bool) {
 746         cl.lock()
 747         defer cl.unlock()
 748         if false { // Emulate the bug from #114
 749                 var firstIh InfoHash
 750                 for ih := range cl.torrents {
 751                         firstIh = ih
 752                         break
 753                 }
 754                 for range cl.torrents {
 755                         if !f(firstIh[:]) {
 756                                 break
 757                         }
 758                 }
 759                 return
 760         }
 761         for ih := range cl.torrents {
 762                 if !f(ih[:]) {
 763                         break
 764                 }
 765         }
 766 }
 767
 768 // Do encryption and bittorrent handshakes as receiver.
 769 func (cl *Client) receiveHandshakes(c *connection) (t *Torrent, err error) {
 770         defer perf.ScopeTimerErr(&err)()
 771         var rw io.ReadWriter
 772         rw, c.headerEncrypted, c.cryptoMethod, err = handleEncryption(c.rw(), cl.forSkeys, cl.config.HeaderObfuscationPolicy, cl.config.CryptoSelector)
 773         c.setRW(rw)
 774         if err == nil || err == mse.ErrNoSecretKeyMatch {
 775                 if c.headerEncrypted {
 776                         torrent.Add("handshakes received encrypted", 1)
 777                 } else {
 778                         torrent.Add("handshakes received unencrypted", 1)
 779                 }
 780         } else {
 781                 torrent.Add("handshakes received with error while handling encryption", 1)
 782         }
 783         if err != nil {
 784                 if err == mse.ErrNoSecretKeyMatch {
 785                         err = nil
 786                 }
 787                 return
 788         }
 789         if cl.config.HeaderObfuscationPolicy.RequirePreferred && c.headerEncrypted != cl.config.HeaderObfuscationPolicy.Preferred {
 790                 err = errors.New("connection not have required header obfuscation")
 791                 return
 792         }
 793         ih, err := cl.connBtHandshake(c, nil)
 794         if err != nil {
 795                 err = xerrors.Errorf("during bt handshake: %w", err)
 796                 return
 797         }
 798         cl.lock()
 799         t = cl.torrents[ih]
 800         cl.unlock()
 801         return
 802 }
 803
 804 func (cl *Client) connBtHandshake(c *connection, ih *metainfo.Hash) (ret metainfo.Hash, err error) {
 805         res, err := pp.Handshake(c.rw(), ih, cl.peerID, cl.extensionBytes)
 806         if err != nil {
 807                 return
 808         }
 809         ret = res.Hash
 810         c.PeerExtensionBytes = res.PeerExtensionBits
 811         c.PeerID = res.PeerID
 812         c.completedHandshake = time.Now()
 813         return
 814 }
 815
 816 func (cl *Client) runReceivedConn(c *connection) {
 817         err := c.conn.SetDeadline(time.Now().Add(cl.config.HandshakesTimeout))
 818         if err != nil {
 819                 panic(err)
 820         }
 821         t, err := cl.receiveHandshakes(c)
 822         if err != nil {
 823                 log.Fmsg(
 824                         "error receiving handshakes: %s", err,
 825                 ).AddValue(
 826                         debugLogValue,
 827                 ).Add(
 828                         "network", c.network,
 829                 ).Log(cl.logger)
 830                 torrent.Add("error receiving handshake", 1)
 831                 cl.lock()
 832                 cl.onBadAccept(c.remoteAddr)
 833                 cl.unlock()
 834                 return
 835         }
 836         if t == nil {
 837                 torrent.Add("received handshake for unloaded torrent", 1)
 838                 cl.lock()
 839                 cl.onBadAccept(c.remoteAddr)
 840                 cl.unlock()
 841                 return
 842         }
 843         torrent.Add("received handshake for loaded torrent", 1)
 844         cl.lock()
 845         defer cl.unlock()
 846         cl.runHandshookConn(c, t)
 847 }
 848
 849 func (cl *Client) runHandshookConn(c *connection, t *Torrent) {
 850         c.setTorrent(t)
 851         if c.PeerID == cl.peerID {
 852                 if c.outgoing {
 853                         connsToSelf.Add(1)
 854                         addr := c.conn.RemoteAddr().String()
 855                         cl.dopplegangerAddrs[addr] = struct{}{}
 856                 } else {
 857                         // Because the remote address is not necessarily the same as its
 858                         // client's torrent listen address, we won't record the remote address
 859                         // as a doppleganger. Instead, the initiator can record *us* as the
 860                         // doppleganger.
 861                 }
 862                 return
 863         }
 864         c.conn.SetWriteDeadline(time.Time{})
 865         c.r = deadlineReader{c.conn, c.r}
 866         completedHandshakeConnectionFlags.Add(c.connectionFlags(), 1)
 867         if connIsIpv6(c.conn) {
 868                 torrent.Add("completed handshake over ipv6", 1)
 869         }
 870         if err := t.addConnection(c); err != nil {
 871                 log.Fmsg("error adding connection: %s", err).AddValues(c, debugLogValue).Log(t.logger)
 872                 return
 873         }
 874         defer t.dropConnection(c)
 875         go c.writer(time.Minute)
 876         cl.sendInitialMessages(c, t)
 877         err := c.mainReadLoop()
 878         if err != nil && cl.config.Debug {
 879                 cl.logger.Printf("error during connection main read loop: %s", err)
 880         }
 881 }
 882
 883 // See the order given in Transmission's tr_peerMsgsNew.
 884 func (cl *Client) sendInitialMessages(conn *connection, torrent *Torrent) {
 885         if conn.PeerExtensionBytes.SupportsExtended() && cl.extensionBytes.SupportsExtended() {
 886                 conn.Post(pp.Message{
 887                         Type:       pp.Extended,
 888                         ExtendedID: pp.HandshakeExtendedID,
 889                         ExtendedPayload: func() []byte {
 890                                 msg := pp.ExtendedHandshakeMessage{
 891                                         M: map[pp.ExtensionName]pp.ExtensionNumber{
 892                                                 pp.ExtensionNameMetadata: metadataExtendedId,
 893                                         },
 894                                         V:            cl.config.ExtendedHandshakeClientVersion,
 895                                         Reqq:         64, // TODO: Really?
 896                                         YourIp:       pp.CompactIp(conn.remoteAddr.IP),
 897                                         Encryption:   cl.config.HeaderObfuscationPolicy.Preferred || !cl.config.HeaderObfuscationPolicy.RequirePreferred,
 898                                         Port:         cl.incomingPeerPort(),
 899                                         MetadataSize: torrent.metadataSize(),
 900                                         // TODO: We can figured these out specific to the socket
 901                                         // used.
 902                                         Ipv4: pp.CompactIp(cl.config.PublicIp4.To4()),
 903                                         Ipv6: cl.config.PublicIp6.To16(),
 904                                 }
 905                                 if !cl.config.DisablePEX {
 906                                         msg.M[pp.ExtensionNamePex] = pexExtendedId
 907                                 }
 908                                 return bencode.MustMarshal(msg)
 909                         }(),
 910                 })
 911         }
 912         func() {
 913                 if conn.fastEnabled() {
 914                         if torrent.haveAllPieces() {
 915                                 conn.Post(pp.Message{Type: pp.HaveAll})
 916                                 conn.sentHaves.AddRange(0, bitmap.BitIndex(conn.t.NumPieces()))
 917                                 return
 918                         } else if !torrent.haveAnyPieces() {
 919                                 conn.Post(pp.Message{Type: pp.HaveNone})
 920                                 conn.sentHaves.Clear()
 921                                 return
 922                         }
 923                 }
 924                 conn.PostBitfield()
 925         }()
 926         if conn.PeerExtensionBytes.SupportsDHT() && cl.extensionBytes.SupportsDHT() && cl.haveDhtServer() {
 927                 conn.Post(pp.Message{
 928                         Type: pp.Port,
 929                         Port: cl.dhtPort(),
 930                 })
 931         }
 932 }
 933
 934 func (cl *Client) dhtPort() (ret uint16) {
 935         cl.eachDhtServer(func(s *dht.Server) {
 936                 ret = uint16(missinggo.AddrPort(s.Addr()))
 937         })
 938         return
 939 }
 940
 941 func (cl *Client) haveDhtServer() (ret bool) {
 942         cl.eachDhtServer(func(_ *dht.Server) {
 943                 ret = true
 944         })
 945         return
 946 }
 947
 948 // Process incoming ut_metadata message.
 949 func (cl *Client) gotMetadataExtensionMsg(payload []byte, t *Torrent, c *connection) error {
 950         var d map[string]int
 951         err := bencode.Unmarshal(payload, &d)
 952         if _, ok := err.(bencode.ErrUnusedTrailingBytes); ok {
 953         } else if err != nil {
 954                 return fmt.Errorf("error unmarshalling bencode: %s", err)
 955         }
 956         msgType, ok := d["msg_type"]
 957         if !ok {
 958                 return errors.New("missing msg_type field")
 959         }
 960         piece := d["piece"]
 961         switch msgType {
 962         case pp.DataMetadataExtensionMsgType:
 963                 c.allStats(add(1, func(cs *ConnStats) *Count { return &cs.MetadataChunksRead }))
 964                 if !c.requestedMetadataPiece(piece) {
 965                         return fmt.Errorf("got unexpected piece %d", piece)
 966                 }
 967                 c.metadataRequests[piece] = false
 968                 begin := len(payload) - metadataPieceSize(d["total_size"], piece)
 969                 if begin < 0 || begin >= len(payload) {
 970                         return fmt.Errorf("data has bad offset in payload: %d", begin)
 971                 }
 972                 t.saveMetadataPiece(piece, payload[begin:])
 973                 c.lastUsefulChunkReceived = time.Now()
 974                 return t.maybeCompleteMetadata()
 975         case pp.RequestMetadataExtensionMsgType:
 976                 if !t.haveMetadataPiece(piece) {
 977                         c.Post(t.newMetadataExtensionMessage(c, pp.RejectMetadataExtensionMsgType, d["piece"], nil))
 978                         return nil
 979                 }
 980                 start := (1 << 14) * piece
 981                 c.Post(t.newMetadataExtensionMessage(c, pp.DataMetadataExtensionMsgType, piece, t.metadataBytes[start:start+t.metadataPieceSize(piece)]))
 982                 return nil
 983         case pp.RejectMetadataExtensionMsgType:
 984                 return nil
 985         default:
 986                 return errors.New("unknown msg_type value")
 987         }
 988 }
 989
 990 func (cl *Client) badPeerIPPort(ip net.IP, port int) bool {
 991         if port == 0 {
 992                 return true
 993         }
 994         if cl.dopplegangerAddr(net.JoinHostPort(ip.String(), strconv.FormatInt(int64(port), 10))) {
 995                 return true
 996         }
 997         if _, ok := cl.ipBlockRange(ip); ok {
 998                 return true
 999         }
1000         if _, ok := cl.badPeerIPs[ip.String()]; ok {
1001                 return true
1002         }
1003         return false
1004 }
1005
1006 // Return a Torrent ready for insertion into a Client.
1007 func (cl *Client) newTorrent(ih metainfo.Hash, specStorage storage.ClientImpl) (t *Torrent) {
1008         // use provided storage, if provided
1009         storageClient := cl.defaultStorage
1010         if specStorage != nil {
1011                 storageClient = storage.NewClient(specStorage)
1012         }
1013
1014         t = &Torrent{
1015                 cl:       cl,
1016                 infoHash: ih,
1017                 peers: prioritizedPeers{
1018                         om: btree.New(32),
1019                         getPrio: func(p Peer) peerPriority {
1020                                 return bep40PriorityIgnoreError(cl.publicAddr(p.IP), p.addr())
1021                         },
1022                 },
1023                 conns: make(map[*connection]struct{}, 2*cl.config.EstablishedConnsPerTorrent),
1024
1025                 halfOpen:          make(map[string]Peer),
1026                 pieceStateChanges: pubsub.NewPubSub(),
1027
1028                 storageOpener:       storageClient,
1029                 maxEstablishedConns: cl.config.EstablishedConnsPerTorrent,
1030
1031                 networkingEnabled: true,
1032                 requestStrategy:   2,
1033                 metadataChanged: sync.Cond{
1034                         L: cl.locker(),
1035                 },
1036                 duplicateRequestTimeout: 1 * time.Second,
1037         }
1038         t.logger = cl.logger.WithValues(t)
1039         t.setChunkSize(defaultChunkSize)
1040         return
1041 }
1042
1043 // A file-like handle to some torrent data resource.
1044 type Handle interface {
1045         io.Reader
1046         io.Seeker
1047         io.Closer
1048         io.ReaderAt
1049 }
1050
1051 func (cl *Client) AddTorrentInfoHash(infoHash metainfo.Hash) (t *Torrent, new bool) {
1052         return cl.AddTorrentInfoHashWithStorage(infoHash, nil)
1053 }
1054
1055 // Adds a torrent by InfoHash with a custom Storage implementation.
1056 // If the torrent already exists then this Storage is ignored and the
1057 // existing torrent returned with `new` set to `false`
1058 func (cl *Client) AddTorrentInfoHashWithStorage(infoHash metainfo.Hash, specStorage storage.ClientImpl) (t *Torrent, new bool) {
1059         cl.lock()
1060         defer cl.unlock()
1061         t, ok := cl.torrents[infoHash]
1062         if ok {
1063                 return
1064         }
1065         new = true
1066
1067         t = cl.newTorrent(infoHash, specStorage)
1068         cl.eachDhtServer(func(s *dht.Server) {
1069                 go t.dhtAnnouncer(s)
1070         })
1071         cl.torrents[infoHash] = t
1072         cl.clearAcceptLimits()
1073         t.updateWantPeersEvent()
1074         // Tickle Client.waitAccept, new torrent may want conns.
1075         cl.event.Broadcast()
1076         return
1077 }
1078
1079 // Add or merge a torrent spec. If the torrent is already present, the
1080 // trackers will be merged with the existing ones. If the Info isn't yet
1081 // known, it will be set. The display name is replaced if the new spec
1082 // provides one. Returns new if the torrent wasn't already in the client.
1083 // Note that any `Storage` defined on the spec will be ignored if the
1084 // torrent is already present (i.e. `new` return value is `true`)
1085 func (cl *Client) AddTorrentSpec(spec *TorrentSpec) (t *Torrent, new bool, err error) {
1086         t, new = cl.AddTorrentInfoHashWithStorage(spec.InfoHash, spec.Storage)
1087         if spec.DisplayName != "" {
1088                 t.SetDisplayName(spec.DisplayName)
1089         }
1090         if spec.InfoBytes != nil {
1091                 err = t.SetInfoBytes(spec.InfoBytes)
1092                 if err != nil {
1093                         return
1094                 }
1095         }
1096         cl.lock()
1097         defer cl.unlock()
1098         if spec.ChunkSize != 0 {
1099                 t.setChunkSize(pp.Integer(spec.ChunkSize))
1100         }
1101         t.addTrackers(spec.Trackers)
1102         t.maybeNewConns()
1103         return
1104 }
1105
1106 func (cl *Client) dropTorrent(infoHash metainfo.Hash) (err error) {
1107         t, ok := cl.torrents[infoHash]
1108         if !ok {
1109                 err = fmt.Errorf("no such torrent")
1110                 return
1111         }
1112         err = t.close()
1113         if err != nil {
1114                 panic(err)
1115         }
1116         delete(cl.torrents, infoHash)
1117         return
1118 }
1119
1120 func (cl *Client) allTorrentsCompleted() bool {
1121         for _, t := range cl.torrents {
1122                 if !t.haveInfo() {
1123                         return false
1124                 }
1125                 if !t.haveAllPieces() {
1126                         return false
1127                 }
1128         }
1129         return true
1130 }
1131
1132 // Returns true when all torrents are completely downloaded and false if the
1133 // client is stopped before that.
1134 func (cl *Client) WaitAll() bool {
1135         cl.lock()
1136         defer cl.unlock()
1137         for !cl.allTorrentsCompleted() {
1138                 if cl.closed.IsSet() {
1139                         return false
1140                 }
1141                 cl.event.Wait()
1142         }
1143         return true
1144 }
1145
1146 // Returns handles to all the torrents loaded in the Client.
1147 func (cl *Client) Torrents() []*Torrent {
1148         cl.lock()
1149         defer cl.unlock()
1150         return cl.torrentsAsSlice()
1151 }
1152
1153 func (cl *Client) torrentsAsSlice() (ret []*Torrent) {
1154         for _, t := range cl.torrents {
1155                 ret = append(ret, t)
1156         }
1157         return
1158 }
1159
1160 func (cl *Client) AddMagnet(uri string) (T *Torrent, err error) {
1161         spec, err := TorrentSpecFromMagnetURI(uri)
1162         if err != nil {
1163                 return
1164         }
1165         T, _, err = cl.AddTorrentSpec(spec)
1166         return
1167 }
1168
1169 func (cl *Client) AddTorrent(mi *metainfo.MetaInfo) (T *Torrent, err error) {
1170         T, _, err = cl.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
1171         var ss []string
1172         slices.MakeInto(&ss, mi.Nodes)
1173         cl.AddDHTNodes(ss)
1174         return
1175 }
1176
1177 func (cl *Client) AddTorrentFromFile(filename string) (T *Torrent, err error) {
1178         mi, err := metainfo.LoadFromFile(filename)
1179         if err != nil {
1180                 return
1181         }
1182         return cl.AddTorrent(mi)
1183 }
1184
1185 func (cl *Client) DhtServers() []*dht.Server {
1186         return cl.dhtServers
1187 }
1188
1189 func (cl *Client) AddDHTNodes(nodes []string) {
1190         for _, n := range nodes {
1191                 hmp := missinggo.SplitHostMaybePort(n)
1192                 ip := net.ParseIP(hmp.Host)
1193                 if ip == nil {
1194                         cl.logger.Printf("won't add DHT node with bad IP: %q", hmp.Host)
1195                         continue
1196                 }
1197                 ni := krpc.NodeInfo{
1198                         Addr: krpc.NodeAddr{
1199                                 IP:   ip,
1200                                 Port: hmp.Port,
1201                         },
1202                 }
1203                 cl.eachDhtServer(func(s *dht.Server) {
1204                         s.AddNode(ni)
1205                 })
1206         }
1207 }
1208
1209 func (cl *Client) banPeerIP(ip net.IP) {
1210         if cl.badPeerIPs == nil {
1211                 cl.badPeerIPs = make(map[string]struct{})
1212         }
1213         cl.badPeerIPs[ip.String()] = struct{}{}
1214 }
1215
1216 func (cl *Client) newConnection(nc net.Conn, outgoing bool, remoteAddr IpPort, network string) (c *connection) {
1217         c = &connection{
1218                 conn:            nc,
1219                 outgoing:        outgoing,
1220                 Choked:          true,
1221                 PeerChoked:      true,
1222                 PeerMaxRequests: 250,
1223                 writeBuffer:     new(bytes.Buffer),
1224                 remoteAddr:      remoteAddr,
1225                 network:         network,
1226         }
1227         c.writerCond.L = cl.locker()
1228         c.setRW(connStatsReadWriter{nc, c})
1229         c.r = &rateLimitedReader{
1230                 l: cl.config.DownloadRateLimiter,
1231                 r: c.r,
1232         }
1233         return
1234 }
1235
1236 func (cl *Client) onDHTAnnouncePeer(ih metainfo.Hash, p dht.Peer) {
1237         cl.lock()
1238         defer cl.unlock()
1239         t := cl.torrent(ih)
1240         if t == nil {
1241                 return
1242         }
1243         t.addPeers([]Peer{{
1244                 IP:     p.IP,
1245                 Port:   p.Port,
1246                 Source: peerSourceDHTAnnouncePeer,
1247         }})
1248 }
1249
1250 func firstNotNil(ips ...net.IP) net.IP {
1251         for _, ip := range ips {
1252                 if ip != nil {
1253                         return ip
1254                 }
1255         }
1256         return nil
1257 }
1258
1259 func (cl *Client) eachListener(f func(socket) bool) {
1260         for _, s := range cl.conns {
1261                 if !f(s) {
1262                         break
1263                 }
1264         }
1265 }
1266
1267 func (cl *Client) findListener(f func(net.Listener) bool) (ret net.Listener) {
1268         cl.eachListener(func(l socket) bool {
1269                 ret = l
1270                 return !f(l)
1271         })
1272         return
1273 }
1274
1275 func (cl *Client) publicIp(peer net.IP) net.IP {
1276         // TODO: Use BEP 10 to determine how peers are seeing us.
1277         if peer.To4() != nil {
1278                 return firstNotNil(
1279                         cl.config.PublicIp4,
1280                         cl.findListenerIp(func(ip net.IP) bool { return ip.To4() != nil }),
1281                 )
1282         } else {
1283                 return firstNotNil(
1284                         cl.config.PublicIp6,
1285                         cl.findListenerIp(func(ip net.IP) bool { return ip.To4() == nil }),
1286                 )
1287         }
1288 }
1289
1290 func (cl *Client) findListenerIp(f func(net.IP) bool) net.IP {
1291         return missinggo.AddrIP(cl.findListener(func(l net.Listener) bool {
1292                 return f(missinggo.AddrIP(l.Addr()))
1293         }).Addr())
1294 }
1295
1296 // Our IP as a peer should see it.
1297 func (cl *Client) publicAddr(peer net.IP) IpPort {
1298         return IpPort{cl.publicIp(peer), uint16(cl.incomingPeerPort())}
1299 }
1300
1301 func (cl *Client) ListenAddrs() (ret []net.Addr) {
1302         cl.lock()
1303         defer cl.unlock()
1304         cl.eachListener(func(l socket) bool {
1305                 ret = append(ret, l.Addr())
1306                 return true
1307         })
1308         return
1309 }
1310
1311 func (cl *Client) onBadAccept(addr IpPort) {
1312         ip := maskIpForAcceptLimiting(addr.IP)
1313         if cl.acceptLimiter == nil {
1314                 cl.acceptLimiter = make(map[ipStr]int)
1315         }
1316         cl.acceptLimiter[ipStr(ip.String())]++
1317 }
1318
1319 func maskIpForAcceptLimiting(ip net.IP) net.IP {
1320         if ip4 := ip.To4(); ip4 != nil {
1321                 return ip4.Mask(net.CIDRMask(24, 32))
1322         }
1323         return ip
1324 }
1325
1326 func (cl *Client) clearAcceptLimits() {
1327         cl.acceptLimiter = nil
1328 }
1329
1330 func (cl *Client) acceptLimitClearer() {
1331         for {
1332                 select {
1333                 case <-cl.closed.LockedChan(cl.locker()):
1334                         return
1335                 case <-time.After(15 * time.Minute):
1336                         cl.lock()
1337                         cl.clearAcceptLimits()
1338                         cl.unlock()
1339                 }
1340         }
1341 }
1342
1343 func (cl *Client) rateLimitAccept(ip net.IP) bool {
1344         if cl.config.DisableAcceptRateLimiting {
1345                 return false
1346         }
1347         return cl.acceptLimiter[ipStr(maskIpForAcceptLimiting(ip).String())] > 0
1348 }
1349
1350 func (cl *Client) rLock() {
1351         cl._mu.RLock()
1352 }
1353
1354 func (cl *Client) rUnlock() {
1355         cl._mu.RUnlock()
1356 }
1357
1358 func (cl *Client) lock() {
1359         cl._mu.Lock()
1360 }
1361
1362 func (cl *Client) unlock() {
1363         cl._mu.Unlock()
1364 }
1365
1366 func (cl *Client) locker() sync.Locker {
1367         return clientLocker{cl}
1368 }
1369
1370 type clientLocker struct {
1371         *Client
1372 }
1373
1374 func (cl clientLocker) Lock() {
1375         cl.lock()
1376 }
1377
1378 func (cl clientLocker) Unlock() {
1379         cl.unlock()
1380 }