// Package bt downloads a torrent or magnet link. The peer wire protocol, DHT, // trackers, magnet metadata, piece selection and choking are all carried by // github.com/anacrolix/torrent; this package's job is to present that engine as // a download.Download. Within a single run, one shared torrent.Client serves // every download, so they share one listen port, DHT node and piece-store // rather than fighting over them. package bt import ( "context" "fmt" "path/filepath" "strconv" "strings" "sync/atomic" "time" missinggo "github.com/anacrolix/missinggo/v2" "github.com/anacrolix/torrent" "github.com/anacrolix/torrent/metainfo" "github.com/hanbok/got/download" "golang.org/x/time/rate" ) // ClientConfig holds the run-wide settings used to build the shared client. type ClientConfig struct { Dir string ListenPortSpec string // raw --listen-port spec ("6881-6999,7000"); empty lets the OS choose. DHT bool MaxPeers int // max peer connections per torrent (0 = library default) OverallDown *rate.Limiter // global down limit (may be nil) OverallUp *rate.Limiter // global up limit (may be nil) DownLimit int64 // per-run bytes/s, 0 = unlimited (used only without a global limiter) UpLimit int64 UserAgent string DisableIPv6 bool // force IPv4-only (--disable-ipv6) } // ParsePorts expands a --listen-port spec into a flat list of candidate ports. // The spec is a comma-separated list of single ports and a-b ranges // (e.g. "6881-6889,6999"). Out-of-range and unparsable entries are // skipped; an empty or all-bad spec yields nil. func ParsePorts(spec string) []int { var ports []int for _, part := range strings.Split(spec, ",") { part = strings.TrimSpace(part) if part == "" { continue } lo, hi, isRange := strings.Cut(part, "-") a, err := strconv.Atoi(strings.TrimSpace(lo)) if err != nil { // Unparsable low bound: drop the whole entry, symmetrically with the // high bound below (an out-of-range a is still filtered per-port). continue } if !isRange { if a >= 1 && a <= 65535 { ports = append(ports, a) } continue } b, err := strconv.Atoi(strings.TrimSpace(hi)) if err != nil { continue } for i := a; i <= b; i++ { if i >= 1 && i <= 65535 { ports = append(ports, i) } } } return ports } // NewClient builds the shared anacrolix client from the run-wide config. When a // listen-port spec is given it tries each candidate port in turn, advancing past // any that is already in use, and falls back to an OS-chosen port if none bind, // so a single busy port does not disable BitTorrent entirely. // // Two per-torrent features have no equivalent in anacrolix/torrent v1.61.0, // which exposes only client-wide knobs; both must therefore be configured run-wide // here rather than attached to an individual Torrent: // // 1. No per-Torrent rate limiter. Throttles are read from the client-wide // DownloadRateLimiter/UploadRateLimiter (no TorrentSpec rate field exists), so // per-torrent --max-download-limit / --max-upload-limit cannot be // honored individually; the effective cap is the shared limiter set below. // 2. No per-Torrent DHT/PEX/LSD toggle for BEP27 private torrents. Those are // client-wide only (NoDHT, DisablePEX, etc.), so a private torrent's leak // prevention must come from a run-wide setting (e.g. --enable-dht=false). func NewClient(cfg ClientConfig) (*torrent.Client, error) { c := torrent.NewDefaultClientConfig() c.DataDir = cfg.Dir if c.DataDir == "" { c.DataDir = "." } c.Seed = true c.NoDHT = !cfg.DHT if cfg.MaxPeers > 0 { c.EstablishedConnsPerTorrent = cfg.MaxPeers } if cfg.UserAgent != "" { c.HTTPUserAgent = cfg.UserAgent } c.DisableIPv6 = cfg.DisableIPv6 switch { case cfg.OverallDown != nil: c.DownloadRateLimiter = cfg.OverallDown case cfg.DownLimit > 0: c.DownloadRateLimiter = rate.NewLimiter(rate.Limit(cfg.DownLimit), download.LimiterBurst(cfg.DownLimit)) } switch { case cfg.OverallUp != nil: c.UploadRateLimiter = cfg.OverallUp case cfg.UpLimit > 0: c.UploadRateLimiter = rate.NewLimiter(rate.Limit(cfg.UpLimit), download.LimiterBurst(cfg.UpLimit)) } // Candidate ports from the --listen-port spec. ports := ParsePorts(cfg.ListenPortSpec) if len(ports) == 0 { // No usable spec: let the OS choose (anacrolix already retries on 0). return torrent.NewClient(c) } // Try each candidate; the library only auto-retries when the port is 0, so // for a fixed busy port it returns an address-in-use error and we advance. var lastErr error for _, p := range ports { c.ListenPort = p cl, err := torrent.NewClient(c) if err == nil { return cl, nil } if !isAddrInUse(err) { return nil, err } lastErr = err } // Every named port was busy; fall back to an OS-chosen port rather than // giving up on BitTorrent, so the run keeps going with whatever it can bind. c.ListenPort = 0 cl, err := torrent.NewClient(c) if err != nil { return nil, fmt.Errorf("all listen ports busy (%w); fallback failed: %v", lastErr, err) } return cl, nil } // isAddrInUse reports whether err is an "address already in use" bind failure, // matching either missinggo's helper or the raw string the library wraps. func isAddrInUse(err error) bool { return err != nil && (missinggo.IsAddrInUse(err) || strings.Contains(err.Error(), "address already in use")) } // Options are the per-download settings. type Options struct { SeedTimeSet bool // was --seed-time given? SeedTime time.Duration // how long to seed (0 with SeedTimeSet means no seeding) SeedRatio float64 // stop seeding at this ratio; checked alongside SeedTime StopTimeout time.Duration // abort if no download progress for this long (0 = off) MetaTimeout time.Duration // abort a magnet that can't fetch metadata in this long (0 = off) SelectFiles map[int]bool // 1-based file indexes to fetch; empty = all CheckIntegrity bool // re-verify data against piece hashes before downloading DryRun bool // fetch metadata only, then stop without downloading (--dry-run) } // Download implements download.Download for a single torrent or magnet, added // to a shared client. type Download struct { client *torrent.Client source string // magnet URI or path to a .torrent file isFile bool opts Options name atomic.Pointer[string] status int32 // download.Status // t is published once Run adds the torrent to the client. It is atomic so // Stat stays lock-free, exactly like httpdl.Stat and bt's own name field. t atomic.Pointer[torrent.Torrent] } // New builds a torrent download on the shared client. source is a .torrent path // when isFile is true, otherwise a magnet URI. func New(client *torrent.Client, source string, isFile bool, opts Options) *Download { d := &Download{client: client, source: source, isFile: isFile, opts: opts} if isFile { d.setName(strings.TrimSuffix(filepath.Base(source), ".torrent")) } else { d.setName(magnetName(source)) } return d } func (d *Download) Name() string { if p := d.name.Load(); p != nil { return *p } return "torrent" } func (d *Download) setName(s string) { d.name.Store(&s) } // setStatus stores the lifecycle state with a single atomic op so Stat stays // lock-free, mirroring httpdl. Errored is set once, by Run's deferred guard, when // Run returns a non-nil error. func (d *Download) setStatus(s download.Status) { atomic.StoreInt32(&d.status, int32(s)) } // id is the process-unique identity for this download: the infohash once known, // otherwise the source string (a pre-metadata magnet has no infohash yet, and // two such magnets would otherwise collide on Name). func (d *Download) id(t *torrent.Torrent) string { if t != nil { if h := t.InfoHash(); h != (metainfo.Hash{}) { return h.HexString() } } return d.source } func (d *Download) Stat() download.Stat { t := d.t.Load() st := download.Status(atomic.LoadInt32(&d.status)) if t == nil || t.Info() == nil { return download.Stat{Name: d.Name(), ID: d.id(t), IsBT: true, Status: st, Total: -1} } s := t.Stats() total, completed := t.Length(), t.BytesCompleted() // With --select-file, report progress against the selected subset, not the // whole torrent, so the percentage and ETA mean something. if len(d.opts.SelectFiles) > 0 { total, completed = 0, 0 for i, f := range t.Files() { if d.opts.SelectFiles[i+1] { total += f.Length() completed += f.BytesCompleted() } } } return download.Stat{ Name: d.Name(), ID: d.id(t), IsBT: true, Status: st, Total: total, Completed: completed, Uploaded: s.BytesWrittenData.Int64(), Conns: s.ActivePeers, Seeders: s.ConnectedSeeders, } } // Run adds the torrent to the shared client, downloads it, then seeds. On any // failure the named return is non-nil, and the deferred guard flips the status // to Errored exactly once — so each failure site below just returns its error, // mirroring httpdl.Run's idiom. func (d *Download) Run(ctx context.Context) (err error) { defer func() { if err != nil { d.setStatus(download.Errored) } }() d.setStatus(download.Active) if d.client == nil { return fmt.Errorf("bittorrent is unavailable") } t, err := d.add() if err != nil { return err } defer t.Drop() // release this torrent; the client lives on for others d.t.Store(t) if err := d.awaitInfo(ctx, t); err != nil { return err } d.setName(t.Name()) // --dry-run: fetching the metadata is all a torrent does under a dry run // (the content download is cancelled). Report success without downloading. if d.opts.DryRun { d.setStatus(download.Complete) return nil } if err := d.choose(t); err != nil { return err } if d.opts.CheckIntegrity { if err := t.VerifyDataContext(ctx); err != nil { return err } } if err := d.wait(ctx, t); err != nil { return err } d.setStatus(download.Seeding) d.seed(ctx, t) d.setStatus(download.Complete) return nil } // add hands the source to the shared client. It recovers from the library's // panics on malformed input (e.g. a magnet with a zero infohash) so a bad // argument fails as an ordinary error instead of crashing the whole program. func (d *Download) add() (t *torrent.Torrent, err error) { defer func() { if r := recover(); r != nil { t, err = nil, fmt.Errorf("invalid torrent/magnet: %v", r) } }() if d.isFile { return d.client.AddTorrentFromFile(d.source) } return d.client.AddMagnet(d.source) } // choose applies --select-file: when set, only the listed files are fetched. // With no selection we ask for everything. It is an error for a selection to // match no file, so a typo'd index fails loudly instead of "completing" nothing. func (d *Download) choose(t *torrent.Torrent) error { if len(d.opts.SelectFiles) == 0 { t.DownloadAll() return nil } files := t.Files() matched := 0 for i, f := range files { if d.opts.SelectFiles[i+1] { f.SetPriority(torrent.PiecePriorityNormal) matched++ } else { f.SetPriority(torrent.PiecePriorityNone) } } if matched == 0 { return fmt.Errorf("--select-file: no file index in 1..%d", len(files)) } return nil } // awaitInfo blocks until the torrent metadata arrives. It gives up after // --bt-metadata-timeout (default 60s) so a magnet with no reachable peers — a // dead link, or UDP trackers behind a firewall with no DHT — fails fast instead // of hanging forever. Setting --bt-metadata-timeout=0 disables the timeout and // waits indefinitely. func (d *Download) awaitInfo(ctx context.Context, t *torrent.Torrent) error { var deadline <-chan time.Time if d.opts.MetaTimeout > 0 { tm := time.NewTimer(d.opts.MetaTimeout) defer tm.Stop() deadline = tm.C } select { case <-t.GotInfo(): return nil case <-ctx.Done(): return ctx.Err() case <-deadline: return fmt.Errorf("timed out fetching metadata after %s", d.opts.MetaTimeout) } } // wait blocks until the wanted data is complete, ctx is cancelled, or the // download stalls past --bt-stop-timeout. func (d *Download) wait(ctx context.Context, t *torrent.Torrent) error { tick := time.NewTicker(500 * time.Millisecond) defer tick.Stop() last := t.BytesCompleted() lastChange := time.Now() for { if d.complete(t) { return nil } select { case <-ctx.Done(): return ctx.Err() case <-tick.C: if n := t.BytesCompleted(); n != last { last, lastChange = n, time.Now() } else if d.opts.StopTimeout > 0 && time.Since(lastChange) > d.opts.StopTimeout { return fmt.Errorf("no progress for %s", d.opts.StopTimeout) } } } } // complete reports whether everything we asked for is downloaded AND verified. // For a selection we check each wanted file's pieces are hash-complete rather // than trusting byte counts, which include not-yet-verified (dirty) bytes. func (d *Download) complete(t *torrent.Torrent) bool { if len(d.opts.SelectFiles) == 0 { return t.Complete().Bool() } for i, f := range t.Files() { if !d.opts.SelectFiles[i+1] { continue } for _, ps := range f.State() { if !ps.Complete { return false } } } return true } // seed keeps the torrent up after completion. Seeding stops as soon as EITHER // criterion fires: the --seed-time elapses or the --seed-ratio is reached. Both // are active at once when both are set; with neither set we seed forever (until // ctx is cancelled). One short-circuit applies: --seed-time=0 means do not seed. func (d *Download) seed(ctx context.Context, t *torrent.Torrent) { if d.opts.SeedTimeSet && d.opts.SeedTime == 0 { return } // Arm the time criterion whenever --seed-time was given (independently of the // ratio criterion below). var timer <-chan time.Time if d.opts.SeedTimeSet { tm := time.NewTimer(d.opts.SeedTime) defer tm.Stop() timer = tm.C } tick := time.NewTicker(time.Second) defer tick.Stop() // Share ratio is uploaded/downloaded. The denominator is the torrent's // completed bytes at the moment seeding starts — under --select-file only the // selected files are ever fetched, so this is already close to the subset, but // it is not subset-filtered: any boundary/shared pieces count toward it too. dl := t.BytesCompleted() for { select { case <-ctx.Done(): return case <-timer: return case <-tick.C: // Ratio criterion is checked every tick whenever a ratio is set, // regardless of whether --seed-time is also set. if d.opts.SeedRatio > 0 && dl > 0 { st := t.Stats() up := st.BytesWrittenData.Int64() if float64(up)/float64(dl) >= d.opts.SeedRatio { return } } } } } // FileInfo describes one file inside a torrent, for --show-files. type FileInfo struct { Index int Path string Length int64 } // Files lists the files in a .torrent without starting a download. func Files(torrentPath string) ([]FileInfo, error) { mi, err := metainfo.LoadFromFile(torrentPath) if err != nil { return nil, err } info, err := mi.UnmarshalInfo() if err != nil { return nil, err } if len(info.Files) == 0 { return []FileInfo{{1, info.Name, info.Length}}, nil } out := make([]FileInfo, len(info.Files)) for i, f := range info.Files { out[i] = FileInfo{i + 1, strings.Join(f.Path, "/"), f.Length} } return out, nil } // SourceInfoHash returns the v1 infohash of a .torrent file (isFile) or a magnet // URI, with ok=false when it can't be known without a network fetch (an HTTP // URL), the source is malformed, or the v1 infohash is zero — so a BitTorrent // v2-only source is treated as unique (not deduped). Callers use it to collapse // two sources that name the same torrent into one download: the shared client // keys torrents by infohash and does not refcount, so two jobs sharing one // torrent would have the first to finish Drop it out from under the other. func SourceInfoHash(source string, isFile bool) (metainfo.Hash, bool) { var h metainfo.Hash if isFile { mi, err := metainfo.LoadFromFile(source) if err != nil { return metainfo.Hash{}, false } h = mi.HashInfoBytes() } else { m, err := metainfo.ParseMagnetUri(source) if err != nil { return metainfo.Hash{}, false } h = m.InfoHash } if h == (metainfo.Hash{}) { return metainfo.Hash{}, false } return h, true } func magnetName(uri string) string { if m, err := metainfo.ParseMagnetUri(uri); err == nil && m.DisplayName != "" { return m.DisplayName } return "magnet" }