nprasad2077/gopeed
1
0 Fork 0
mirror of https://github.com/GopeedLab/gopeed.git synced 2026-11-09 10:00:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(protocol): add FTP, FTPS and FTPES downloads (#1569)

Browse Source
This commit is contained in:
Silas Mariusz authored and GitHub committed 2026-10-30 17:01:46 +08:00
1 parent e0951de270
commit c0b70d319d
54 files changed
+3850 -38
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/protocol/ftp/conn.go
+404
View File
@@ -0,0 +1,404 @@
package ftp
import (
"context"
"crypto/tls"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/textproto"
"net/url"
"strings"
"sync"
"time"
"github.com/jlaffaye/ftp"
"golang.org/x/net/proxy"
pftp "github.com/GopeedLab/gopeed/pkg/protocol/ftp"
)
const (
connectTimeout = 15 * time.Second
readTimeout = 15 * time.Second
quitTimeout = 2 * time.Second
)
// target is a parsed FTP URL. It holds the password, so it must never be
// formatted into an error or a log line.
type target struct {
host string // host name or IP, the TLS server name
addr string // host:port
user string
pass string
path string // path relative to the login directory, "" for the directory itself
tlsMode string
}
// parseTarget reads an ftp, ftps or ftpes URL. The path follows RFC 1738: it
// is relative to the login directory, and an encoded leading slash
// (ftp://host/%2Fpub) makes it absolute.
func parseTarget(rawURL, optTLS string) (*target, error) {
u, err := url.Parse(rawURL)
if err != nil {
// url.Error repeats the whole URL, password included.
return nil, errors.New("ftp: invalid URL")
}
t := &target{}
switch strings.ToLower(u.Scheme) {
case "ftps":
t.tlsMode = pftp.TLSImplicit
case "ftpes":
t.tlsMode = pftp.TLSExplicit
case "ftp":
switch strings.ToLower(optTLS) {
case "", pftp.TLSNone:
t.tlsMode = pftp.TLSNone
case pftp.TLSExplicit:
t.tlsMode = pftp.TLSExplicit
case pftp.TLSImplicit:
t.tlsMode = pftp.TLSImplicit
default:
return nil, fmt.Errorf("ftp: unknown tls mode %q", optTLS)
}
default:
return nil, fmt.Errorf("ftp: unsupported scheme %q", u.Scheme)
}
t.host = u.Hostname()
if t.host == "" {
return nil, errors.New("ftp: URL has no host")
}
port := u.Port()
if port == "" {
port = "21"
if t.tlsMode == pftp.TLSImplicit {
port = "990"
}
}
t.addr = net.JoinHostPort(t.host, port)
if u.User != nil {
// Username and Password are already percent-decoded.
t.user = u.User.Username()
t.pass, _ = u.User.Password()
}
if t.user == "" {
t.user, t.pass = "anonymous", "anonymous@"
}
t.path = strings.TrimPrefix(u.Path, "/")
return t, nil
}
type dialContextFunc func(ctx context.Context, network, addr string) (net.Conn, error)
// session is one logged-in control connection. It owns the network
// connections it opens, so abort can unblock any read or write in progress.
type session struct {
*ftp.ServerConn
ctrl net.Conn // control connection as dialed, under any TLS layer
dataHost string // where data connections go
mu sync.Mutex
data net.Conn // current data connection as dialed
closed bool
stop func() bool
}
// deadline bounds the next control-connection exchange. The control
// connection is idle during a transfer, so every exchange sets a new one.
func (s *session) deadline() {
_ = s.ctrl.SetDeadline(time.Now().Add(readTimeout))
}
// clearDeadline lets the control connection stay idle through a transfer.
func (s *session) clearDeadline() {
_ = s.ctrl.SetDeadline(time.Time{})
}
// abort closes every connection at once. Pending calls fail immediately.
func (s *session) abort() {
s.mu.Lock()
defer s.mu.Unlock()
s.closed = true
_ = s.ctrl.Close()
if s.data != nil {
_ = s.data.Close()
}
}
// quit logs out and waits, briefly, for the server to hang up. Servers that
// cap logins count a session until they have seen it end, so the next login
// would otherwise be refused.
func (s *session) quit() {
if s.stop != nil {
s.stop()
}
_ = s.ctrl.SetDeadline(time.Now().Add(quitTimeout))
if s.ServerConn != nil {
_ = s.ServerConn.Quit()
}
s.abort()
}
func (s *session) setData(c net.Conn) error {
s.mu.Lock()
defer s.mu.Unlock()
if s.closed {
_ = c.Close()
return net.ErrClosed
}
s.data = c
return nil
}
// ctrlConn is the control connection handed to the ftp library.
type ctrlConn struct {
net.Conn
}
// RemoteAddr must be a *net.TCPAddr: the library asserts it. A connection
// through a proxy may report something else, and its address is not used
// for data connections anyway (see dialData).
func (c ctrlConn) RemoteAddr() net.Addr {
if a, ok := c.Conn.RemoteAddr().(*net.TCPAddr); ok {
return a
}
return &net.TCPAddr{}
}
// Close is called by the library's Quit right after it sends QUIT. Reading
// until the server hangs up (bounded by the deadline quit set) tells us the
// server has ended the session.
func (c ctrlConn) Close() error {
_, _ = io.Copy(io.Discard, c.Conn)
return c.Conn.Close()
}
// dial opens a control connection, upgrades it to TLS when asked, and logs in.
//
// The library is given a dial function for every connection. That is how
// data connections go through the same proxy as the control connection, but
// it also means the library does no TLS on them: with a dial function it
// neither wraps an implicit-TLS control connection nor any data connection.
// We do both here. Explicit TLS (AUTH TLS) is still done by the library, and
// so are PBSZ and PROT P after login. For the same reason the library's
// DialWithTimeout and DialWithContext would have no effect: connectTimeout
// and ctx are applied to our own dials instead.
func (f *Fetcher) dial(ctx context.Context, t *target) (*session, error) {
netDial, proxied, err := f.netDialer(t)
if err != nil {
return nil, err
}
var tlsConf *tls.Config
if t.tlsMode != pftp.TLSNone {
tlsConf = f.tlsConfig(t.host)
}
dialCtx, cancel := context.WithTimeout(ctx, connectTimeout)
defer cancel()
raw, err := netDial(dialCtx, "tcp", t.addr)
if err != nil {
return nil, err
}
ctrl := raw
if t.tlsMode == pftp.TLSImplicit {
tc := tls.Client(raw, tlsConf)
if err := tc.HandshakeContext(dialCtx); err != nil {
_ = raw.Close()
return nil, err
}
ctrl = tc
}
s := &session{ctrl: raw, dataHost: t.host}
if a, ok := raw.RemoteAddr().(*net.TCPAddr); ok && !proxied {
// Same IP as the control connection, as the library itself does,
// so a name with several addresses cannot send us elsewhere.
s.dataHost = a.IP.String()
}
s.stop = context.AfterFunc(ctx, s.abort)
s.deadline()
handedOut := false
opts := []ftp.DialOption{
ftp.DialWithShutTimeout(readTimeout),
ftp.DialWithDialFunc(func(network, addr string) (net.Conn, error) {
if !handedOut {
// Dial asks for the control connection first, once.
handedOut = true
return ctrlConn{ctrl}, nil
}
return s.dialData(ctx, netDial, tlsConf, addr)
}),
}
switch t.tlsMode {
case pftp.TLSImplicit:
opts = append(opts, ftp.DialWithTLS(tlsConf))
case pftp.TLSExplicit:
opts = append(opts, ftp.DialWithExplicitTLS(tlsConf))
}
c, err := ftp.Dial(t.addr, opts...)
if err != nil {
s.stop()
s.abort()
return nil, err
}
s.ServerConn = c
s.deadline()
if err := c.Login(t.user, t.pass); err != nil {
s.stop()
s.abort()
return nil, err
}
return s, nil
}
// dialData opens a passive data connection. The library takes the host from
// the control connection's peer, which through a proxy is the proxy, so only
// the port is used here.
func (s *session) dialData(ctx context.Context, netDial dialContextFunc, tlsConf *tls.Config, addr string) (net.Conn, error) {
_, port, err := net.SplitHostPort(addr)
if err != nil {
return nil, err
}
dialCtx, cancel := context.WithTimeout(ctx, connectTimeout)
defer cancel()
raw, err := netDial(dialCtx, "tcp", net.JoinHostPort(s.dataHost, port))
if err != nil {
return nil, err
}
if err := s.setData(raw); err != nil {
return nil, err
}
if tlsConf != nil {
// The handshake runs on the first read, as in the library, because
// some servers only start it once the transfer command is answered.
return tls.Client(raw, tlsConf), nil
}
return raw, nil
}
// tlsConfig returns the TLS settings for one control connection and its data
// connections. Certificates are verified unless the request sets
// skipVerifyCert, as for HTTP.
//
// Each call gets its own one-entry session cache, so a data connection
// resumes the TLS session of its own control connection. vsftpd
// (require_ssl_reuse) only checks that some session was resumed, but ProFTPD's
// mod_tls also checks that it is the control connection's; a cache shared by
// several workers would hand a data connection another worker's session.
// A cache in the manager's base config is kept as given.
func (f *Fetcher) tlsConfig(host string) *tls.Config {
cfg := &tls.Config{}
if f.manager != nil && f.manager.baseTLS != nil {
cfg = f.manager.baseTLS.Clone()
}
cfg.ServerName = host
if f.meta.Req != nil && f.meta.Req.SkipVerifyCert {
cfg.InsecureSkipVerify = true
}
if cfg.ClientSessionCache == nil {
cfg.ClientSessionCache = tls.NewLRUClientSessionCache(1)
}
return cfg
}
// netDialer picks how to reach the server, from the same proxy settings the
// HTTP protocol reads (the task's proxy, otherwise the global one).
func (f *Fetcher) netDialer(t *target) (dialContextFunc, bool, error) {
direct := (&net.Dialer{Timeout: connectTimeout}).DialContext
pu, err := f.proxyURL(t)
if err != nil {
return nil, false, err
}
if pu == nil {
return direct, false, nil
}
switch strings.ToLower(pu.Scheme) {
case "socks5", "socks5h":
d, err := proxy.FromURL(pu, &net.Dialer{Timeout: connectTimeout})
if err != nil {
return nil, false, fmt.Errorf("ftp: socks5 proxy: %w", err)
}
if cd, ok := d.(proxy.ContextDialer); ok {
return cd.DialContext, true, nil
}
return func(_ context.Context, network, addr string) (net.Conn, error) {
return d.Dial(network, addr)
}, true, nil
default:
// An HTTP proxy is not used. Every passive data connection goes to
// a port the server picks, and HTTP proxies usually allow CONNECT to
// port 443 only, so the transfer would fail after the login worked.
f.warnOnce.Do(func() {
if l := f.logger(); l != nil {
l.Warn().Msgf("ftp: the %s proxy at %s cannot carry FTP data connections (HTTP proxy), connecting directly; use a SOCKS5 proxy for FTP", strings.ToLower(pu.Scheme), pu.Host)
}
})
return direct, false, nil
}
}
func (f *Fetcher) proxyURL(t *target) (*url.URL, error) {
if f.ctl == nil || f.ctl.GetProxy == nil {
return nil, nil
}
handler := f.ctl.GetProxy(f.meta.Req.Proxy)
if handler == nil {
return nil, nil
}
// The proxy chooser reads only the URL; the credentials stay out.
req := &http.Request{URL: &url.URL{Scheme: "ftp", Host: t.addr, Path: "/" + t.path}, Header: http.Header{}}
pu, err := handler(req)
if err != nil {
return nil, errors.New("ftp: cannot read the proxy settings")
}
return pu, nil
}
// isLoginRefusal tells a server that is full apart from bad credentials.
// 421 always means "not now". 530 means a refused login; it is only a
// capacity limit when these credentials have worked before. The library
// drops the code when USER itself is refused, so a "too many" message counts
// as well.
func isLoginRefusal(err error, credentialsWorked bool) bool {
var te *textproto.Error
if errors.As(err, &te) {
switch te.Code {
case ftp.StatusNotAvailable:
return true
case ftp.StatusNotLoggedIn:
return credentialsWorked || strings.Contains(strings.ToLower(te.Msg), "too many")
}
return false
}
return strings.Contains(strings.ToLower(err.Error()), "too many")
}
// isPermanent reports errors a retry cannot fix: a 5xx reply and a
// certificate that does not verify.
func isPermanent(err error) bool {
var te *textproto.Error
if errors.As(err, &te) {
return te.Code >= 500
}
var verifyErr *tls.CertificateVerificationError
if errors.As(err, &verifyErr) {
return true
}
var recordErr tls.RecordHeaderError
if errors.As(err, &recordErr) {
return true
}
return errors.Is(err, errSizeChanged)
}
func replyCode(err error) int {
var te *textproto.Error
if errors.As(err, &te) {
return te.Code
}
return 0
}
internal/protocol/ftp/download_test.go
+146
View File
@@ -0,0 +1,146 @@
package ftp_test
import (
"bytes"
"strings"
"sync"
"testing"
"time"
"github.com/rs/zerolog"
"github.com/GopeedLab/gopeed/internal/logger"
iftp "github.com/GopeedLab/gopeed/internal/protocol/ftp"
"github.com/GopeedLab/gopeed/pkg/base"
"github.com/GopeedLab/gopeed/pkg/download"
pftp "github.com/GopeedLab/gopeed/pkg/protocol/ftp"
)
// syncBuffer is a log sink that the downloader's goroutines can share.
type syncBuffer struct {
mu sync.Mutex
buf bytes.Buffer
}
func (b *syncBuffer) Write(p []byte) (int, error) {
b.mu.Lock()
defer b.mu.Unlock()
return b.buf.Write(p)
}
func (b *syncBuffer) String() string {
b.mu.Lock()
defer b.mu.Unlock()
return b.buf.String()
}
// taskEvents records the first done or error event of every task, so the
// test never reads a task's status while the downloader writes it.
type taskEvents struct {
mu sync.Mutex
final map[string]download.EventKey
}
func (e *taskEvents) listen(ev *download.Event) {
if ev.Key != download.EventKeyDone && ev.Key != download.EventKeyError {
return
}
e.mu.Lock()
defer e.mu.Unlock()
if _, seen := e.final[ev.Task.ID]; !seen {
e.final[ev.Task.ID] = ev.Key
}
}
func (e *taskEvents) wait(t *testing.T, id string) download.EventKey {
t.Helper()
deadline := time.Now().Add(60 * time.Second)
for time.Now().Before(deadline) {
e.mu.Lock()
key, ok := e.final[id]
e.mu.Unlock()
if ok {
return key
}
time.Sleep(20 * time.Millisecond)
}
t.Fatalf("task %s did not finish", id)
return ""
}
// TestPasswordNeverLogged runs FTP tasks through the downloader with its core
// logger captured: a success, a missing file and a wrong password, all behind
// an HTTP proxy that FTP must skip with a warning.
func TestPasswordNeverLogged(t *testing.T) {
srv := iftp.StartFTPServer(t, iftp.ServerOptions{User: "u", Pass: "p@ss"})
want := srv.WriteFile(t, "secret.bin", iftp.RandomBytes(300000, 21))
d := download.NewDownloader(&download.DownloaderConfig{Storage: download.NewMemStorage(), StorageDir: t.TempDir()})
logs := &syncBuffer{}
d.Logger = &logger.Logger{Logger: zerolog.New(logs)}
events := &taskEvents{final: map[string]download.EventKey{}}
d.Listener(events.listen)
if err := d.Setup(); err != nil {
t.Fatal(err)
}
// Close, not Clear: Clear resets the extension list without a lock while
// the last task's done hooks may still read it.
defer d.Close()
cfg, err := d.GetConfig()
if err != nil {
t.Fatal(err)
}
cfg.Proxy = &base.DownloaderProxyConfig{Enable: true, Scheme: "http", Host: "127.0.0.1:1", Usr: "proxyuser", Pwd: "proxysecret"}
if err := d.PutConfig(cfg); err != nil {
t.Fatal(err)
}
dir := t.TempDir()
create := func(rawURL string) string {
id, err := d.CreateDirect(&base.Request{URL: rawURL}, &base.Options{Path: dir, Extra: &pftp.OptsExtra{Connections: 2}})
if err != nil {
if strings.Contains(err.Error(), "p@ss") || strings.Contains(err.Error(), "p%40ss") {
t.Fatalf("the password leaked into an error: %v", err)
}
t.Fatal(err)
}
return id
}
root := "ftp://u:p%40ss@" + srv.Addr + "/"
okID := create(root + "secret.bin")
if key := events.wait(t, okID); key != download.EventKeyDone {
t.Fatalf("download ended with %q, logs:\n%s", key, logs.String())
}
ok := d.GetTask(okID)
if ok.Protocol != "ftp" {
t.Fatalf("protocol %q, want ftp", ok.Protocol)
}
if got := iftp.FileSHA256(t, ok.Meta.SingleFilepath()); got != want {
t.Fatalf("sha256 %s, want %s", got, want)
}
missingID := create(root + "missing.bin")
if key := events.wait(t, missingID); key != download.EventKeyError {
t.Fatalf("a missing file must fail, got %q", key)
}
wrongID := create("ftp://u:wr0ng%40pw@" + srv.Addr + "/secret.bin")
if key := events.wait(t, wrongID); key != download.EventKeyError {
t.Fatalf("a wrong password must fail, got %q", key)
}
missing, wrong := d.GetTask(missingID), d.GetTask(wrongID)
for _, task := range []*download.Task{ok, missing, wrong} {
if name := task.Name(); strings.Contains(name, "p@ss") || strings.Contains(name, "p%40ss") || strings.Contains(name, "wr0ng") {
t.Fatalf("task name %q shows the password", name)
}
}
out := logs.String()
if !strings.Contains(out, "task id") || !strings.Contains(out, "HTTP proxy") {
t.Fatalf("the captured log misses the expected lines, so it proves nothing:\n%s", out)
}
for _, secret := range []string{"p@ss", "p%40ss", "wr0ng", "proxysecret"} {
if strings.Contains(out, secret) {
t.Fatalf("%q appears in the log:\n%s", secret, out)
}
}
}
internal/protocol/ftp/fetcher.go
+810
View File
@@ -0,0 +1,810 @@
// Package ftp downloads files and directory trees over FTP, FTPS (implicit
// TLS) and FTPES (explicit TLS, AUTH TLS).
//
// A file is split into segments of at least 1 MiB. Each segment is fetched on
// its own login with REST and RETR, and its data connection is closed once
// the segment's last byte has arrived. A server that refuses a login lowers
// the task's connection ceiling to the logins that worked, and a server that
// refuses REST gets one sequential stream per file.
package ftp
import (
"context"
"errors"
"fmt"
"io"
"os"
"path"
"sync"
"sync/atomic"
"time"
"github.com/jlaffaye/ftp"
"github.com/rs/zerolog"
"github.com/GopeedLab/gopeed/internal/controller"
"github.com/GopeedLab/gopeed/internal/fetcher"
"github.com/GopeedLab/gopeed/pkg/base"
pftp "github.com/GopeedLab/gopeed/pkg/protocol/ftp"
)
const (
// Many servers cap logins per user, so FTP starts lower than HTTP.
defaultConnections = 4
maxConnections = 256
minSegmentSize = 1 << 20
maxRetries = 5
copyBufferSize = 256 << 10
)
// retryDelay is the first wait before a retry. Later retries wait longer.
var retryDelay = 500 * time.Millisecond
var (
errSizeChanged = errors.New("ftp: the remote file changed size since the task was resolved")
errShortTransfer = errors.New("ftp: the transfer ended before the segment was complete")
errRestUnsupported = errors.New("ftp: the server does not support REST")
errDiskWrite = errors.New("ftp: cannot write the downloaded data")
)
type config struct {
Connections int `json:"connections"`
}
// segment is a byte range of one file. End is inclusive, so an empty file
// has End == Begin-1 and nothing to fetch.
type segment struct {
File int `json:",omitempty"` // index in Res.Files
Begin int64
End int64
Downloaded int64
}
func (s *segment) remain() int64 {
return s.End - s.Begin + 1 - s.Downloaded
}
// lane is one worker that logged in, shown as one connection in the stats.
type lane struct {
downloaded int64
retryTimes int
failed bool
completed bool
}
// loginError marks a failure to connect or log in, as opposed to a failure
// during a transfer.
type loginError struct{ err error }
func (e *loginError) Error() string { return e.err.Error() }
func (e *loginError) Unwrap() error { return e.err }
type Fetcher struct {
manager *FetcherManager
ctl *controller.Controller
config *config
meta *fetcher.FetcherMeta
doneCh chan error
warnOnce sync.Once
// loginOK is set once these credentials have logged in, after which a
// 530 on another login means the server is full.
loginOK atomic.Bool
// mu guards the download state below.
mu sync.Mutex
segments []*segment
maxConns int
noRest bool
lanes []*lane
done bool
// runMu serializes Start and Pause.
runMu sync.Mutex
cancel context.CancelFunc
runDone chan struct{}
}
func (f *Fetcher) Setup(ctl *controller.Controller) {
f.ctl = ctl
f.doneCh = make(chan error, 1)
if f.meta == nil {
f.meta = &fetcher.FetcherMeta{}
}
if ctl != nil && ctl.GetConfig != nil {
ctl.GetConfig(&f.config)
}
if f.config == nil {
f.config = &config{Connections: defaultConnections}
}
}
func (f *Fetcher) logger() *zerolog.Logger {
if f.ctl == nil {
return nil
}
return f.ctl.Logger
}
func (f *Fetcher) warn(format string, args ...any) {
if l := f.logger(); l != nil {
l.Warn().Msgf(format, args...)
}
}
func (f *Fetcher) initOptions() error {
if f.meta.Opts == nil {
f.meta.Opts = &base.Options{}
}
opts := f.meta.Opts
if err := base.ParseOptExtra[pftp.OptsExtra](opts); err != nil {
return err
}
if opts.Extra == nil {
opts.Extra = &pftp.OptsExtra{}
}
return nil
}
func (f *Fetcher) extra() *pftp.OptsExtra {
if f.meta.Opts != nil {
if e, ok := f.meta.Opts.Extra.(*pftp.OptsExtra); ok && e != nil {
return e
}
}
return &pftp.OptsExtra{}
}
// tlsMode is the request's "tls" field, empty when the request has none.
func (f *Fetcher) tlsMode() string {
if f.meta.Req != nil {
if e, ok := f.meta.Req.Extra.(*pftp.ReqExtra); ok && e != nil {
return e.TLS
}
}
return ""
}
// connections is the number of parallel logins to use. The caller holds mu.
func (f *Fetcher) connections() int {
if f.noRest {
return 1
}
n := f.extra().Connections
if n <= 0 && f.config != nil {
n = f.config.Connections
}
if n <= 0 {
n = defaultConnections
}
n = min(n, maxConnections)
if f.maxConns > 0 {
n = min(n, f.maxConns)
}
return n
}
func (f *Fetcher) Resolve(req *base.Request) error {
if err := base.ParseReqExtra[pftp.ReqExtra](req); err != nil {
return err
}
f.meta.Req = req
t, err := parseTarget(req.URL, f.tlsMode())
if err != nil {
return err
}
s, err := f.dial(context.Background(), t)
if err != nil {
return err
}
// quit, not abort: a server that caps logins must see this session end
// before the download's own logins arrive.
defer s.quit()
f.loginOK.Store(true)
res, err := f.resolveTarget(s, t)
if err != nil {
return err
}
f.meta.Res = res
f.mu.Lock()
f.segments = nil
f.done = false
f.mu.Unlock()
return nil
}
// selected returns the indexes of the files to download, in SelectFiles
// order. The caller holds mu or owns the fetcher.
func (f *Fetcher) selected() []int {
if f.meta == nil || f.meta.Res == nil {
return nil
}
if f.meta.Opts != nil && len(f.meta.Opts.SelectFiles) > 0 {
return f.meta.Opts.SelectFiles
}
all := make([]int, len(f.meta.Res.Files))
for i := range all {
all[i] = i
}
return all
}
// localPath is where a file lands: under the task path, with the file's
// directories below the URL kept.
func (f *Fetcher) localPath(index int) string {
if f.meta.Res.Name == "" {
return f.meta.SingleFilepath()
}
file := f.meta.Res.Files[index]
return path.Join(f.meta.FolderPath(), file.Path, file.Name)
}
// remotePath is the server path of a file. A single file uses the URL's own
// path, which a local rename never touches.
func (f *Fetcher) remotePath(t *target, index int) string {
if f.meta.Res.Name == "" {
return t.path
}
file := f.meta.Res.Files[index]
return path.Join(t.path, file.Path, file.Name)
}
// plan splits every selected file into segments of at least minSegmentSize,
// one per connection at most. The caller holds mu.
func (f *Fetcher) plan() {
conns := f.connections()
f.segments = nil
for _, index := range f.selected() {
size := f.meta.Res.Files[index].Size
n := int64(1)
if !f.noRest {
n = max(1, min(int64(conns), size/minSegmentSize))
}
for i := int64(0); i < n; i++ {
f.segments = append(f.segments, &segment{
File: index,
Begin: i * size / n,
End: (i+1)*size/n - 1,
})
}
}
}
// planSequential gives every unfinished file one segment from byte 0,
// because without REST a transfer cannot start in the middle. The caller
// holds mu.
func (f *Fetcher) planSequential() {
complete := map[int]bool{}
for _, s := range f.segments {
if _, seen := complete[s.File]; !seen {
complete[s.File] = true
}
if s.remain() > 0 {
complete[s.File] = false
}
}
var out []*segment
for _, index := range f.selected() {
size := f.meta.Res.Files[index].Size
if complete[index] {
out = append(out, &segment{File: index, Begin: 0, End: size - 1, Downloaded: size})
continue
}
out = append(out, &segment{File: index, Begin: 0, End: size - 1})
}
f.segments = out
// Bytes counted by the old segments are fetched again.
f.lanes = nil
}
// openFiles opens every selected file for writing. A fresh plan creates and
// sizes them; a resumed one keeps their content, and a file that has gone
// missing starts over.
func (f *Fetcher) openFiles(fresh bool) (map[int]*os.File, error) {
files := map[int]*os.File{}
closeAll := func() {
for _, file := range files {
_ = file.Close()
}
}
for _, index := range f.selected() {
name := f.localPath(index)
var (
file *os.File
err error
)
_, statErr := os.Stat(name)
switch {
case fresh || os.IsNotExist(statErr):
file, err = f.ctl.Touch(name, f.meta.Res.Files[index].Size)
if err == nil && !fresh {
f.mu.Lock()
for _, s := range f.segments {
if s.File == index {
s.Downloaded = 0
}
}
f.mu.Unlock()
}
case statErr != nil:
err = statErr
default:
file, err = os.OpenFile(name, os.O_RDWR, 0)
}
if err != nil {
closeAll()
return nil, err
}
files[index] = file
}
return files, nil
}
func (f *Fetcher) Start() error {
f.runMu.Lock()
defer f.runMu.Unlock()
if f.runDone != nil {
select {
case <-f.runDone:
default:
return nil // already running
}
}
if err := f.initOptions(); err != nil {
return err
}
if err := base.ParseReqExtra[pftp.ReqExtra](f.meta.Req); err != nil {
return err
}
t, err := parseTarget(f.meta.Req.URL, f.tlsMode())
if err != nil {
return err
}
// A paused or failed run may have left a result nobody read.
select {
case <-f.doneCh:
default:
}
f.mu.Lock()
if f.done {
f.mu.Unlock()
f.doneCh <- nil
return nil
}
fresh := len(f.segments) == 0
if fresh {
f.plan()
} else if f.noRest {
f.planSequential()
}
f.mu.Unlock()
files, err := f.openFiles(fresh)
if err != nil {
return err
}
ctx, cancel := context.WithCancel(context.Background())
f.cancel = cancel
f.runDone = make(chan struct{})
go f.run(ctx, t, files, f.runDone)
return nil
}
func (f *Fetcher) run(ctx context.Context, t *target, files map[int]*os.File, done chan struct{}) {
defer close(done)
err := f.download(ctx, t, files)
for _, file := range files {
if cerr := file.Close(); cerr != nil && err == nil {
err = cerr
}
}
f.mu.Lock()
if err == nil {
f.done = true
for _, l := range f.lanes {
l.completed = true
}
}
f.mu.Unlock()
if err != nil && ctx.Err() != nil {
return // paused
}
select {
case f.doneCh <- err:
default:
}
}
func (f *Fetcher) download(ctx context.Context, t *target, files map[int]*os.File) error {
for {
err := f.runWorkers(ctx, t, files)
if !errors.Is(err, errRestUnsupported) {
return err
}
f.mu.Lock()
f.noRest = true
f.planSequential()
f.mu.Unlock()
f.warn("ftp: the server refused REST, so each file is downloaded in one stream")
}
}
// pool hands out the segments of one run to its workers. Every decision to
// leave the run is taken under mu together with the queue it depends on, so
// a segment put back can never be left without a worker.
type pool struct {
mu sync.Mutex
queue []*segment
running int // workers still in the run
}
// take returns the next segment. A worker that finds the queue empty leaves
// the run.
func (p *pool) take() *segment {
p.mu.Lock()
defer p.mu.Unlock()
if len(p.queue) == 0 {
p.running--
return nil
}
s := p.queue[0]
p.queue = p.queue[1:]
return s
}
func (p *pool) putBack(s *segment) {
p.mu.Lock()
defer p.mu.Unlock()
p.queue = append([]*segment{s}, p.queue...)
}
// refused puts back the segment of a worker whose login the server turned
// away. The worker leaves when another worker is still in the run to take
// the segment, and left is the number of workers that remain; the last
// worker stays and retries.
func (p *pool) refused(s *segment) (leave bool, left int) {
p.mu.Lock()
defer p.mu.Unlock()
p.queue = append([]*segment{s}, p.queue...)
if p.running > 1 {
p.running--
return true, p.running
}
return false, p.running
}
// leave takes a worker out of the run when it stops for any other reason.
func (p *pool) leave() {
p.mu.Lock()
defer p.mu.Unlock()
p.running--
}
func (f *Fetcher) runWorkers(ctx context.Context, t *target, files map[int]*os.File) error {
runCtx, cancel := context.WithCancel(ctx)
defer cancel()
f.mu.Lock()
p := &pool{}
for _, s := range f.segments {
if s.remain() > 0 {
p.queue = append(p.queue, s)
}
}
workers := min(f.connections(), len(p.queue))
f.mu.Unlock()
if workers == 0 {
return nil
}
p.running = workers
var (
wg sync.WaitGroup
errMu sync.Mutex
firstErr error
)
for id := 0; id < workers; id++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
if err := f.worker(runCtx, t, files, p, id); err != nil {
errMu.Lock()
if firstErr == nil {
firstErr = err
}
errMu.Unlock()
cancel()
}
}(id)
}
wg.Wait()
if err := ctx.Err(); err != nil {
return err
}
if firstErr != nil {
return firstErr
}
f.mu.Lock()
defer f.mu.Unlock()
for _, s := range f.segments {
if s.remain() > 0 {
return errors.New("ftp: the download stopped with data missing")
}
}
return nil
}
// worker fetches segments from the pool until it is empty. A worker whose
// login the server refuses (421, or 530 once these credentials have worked)
// hands its segment back and leaves while other workers remain, lowering the
// task's ceiling to the workers left. This holds whether or not the worker
// had logged in before: a slot can also go to another client mid-download.
func (f *Fetcher) worker(ctx context.Context, t *target, files map[int]*os.File, p *pool, id int) error {
var ln *lane
f.mu.Lock()
if id < len(f.lanes) {
ln = f.lanes[id]
}
f.mu.Unlock()
onLogin := func() {
f.mu.Lock()
if ln == nil {
ln = &lane{}
f.lanes = append(f.lanes, ln)
}
f.mu.Unlock()
}
failures := 0
pendingRetries := 0
for {
seg := p.take()
if seg == nil {
return nil
}
err := f.fetchSegment(ctx, t, files, seg, &ln, onLogin)
if err == nil {
failures = 0
continue
}
if ctx.Err() != nil {
p.putBack(seg)
p.leave()
return nil
}
var le *loginError
if errors.As(err, &le) && isLoginRefusal(le.err, f.loginOK.Load()) {
if leave, left := p.refused(seg); leave {
f.lowerCeiling(left)
return nil
}
// The last worker: wait for a slot.
} else {
p.putBack(seg)
if errors.Is(err, errRestUnsupported) {
p.leave()
return err
}
if isPermanent(err) || errors.Is(err, errDiskWrite) {
f.markFailed(ln)
p.leave()
return err
}
}
failures++
f.mu.Lock()
if ln != nil {
ln.retryTimes += pendingRetries + 1
pendingRetries = 0
} else {
pendingRetries++
}
f.mu.Unlock()
if failures > maxRetries {
f.markFailed(ln)
p.leave()
return fmt.Errorf("ftp: giving up after %d retries: %w", maxRetries, err)
}
select {
case <-ctx.Done():
p.leave()
return nil
case <-time.After(retryDelay * time.Duration(failures)):
}
}
}
func (f *Fetcher) lowerCeiling(n int) {
f.mu.Lock()
defer f.mu.Unlock()
if f.maxConns == 0 || n < f.maxConns {
f.maxConns = n
}
}
func (f *Fetcher) markFailed(ln *lane) {
if ln == nil {
return
}
f.mu.Lock()
ln.failed = true
f.mu.Unlock()
}
// isRestRefused tells a REST the server does not implement. RETR itself
// fails with 550 and the like, never with these codes.
func isRestRefused(err error) bool {
switch replyCode(err) {
case ftp.StatusBadCommand, ftp.StatusBadArguments, ftp.StatusNotImplemented, ftp.StatusNotImplementedParameter:
return true
}
return false
}
// fetchSegment logs in, retrieves the segment from its first missing byte
// and writes it in place. It stops reading at the segment's end and closes
// the data connection; the server then reports the transfer aborted, which
// is expected.
func (f *Fetcher) fetchSegment(ctx context.Context, t *target, files map[int]*os.File, seg *segment, ln **lane, onLogin func()) error {
s, err := f.dial(ctx, t)
if err != nil {
return &loginError{err}
}
defer s.quit()
f.loginOK.Store(true)
onLogin()
remote := f.remotePath(t, seg.File)
f.mu.Lock()
noRest := f.noRest
wantSize := f.meta.Res.Files[seg.File].Size
if noRest {
seg.Downloaded = 0
}
pos := seg.Begin + seg.Downloaded
end := seg.End
f.mu.Unlock()
// A file that changed on the server would mix old and new bytes.
s.deadline()
if size, err := s.FileSize(remote); err == nil && size != wantSize {
return errSizeChanged
} else if err != nil && replyCode(err) == 0 {
return err
}
s.deadline()
var resp *ftp.Response
if noRest {
resp, err = s.Retr(remote)
} else {
resp, err = s.RetrFrom(remote, uint64(pos))
}
if err != nil {
if !noRest && pos > 0 && isRestRefused(err) {
return errRestUnsupported
}
return err
}
defer resp.Close()
s.clearDeadline()
file := files[seg.File]
buf := make([]byte, copyBufferSize)
for pos <= end {
n := int(min(int64(len(buf)), end-pos+1))
_ = resp.SetDeadline(time.Now().Add(readTimeout))
nr, rerr := resp.Read(buf[:n])
if nr > 0 {
if _, werr := file.WriteAt(buf[:nr], pos); werr != nil {
return fmt.Errorf("%w: %w", errDiskWrite, werr)
}
pos += int64(nr)
f.mu.Lock()
seg.Downloaded += int64(nr)
if *ln != nil {
(*ln).downloaded += int64(nr)
}
f.mu.Unlock()
}
if rerr != nil {
if pos > end {
break
}
if errors.Is(rerr, io.EOF) {
return errShortTransfer
}
return rerr
}
}
return nil
}
func (f *Fetcher) Patch(req *base.Request, opts *base.Options) error {
if req == nil {
return nil
}
if req.URL != "" {
f.meta.Req.URL = req.URL
}
if req.Labels != nil {
if f.meta.Req.Labels == nil {
f.meta.Req.Labels = make(map[string]string)
}
for k, v := range req.Labels {
f.meta.Req.Labels[k] = v
}
}
if req.Proxy != nil {
f.meta.Req.Proxy = req.Proxy
}
return nil
}
func (f *Fetcher) Pause() error {
f.runMu.Lock()
defer f.runMu.Unlock()
if f.cancel != nil {
f.cancel()
}
if f.runDone != nil {
<-f.runDone
}
return nil
}
func (f *Fetcher) Close() error {
return f.Pause()
}
func (f *Fetcher) Meta() *fetcher.FetcherMeta {
return f.meta
}
// Stats reports one connection per worker that logged in, in the HTTP
// protocol's shape.
func (f *Fetcher) Stats() *fetcher.Stats {
f.mu.Lock()
defer f.mu.Unlock()
var total int64
if n := int64(len(f.lanes)); n > 0 && f.meta != nil && f.meta.Res != nil && f.meta.Res.Size > 0 {
total = (f.meta.Res.Size + n - 1) / n
}
conns := make([]*pftp.StatsConnection, 0, len(f.lanes))
for _, l := range f.lanes {
conns = append(conns, &pftp.StatsConnection{
Downloaded: l.downloaded,
Total: total,
Completed: l.completed || (f.done && !l.failed),
Failed: l.failed,
RetryTimes: l.retryTimes,
})
}
return &fetcher.Stats{Snapshot: &pftp.Stats{Connections: conns}}
}
// Progress returns the bytes downloaded of each selected file, in
// SelectFiles order.
func (f *Fetcher) Progress() fetcher.Progress {
f.mu.Lock()
defer f.mu.Unlock()
perFile := map[int]int64{}
for _, s := range f.segments {
perFile[s.File] += s.Downloaded
}
sel := f.selected()
p := make(fetcher.Progress, len(sel))
for i, index := range sel {
p[i] = perFile[index]
}
return p
}
func (f *Fetcher) Wait() error {
return <-f.doneCh
}
internal/protocol/ftp/fetcher_manager.go
+109
View File
@@ -0,0 +1,109 @@
package ftp
import (
"crypto/tls"
"net/url"
"path"
"strings"
"github.com/GopeedLab/gopeed/internal/fetcher"
"github.com/GopeedLab/gopeed/pkg/base"
pftp "github.com/GopeedLab/gopeed/pkg/protocol/ftp"
)
// fetcherData is what a task saves to resume later. Segments has the same
// Begin, End and Downloaded fields every version writes; the other fields are
// optional, so a save without them still restores.
type fetcherData struct {
Segments []*segment
// MaxConns is the login ceiling the server enforced, 0 when none was hit.
MaxConns int `json:",omitempty"`
// NoRest records that the server refused REST, so the task downloads each
// file in one sequential stream.
NoRest bool `json:",omitempty"`
}
// FetcherManager handles ftp://, ftps:// (implicit TLS) and ftpes:// (explicit
// TLS) URLs.
type FetcherManager struct {
// baseTLS is the base for every TLS connection. Nil means the system
// roots; the package's tests set it to trust their own certificate. The
// task's skipVerifyCert still turns verification off.
baseTLS *tls.Config
}
func (fm *FetcherManager) Name() string {
return "ftp"
}
func (fm *FetcherManager) Filters() []*fetcher.SchemeFilter {
return []*fetcher.SchemeFilter{
{Type: fetcher.FilterTypeUrl, Pattern: "FTP"},
{Type: fetcher.FilterTypeUrl, Pattern: "FTPS"},
{Type: fetcher.FilterTypeUrl, Pattern: "FTPES"},
}
}
func (fm *FetcherManager) Build() fetcher.Fetcher {
return &Fetcher{manager: fm}
}
// ParseName names a task before it is resolved. It never includes the
// userinfo of the URL.
func (fm *FetcherManager) ParseName(u string) string {
parsed, err := url.Parse(u)
if err != nil {
return ""
}
name := path.Base(strings.TrimSuffix(parsed.Path, "/"))
if name == "" || name == "/" || name == "." {
name = parsed.Hostname()
}
return name
}
func (fm *FetcherManager) AutoRename() bool {
return true
}
func (fm *FetcherManager) DefaultConfig() any {
return &config{Connections: defaultConnections}
}
func (fm *FetcherManager) Store(f fetcher.Fetcher) (any, error) {
ff := f.(*Fetcher)
ff.mu.Lock()
defer ff.mu.Unlock()
segments := make([]*segment, len(ff.segments))
for i, s := range ff.segments {
c := *s
segments[i] = &c
}
return &fetcherData{
Segments: segments,
MaxConns: ff.maxConns,
NoRest: ff.noRest,
}, nil
}
func (fm *FetcherManager) Restore() (v any, f func(meta *fetcher.FetcherMeta, v any) fetcher.Fetcher) {
return &fetcherData{}, func(meta *fetcher.FetcherMeta, v any) fetcher.Fetcher {
fd := v.(*fetcherData)
ff := fm.Build().(*Fetcher)
ff.meta = meta
if meta != nil && meta.Req != nil {
_ = base.ParseReqExtra[pftp.ReqExtra](meta.Req)
}
if meta != nil && meta.Opts != nil {
_ = base.ParseOptExtra[pftp.OptsExtra](meta.Opts)
}
ff.segments = fd.Segments
ff.maxConns = fd.MaxConns
ff.noRest = fd.NoRest
return ff
}
}
func (fm *FetcherManager) Close() error {
return nil
}
internal/protocol/ftp/fetcher_test.go
+818
View File
@@ -0,0 +1,818 @@
package ftp
import (
"bytes"
"context"
"crypto/tls"
"encoding/json"
"errors"
"fmt"
"net"
"os"
"path/filepath"
"slices"
"strings"
"sync/atomic"
"testing"
"time"
"github.com/armon/go-socks5"
"github.com/rs/zerolog"
"github.com/GopeedLab/gopeed/internal/controller"
"github.com/GopeedLab/gopeed/internal/fetcher"
"github.com/GopeedLab/gopeed/pkg/base"
pftp "github.com/GopeedLab/gopeed/pkg/protocol/ftp"
)
const mib = 1 << 20
// testManager trusts the test server's certificate, the way an embedding
// application could add private roots.
func testManager(srv *FTPServer) *FetcherManager {
fm := &FetcherManager{}
if srv != nil {
fm.baseTLS = &tls.Config{RootCAs: srv.CA}
}
return fm
}
func testController(fm *FetcherManager) *controller.Controller {
ctl := controller.NewController()
ctl.GetConfig = func(v any) {
data, _ := json.Marshal(fm.DefaultConfig())
_ = json.Unmarshal(data, v)
}
return ctl
}
func buildFetcher(fm *FetcherManager) *Fetcher {
f := fm.Build().(*Fetcher)
f.Setup(testController(fm))
return f
}
func ftpURL(srv *FTPServer, scheme, userinfo, p string) string {
if userinfo != "" {
userinfo += "@"
}
return scheme + "://" + userinfo + srv.Addr + "/" + strings.TrimPrefix(p, "/")
}
// resolveOpts resolves req and then gives the fetcher its options, in the
// order the downloader uses.
func resolveOpts(f *Fetcher, req *base.Request, opts *base.Options) error {
if err := f.Resolve(req); err != nil {
return err
}
f.Meta().Opts = opts
return nil
}
func resolveTask(t *testing.T, f *Fetcher, rawURL string, extra *pftp.OptsExtra) {
t.Helper()
resolveRequest(t, f, &base.Request{URL: rawURL}, extra)
}
func resolveRequest(t *testing.T, f *Fetcher, req *base.Request, extra *pftp.OptsExtra) {
t.Helper()
opts := &base.Options{Path: t.TempDir()}
if extra != nil {
opts.Extra = extra
}
if err := resolveOpts(f, req, opts); err != nil {
t.Fatalf("resolve %s: %v", req.URL, err)
}
}
func waitFetcher(t *testing.T, f fetcher.Fetcher) error {
t.Helper()
done := make(chan error, 1)
go func() { done <- f.Wait() }()
select {
case err := <-done:
return err
case <-time.After(60 * time.Second):
t.Fatal("download did not finish in 60s")
return nil
}
}
func startAndWait(t *testing.T, f *Fetcher) {
t.Helper()
if err := f.Start(); err != nil {
t.Fatalf("start: %v", err)
}
if err := waitFetcher(t, f); err != nil {
t.Fatalf("download: %v", err)
}
}
func checkFile(t *testing.T, name, want string) {
t.Helper()
if got := FileSHA256(t, name); got != want {
t.Fatalf("%s: sha256 %s, want %s", name, got, want)
}
}
func ftpStats(t *testing.T, f *Fetcher) *pftp.Stats {
t.Helper()
stats, ok := f.Stats().Snapshot.(*pftp.Stats)
if !ok {
t.Fatalf("stats snapshot is %T, want *ftp.Stats", f.Stats().Snapshot)
}
return stats
}
func storeData(t *testing.T, fm *FetcherManager, f *Fetcher) []byte {
t.Helper()
data, err := fm.Store(f)
if err != nil {
t.Fatal(err)
}
blob, err := json.Marshal(data)
if err != nil {
t.Fatal(err)
}
return blob
}
// restoreFetcher rebuilds a fetcher the way the downloader does after a
// restart: the meta and the save blob both come back from JSON.
func restoreFetcher(t *testing.T, fm *FetcherManager, meta *fetcher.FetcherMeta, blob []byte) *Fetcher {
t.Helper()
metaJSON, err := json.Marshal(meta)
if err != nil {
t.Fatal(err)
}
var restoredMeta fetcher.FetcherMeta
if err := json.Unmarshal(metaJSON, &restoredMeta); err != nil {
t.Fatal(err)
}
v, build := fm.Restore()
if err := json.Unmarshal(blob, v); err != nil {
t.Fatal(err)
}
f := build(&restoredMeta, v).(*Fetcher)
f.Setup(testController(fm))
return f
}
func TestResolveFile(t *testing.T) {
for _, noMLST := range []bool{false, true} {
t.Run(fmt.Sprintf("noMLST=%v", noMLST), func(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{NoMLST: noMLST})
data := RandomBytes(100*1024, 1)
srv.WriteFile(t, "pub/file name.bin", data)
f := buildFetcher(testManager(srv))
resolveTask(t, f, ftpURL(srv, "ftp", "", "pub/file%20name.bin"), nil)
res := f.Meta().Res
if res.Name != "" {
t.Fatalf("a file must not be a folder resource, got name %q", res.Name)
}
if len(res.Files) != 1 {
t.Fatalf("got %d files, want 1", len(res.Files))
}
file := res.Files[0]
if file.Name != "file name.bin" || file.Path != "" || file.Size != int64(len(data)) {
t.Fatalf("got file %+v, want name %q, empty path, size %d", file, "file name.bin", len(data))
}
if res.Size != int64(len(data)) {
t.Fatalf("resource size %d, want %d", res.Size, len(data))
}
if file.Ctime == nil || time.Since(*file.Ctime) > time.Hour {
t.Fatalf("modification time %v is missing or wrong", file.Ctime)
}
})
}
}
func writeTree(t *testing.T, srv *FTPServer) map[string]string {
t.Helper()
return map[string]string{
"a.txt": srv.WriteFile(t, "pub/tree/a.txt", []byte("hello")),
"sub/b.bin": srv.WriteFile(t, "pub/tree/sub/b.bin", RandomBytes(3*mib+17, 2)),
"sub/deeper/c.dat": srv.WriteFile(t, "pub/tree/sub/deeper/c.dat", RandomBytes(2000, 3)),
"sub/empty.txt": srv.WriteFile(t, "pub/tree/sub/empty.txt", nil),
}
}
func TestResolveDir(t *testing.T) {
for _, noMLST := range []bool{false, true} {
for _, p := range []string{"pub/tree", "pub/tree/"} {
t.Run(fmt.Sprintf("noMLST=%v/%s", noMLST, p), func(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{NoMLST: noMLST})
writeTree(t, srv)
if err := os.MkdirAll(filepath.Join(srv.Root, "pub/tree/void"), 0o755); err != nil {
t.Fatal(err)
}
f := buildFetcher(testManager(srv))
resolveTask(t, f, ftpURL(srv, "ftp", "", p), nil)
res := f.Meta().Res
if res.Name != "tree" {
t.Fatalf("folder name %q, want tree", res.Name)
}
var got []string
for _, file := range res.Files {
got = append(got, fmt.Sprintf("%s|%s|%d", file.Path, file.Name, file.Size))
}
want := []string{
"|a.txt|5",
"sub|b.bin|3145745",
"sub|empty.txt|0",
"sub/deeper|c.dat|2000",
}
if !slices.Equal(got, want) {
t.Fatalf("files\n got %q\nwant %q", got, want)
}
if res.Size != 5+3145745+2000 {
t.Fatalf("resource size %d", res.Size)
}
})
}
}
}
func TestResolveDirLimits(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{})
srv.WriteFile(t, "deep/1/2/3/4/5/6/7/8/f.txt", []byte("ok"))
srv.WriteFile(t, "deep9/1/2/3/4/5/6/7/8/9/f.txt", []byte("too deep"))
for i := 0; i < 4; i++ {
srv.WriteFile(t, fmt.Sprintf("many/f%d", i), []byte("x"))
}
f := buildFetcher(testManager(srv))
resolveTask(t, f, ftpURL(srv, "ftp", "", "deep"), nil)
if got := f.Meta().Res.Files[0].Path; got != "1/2/3/4/5/6/7/8" {
t.Fatalf("depth 8 path %q", got)
}
f = buildFetcher(testManager(srv))
err := resolveOpts(f, &base.Request{URL: ftpURL(srv, "ftp", "", "deep9")}, &base.Options{Path: t.TempDir()})
if err == nil || !strings.Contains(err.Error(), "8 levels") {
t.Fatalf("a tree deeper than 8 levels must fail and name the limit, got %v", err)
}
saved := maxWalkFiles
maxWalkFiles = 3
defer func() { maxWalkFiles = saved }()
f = buildFetcher(testManager(srv))
err = resolveOpts(f, &base.Request{URL: ftpURL(srv, "ftp", "", "many")}, &base.Options{Path: t.TempDir()})
if err == nil || !strings.Contains(err.Error(), "3 files") {
t.Fatalf("a tree over the file cap must fail and name the limit, got %v", err)
}
}
func TestDownloadPlain(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{})
data := RandomBytes(3*mib+12345, 4)
want := srv.WriteFile(t, "pub/plain.bin", data)
f := buildFetcher(testManager(srv))
resolveTask(t, f, ftpURL(srv, "ftp", "", "pub/plain.bin"), &pftp.OptsExtra{Connections: 4})
startAndWait(t, f)
checkFile(t, f.Meta().SingleFilepath(), want)
if got := f.Progress().TotalDownloaded(); got != int64(len(data)) {
t.Fatalf("progress %d, want %d", got, len(data))
}
stats := ftpStats(t, f)
if len(stats.Connections) == 0 || len(stats.Connections) > 4 {
t.Fatalf("got %d connections in stats", len(stats.Connections))
}
var sum int64
for _, c := range stats.Connections {
sum += c.Downloaded
if !c.Completed || c.Failed {
t.Fatalf("connection stats %+v after success", c)
}
}
if sum != int64(len(data)) {
t.Fatalf("connections downloaded %d bytes, want %d", sum, len(data))
}
}
func TestDownloadDir(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{})
want := writeTree(t, srv)
f := buildFetcher(testManager(srv))
resolveTask(t, f, ftpURL(srv, "ftp", "", "pub/tree"), &pftp.OptsExtra{Connections: 2})
startAndWait(t, f)
root := f.Meta().FolderPath()
if root != filepath.Join(f.Meta().Opts.Path, "tree") {
t.Fatalf("folder path %q", root)
}
for rel, sum := range want {
checkFile(t, filepath.Join(root, filepath.FromSlash(rel)), sum)
}
}
func testDownloadTLS(t *testing.T, mode string, schemes map[string]string) {
srv := StartFTPServer(t, ServerOptions{TLS: mode})
data := RandomBytes(2*mib+999, 5)
want := srv.WriteFile(t, "secure.bin", data)
for scheme, reqTLS := range schemes {
t.Run(scheme+"+"+reqTLS, func(t *testing.T) {
f := buildFetcher(testManager(srv))
req := &base.Request{URL: ftpURL(srv, scheme, "", "secure.bin"), Extra: &pftp.ReqExtra{TLS: reqTLS}}
resolveRequest(t, f, req, &pftp.OptsExtra{Connections: 2})
startAndWait(t, f)
checkFile(t, f.Meta().SingleFilepath(), want)
})
}
}
func TestDownloadExplicitTLS(t *testing.T) {
testDownloadTLS(t, pftp.TLSExplicit, map[string]string{
"ftpes": "",
"ftp": pftp.TLSExplicit,
})
}
func TestDownloadImplicitTLS(t *testing.T) {
testDownloadTLS(t, pftp.TLSImplicit, map[string]string{
"ftps": "",
"ftp": pftp.TLSImplicit,
})
}
// The TLS mode of a plain ftp:// link lives on the request, so a task that is
// saved and restored from JSON still logs in over TLS.
func TestRestoredRequestKeepsTLSMode(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{TLS: pftp.TLSExplicit})
want := srv.WriteFile(t, "keep.bin", RandomBytes(2*mib+5, 12))
fm := testManager(srv)
f := buildFetcher(fm)
req := &base.Request{URL: ftpURL(srv, "ftp", "", "keep.bin"), Extra: &pftp.ReqExtra{TLS: pftp.TLSExplicit}}
resolveRequest(t, f, req, &pftp.OptsExtra{Connections: 2})
restored := restoreFetcher(t, fm, f.Meta(), storeData(t, fm, f))
if _, ok := restored.Meta().Req.Extra.(*pftp.ReqExtra); !ok {
t.Fatalf("restored request extra is %T, want *ftp.ReqExtra", restored.Meta().Req.Extra)
}
// Explicit mode refuses plain logins, so the download only finishes over TLS.
startAndWait(t, restored)
checkFile(t, restored.Meta().SingleFilepath(), want)
}
func TestBadCertRejectedByDefault(t *testing.T) {
for mode, scheme := range map[string]string{pftp.TLSExplicit: "ftpes", pftp.TLSImplicit: "ftps"} {
t.Run(mode, func(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{TLS: mode})
want := srv.WriteFile(t, "f.bin", RandomBytes(4096, 6))
rawURL := ftpURL(srv, scheme, "", "f.bin")
// No injected roots: the self-signed certificate is unknown.
f := buildFetcher(testManager(nil))
err := resolveOpts(f, &base.Request{URL: rawURL}, &base.Options{Path: t.TempDir()})
var verifyErr *tls.CertificateVerificationError
if !errors.As(err, &verifyErr) {
t.Fatalf("want a certificate verification error, got %v", err)
}
// The request's existing skipVerifyCert switch accepts it.
f = buildFetcher(testManager(nil))
if err := resolveOpts(f, &base.Request{URL: rawURL, SkipVerifyCert: true}, &base.Options{Path: t.TempDir()}); err != nil {
t.Fatalf("skipVerifyCert resolve: %v", err)
}
startAndWait(t, f)
checkFile(t, f.Meta().SingleFilepath(), want)
})
}
}
func TestSegmentsUseRest(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{})
srv.ReadDelay.Store(int64(time.Millisecond))
size := 8 * mib
want := srv.WriteFile(t, "big.bin", RandomBytes(size, 7))
f := buildFetcher(testManager(srv))
resolveTask(t, f, ftpURL(srv, "ftp", "", "big.bin"), &pftp.OptsExtra{Connections: 4})
startAndWait(t, f)
checkFile(t, f.Meta().SingleFilepath(), want)
rests := srv.RestOffsets()
for _, off := range []int64{2 * mib, 4 * mib, 6 * mib} {
if !slices.Contains(rests, off) {
t.Fatalf("REST offsets %v miss the segment start %d", rests, off)
}
}
for _, off := range rests {
if off <= 0 || off >= int64(size) {
t.Fatalf("REST offset %d out of range", off)
}
}
if got := len(ftpStats(t, f).Connections); got != 4 {
t.Fatalf("stats report %d connections, want 4", got)
}
}
func TestLoginCapFallsBack(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{MaxLogins: 1})
want := srv.WriteFile(t, "capped.bin", RandomBytes(4*mib+5, 8))
fm := testManager(srv)
f := buildFetcher(fm)
resolveTask(t, f, ftpURL(srv, "ftp", "", "capped.bin"), &pftp.OptsExtra{Connections: 4})
startAndWait(t, f)
checkFile(t, f.Meta().SingleFilepath(), want)
if srv.Refused() == 0 {
t.Fatal("the server never refused a login, so the cap was not exercised")
}
if got := len(ftpStats(t, f).Connections); got != 1 {
t.Fatalf("stats report %d connections, want 1", got)
}
var saved fetcherData
if err := json.Unmarshal(storeData(t, fm, f), &saved); err != nil {
t.Fatal(err)
}
if saved.MaxConns != 1 {
t.Fatalf("saved login ceiling %d, want 1", saved.MaxConns)
}
}
func TestNoRestSequential(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{NoRest: true})
srv.ReadDelay.Store(int64(time.Millisecond))
want := srv.WriteFile(t, "norest.bin", RandomBytes(4*mib+3, 9))
fm := testManager(srv)
f := buildFetcher(fm)
resolveTask(t, f, ftpURL(srv, "ftp", "", "norest.bin"), &pftp.OptsExtra{Connections: 4})
startAndWait(t, f)
checkFile(t, f.Meta().SingleFilepath(), want)
if rests := srv.RestOffsets(); len(rests) != 0 {
t.Fatalf("the server cannot have seen a REST, got %v", rests)
}
var saved fetcherData
if err := json.Unmarshal(storeData(t, fm, f), &saved); err != nil {
t.Fatal(err)
}
if !saved.NoRest {
t.Fatal("the missing REST support was not recorded")
}
}
// TestResumeAfterKill stops the server's transfers at a fixed byte count,
// waits until the client has written exactly those bytes, then pauses, saves,
// restores and finishes. No timing decides where the pause lands.
func TestResumeAfterKill(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{})
size := int64(4*mib + 777)
want := srv.WriteFile(t, "resume.bin", RandomBytes(int(size), 10))
stallAt := size / 2
srv.StallAfter(stallAt)
fm := testManager(srv)
f := buildFetcher(fm)
resolveTask(t, f, ftpURL(srv, "ftp", "", "resume.bin"), &pftp.OptsExtra{Connections: 2})
if err := f.Start(); err != nil {
t.Fatal(err)
}
select {
case <-srv.Stalled():
case <-time.After(30 * time.Second):
t.Fatal("the server never reached the stall point")
}
deadline := time.Now().Add(30 * time.Second)
for f.Progress().TotalDownloaded() != stallAt {
if time.Now().After(deadline) {
t.Fatalf("progress %d, want the %d bytes the server sent", f.Progress().TotalDownloaded(), stallAt)
}
time.Sleep(5 * time.Millisecond)
}
if err := f.Pause(); err != nil {
t.Fatal(err)
}
srv.Release()
blob := storeData(t, fm, f)
var saved fetcherData
if err := json.Unmarshal(blob, &saved); err != nil {
t.Fatal(err)
}
var savedTotal int64
resumeAt := map[int64]bool{}
for _, seg := range saved.Segments {
savedTotal += seg.Downloaded
if seg.Begin+seg.Downloaded <= seg.End {
resumeAt[seg.Begin+seg.Downloaded] = true
}
}
if savedTotal != stallAt {
t.Fatalf("saved progress %d, want %d", savedTotal, stallAt)
}
before := len(srv.RetrOffsets())
restored := restoreFetcher(t, fm, f.Meta(), blob)
if got := restored.Progress().TotalDownloaded(); got != savedTotal {
t.Fatalf("restored progress %d, want %d", got, savedTotal)
}
startAndWait(t, restored)
checkFile(t, restored.Meta().SingleFilepath(), want)
resumed := srv.RetrOffsets()[before:]
if len(resumed) == 0 {
t.Fatal("no transfer after resume")
}
for _, off := range resumed {
if !resumeAt[off] {
t.Fatalf("resumed RETR at %d, want one of the saved offsets %v", off, resumeAt)
}
}
}
// TestLateRefusalAfterQueueDrained: worker B holds the second file's segment
// while the server takes its time to refuse B's login. Worker A finishes the
// first file meanwhile and finds the queue empty. B must not leave too, or
// nobody fetches the second file.
func TestLateRefusalAfterQueueDrained(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{MaxLogins: 1, RefuseDelay: 500 * time.Millisecond})
want := map[string]string{
"a.bin": srv.WriteFile(t, "dir/a.bin", RandomBytes(1000, 16)),
"b.bin": srv.WriteFile(t, "dir/b.bin", RandomBytes(1000, 17)),
}
f := buildFetcher(testManager(srv))
resolveTask(t, f, ftpURL(srv, "ftp", "", "dir"), &pftp.OptsExtra{Connections: 2})
startAndWait(t, f)
for name, sum := range want {
checkFile(t, filepath.Join(f.Meta().FolderPath(), name), sum)
}
if srv.Refused() == 0 {
t.Fatal("the server never refused a login, so the late refusal was not exercised")
}
}
// TestLoginCapLoweredMidway: both workers have logged in when the server's
// cap drops to one, as when another client takes a slot. The refused worker
// leaves, the ceiling drops to one, and the task finishes.
func TestLoginCapLoweredMidway(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{})
srv.ReadDelay.Store(int64(time.Millisecond))
want := map[string]string{}
for i := 0; i < 8; i++ {
name := fmt.Sprintf("f%d.bin", i)
want[name] = srv.WriteFile(t, "dir/"+name, RandomBytes(512<<10, byte(30+i)))
}
fm := testManager(srv)
f := buildFetcher(fm)
resolveTask(t, f, ftpURL(srv, "ftp", "", "dir"), &pftp.OptsExtra{Connections: 2})
if err := f.Start(); err != nil {
t.Fatal(err)
}
deadline := time.Now().Add(30 * time.Second)
for len(ftpStats(t, f).Connections) < 2 || f.Progress().TotalDownloaded() < 512<<10 {
if time.Now().After(deadline) {
t.Fatal("the two workers never got going")
}
time.Sleep(2 * time.Millisecond)
}
srv.SetMaxLogins(1)
if err := waitFetcher(t, f); err != nil {
t.Fatalf("download: %v", err)
}
for name, sum := range want {
checkFile(t, filepath.Join(f.Meta().FolderPath(), name), sum)
}
if srv.Refused() == 0 {
t.Fatal("the server never refused a login, so the lowered cap was not exercised")
}
for _, c := range ftpStats(t, f).Connections {
if c.Failed {
t.Fatalf("a refused worker failed: %+v", c)
}
}
var saved fetcherData
if err := json.Unmarshal(storeData(t, fm, f), &saved); err != nil {
t.Fatal(err)
}
if saved.MaxConns != 1 {
t.Fatalf("saved login ceiling %d, want 1", saved.MaxConns)
}
}
// TestDataConnectionsResumeOwnSession runs four FTPS workers against a server
// that, like ProFTPD, accepts a data connection only when it resumes the TLS
// session of its own control connection.
func TestDataConnectionsResumeOwnSession(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{TLS: pftp.TLSImplicit, RequireSessionReuse: true})
srv.ReadDelay.Store(int64(200 * time.Microsecond))
want := srv.WriteFile(t, "reuse.bin", RandomBytes(8*mib, 18))
f := buildFetcher(testManager(srv))
resolveTask(t, f, ftpURL(srv, "ftps", "", "reuse.bin"), &pftp.OptsExtra{Connections: 4})
startAndWait(t, f)
checkFile(t, f.Meta().SingleFilepath(), want)
ok, bad := srv.SessionReuse()
if ok < 4 || bad != 0 {
t.Fatalf("data connections: %d resumed their own session, %d did not", ok, bad)
}
for _, c := range ftpStats(t, f).Connections {
if c.RetryTimes != 0 {
t.Fatalf("a worker needed retries: %+v", c)
}
}
}
func TestRetryAfterDroppedConnection(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{})
want := srv.WriteFile(t, "drop.bin", RandomBytes(3*mib, 11))
srv.DropAfter.Store(mib + 100)
f := buildFetcher(testManager(srv))
resolveTask(t, f, ftpURL(srv, "ftp", "", "drop.bin"), &pftp.OptsExtra{Connections: 1})
startAndWait(t, f)
checkFile(t, f.Meta().SingleFilepath(), want)
offsets := srv.RetrOffsets()
if len(offsets) != 2 || offsets[0] != 0 || offsets[1] <= 0 || offsets[1] > mib+100 {
t.Fatalf("RETR offsets %v, want a restart at the dropped offset", offsets)
}
if got := ftpStats(t, f).Connections[0].RetryTimes; got != 1 {
t.Fatalf("retry count %d, want 1", got)
}
}
func TestPercentEncodedPassword(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{User: "u", Pass: "p@ss"})
want := srv.WriteFile(t, "secret.bin", RandomBytes(5000, 12))
f := buildFetcher(testManager(srv))
resolveTask(t, f, ftpURL(srv, "ftp", "u:p%40ss", "secret.bin"), nil)
startAndWait(t, f)
checkFile(t, f.Meta().SingleFilepath(), want)
f = buildFetcher(testManager(srv))
err := resolveOpts(f, &base.Request{URL: ftpURL(srv, "ftp", "u:wrong", "secret.bin")}, &base.Options{Path: t.TempDir()})
if err == nil || !strings.Contains(err.Error(), "530") {
t.Fatalf("a wrong password must fail with 530, got %v", err)
}
}
func TestOldSaveFormatResumes(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{})
data := RandomBytes(2*mib+1, 13)
want := srv.WriteFile(t, "old.bin", data)
fm := testManager(srv)
f := buildFetcher(fm)
resolveTask(t, f, ftpURL(srv, "ftp", "", "old.bin"), &pftp.OptsExtra{Connections: 1})
// A save blob that has only the segment fields every version writes.
const done = 700000
blob := []byte(fmt.Sprintf(`{"Segments":[{"Begin":0,"End":%d,"Downloaded":%d}]}`, len(data)-1, done))
name := f.Meta().SingleFilepath()
if err := os.WriteFile(name, append(data[:done:done], make([]byte, len(data)-done)...), 0o644); err != nil {
t.Fatal(err)
}
restored := restoreFetcher(t, fm, f.Meta(), blob)
if got := restored.Progress().TotalDownloaded(); got != done {
t.Fatalf("restored progress %d, want %d", got, done)
}
startAndWait(t, restored)
checkFile(t, name, want)
if got := srv.RetrOffsets(); !slices.Equal(got, []int64{done}) {
t.Fatalf("RETR offsets %v, want [%d]", got, done)
}
}
func TestDownloadThroughSOCKS5(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{TLS: pftp.TLSExplicit})
want := srv.WriteFile(t, "proxied.bin", RandomBytes(2*mib+1, 14))
var dials atomic.Int32
proxySrv, err := socks5.New(&socks5.Config{
Dial: func(ctx context.Context, network, addr string) (net.Conn, error) {
dials.Add(1)
var d net.Dialer
return d.DialContext(ctx, network, addr)
},
})
if err != nil {
t.Fatal(err)
}
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
defer ln.Close()
go proxySrv.Serve(ln)
f := buildFetcher(testManager(srv))
req := &base.Request{
URL: ftpURL(srv, "ftpes", "", "proxied.bin"),
Proxy: &base.RequestProxy{
Mode: base.RequestProxyModeCustom,
Scheme: "socks5",
Host: ln.Addr().String(),
},
}
if err := resolveOpts(f, req, &base.Options{Path: t.TempDir(), Extra: &pftp.OptsExtra{Connections: 2}}); err != nil {
t.Fatal(err)
}
startAndWait(t, f)
checkFile(t, f.Meta().SingleFilepath(), want)
// Resolve's control and listing, then per worker one control and one
// data connection at least: every one of them went through the proxy.
if got := dials.Load(); got < 4 {
t.Fatalf("the proxy saw %d connections, want every control and data connection", got)
}
}
func TestHTTPProxyFallsBackToDirect(t *testing.T) {
srv := StartFTPServer(t, ServerOptions{})
want := srv.WriteFile(t, "direct.bin", RandomBytes(100000, 15))
var logs bytes.Buffer
logger := zerolog.New(&logs)
fm := testManager(srv)
f := fm.Build().(*Fetcher)
ctl := testController(fm)
ctl.Logger = &logger
f.Setup(ctl)
req := &base.Request{
URL: ftpURL(srv, "ftp", "", "direct.bin"),
Proxy: &base.RequestProxy{
Mode: base.RequestProxyModeCustom,
Scheme: "http",
Host: "127.0.0.1:1",
Usr: "proxyuser",
Pwd: "proxysecret",
},
}
if err := resolveOpts(f, req, &base.Options{Path: t.TempDir(), Extra: &pftp.OptsExtra{Connections: 2}}); err != nil {
t.Fatal(err)
}
startAndWait(t, f)
checkFile(t, f.Meta().SingleFilepath(), want)
out := logs.String()
if n := strings.Count(out, "HTTP proxy"); n != 1 {
t.Fatalf("want exactly one HTTP proxy warning, got %d in %q", n, out)
}
if strings.Contains(out, "proxysecret") {
t.Fatalf("the proxy password leaked into the log: %q", out)
}
}
func TestTorrentNameIsFile(t *testing.T) {
fm := testManager(nil)
for _, u := range []string{"ftp://h/x.torrent", "ftps://h/x.torrent", "ftpes://h/x.torrent", "FTP://h/v.m3u8"} {
matched := false
for _, filter := range fm.Filters() {
if filter.Match(u) {
matched = true
}
}
if !matched {
t.Fatalf("%s is not matched by the FTP filters", u)
}
}
if fm.Name() != "ftp" {
t.Fatalf("manager name %q", fm.Name())
}
srv := StartFTPServer(t, ServerOptions{})
want := srv.WriteFile(t, "x.torrent", []byte("d8:announce0:e"))
f := buildFetcher(testManager(srv))
resolveTask(t, f, ftpURL(srv, "ftp", "", "x.torrent"), nil)
res := f.Meta().Res
if res.Name != "" || len(res.Files) != 1 || res.Files[0].Name != "x.torrent" {
t.Fatalf("x.torrent must resolve to one FTP file, got name %q and %d files", res.Name, len(res.Files))
}
startAndWait(t, f)
checkFile(t, f.Meta().SingleFilepath(), want)
}
func TestParseNameHidesCredentials(t *testing.T) {
fm := testManager(nil)
for u, want := range map[string]string{
"ftp://u:p%[email protected]/pub/a.iso": "a.iso",
"ftp://u:p%[email protected]/": "example.com",
"ftps://u:p%[email protected]": "example.com",
"ftp://u:p%[email protected]/pub/dir/": "dir",
} {
if got := fm.ParseName(u); got != want {
t.Fatalf("ParseName(%q) = %q, want %q", u, got, want)
}
}
}
func TestUnsafeEntryNames(t *testing.T) {
for name, ok := range map[string]bool{
"a.txt": true, "..a": true, "a b": true,
"": false, ".": false, "..": false, "a/b": false, `a\b`: false, "../x": false, "x\x00y": false,
} {
if got := safeEntryName(name); got != ok {
t.Fatalf("safeEntryName(%q) = %v, want %v", name, got, ok)
}
}
}
internal/protocol/ftp/testserver_test.go
+587
View File
@@ -0,0 +1,587 @@
package ftp
import (
"bufio"
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"crypto/sha256"
"crypto/tls"
"crypto/x509"
"crypto/x509/pkix"
"encoding/hex"
"errors"
"io"
"math/big"
"net"
"os"
"path/filepath"
"regexp"
"strconv"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
ftpserver "github.com/fclairamb/ftpserverlib"
"github.com/spf13/afero"
pftp "github.com/GopeedLab/gopeed/pkg/protocol/ftp"
)
// ServerOptions are the knobs of the in-process FTP test server.
type ServerOptions struct {
// TLS is pftp.TLSNone (or empty), pftp.TLSExplicit or pftp.TLSImplicit.
// Explicit mode refuses plain logins, so a passing test proves TLS was used.
TLS string
// MaxLogins caps concurrent control connections. Beyond it the server
// greets with "421 Too many connections" and hangs up. Zero means no cap.
MaxLogins int
// NoRest answers "502" to REST. It filters the plain control stream, so it
// works with TLSNone and TLSImplicit only.
NoRest bool
// NoMLST disables MLST and MLSD, so clients fall back to SIZE and LIST.
NoMLST bool
// User and Pass are the only accepted credentials. An empty User accepts
// any login, anonymous included.
User string
Pass string
// RefuseDelay holds every 421 refusal back this long, like a server
// that answers slowly when it is full.
RefuseDelay time.Duration
// RequireSessionReuse makes every data connection resume the TLS
// session of the control connection that opened it, as ProFTPD's
// mod_tls does by default. A data connection that does not is refused.
// It needs TLSImplicit, where this test's wrapper sees the plain
// control stream and so learns which passive port belongs to whom.
RequireSessionReuse bool
}
// FTPServer is an ftpserverlib server over a temp directory.
type FTPServer struct {
Addr string // host:port
Root string // directory served as "/"
// CA trusts the server's self-signed certificate.
CA *x509.CertPool
opts ServerOptions
tlsConf *tls.Config
server *ftpserver.FtpServer
// ReadDelay is slept before every read of a transferred file, in
// nanoseconds, to slow downloads down. Tests may change it at any time.
ReadDelay atomic.Int64
// DropAfter, when positive, makes the next transfer fail once after it
// has sent that many bytes, as if the data connection were killed.
DropAfter atomic.Int64
mu sync.Mutex
active int
retrs []int64 // offset of every RETR, 0 when no REST came first
refused int
logins int
// Passive port to the control connection that announced it, and the
// outcome of every data connection's session check.
pasvOwner map[int]string
reuseOK int
reuseBad int
// Stall state, see StallAfter. stallMu is held across file reads.
stallMu sync.Mutex
stallAt int64
sent int64
stallCh chan struct{}
stalled chan struct{}
stalledOnce *sync.Once
}
// StartFTPServer starts a server and stops it when the test ends.
func StartFTPServer(t *testing.T, opts ServerOptions) *FTPServer {
t.Helper()
s := &FTPServer{Root: t.TempDir(), opts: opts}
s.tlsConf, s.CA = selfSignedTLS(t)
base, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
var ln net.Listener = base
if opts.RequireSessionReuse {
if opts.TLS != pftp.TLSImplicit {
t.Fatal("RequireSessionReuse needs TLSImplicit")
}
s.requireSessionReuse(t, base.Addr().(*net.TCPAddr).Port)
}
if opts.TLS == pftp.TLSImplicit {
ln = tls.NewListener(ln, s.tlsConf)
}
ln = &limitListener{Listener: ln, srv: s}
s.Addr = base.Addr().String()
s.server = ftpserver.NewFtpServer(&mainDriver{srv: s, listener: ln})
if err := s.server.Listen(); err != nil {
t.Fatal(err)
}
go s.server.Serve()
t.Cleanup(func() { _ = s.server.Stop() })
return s
}
// WriteFile creates a file under the served root and returns its SHA-256.
func (s *FTPServer) WriteFile(t *testing.T, rel string, data []byte) string {
t.Helper()
name := filepath.Join(s.Root, filepath.FromSlash(rel))
if err := os.MkdirAll(filepath.Dir(name), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(name, data, 0o644); err != nil {
t.Fatal(err)
}
return SHA256Hex(data)
}
// RetrOffsets returns the start offset of every RETR the server received.
func (s *FTPServer) RetrOffsets() []int64 {
s.mu.Lock()
defer s.mu.Unlock()
return append([]int64(nil), s.retrs...)
}
// RestOffsets returns the non-zero RETR offsets, which only a REST sets.
func (s *FTPServer) RestOffsets() []int64 {
var out []int64
for _, off := range s.RetrOffsets() {
if off != 0 {
out = append(out, off)
}
}
return out
}
// Refused returns how many connections were turned away with 421.
func (s *FTPServer) Refused() int {
s.mu.Lock()
defer s.mu.Unlock()
return s.refused
}
// Logins returns the number of successful logins.
func (s *FTPServer) Logins() int {
s.mu.Lock()
defer s.mu.Unlock()
return s.logins
}
// SetMaxLogins changes the login cap while the server runs, as when other
// clients take slots.
func (s *FTPServer) SetMaxLogins(n int) {
s.mu.Lock()
s.opts.MaxLogins = n
s.mu.Unlock()
}
// SessionReuse returns how many data connections resumed their own control
// connection's TLS session, and how many did not.
func (s *FTPServer) SessionReuse() (ok, bad int) {
s.mu.Lock()
defer s.mu.Unlock()
return s.reuseOK, s.reuseBad
}
// StallAfter makes every transfer stop sending once the server has sent n
// bytes in total, until Release. Stalled is closed when that happens.
func (s *FTPServer) StallAfter(n int64) {
s.stallMu.Lock()
defer s.stallMu.Unlock()
s.stallAt, s.sent = n, 0
s.stallCh = make(chan struct{})
s.stalled = make(chan struct{})
s.stalledOnce = &sync.Once{}
}
// Stalled is closed once a transfer has stopped at the StallAfter limit.
func (s *FTPServer) Stalled() <-chan struct{} {
s.stallMu.Lock()
defer s.stallMu.Unlock()
return s.stalled
}
// Sent returns the bytes sent since StallAfter.
func (s *FTPServer) Sent() int64 {
s.stallMu.Lock()
defer s.stallMu.Unlock()
return s.sent
}
// Release ends the stall. The stalled transfers fail.
func (s *FTPServer) Release() {
s.stallMu.Lock()
defer s.stallMu.Unlock()
if s.stallAt > 0 {
s.stallAt = 0
close(s.stallCh)
}
}
var errStalled = errors.New("test server released a stalled transfer")
// stall limits a read to the StallAfter budget, and blocks once the budget
// is spent.
func (s *FTPServer) stall(read func([]byte) (int, error), p []byte) (int, error) {
s.stallMu.Lock()
if s.stallAt <= 0 {
s.stallMu.Unlock()
return read(p)
}
allowed := s.stallAt - s.sent
if allowed <= 0 {
ch := s.stallCh
s.stalledOnce.Do(func() { close(s.stalled) })
s.stallMu.Unlock()
<-ch
return 0, errStalled
}
if int64(len(p)) > allowed {
p = p[:allowed]
}
n, err := read(p)
s.sent += int64(n)
s.stallMu.Unlock()
return n, err
}
var pasvReply = regexp.MustCompile(`^(?:229 [^(]*\(\|\|\|(\d+)\|\)|227 [^(]*\(\d+,\d+,\d+,\d+,(\d+),(\d+)\))`)
// notePassive records the passive port a control connection announced.
func (s *FTPServer) notePassive(owner string, reply []byte) {
m := pasvReply.FindSubmatch(reply)
if m == nil {
return
}
var port int
if len(m[1]) > 0 {
port, _ = strconv.Atoi(string(m[1]))
} else {
hi, _ := strconv.Atoi(string(m[2]))
lo, _ := strconv.Atoi(string(m[3]))
port = hi<<8 | lo
}
s.mu.Lock()
s.pasvOwner[port] = owner
s.mu.Unlock()
}
// requireSessionReuse stamps each control connection's session tickets with
// its address, and checks that each data connection resumed a ticket of the
// control connection that announced its passive port.
func (s *FTPServer) requireSessionReuse(t *testing.T, controlPort int) {
base := s.tlsConf
var key [32]byte
if _, err := rand.Read(key[:]); err != nil {
t.Fatal(err)
}
base.SetSessionTicketKeys([][32]byte{key})
s.pasvOwner = map[int]string{}
stamp := func(owner string) func(tls.ConnectionState, *tls.SessionState) ([]byte, error) {
return func(cs tls.ConnectionState, ss *tls.SessionState) ([]byte, error) {
ss.Extra = [][]byte{[]byte(owner)}
return base.EncryptTicket(cs, ss)
}
}
base.GetConfigForClient = func(hello *tls.ClientHelloInfo) (*tls.Config, error) {
cfg := base.Clone()
cfg.GetConfigForClient = nil
local, _ := hello.Conn.LocalAddr().(*net.TCPAddr)
if local == nil || local.Port == controlPort {
cfg.WrapSession = stamp(hello.Conn.RemoteAddr().String())
return cfg, nil
}
s.mu.Lock()
owner := s.pasvOwner[local.Port]
s.mu.Unlock()
var resumedFrom string
cfg.UnwrapSession = func(identity []byte, cs tls.ConnectionState) (*tls.SessionState, error) {
ss, err := base.DecryptTicket(identity, cs)
if err != nil || ss == nil {
return nil, nil
}
if len(ss.Extra) > 0 {
resumedFrom = string(ss.Extra[0])
}
return ss, nil
}
cfg.WrapSession = stamp(owner)
cfg.VerifyConnection = func(cs tls.ConnectionState) error {
ok := owner != "" && cs.DidResume && resumedFrom == owner
s.mu.Lock()
if ok {
s.reuseOK++
} else {
s.reuseBad++
}
s.mu.Unlock()
if !ok {
return errors.New("data connection did not resume its control connection's TLS session")
}
return nil
}
return cfg, nil
}
}
func (s *FTPServer) acquire() bool {
s.mu.Lock()
defer s.mu.Unlock()
if s.opts.MaxLogins > 0 && s.active >= s.opts.MaxLogins {
s.refused++
return false
}
s.active++
return true
}
func (s *FTPServer) release() {
s.mu.Lock()
s.active--
s.mu.Unlock()
}
// limitListener counts control connections and refuses those beyond
// MaxLogins with a 421 greeting, like vsftpd's max_clients and max_per_ip.
type limitListener struct {
net.Listener
srv *FTPServer
}
func (l *limitListener) Accept() (net.Conn, error) {
for {
c, err := l.Listener.Accept()
if err != nil {
return nil, err
}
if !l.srv.acquire() {
go func() {
time.Sleep(l.srv.opts.RefuseDelay)
_ = c.SetDeadline(time.Now().Add(5 * time.Second))
_, _ = io.WriteString(c, "421 Too many connections\r\n")
_ = c.Close()
}()
continue
}
cc := &countedConn{Conn: c, release: l.srv.release}
if l.srv.opts.RequireSessionReuse {
owner := c.RemoteAddr().String()
cc.sniff = func(b []byte) { l.srv.notePassive(owner, b) }
}
var conn net.Conn = cc
if l.srv.opts.NoRest {
conn = &noRestConn{countedConn: conn.(*countedConn), in: bufio.NewReader(c)}
}
return conn, nil
}
}
type countedConn struct {
net.Conn
once sync.Once
release func()
sniff func([]byte) // sees every reply the server writes
}
func (c *countedConn) Write(b []byte) (int, error) {
if c.sniff != nil {
c.sniff(b)
}
return c.Conn.Write(b)
}
func (c *countedConn) Close() error {
c.once.Do(c.release)
return c.Conn.Close()
}
// noRestConn answers REST itself with 502 and hides it from the server.
type noRestConn struct {
*countedConn
in *bufio.Reader
pending []byte
}
func (c *noRestConn) Read(p []byte) (int, error) {
for len(c.pending) == 0 {
line, err := c.in.ReadBytes('\n')
if len(line) > 0 {
if strings.HasPrefix(strings.ToUpper(string(line)), "REST ") {
if _, werr := io.WriteString(c.Conn, "502 REST not implemented\r\n"); werr != nil {
return 0, werr
}
} else {
c.pending = line
}
}
if err != nil {
if len(c.pending) > 0 {
break
}
return 0, err
}
}
n := copy(p, c.pending)
c.pending = c.pending[n:]
return n, nil
}
type mainDriver struct {
srv *FTPServer
listener net.Listener
}
func (d *mainDriver) GetSettings() (*ftpserver.Settings, error) {
settings := &ftpserver.Settings{
Listener: d.listener,
DisableMLSD: d.srv.opts.NoMLST,
DisableMLST: d.srv.opts.NoMLST,
}
switch d.srv.opts.TLS {
case pftp.TLSExplicit:
settings.TLSRequired = ftpserver.MandatoryEncryption
case pftp.TLSImplicit:
settings.TLSRequired = ftpserver.ImplicitEncryption
}
return settings, nil
}
func (d *mainDriver) ClientConnected(ftpserver.ClientContext) (string, error) {
return "gopeed test server", nil
}
func (d *mainDriver) ClientDisconnected(ftpserver.ClientContext) {}
func (d *mainDriver) AuthUser(_ ftpserver.ClientContext, user, pass string) (ftpserver.ClientDriver, error) {
if d.srv.opts.User != "" && (user != d.srv.opts.User || pass != d.srv.opts.Pass) {
return nil, errors.New("bad credentials")
}
d.srv.mu.Lock()
d.srv.logins++
d.srv.mu.Unlock()
return &clientDriver{Fs: afero.NewBasePathFs(afero.NewOsFs(), d.srv.Root), srv: d.srv}, nil
}
func (d *mainDriver) GetTLSConfig() (*tls.Config, error) {
return d.srv.tlsConf, nil
}
type clientDriver struct {
afero.Fs
srv *FTPServer
}
// GetHandle sees the REST offset of every transfer. The server seeks the
// returned file to that offset itself.
func (d *clientDriver) GetHandle(name string, flags int, offset int64) (ftpserver.FileTransfer, error) {
f, err := d.Fs.OpenFile(name, flags, 0o644)
if err != nil {
return nil, err
}
if flags == os.O_RDONLY {
d.srv.mu.Lock()
d.srv.retrs = append(d.srv.retrs, offset)
d.srv.mu.Unlock()
}
return &slowFile{File: f, srv: d.srv, dropAfter: d.srv.DropAfter.Swap(0)}, nil
}
var errDropped = errors.New("test server dropped the transfer")
type slowFile struct {
afero.File
srv *FTPServer
sent int64
dropAfter int64
}
func (f *slowFile) Read(p []byte) (int, error) {
if delay := time.Duration(f.srv.ReadDelay.Load()); delay > 0 {
time.Sleep(delay)
if len(p) > 16<<10 {
p = p[:16<<10]
}
}
if f.dropAfter > 0 {
if f.sent >= f.dropAfter {
return 0, errDropped
}
if rest := f.dropAfter - f.sent; int64(len(p)) > rest {
p = p[:rest]
}
}
n, err := f.srv.stall(f.File.Read, p)
f.sent += int64(n)
return n, err
}
// selfSignedTLS returns a server config with a fresh certificate for
// 127.0.0.1 and localhost, and a pool that trusts it.
func selfSignedTLS(t *testing.T) (*tls.Config, *x509.CertPool) {
t.Helper()
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
if err != nil {
t.Fatal(err)
}
tmpl := &x509.Certificate{
SerialNumber: big.NewInt(time.Now().UnixNano()),
Subject: pkix.Name{CommonName: "gopeed ftp test"},
NotBefore: time.Now().Add(-time.Hour),
NotAfter: time.Now().Add(time.Hour),
IPAddresses: []net.IP{net.ParseIP("127.0.0.1")},
DNSNames: []string{"localhost"},
KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
IsCA: true,
BasicConstraintsValid: true,
}
der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
if err != nil {
t.Fatal(err)
}
cert, err := x509.ParseCertificate(der)
if err != nil {
t.Fatal(err)
}
pool := x509.NewCertPool()
pool.AddCert(cert)
return &tls.Config{
Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key, Leaf: cert}},
MinVersion: tls.VersionTLS12,
}, pool
}
func SHA256Hex(data []byte) string {
sum := sha256.Sum256(data)
return hex.EncodeToString(sum[:])
}
func FileSHA256(t *testing.T, name string) string {
t.Helper()
data, err := os.ReadFile(name)
if err != nil {
t.Fatal(err)
}
return SHA256Hex(data)
}
// RandomBytes returns n reproducible pseudo-random bytes.
func RandomBytes(n int, seed byte) []byte {
out := make([]byte, n)
x := uint32(seed) + 1
for i := range out {
x ^= x << 13
x ^= x >> 17
x ^= x << 5
out[i] = byte(x)
}
return out
}
internal/protocol/ftp/walk.go
+187
View File
@@ -0,0 +1,187 @@
package ftp
import (
"cmp"
"fmt"
"path"
"slices"
"strings"
"time"
"github.com/jlaffaye/ftp"
"github.com/GopeedLab/gopeed/pkg/base"
)
// Limits of a directory download. A tree past either one is refused with an
// error that names the limit; it is never cut short.
const maxWalkDepth = 8
// maxWalkFiles is a variable so tests can lower it.
var maxWalkFiles = 10000
// resolveTarget describes the URL's path: one file, or a directory walked
// into a multi-file resource whose paths are relative to it.
func (f *Fetcher) resolveTarget(s *session, t *target) (*base.Resource, error) {
isDir, size, mtime, err := f.stat(s, t.path)
if err != nil {
return nil, err
}
if !isDir {
return &base.Resource{
Size: size,
Range: true,
Files: []*base.FileInfo{{
Name: path.Base(t.path),
Size: size,
Ctime: mtime,
}},
}, nil
}
files, err := f.walk(s, t.path)
if err != nil {
return nil, err
}
name := path.Base(strings.TrimSuffix(t.path, "/"))
if name == "" || name == "." || name == "/" {
name = t.host
}
res := &base.Resource{Name: name, Range: true, Files: files}
res.CalcSize(nil)
return res, nil
}
// stat tells a file from a directory. MLST answers both at once. Without
// it, SIZE answers for a file, and a directory is one we can change into.
func (f *Fetcher) stat(s *session, p string) (isDir bool, size int64, mtime *time.Time, err error) {
if p == "" || strings.HasSuffix(p, "/") {
return true, 0, nil, nil
}
s.deadline()
if entry, mlstErr := s.GetEntry(p); mlstErr == nil {
switch entry.Type {
case ftp.EntryTypeFolder:
return true, 0, nil, nil
case ftp.EntryTypeFile:
if !entry.Time.IsZero() {
t := entry.Time
mtime = &t
}
return false, int64(entry.Size), mtime, nil
}
// A link: SIZE below follows it on the server.
}
// MLST is missing or refused the path: the older commands decide. A
// broken connection fails them too, with its own error.
s.deadline()
size, sizeErr := s.FileSize(p)
if sizeErr == nil {
if s.IsGetTimeSupported() {
s.deadline()
if t, err := s.GetTime(p); err == nil {
mtime = &t
}
}
return false, size, mtime, nil
}
if replyCode(sizeErr) == 0 {
return false, 0, nil, sizeErr
}
// SIZE failed: a directory, or a missing path.
s.deadline()
cwd, err := s.CurrentDir()
if err != nil {
return false, 0, nil, err
}
s.deadline()
if err := s.ChangeDir(p); err != nil {
return false, 0, nil, sizeErr
}
s.deadline()
if err := s.ChangeDir(cwd); err != nil {
return false, 0, nil, err
}
return true, 0, nil, nil
}
// safeEntryName reports whether a name from a listing can become a local
// path element. A hostile server could otherwise list "../x" and write
// outside the task directory.
func safeEntryName(name string) bool {
if name == "" || name == "." || name == ".." {
return false
}
return !strings.ContainsAny(name, "/\\\x00")
}
// walk lists a directory tree, at most maxWalkDepth levels below root and
// maxWalkFiles files. Files come back sorted by path, then name.
func (f *Fetcher) walk(s *session, root string) ([]*base.FileInfo, error) {
type dir struct {
remote string // path on the server
rel string // path below root, "" for root itself
depth int
}
var files []*base.FileInfo
stack := []dir{{remote: strings.TrimSuffix(root, "/"), rel: "", depth: 0}}
for len(stack) > 0 {
d := stack[len(stack)-1]
stack = stack[:len(stack)-1]
s.deadline()
entries, err := s.List(d.remote)
if err != nil {
return nil, err
}
for _, e := range entries {
if e.Name == "." || e.Name == ".." {
continue
}
if !safeEntryName(e.Name) {
f.warn("ftp: skipping a listed name that is not a safe file name: %q", e.Name)
continue
}
remote := path.Join(d.remote, e.Name)
rel := path.Join(d.rel, e.Name)
if d.remote == "" {
remote = e.Name
}
entryType, size := e.Type, int64(e.Size)
if entryType == ftp.EntryTypeLink {
// A link to a file has a size; a link to a directory is
// not followed, so a loop cannot make the walk endless.
s.deadline()
linkSize, err := s.FileSize(remote)
if err != nil {
f.warn("ftp: skipping link %q: %v", rel, err)
continue
}
entryType, size = ftp.EntryTypeFile, linkSize
}
switch entryType {
case ftp.EntryTypeFolder:
if d.depth+1 > maxWalkDepth {
return nil, fmt.Errorf("ftp: the folder is deeper than %d levels, the limit for a folder download", maxWalkDepth)
}
stack = append(stack, dir{remote: remote, rel: rel, depth: d.depth + 1})
case ftp.EntryTypeFile:
if len(files) >= maxWalkFiles {
return nil, fmt.Errorf("ftp: the folder has more than %d files, the limit for a folder download", maxWalkFiles)
}
fi := &base.FileInfo{Name: e.Name, Path: d.rel, Size: size}
if !e.Time.IsZero() {
t := e.Time
fi.Ctime = &t
}
files = append(files, fi)
}
}
}
slices.SortFunc(files, func(a, b *base.FileInfo) int {
return cmp.Or(strings.Compare(a.Path, b.Path), strings.Compare(a.Name, b.Name))
})
return files, nil
}