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         "strconv"
  14         "strings"
  15         "time"
  16
  17         "github.com/anacrolix/dht/v2"
  18         "github.com/anacrolix/dht/v2/krpc"
  19         "github.com/anacrolix/log"
  20         "github.com/anacrolix/missinggo/bitmap"
  21         "github.com/anacrolix/missinggo/perf"
  22         "github.com/anacrolix/missinggo/pproffd"
  23         "github.com/anacrolix/missinggo/pubsub"
  24         "github.com/anacrolix/missinggo/slices"
  25         "github.com/anacrolix/missinggo/v2"
  26         "github.com/anacrolix/missinggo/v2/conntrack"
  27         "github.com/anacrolix/sync"
  28         "github.com/davecgh/go-spew/spew"
  29         "github.com/dustin/go-humanize"
  30         "github.com/google/btree"
  31         "golang.org/x/time/rate"
  32         "golang.org/x/xerrors"
  33
  34         "github.com/anacrolix/torrent/bencode"
  35         "github.com/anacrolix/torrent/iplist"
  36         "github.com/anacrolix/torrent/metainfo"
  37         "github.com/anacrolix/torrent/mse"
  38         pp "github.com/anacrolix/torrent/peer_protocol"
  39         "github.com/anacrolix/torrent/storage"
  40 )
  41
  42 // Clients contain zero or more Torrents. A Client manages a blocklist, the
  43 // TCP/UDP protocol ports, and DHT as desired.
  44 type Client struct {
  45         // An aggregate of stats over all connections. First in struct to ensure
  46         // 64-bit alignment of fields. See #262.
  47         stats ConnStats
  48
  49         _mu    lockWithDeferreds
  50         event  sync.Cond
  51         closed missinggo.Event
  52
  53         config *ClientConfig
  54         logger log.Logger
  55
  56         peerID         PeerID
  57         defaultStorage *storage.Client
  58         onClose        []func()
  59         dialers        []Dialer
  60         listeners      []Listener
  61         dhtServers     []DhtServer
  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 // Returns the port number for the first listener that has one. No longer assumes that all port
  94 // numbers are the same, due to support for custom listeners. Returns zero if no port number is
  95 // found.
  96 func (cl *Client) LocalPort() (port int) {
  97         cl.eachListener(func(l Listener) bool {
  98                 port = addrPortOrZero(l.Addr())
  99                 return port == 0
 100         })
 101         return
 102 }
 103
 104 func writeDhtServerStatus(w io.Writer, s DhtServer) {
 105         dhtStats := s.Stats()
 106         fmt.Fprintf(w, " ID: %x\n", s.ID())
 107         spew.Fdump(w, dhtStats)
 108 }
 109
 110 // Writes out a human readable status of the client, such as for writing to a
 111 // HTTP status page.
 112 func (cl *Client) WriteStatus(_w io.Writer) {
 113         cl.rLock()
 114         defer cl.rUnlock()
 115         w := bufio.NewWriter(_w)
 116         defer w.Flush()
 117         fmt.Fprintf(w, "Listen port: %d\n", cl.LocalPort())
 118         fmt.Fprintf(w, "Peer ID: %+q\n", cl.PeerID())
 119         fmt.Fprintf(w, "Announce key: %x\n", cl.announceKey())
 120         fmt.Fprintf(w, "Banned IPs: %d\n", len(cl.badPeerIPsLocked()))
 121         cl.eachDhtServer(func(s DhtServer) {
 122                 fmt.Fprintf(w, "%s DHT server at %s:\n", s.Addr().Network(), s.Addr().String())
 123                 writeDhtServerStatus(w, s)
 124         })
 125         spew.Fdump(w, &cl.stats)
 126         fmt.Fprintf(w, "# Torrents: %d\n", len(cl.torrentsAsSlice()))
 127         fmt.Fprintln(w)
 128         for _, t := range slices.Sort(cl.torrentsAsSlice(), func(l, r *Torrent) bool {
 129                 return l.InfoHash().AsString() < r.InfoHash().AsString()
 130         }).([]*Torrent) {
 131                 if t.name() == "" {
 132                         fmt.Fprint(w, "<unknown name>")
 133                 } else {
 134                         fmt.Fprint(w, t.name())
 135                 }
 136                 fmt.Fprint(w, "\n")
 137                 if t.info != nil {
 138                         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())))
 139                 } else {
 140                         w.WriteString("<missing metainfo>")
 141                 }
 142                 fmt.Fprint(w, "\n")
 143                 t.writeStatus(w)
 144                 fmt.Fprintln(w)
 145         }
 146 }
 147
 148 const debugLogValue = log.Debug
 149
 150 func (cl *Client) debugLogFilter(m log.Msg) bool {
 151         if cl.config.Debug {
 152                 return true
 153         }
 154         return !m.HasValue(debugLogValue)
 155 }
 156
 157 func (cl *Client) initLogger() {
 158         cl.logger = cl.config.Logger.WithValues(cl).WithFilter(cl.debugLogFilter)
 159 }
 160
 161 func (cl *Client) announceKey() int32 {
 162         return int32(binary.BigEndian.Uint32(cl.peerID[16:20]))
 163 }
 164
 165 func NewClient(cfg *ClientConfig) (cl *Client, err error) {
 166         if cfg == nil {
 167                 cfg = NewDefaultClientConfig()
 168                 cfg.ListenPort = 0
 169         }
 170         defer func() {
 171                 if err != nil {
 172                         cl = nil
 173                 }
 174         }()
 175         cl = &Client{
 176                 config:            cfg,
 177                 dopplegangerAddrs: make(map[string]struct{}),
 178                 torrents:          make(map[metainfo.Hash]*Torrent),
 179                 dialRateLimiter:   rate.NewLimiter(10, 10),
 180         }
 181         go cl.acceptLimitClearer()
 182         cl.initLogger()
 183         defer func() {
 184                 if err == nil {
 185                         return
 186                 }
 187                 cl.Close()
 188         }()
 189         cl.extensionBytes = defaultPeerExtensionBytes()
 190         cl.event.L = cl.locker()
 191         storageImpl := cfg.DefaultStorage
 192         if storageImpl == nil {
 193                 // We'd use mmap by default but HFS+ doesn't support sparse files.
 194                 storageImplCloser := storage.NewFile(cfg.DataDir)
 195                 cl.onClose = append(cl.onClose, func() {
 196                         if err := storageImplCloser.Close(); err != nil {
 197                                 cl.logger.Printf("error closing default storage: %s", err)
 198                         }
 199                 })
 200                 storageImpl = storageImplCloser
 201         }
 202         cl.defaultStorage = storage.NewClient(storageImpl)
 203         if cfg.IPBlocklist != nil {
 204                 cl.ipBlockList = cfg.IPBlocklist
 205         }
 206
 207         if cfg.PeerID != "" {
 208                 missinggo.CopyExact(&cl.peerID, cfg.PeerID)
 209         } else {
 210                 o := copy(cl.peerID[:], cfg.Bep20)
 211                 _, err = rand.Read(cl.peerID[o:])
 212                 if err != nil {
 213                         panic("error generating peer id")
 214                 }
 215         }
 216
 217         sockets, err := listenAll(cl.listenNetworks(), cl.config.ListenHost, cl.config.ListenPort, cl.firewallCallback)
 218         if err != nil {
 219                 return
 220         }
 221
 222         // Check for panics.
 223         cl.LocalPort()
 224
 225         for _, _s := range sockets {
 226                 s := _s // Go is fucking retarded.
 227                 cl.onClose = append(cl.onClose, func() { s.Close() })
 228                 if peerNetworkEnabled(parseNetworkString(s.Addr().Network()), cl.config) {
 229                         cl.dialers = append(cl.dialers, s)
 230                         cl.listeners = append(cl.listeners, s)
 231                         go cl.acceptConnections(s)
 232                 }
 233         }
 234
 235         go cl.forwardPort()
 236         if !cfg.NoDHT {
 237                 for _, s := range sockets {
 238                         if pc, ok := s.(net.PacketConn); ok {
 239                                 ds, err := cl.newAnacrolixDhtServer(pc)
 240                                 if err != nil {
 241                                         panic(err)
 242                                 }
 243                                 cl.dhtServers = append(cl.dhtServers, anacrolixDhtServerWrapper{ds})
 244                                 cl.onClose = append(cl.onClose, func() { ds.Close() })
 245                         }
 246                 }
 247         }
 248
 249         return
 250 }
 251
 252 func (cl *Client) AddDhtServer(d DhtServer) {
 253         cl.dhtServers = append(cl.dhtServers, d)
 254 }
 255
 256 // Adds a Dialer for outgoing connections. All Dialers are used when attempting to connect to a
 257 // given address for any Torrent.
 258 func (cl *Client) AddDialer(d Dialer) {
 259         cl.dialers = append(cl.dialers, d)
 260 }
 261
 262 // Registers a Listener, and starts Accepting on it. You must Close Listeners provided this way
 263 // yourself.
 264 func (cl *Client) AddListener(l Listener) {
 265         cl.listeners = append(cl.listeners, l)
 266         go cl.acceptConnections(l)
 267 }
 268
 269 func (cl *Client) firewallCallback(net.Addr) bool {
 270         cl.rLock()
 271         block := !cl.wantConns()
 272         cl.rUnlock()
 273         if block {
 274                 torrent.Add("connections firewalled", 1)
 275         } else {
 276                 torrent.Add("connections not firewalled", 1)
 277         }
 278         return block
 279 }
 280
 281 func (cl *Client) listenOnNetwork(n network) bool {
 282         if n.Ipv4 && cl.config.DisableIPv4 {
 283                 return false
 284         }
 285         if n.Ipv6 && cl.config.DisableIPv6 {
 286                 return false
 287         }
 288         if n.Tcp && cl.config.DisableTCP {
 289                 return false
 290         }
 291         if n.Udp && cl.config.DisableUTP && cl.config.NoDHT {
 292                 return false
 293         }
 294         return true
 295 }
 296
 297 func (cl *Client) listenNetworks() (ns []network) {
 298         for _, n := range allPeerNetworks {
 299                 if cl.listenOnNetwork(n) {
 300                         ns = append(ns, n)
 301                 }
 302         }
 303         return
 304 }
 305
 306 func (cl *Client) newAnacrolixDhtServer(conn net.PacketConn) (s *dht.Server, err error) {
 307         cfg := dht.ServerConfig{
 308                 IPBlocklist:    cl.ipBlockList,
 309                 Conn:           conn,
 310                 OnAnnouncePeer: cl.onDHTAnnouncePeer,
 311                 PublicIP: func() net.IP {
 312                         if connIsIpv6(conn) && cl.config.PublicIp6 != nil {
 313                                 return cl.config.PublicIp6
 314                         }
 315                         return cl.config.PublicIp4
 316                 }(),
 317                 StartingNodes:      cl.config.DhtStartingNodes,
 318                 ConnectionTracking: cl.config.ConnTracker,
 319                 OnQuery:            cl.config.DHTOnQuery,
 320                 Logger:             cl.logger.WithValues("dht", conn.LocalAddr().String()),
 321         }
 322         s, err = dht.NewServer(&cfg)
 323         if err == nil {
 324                 go func() {
 325                         ts, err := s.Bootstrap()
 326                         if err != nil {
 327                                 cl.logger.Printf("error bootstrapping dht: %s", err)
 328                         }
 329                         log.Fstr("%v completed bootstrap (%v)", s, ts).AddValues(s, ts).Log(cl.logger)
 330                 }()
 331         }
 332         return
 333 }
 334
 335 func (cl *Client) Closed() <-chan struct{} {
 336         cl.lock()
 337         defer cl.unlock()
 338         return cl.closed.C()
 339 }
 340
 341 func (cl *Client) eachDhtServer(f func(DhtServer)) {
 342         for _, ds := range cl.dhtServers {
 343                 f(ds)
 344         }
 345 }
 346
 347 // Stops the client. All connections to peers are closed and all activity will
 348 // come to a halt.
 349 func (cl *Client) Close() {
 350         cl.lock()
 351         defer cl.unlock()
 352         cl.closed.Set()
 353         for _, t := range cl.torrents {
 354                 t.close()
 355         }
 356         for i := range cl.onClose {
 357                 cl.onClose[len(cl.onClose)-1-i]()
 358         }
 359         cl.event.Broadcast()
 360 }
 361
 362 func (cl *Client) ipBlockRange(ip net.IP) (r iplist.Range, blocked bool) {
 363         if cl.ipBlockList == nil {
 364                 return
 365         }
 366         return cl.ipBlockList.Lookup(ip)
 367 }
 368
 369 func (cl *Client) ipIsBlocked(ip net.IP) bool {
 370         _, blocked := cl.ipBlockRange(ip)
 371         return blocked
 372 }
 373
 374 func (cl *Client) wantConns() bool {
 375         for _, t := range cl.torrents {
 376                 if t.wantConns() {
 377                         return true
 378                 }
 379         }
 380         return false
 381 }
 382
 383 func (cl *Client) waitAccept() {
 384         for {
 385                 if cl.closed.IsSet() {
 386                         return
 387                 }
 388                 if cl.wantConns() {
 389                         return
 390                 }
 391                 cl.event.Wait()
 392         }
 393 }
 394
 395 // TODO: Apply filters for non-standard networks, particularly rate-limiting.
 396 func (cl *Client) rejectAccepted(conn net.Conn) error {
 397         ra := conn.RemoteAddr()
 398         if rip := addrIpOrNil(ra); rip != nil {
 399                 if cl.config.DisableIPv4Peers && rip.To4() != nil {
 400                         return errors.New("ipv4 peers disabled")
 401                 }
 402                 if cl.config.DisableIPv4 && len(rip) == net.IPv4len {
 403                         return errors.New("ipv4 disabled")
 404
 405                 }
 406                 if cl.config.DisableIPv6 && len(rip) == net.IPv6len && rip.To4() == nil {
 407                         return errors.New("ipv6 disabled")
 408                 }
 409                 if cl.rateLimitAccept(rip) {
 410                         return errors.New("source IP accepted rate limited")
 411                 }
 412                 if cl.badPeerIPPort(rip, missinggo.AddrPort(ra)) {
 413                         return errors.New("bad source addr")
 414                 }
 415         }
 416         return nil
 417 }
 418
 419 func (cl *Client) acceptConnections(l net.Listener) {
 420         for {
 421                 conn, err := l.Accept()
 422                 torrent.Add("client listener accepts", 1)
 423                 conn = pproffd.WrapNetConn(conn)
 424                 cl.rLock()
 425                 closed := cl.closed.IsSet()
 426                 var reject error
 427                 if conn != nil {
 428                         reject = cl.rejectAccepted(conn)
 429                 }
 430                 cl.rUnlock()
 431                 if closed {
 432                         if conn != nil {
 433                                 conn.Close()
 434                         }
 435                         return
 436                 }
 437                 if err != nil {
 438                         log.Fmsg("error accepting connection: %s", err).AddValue(debugLogValue).Log(cl.logger)
 439                         continue
 440                 }
 441                 go func() {
 442                         if reject != nil {
 443                                 torrent.Add("rejected accepted connections", 1)
 444                                 log.Fmsg("rejecting accepted conn: %v", reject).AddValue(debugLogValue).Log(cl.logger)
 445                                 conn.Close()
 446                         } else {
 447                                 go cl.incomingConnection(conn)
 448                         }
 449                         log.Fmsg("accepted %q connection at %q from %q",
 450                                 l.Addr().Network(),
 451                                 conn.LocalAddr(),
 452                                 conn.RemoteAddr(),
 453                         ).AddValue(debugLogValue).Log(cl.logger)
 454                         torrent.Add(fmt.Sprintf("accepted conn remote IP len=%d", len(addrIpOrNil(conn.RemoteAddr()))), 1)
 455                         torrent.Add(fmt.Sprintf("accepted conn network=%s", conn.RemoteAddr().Network()), 1)
 456                         torrent.Add(fmt.Sprintf("accepted on %s listener", l.Addr().Network()), 1)
 457                 }()
 458         }
 459 }
 460
 461 func (cl *Client) incomingConnection(nc net.Conn) {
 462         defer nc.Close()
 463         if tc, ok := nc.(*net.TCPConn); ok {
 464                 tc.SetLinger(0)
 465         }
 466         c := cl.newConnection(nc, false, nc.RemoteAddr(), nc.RemoteAddr().Network())
 467         c.Discovery = PeerSourceIncoming
 468         cl.runReceivedConn(c)
 469 }
 470
 471 // Returns a handle to the given torrent, if it's present in the client.
 472 func (cl *Client) Torrent(ih metainfo.Hash) (t *Torrent, ok bool) {
 473         cl.lock()
 474         defer cl.unlock()
 475         t, ok = cl.torrents[ih]
 476         return
 477 }
 478
 479 func (cl *Client) torrent(ih metainfo.Hash) *Torrent {
 480         return cl.torrents[ih]
 481 }
 482
 483 type dialResult struct {
 484         Conn    net.Conn
 485         Network string
 486 }
 487
 488 func countDialResult(err error) {
 489         if err == nil {
 490                 torrent.Add("successful dials", 1)
 491         } else {
 492                 torrent.Add("unsuccessful dials", 1)
 493         }
 494 }
 495
 496 func reducedDialTimeout(minDialTimeout, max time.Duration, halfOpenLimit int, pendingPeers int) (ret time.Duration) {
 497         ret = max / time.Duration((pendingPeers+halfOpenLimit)/halfOpenLimit)
 498         if ret < minDialTimeout {
 499                 ret = minDialTimeout
 500         }
 501         return
 502 }
 503
 504 // Returns whether an address is known to connect to a client with our own ID.
 505 func (cl *Client) dopplegangerAddr(addr string) bool {
 506         _, ok := cl.dopplegangerAddrs[addr]
 507         return ok
 508 }
 509
 510 // Returns a connection over UTP or TCP, whichever is first to connect.
 511 func (cl *Client) dialFirst(ctx context.Context, addr string) (res dialResult) {
 512         {
 513                 t := perf.NewTimer(perf.CallerName(0))
 514                 defer func() {
 515                         if res.Conn == nil {
 516                                 t.Mark(fmt.Sprintf("returned no conn (context: %v)", ctx.Err()))
 517                         } else {
 518                                 t.Mark("returned conn over " + res.Network)
 519                         }
 520                 }()
 521         }
 522         ctx, cancel := context.WithCancel(ctx)
 523         // As soon as we return one connection, cancel the others.
 524         defer cancel()
 525         left := 0
 526         resCh := make(chan dialResult, left)
 527         func() {
 528                 cl.lock()
 529                 defer cl.unlock()
 530                 cl.eachDialer(func(s Dialer) bool {
 531                         func() {
 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                                                 s.LocalAddr().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 Dialer, addr string) net.Conn {
 572         network := s.LocalAddr().Network()
 573         cte := cl.config.ConnTracker.Wait(
 574                 ctx,
 575                 conntrack.Entry{network, s.LocalAddr().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 net.Addr, network string) (c *PeerConn, 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 net.Addr, obfuscatedHeader bool) (*PeerConn, 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 net.Addr) (c *PeerConn, 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 net.Addr, ps PeerSource, trusted bool) {
 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         c.trusted = trusted
 708         cl.runHandshookConn(c, t)
 709 }
 710
 711 // The port number for incoming peer connections. 0 if the client isn't listening.
 712 func (cl *Client) incomingPeerPort() int {
 713         return cl.LocalPort()
 714 }
 715
 716 func (cl *Client) initiateHandshakes(c *PeerConn, 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 *PeerConn) (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 *PeerConn, 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 *PeerConn) {
 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 on %v: %s", c, 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                 log.Fmsg("received handshake for unloaded torrent").AddValue(debugLogValue).Log(cl.logger)
 839                 cl.lock()
 840                 cl.onBadAccept(c.remoteAddr)
 841                 cl.unlock()
 842                 return
 843         }
 844         torrent.Add("received handshake for loaded torrent", 1)
 845         cl.lock()
 846         defer cl.unlock()
 847         cl.runHandshookConn(c, t)
 848 }
 849
 850 func (cl *Client) runHandshookConn(c *PeerConn, t *Torrent) {
 851         c.setTorrent(t)
 852         if c.PeerID == cl.peerID {
 853                 if c.outgoing {
 854                         connsToSelf.Add(1)
 855                         addr := c.conn.RemoteAddr().String()
 856                         cl.dopplegangerAddrs[addr] = struct{}{}
 857                 } else {
 858                         // Because the remote address is not necessarily the same as its client's torrent listen
 859                         // address, we won't record the remote address as a doppleganger. Instead, the initiator
 860                         // can record *us* as the 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 *PeerConn, 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(addrIpOrNil(conn.remoteAddr)),
 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 DhtServer) {
 936                 ret = uint16(missinggo.AddrPort(s.Addr()))
 937         })
 938         return
 939 }
 940
 941 func (cl *Client) haveDhtServer() (ret bool) {
 942         cl.eachDhtServer(func(_ DhtServer) {
 943                 ret = true
 944         })
 945         return
 946 }
 947
 948 // Process incoming ut_metadata message.
 949 func (cl *Client) gotMetadataExtensionMsg(payload []byte, t *Torrent, c *PeerConn) 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.logger.Printf("sending metadata piece %d", piece)
 982                 c.post(t.newMetadataExtensionMessage(c, pp.DataMetadataExtensionMsgType, piece, t.metadataBytes[start:start+t.metadataPieceSize(piece)]))
 983                 return nil
 984         case pp.RejectMetadataExtensionMsgType:
 985                 return nil
 986         default:
 987                 return errors.New("unknown msg_type value")
 988         }
 989 }
 990
 991 func (cl *Client) badPeerAddr(addr net.Addr) bool {
 992         if ipa, ok := tryIpPortFromNetAddr(addr); ok {
 993                 return cl.badPeerIPPort(ipa.IP, ipa.Port)
 994         }
 995         return false
 996 }
 997
 998 func (cl *Client) badPeerIPPort(ip net.IP, port int) bool {
 999         if port == 0 {
1000                 return true
1001         }
1002         if cl.dopplegangerAddr(net.JoinHostPort(ip.String(), strconv.FormatInt(int64(port), 10))) {
1003                 return true
1004         }
1005         if _, ok := cl.ipBlockRange(ip); ok {
1006                 return true
1007         }
1008         if _, ok := cl.badPeerIPs[ip.String()]; ok {
1009                 return true
1010         }
1011         return false
1012 }
1013
1014 // Return a Torrent ready for insertion into a Client.
1015 func (cl *Client) newTorrent(ih metainfo.Hash, specStorage storage.ClientImpl) (t *Torrent) {
1016         // use provided storage, if provided
1017         storageClient := cl.defaultStorage
1018         if specStorage != nil {
1019                 storageClient = storage.NewClient(specStorage)
1020         }
1021
1022         t = &Torrent{
1023                 cl:       cl,
1024                 infoHash: ih,
1025                 peers: prioritizedPeers{
1026                         om: btree.New(32),
1027                         getPrio: func(p Peer) peerPriority {
1028                                 return bep40PriorityIgnoreError(cl.publicAddr(addrIpOrNil(p.Addr)), p.addr())
1029                         },
1030                 },
1031                 conns: make(map[*PeerConn]struct{}, 2*cl.config.EstablishedConnsPerTorrent),
1032
1033                 halfOpen:          make(map[string]Peer),
1034                 pieceStateChanges: pubsub.NewPubSub(),
1035
1036                 storageOpener:       storageClient,
1037                 maxEstablishedConns: cl.config.EstablishedConnsPerTorrent,
1038
1039                 networkingEnabled: true,
1040                 metadataChanged: sync.Cond{
1041                         L: cl.locker(),
1042                 },
1043         }
1044         t.requestStrategy = cl.config.DefaultRequestStrategy(t.requestStrategyCallbacks(), &cl._mu)
1045         t.logger = cl.logger.WithValues(t).WithText(func(m log.Msg) string {
1046                 return fmt.Sprintf("%v: %s", t, m.Text())
1047         })
1048         t.setChunkSize(defaultChunkSize)
1049         return
1050 }
1051
1052 // A file-like handle to some torrent data resource.
1053 type Handle interface {
1054         io.Reader
1055         io.Seeker
1056         io.Closer
1057         io.ReaderAt
1058 }
1059
1060 func (cl *Client) AddTorrentInfoHash(infoHash metainfo.Hash) (t *Torrent, new bool) {
1061         return cl.AddTorrentInfoHashWithStorage(infoHash, nil)
1062 }
1063
1064 // Adds a torrent by InfoHash with a custom Storage implementation.
1065 // If the torrent already exists then this Storage is ignored and the
1066 // existing torrent returned with `new` set to `false`
1067 func (cl *Client) AddTorrentInfoHashWithStorage(infoHash metainfo.Hash, specStorage storage.ClientImpl) (t *Torrent, new bool) {
1068         cl.lock()
1069         defer cl.unlock()
1070         t, ok := cl.torrents[infoHash]
1071         if ok {
1072                 return
1073         }
1074         new = true
1075
1076         t = cl.newTorrent(infoHash, specStorage)
1077         cl.eachDhtServer(func(s DhtServer) {
1078                 go t.dhtAnnouncer(s)
1079         })
1080         cl.torrents[infoHash] = t
1081         cl.clearAcceptLimits()
1082         t.updateWantPeersEvent()
1083         // Tickle Client.waitAccept, new torrent may want conns.
1084         cl.event.Broadcast()
1085         return
1086 }
1087
1088 // Add or merge a torrent spec. If the torrent is already present, the
1089 // trackers will be merged with the existing ones. If the Info isn't yet
1090 // known, it will be set. The display name is replaced if the new spec
1091 // provides one. Returns new if the torrent wasn't already in the client.
1092 // Note that any `Storage` defined on the spec will be ignored if the
1093 // torrent is already present (i.e. `new` return value is `true`)
1094 func (cl *Client) AddTorrentSpec(spec *TorrentSpec) (t *Torrent, new bool, err error) {
1095         t, new = cl.AddTorrentInfoHashWithStorage(spec.InfoHash, spec.Storage)
1096         if spec.DisplayName != "" {
1097                 t.SetDisplayName(spec.DisplayName)
1098         }
1099         if spec.InfoBytes != nil {
1100                 err = t.SetInfoBytes(spec.InfoBytes)
1101                 if err != nil {
1102                         return
1103                 }
1104         }
1105         cl.lock()
1106         defer cl.unlock()
1107         if spec.ChunkSize != 0 {
1108                 t.setChunkSize(pp.Integer(spec.ChunkSize))
1109         }
1110         t.addTrackers(spec.Trackers)
1111         t.maybeNewConns()
1112         return
1113 }
1114
1115 func (cl *Client) dropTorrent(infoHash metainfo.Hash) (err error) {
1116         t, ok := cl.torrents[infoHash]
1117         if !ok {
1118                 err = fmt.Errorf("no such torrent")
1119                 return
1120         }
1121         err = t.close()
1122         if err != nil {
1123                 panic(err)
1124         }
1125         delete(cl.torrents, infoHash)
1126         return
1127 }
1128
1129 func (cl *Client) allTorrentsCompleted() bool {
1130         for _, t := range cl.torrents {
1131                 if !t.haveInfo() {
1132                         return false
1133                 }
1134                 if !t.haveAllPieces() {
1135                         return false
1136                 }
1137         }
1138         return true
1139 }
1140
1141 // Returns true when all torrents are completely downloaded and false if the
1142 // client is stopped before that.
1143 func (cl *Client) WaitAll() bool {
1144         cl.lock()
1145         defer cl.unlock()
1146         for !cl.allTorrentsCompleted() {
1147                 if cl.closed.IsSet() {
1148                         return false
1149                 }
1150                 cl.event.Wait()
1151         }
1152         return true
1153 }
1154
1155 // Returns handles to all the torrents loaded in the Client.
1156 func (cl *Client) Torrents() []*Torrent {
1157         cl.lock()
1158         defer cl.unlock()
1159         return cl.torrentsAsSlice()
1160 }
1161
1162 func (cl *Client) torrentsAsSlice() (ret []*Torrent) {
1163         for _, t := range cl.torrents {
1164                 ret = append(ret, t)
1165         }
1166         return
1167 }
1168
1169 func (cl *Client) AddMagnet(uri string) (T *Torrent, err error) {
1170         spec, err := TorrentSpecFromMagnetURI(uri)
1171         if err != nil {
1172                 return
1173         }
1174         T, _, err = cl.AddTorrentSpec(spec)
1175         return
1176 }
1177
1178 func (cl *Client) AddTorrent(mi *metainfo.MetaInfo) (T *Torrent, err error) {
1179         T, _, err = cl.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
1180         var ss []string
1181         slices.MakeInto(&ss, mi.Nodes)
1182         cl.AddDHTNodes(ss)
1183         return
1184 }
1185
1186 func (cl *Client) AddTorrentFromFile(filename string) (T *Torrent, err error) {
1187         mi, err := metainfo.LoadFromFile(filename)
1188         if err != nil {
1189                 return
1190         }
1191         return cl.AddTorrent(mi)
1192 }
1193
1194 func (cl *Client) DhtServers() []DhtServer {
1195         return cl.dhtServers
1196 }
1197
1198 func (cl *Client) AddDHTNodes(nodes []string) {
1199         for _, n := range nodes {
1200                 hmp := missinggo.SplitHostMaybePort(n)
1201                 ip := net.ParseIP(hmp.Host)
1202                 if ip == nil {
1203                         cl.logger.Printf("won't add DHT node with bad IP: %q", hmp.Host)
1204                         continue
1205                 }
1206                 ni := krpc.NodeInfo{
1207                         Addr: krpc.NodeAddr{
1208                                 IP:   ip,
1209                                 Port: hmp.Port,
1210                         },
1211                 }
1212                 cl.eachDhtServer(func(s DhtServer) {
1213                         s.AddNode(ni)
1214                 })
1215         }
1216 }
1217
1218 func (cl *Client) banPeerIP(ip net.IP) {
1219         cl.logger.Printf("banning ip %v", ip)
1220         if cl.badPeerIPs == nil {
1221                 cl.badPeerIPs = make(map[string]struct{})
1222         }
1223         cl.badPeerIPs[ip.String()] = struct{}{}
1224 }
1225
1226 func (cl *Client) newConnection(nc net.Conn, outgoing bool, remoteAddr net.Addr, network string) (c *PeerConn) {
1227         c = &PeerConn{
1228                 conn:            nc,
1229                 outgoing:        outgoing,
1230                 choking:         true,
1231                 peerChoking:     true,
1232                 PeerMaxRequests: 250,
1233                 writeBuffer:     new(bytes.Buffer),
1234                 remoteAddr:      remoteAddr,
1235                 network:         network,
1236         }
1237         c.logger = cl.logger.WithValues(c,
1238                 log.Debug, // I want messages to default to debug, and can set it here as it's only used by new code
1239         ).WithText(func(m log.Msg) string {
1240                 return fmt.Sprintf("%v: %s", c, m.Text())
1241         })
1242         c.writerCond.L = cl.locker()
1243         c.setRW(connStatsReadWriter{nc, c})
1244         c.r = &rateLimitedReader{
1245                 l: cl.config.DownloadRateLimiter,
1246                 r: c.r,
1247         }
1248         c.logger.Printf("initialized with remote %v over network %v (outgoing=%t)", remoteAddr, network, outgoing)
1249         return
1250 }
1251
1252 func (cl *Client) onDHTAnnouncePeer(ih metainfo.Hash, ip net.IP, port int, portOk bool) {
1253         cl.lock()
1254         defer cl.unlock()
1255         t := cl.torrent(ih)
1256         if t == nil {
1257                 return
1258         }
1259         t.addPeers([]Peer{{
1260                 Addr:   ipPortAddr{ip, port},
1261                 Source: PeerSourceDhtAnnouncePeer,
1262         }})
1263 }
1264
1265 func firstNotNil(ips ...net.IP) net.IP {
1266         for _, ip := range ips {
1267                 if ip != nil {
1268                         return ip
1269                 }
1270         }
1271         return nil
1272 }
1273
1274 func (cl *Client) eachDialer(f func(Dialer) bool) {
1275         for _, s := range cl.dialers {
1276                 if !f(s) {
1277                         break
1278                 }
1279         }
1280 }
1281
1282 func (cl *Client) eachListener(f func(Listener) bool) {
1283         for _, s := range cl.listeners {
1284                 if !f(s) {
1285                         break
1286                 }
1287         }
1288 }
1289
1290 func (cl *Client) findListener(f func(net.Listener) bool) (ret net.Listener) {
1291         cl.eachListener(func(l Listener) bool {
1292                 ret = l
1293                 return !f(l)
1294         })
1295         return
1296 }
1297
1298 func (cl *Client) publicIp(peer net.IP) net.IP {
1299         // TODO: Use BEP 10 to determine how peers are seeing us.
1300         if peer.To4() != nil {
1301                 return firstNotNil(
1302                         cl.config.PublicIp4,
1303                         cl.findListenerIp(func(ip net.IP) bool { return ip.To4() != nil }),
1304                 )
1305         }
1306
1307         return firstNotNil(
1308                 cl.config.PublicIp6,
1309                 cl.findListenerIp(func(ip net.IP) bool { return ip.To4() == nil }),
1310         )
1311 }
1312
1313 func (cl *Client) findListenerIp(f func(net.IP) bool) net.IP {
1314         return addrIpOrNil(
1315                 cl.findListener(
1316                         func(l net.Listener) bool {
1317                                 return f(addrIpOrNil(l.Addr()))
1318                         },
1319                 ).Addr(),
1320         )
1321 }
1322
1323 // Our IP as a peer should see it.
1324 func (cl *Client) publicAddr(peer net.IP) IpPort {
1325         return IpPort{IP: cl.publicIp(peer), Port: uint16(cl.incomingPeerPort())}
1326 }
1327
1328 // ListenAddrs addresses currently being listened to.
1329 func (cl *Client) ListenAddrs() (ret []net.Addr) {
1330         cl.lock()
1331         defer cl.unlock()
1332         cl.eachListener(func(l Listener) bool {
1333                 ret = append(ret, l.Addr())
1334                 return true
1335         })
1336         return
1337 }
1338
1339 func (cl *Client) onBadAccept(addr net.Addr) {
1340         ipa, ok := tryIpPortFromNetAddr(addr)
1341         if !ok {
1342                 return
1343         }
1344         ip := maskIpForAcceptLimiting(ipa.IP)
1345         if cl.acceptLimiter == nil {
1346                 cl.acceptLimiter = make(map[ipStr]int)
1347         }
1348         cl.acceptLimiter[ipStr(ip.String())]++
1349 }
1350
1351 func maskIpForAcceptLimiting(ip net.IP) net.IP {
1352         if ip4 := ip.To4(); ip4 != nil {
1353                 return ip4.Mask(net.CIDRMask(24, 32))
1354         }
1355         return ip
1356 }
1357
1358 func (cl *Client) clearAcceptLimits() {
1359         cl.acceptLimiter = nil
1360 }
1361
1362 func (cl *Client) acceptLimitClearer() {
1363         for {
1364                 select {
1365                 case <-cl.closed.LockedChan(cl.locker()):
1366                         return
1367                 case <-time.After(15 * time.Minute):
1368                         cl.lock()
1369                         cl.clearAcceptLimits()
1370                         cl.unlock()
1371                 }
1372         }
1373 }
1374
1375 func (cl *Client) rateLimitAccept(ip net.IP) bool {
1376         if cl.config.DisableAcceptRateLimiting {
1377                 return false
1378         }
1379         return cl.acceptLimiter[ipStr(maskIpForAcceptLimiting(ip).String())] > 0
1380 }
1381
1382 func (cl *Client) rLock() {
1383         cl._mu.RLock()
1384 }
1385
1386 func (cl *Client) rUnlock() {
1387         cl._mu.RUnlock()
1388 }
1389
1390 func (cl *Client) lock() {
1391         cl._mu.Lock()
1392 }
1393
1394 func (cl *Client) unlock() {
1395         cl._mu.Unlock()
1396 }
1397
1398 func (cl *Client) locker() *lockWithDeferreds {
1399         return &cl._mu
1400 }
1401
1402 func (cl *Client) String() string {
1403         return fmt.Sprintf("<%[1]T %[1]p>", cl)
1404 }