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