"bufio"
"context"
"crypto/rand"
+ "crypto/sha1"
"encoding/binary"
"errors"
"expvar"
"math"
"net"
"net/http"
+ "net/netip"
"sort"
"strconv"
"strings"
"time"
+ "github.com/davecgh/go-spew/spew"
+ "github.com/dustin/go-humanize"
+ gbtree "github.com/google/btree"
+ "github.com/pion/datachannel"
+ "golang.org/x/time/rate"
+
+ "github.com/anacrolix/chansync"
"github.com/anacrolix/chansync/events"
"github.com/anacrolix/dht/v2"
"github.com/anacrolix/dht/v2/krpc"
+ "github.com/anacrolix/generics"
+ . "github.com/anacrolix/generics"
"github.com/anacrolix/log"
"github.com/anacrolix/missinggo/perf"
- "github.com/anacrolix/missinggo/pubsub"
"github.com/anacrolix/missinggo/v2"
"github.com/anacrolix/missinggo/v2/bitmap"
"github.com/anacrolix/missinggo/v2/pproffd"
"github.com/anacrolix/sync"
- request_strategy "github.com/anacrolix/torrent/request-strategy"
- "github.com/davecgh/go-spew/spew"
- "github.com/dustin/go-humanize"
- "github.com/google/btree"
- "github.com/pion/datachannel"
- "golang.org/x/time/rate"
-
- "github.com/anacrolix/chansync"
"github.com/anacrolix/torrent/bencode"
"github.com/anacrolix/torrent/internal/limiter"
"github.com/anacrolix/torrent/metainfo"
"github.com/anacrolix/torrent/mse"
pp "github.com/anacrolix/torrent/peer_protocol"
+ request_strategy "github.com/anacrolix/torrent/request-strategy"
"github.com/anacrolix/torrent/storage"
"github.com/anacrolix/torrent/tracker"
"github.com/anacrolix/torrent/webtorrent"
// include ourselves if we end up trying to connect to our own address
// through legitimate channels.
dopplegangerAddrs map[string]struct{}
- badPeerIPs map[string]struct{}
+ badPeerIPs map[netip.Addr]struct{}
torrents map[InfoHash]*Torrent
pieceRequestOrder map[interface{}]*request_strategy.PieceRequestOrder
websocketTrackers websocketTrackers
activeAnnounceLimiter limiter.Instance
- webseedHttpClient *http.Client
+ httpClient *http.Client
}
type ipStr string
ips = make([]string, len(cl.badPeerIPs))
i := 0
for k := range cl.badPeerIPs {
- ips[i] = k
+ ips[i] = k.String()
i += 1
}
return
w,
"%f%% of %d bytes (%s)",
100*(1-float64(t.bytesMissingLocked())/float64(t.info.TotalLength())),
- *t.length,
- humanize.Bytes(uint64(*t.length)))
+ t.length(),
+ humanize.Bytes(uint64(t.length())))
} else {
w.WriteString("<missing metainfo>")
}
}
}
-// Filters things that are less than warning from UPnP discovery.
-func upnpDiscoverLogFilter(m log.Msg) bool {
- level, ok := m.GetLevel()
- return !m.HasValue(UpnpDiscoverLogTag) || (!level.LessThan(log.Warning) && ok)
-}
-
func (cl *Client) initLogger() {
logger := cl.config.Logger
if logger.IsZero() {
logger = log.Default
- if !cl.config.Debug {
- logger = logger.FilterLevel(log.Info).WithFilter(upnpDiscoverLogFilter)
- }
+ }
+ if cl.config.Debug {
+ logger = logger.FilterLevel(log.Debug)
}
cl.logger = logger.WithValues(cl)
}
// Initializes a bare minimum Client. *Client and *ClientConfig must not be nil.
func (cl *Client) init(cfg *ClientConfig) {
cl.config = cfg
- cl.dopplegangerAddrs = make(map[string]struct{})
+ generics.MakeMap(&cl.dopplegangerAddrs)
cl.torrents = make(map[metainfo.Hash]*Torrent)
cl.dialRateLimiter = rate.NewLimiter(10, 10)
cl.activeAnnounceLimiter.SlotsPerKey = 2
cl.event.L = cl.locker()
cl.ipBlockList = cfg.IPBlocklist
- cl.webseedHttpClient = &http.Client{
+ cl.httpClient = &http.Client{
Transport: &http.Transport{
- Proxy: cfg.HTTPProxy,
+ Proxy: cfg.HTTPProxy,
+ // I think this value was observed from some webseeds. It seems reasonable to extend it
+ // to other uses of HTTP from the client.
MaxConnsPerHost: 10,
},
}
}
}
- sockets, err := listenAll(cl.listenNetworks(), cl.config.ListenHost, cl.config.ListenPort, cl.firewallCallback)
+ sockets, err := listenAll(cl.listenNetworks(), cl.config.ListenHost, cl.config.ListenPort, cl.firewallCallback, cl.logger)
if err != nil {
return
}
// Creates an anacrolix/dht Server, as would be done internally in NewClient, for the given conn.
func (cl *Client) NewAnacrolixDhtServer(conn net.PacketConn) (s *dht.Server, err error) {
+ logger := cl.logger.WithNames("dht", conn.LocalAddr().String())
cfg := dht.ServerConfig{
IPBlocklist: cl.ipBlockList,
Conn: conn,
}(),
StartingNodes: cl.config.DhtStartingNodes(conn.LocalAddr().Network()),
OnQuery: cl.config.DHTOnQuery,
- Logger: cl.logger.WithContextText(fmt.Sprintf("dht server on %v", conn.LocalAddr().String())),
+ Logger: logger,
}
if f := cl.config.ConfigureAnacrolixDhtServer; f != nil {
f(&cfg)
go func() {
ts, err := s.Bootstrap()
if err != nil {
- cl.logger.Printf("error bootstrapping dht: %s", err)
+ logger.Levelf(log.Warning, "error bootstrapping dht: %s", err)
}
- log.Fstr("%v completed bootstrap (%+v)", s, ts).AddValues(s, ts).Log(cl.logger)
+ logger.Levelf(log.Debug, "completed bootstrap: %+v", ts)
}()
}
return
return
}
if err != nil {
- log.Fmsg("error accepting connection: %s", err).SetLevel(log.Debug).Log(cl.logger)
+ log.Fmsg("error accepting connection: %s", err).LogLevel(log.Debug, cl.logger)
continue
}
go func() {
if reject != nil {
torrent.Add("rejected accepted connections", 1)
- log.Fmsg("rejecting accepted conn: %v", reject).SetLevel(log.Debug).Log(cl.logger)
+ cl.logger.LazyLog(log.Debug, func() log.Msg {
+ return log.Fmsg("rejecting accepted conn: %v", reject)
+ })
conn.Close()
} else {
go cl.incomingConnection(conn)
}
- log.Fmsg("accepted %q connection at %q from %q",
- l.Addr().Network(),
- conn.LocalAddr(),
- conn.RemoteAddr(),
- ).SetLevel(log.Debug).Log(cl.logger)
+ cl.logger.LazyLog(log.Debug, func() log.Msg {
+ return log.Fmsg("accepted %q connection at %q from %q",
+ l.Addr().Network(),
+ conn.LocalAddr(),
+ conn.RemoteAddr(),
+ )
+ })
torrent.Add(fmt.Sprintf("accepted conn remote IP len=%d", len(addrIpOrNil(conn.RemoteAddr()))), 1)
torrent.Add(fmt.Sprintf("accepted conn network=%s", conn.RemoteAddr().Network()), 1)
torrent.Add(fmt.Sprintf("accepted on %s listener", l.Addr().Network()), 1)
cl.noLongerHalfOpen(t, addr.String())
if err != nil {
if cl.config.Debug {
- cl.logger.Printf("error establishing outgoing connection to %v: %v", addr, err)
+ cl.logger.Levelf(log.Debug, "error establishing outgoing connection to %v: %v", addr, err)
}
return
}
}
t, err := cl.receiveHandshakes(c)
if err != nil {
- log.Fmsg(
- "error receiving handshakes on %v: %s", c, err,
- ).SetLevel(log.Debug).
- Add(
+ cl.logger.LazyLog(log.Debug, func() log.Msg {
+ return log.Fmsg(
+ "error receiving handshakes on %v: %s", c, err,
+ ).Add(
"network", c.Network,
- ).Log(cl.logger)
+ )
+ })
torrent.Add("error receiving handshake", 1)
cl.lock()
cl.onBadAccept(c.RemoteAddr)
}
if t == nil {
torrent.Add("received handshake for unloaded torrent", 1)
- log.Fmsg("received handshake for unloaded torrent").SetLevel(log.Debug).Log(cl.logger)
+ cl.logger.LazyLog(log.Debug, func() log.Msg {
+ return log.Fmsg("received handshake for unloaded torrent")
+ })
cl.lock()
cl.onBadAccept(c.RemoteAddr)
cl.unlock()
// address, we won't record the remote address as a doppleganger. Instead, the initiator
// can record *us* as the doppleganger.
} */
- t.logger.WithLevel(log.Debug).Printf("local and remote peer ids are the same")
+ t.logger.Levelf(log.Debug, "local and remote peer ids are the same")
return nil
}
c.r = deadlineReader{c.conn, c.r}
panic(p.updateRequestsTimer)
}
}
- p.updateRequestsTimer = time.AfterFunc(math.MaxInt64, p.updateRequestsTimerFunc)
+ if enableUpdateRequestsTimer {
+ p.updateRequestsTimer = time.AfterFunc(math.MaxInt64, p.updateRequestsTimerFunc)
+ }
}
const peerUpdateRequestsTimerReason = "updateRequestsTimer"
return false
}
+// Returns whether the IP address and port are considered "bad".
func (cl *Client) badPeerIPPort(ip net.IP, port int) bool {
- if port == 0 {
+ if port == 0 || ip == nil {
return true
}
if cl.dopplegangerAddr(net.JoinHostPort(ip.String(), strconv.FormatInt(int64(port), 10))) {
if _, ok := cl.ipBlockRange(ip); ok {
return true
}
- if _, ok := cl.badPeerIPs[ip.String()]; ok {
+ ipAddr, ok := netip.AddrFromSlice(ip)
+ if !ok {
+ panic(ip)
+ }
+ if _, ok := cl.badPeerIPs[ipAddr]; ok {
return true
}
return false
cl: cl,
infoHash: opts.InfoHash,
peers: prioritizedPeers{
- om: btree.New(32),
+ om: gbtree.New(32),
getPrio: func(p PeerInfo) peerPriority {
ipPort := p.addr()
return bep40PriorityIgnoreError(cl.publicAddr(ipPort.IP), ipPort)
},
conns: make(map[*PeerConn]struct{}, 2*cl.config.EstablishedConnsPerTorrent),
- halfOpen: make(map[string]PeerInfo),
- pieceStateChanges: pubsub.NewPubSub(),
+ halfOpen: make(map[string]PeerInfo),
storageOpener: storageClient,
maxEstablishedConns: cl.config.EstablishedConnsPerTorrent,
webSeeds: make(map[string]*Peer),
gotMetainfoC: make(chan struct{}),
}
+ t.smartBanCache.Hash = sha1.Sum
+ t.smartBanCache.Init()
t.networkingEnabled.Set()
- t.logger = cl.logger.WithContextValue(t)
+ t.logger = cl.logger.WithContextValue(t).WithNames("torrent", t.infoHash.HexString())
+ t.sourcesLogger = t.logger.WithNames("sources")
if opts.ChunkSize == 0 {
opts.ChunkSize = defaultChunkSize
}
if spec.DisplayName != "" {
t.SetDisplayName(spec.DisplayName)
}
- t.initialPieceCheckDisabled = spec.DisableInitialPieceCheck
if spec.InfoBytes != nil {
err := t.SetInfoBytes(spec.InfoBytes)
if err != nil {
}
cl := t.cl
cl.AddDhtNodes(spec.DhtNodes)
+ t.UseSources(spec.Sources)
cl.lock()
defer cl.unlock()
- useTorrentSources(spec.Sources, t)
+ t.initialPieceCheckDisabled = spec.DisableInitialPieceCheck
for _, url := range spec.Webseeds {
t.addWebSeed(url)
}
return nil
}
-func useTorrentSources(sources []string, t *Torrent) {
- // TODO: bind context to the lifetime of *Torrent so that it's cancelled if the torrent closes
- ctx := context.Background()
- for i := 0; i < len(sources); i += 1 {
- s := sources[i]
- go func() {
- if err := useTorrentSource(ctx, s, t); err != nil {
- t.logger.WithDefaultLevel(log.Warning).Printf("using torrent source %q: %v", s, err)
- } else {
- t.logger.Printf("successfully used source %q", s)
- }
- }()
- }
-}
-
-func useTorrentSource(ctx context.Context, source string, t *Torrent) (err error) {
- ctx, cancel := context.WithCancel(ctx)
- defer cancel()
- go func() {
- select {
- case <-t.GotInfo():
- case <-t.Closed():
- case <-ctx.Done():
- }
- cancel()
- }()
- var req *http.Request
- if req, err = http.NewRequestWithContext(ctx, http.MethodGet, source, nil); err != nil {
- panic(err)
- }
- var resp *http.Response
- if resp, err = http.DefaultClient.Do(req); err != nil {
- return
- }
- var mi metainfo.MetaInfo
- err = bencode.NewDecoder(resp.Body).Decode(&mi)
- resp.Body.Close()
- if err != nil {
- if ctx.Err() != nil {
- return nil
- }
- return
- }
- return t.MergeSpec(TorrentSpecFromMetaInfo(&mi))
-}
-
func (cl *Client) dropTorrent(infoHash metainfo.Hash, wg *sync.WaitGroup) (err error) {
t, ok := cl.torrents[infoHash]
if !ok {
return
}
err = t.close(wg)
- if err != nil {
- panic(err)
- }
delete(cl.torrents, infoHash)
return
}
}
func (cl *Client) banPeerIP(ip net.IP) {
- cl.logger.Printf("banning ip %v", ip)
- if cl.badPeerIPs == nil {
- cl.badPeerIPs = make(map[string]struct{})
+ // We can't take this from string, because it will lose netip's v4on6. net.ParseIP parses v4
+ // addresses directly to v4on6, which doesn't compare equal with v4.
+ ipAddr, ok := netip.AddrFromSlice(ip)
+ if !ok {
+ panic(ip)
+ }
+ generics.MakeMapIfNilAndSet(&cl.badPeerIPs, ipAddr, struct{}{})
+ for _, t := range cl.torrents {
+ t.iterPeers(func(p *Peer) {
+ if p.remoteIp().Equal(ip) {
+ t.logger.Levelf(log.Warning, "dropping peer %v with banned ip %v", p, ip)
+ // Should this be a close?
+ p.drop()
+ }
+ })
}
- cl.badPeerIPs[ip.String()] = struct{}{}
}
func (cl *Client) newConnection(nc net.Conn, outgoing bool, remoteAddr PeerRemoteAddr, network, connString string) (c *PeerConn) {
connString: connString,
conn: nc,
}
+ c.initRequestState()
+ // TODO: Need to be much more explicit about this, including allowing non-IP bannable addresses.
+ if remoteAddr != nil {
+ netipAddrPort, err := netip.ParseAddrPort(remoteAddr.String())
+ if err == nil {
+ c.bannableAddr = Some(netipAddrPort.Addr())
+ }
+ }
c.peerImpl = c
c.logger = cl.logger.WithDefaultLevel(log.Warning).WithContextValue(c)
c.setRW(connStatsReadWriter{nc, c})