diff --git a/cmd/serve/ftp/ftp.go b/cmd/serve/ftp/ftp.go index d556ccca3..83a917761 100644 --- a/cmd/serve/ftp/ftp.go +++ b/cmd/serve/ftp/ftp.go @@ -170,8 +170,7 @@ type driver struct { srv *ftp.Server ctx context.Context // for global config opt Options - globalVFS *vfs.VFS // the VFS if not using auth proxy - proxy *proxy.Proxy // may be nil if not in use + provider *proxy.Provider useTLS bool userPassMu sync.Mutex // to protect userPass userPass map[string]string // cache of username => password when using vfs proxy @@ -195,15 +194,19 @@ func newServer(ctx context.Context, f fs.Fs, opt *Options, vfsOpt *vfscommon.Opt } d := &driver{ - f: f, - ctx: ctx, - opt: *opt, + f: f, + ctx: ctx, + opt: *opt, + provider: proxy.NewProvider(ctx, f, vfsOpt, proxyOpt), } - if proxy.Opt.AuthProxy != "" { - d.proxy = proxy.New(ctx, proxyOpt, vfsOpt) + defer func() { + if err != nil { + d.provider.Shutdown() + } + }() + + if d.provider.IsProxy() { d.userPass = make(map[string]string, 16) - } else { - d.globalVFS = vfs.New(ctx, f, vfsOpt) } d.useTLS = d.opt.TLSKey != "" @@ -250,7 +253,9 @@ func (d *driver) Serve() error { //lint:ignore U1000 unused when not building linux func (d *driver) Shutdown() error { fs.Logf(d.f, "Stopping FTP on %s", d.srv.Hostname+":"+strconv.Itoa(d.srv.Port)) - return d.srv.Shutdown() + err := d.srv.Shutdown() + d.provider.Shutdown() + return err } // Return the first address of the server @@ -316,8 +321,8 @@ func (l *Logger) PrintResponse(sessionID string, code int, message string) { // CheckPasswd handle auth based on configuration func (d *driver) CheckPasswd(sctx *ftp.Context, user, pass string) (ok bool, err error) { - if d.proxy != nil { - _, _, err = d.proxy.Call(user, pass, false, sctx.Sess.RemoteAddr().String()) + if d.provider.IsProxy() { + _, _, err = d.provider.Proxy().Call(user, pass, false, sctx.Sess.RemoteAddr().String()) if err != nil { fs.Infof(nil, "proxy login failed: %v", err) return false, nil @@ -351,9 +356,9 @@ func (d *driver) CheckPasswd(sctx *ftp.Context, user, pass string) (ok bool, err // Get the VFS for this connection func (d *driver) getVFS(sctx *ftp.Context) (VFS *vfs.VFS, err error) { - if d.proxy == nil { + if !d.provider.IsProxy() { // If no proxy always use the same VFS - return d.globalVFS, nil + return d.provider.VFS(), nil } user := sctx.Sess.LoginUser() d.userPassMu.Lock() @@ -366,7 +371,7 @@ func (d *driver) getVFS(sctx *ftp.Context) (VFS *vfs.VFS, err error) { if err != nil { return nil, err } - VFS, _, err = d.proxy.Call(user, pass, false, sctx.Sess.RemoteAddr().String()) + VFS, _, err = d.provider.Proxy().Call(user, pass, false, sctx.Sess.RemoteAddr().String()) if err != nil { return nil, fmt.Errorf("proxy login failed: %w", err) } diff --git a/cmd/serve/http/http.go b/cmd/serve/http/http.go index 776e96a29..e867928bf 100644 --- a/cmd/serve/http/http.go +++ b/cmd/serve/http/http.go @@ -4,7 +4,6 @@ package http import ( "context" _ "embed" - "errors" "fmt" "io" "net" @@ -150,33 +149,21 @@ control the stats printing. // HTTP contains everything to run the server type HTTP struct { - f fs.Fs - _vfs *vfs.VFS // don't use directly, use getVFS - server *libhttp.Server - opt Options - proxy *proxy.Proxy - ctx context.Context // for global config + f fs.Fs + provider *proxy.Provider + server *libhttp.Server + opt Options + ctx context.Context // for global config } // Gets the VFS in use for this request func (s *HTTP) getVFS(ctx context.Context) (VFS *vfs.VFS, err error) { - if s._vfs != nil { - return s._vfs, nil - } - value := libhttp.CtxGetAuth(ctx) - if value == nil { - return nil, errors.New("no VFS found in context") - } - VFS, ok := value.(*vfs.VFS) - if !ok { - return nil, fmt.Errorf("context value is not VFS: %#v", value) - } - return VFS, nil + return s.provider.Get(ctx) } // auth does proxy authorization func (s *HTTP) auth(r *http.Request, user, pass string) (value any, err error) { - VFS, _, err := s.proxy.Call(user, pass, false, r.RemoteAddr) + VFS, _, err := s.provider.Proxy().Call(user, pass, false, r.RemoteAddr) if err != nil { return nil, err } @@ -185,17 +172,19 @@ func (s *HTTP) auth(r *http.Request, user, pass string) (value any, err error) { func newServer(ctx context.Context, f fs.Fs, opt *Options, vfsOpt *vfscommon.Options, proxyOpt *proxy.Options) (s *HTTP, err error) { s = &HTTP{ - f: f, - ctx: ctx, - opt: *opt, + f: f, + ctx: ctx, + opt: *opt, + provider: proxy.NewProvider(ctx, f, vfsOpt, proxyOpt), } + defer func() { + if err != nil { + s.provider.Shutdown() + } + }() - if proxyOpt.AuthProxy != "" { - s.proxy = proxy.New(ctx, proxyOpt, vfsOpt) - // override auth + if s.provider.IsProxy() { s.opt.Auth.CustomAuthFn = s.auth - } else { - s._vfs = vfs.New(ctx, f, vfsOpt) } s.server, err = libhttp.NewServer(ctx, @@ -235,7 +224,9 @@ func (s *HTTP) Addr() net.Addr { // Shutdown the server func (s *HTTP) Shutdown() error { - return s.server.Shutdown() + err := s.server.Shutdown() + s.provider.Shutdown() + return err } // serveFavicon serves the remote's favicon.ico if it exists, otherwise diff --git a/cmd/serve/proxy/proxy.go b/cmd/serve/proxy/proxy.go index 25fe679fd..29532bf04 100644 --- a/cmd/serve/proxy/proxy.go +++ b/cmd/serve/proxy/proxy.go @@ -22,6 +22,7 @@ import ( "github.com/rclone/rclone/fs/config/configmap" "github.com/rclone/rclone/fs/config/obscure" libcache "github.com/rclone/rclone/lib/cache" + libhttp "github.com/rclone/rclone/lib/http" "github.com/rclone/rclone/vfs" "github.com/rclone/rclone/vfs/vfscommon" ) @@ -161,13 +162,19 @@ type cacheEntry struct { // // Any VFS are created with the vfsOpt passed in. func New(ctx context.Context, opt *Options, vfsOpt *vfscommon.Options) *Proxy { - return &Proxy{ + p := &Proxy{ ctx: ctx, Opt: *opt, cmdLine: strings.Fields(opt.AuthProxy), vfsCache: libcache.New(), vfsOpt: *vfsOpt, } + p.vfsCache.SetFinalizer(func(value any) { + if entry, ok := value.(cacheEntry); ok && entry.vfs != nil { + entry.vfs.Shutdown() + } + }) + return p } // run the proxy command returning a config map @@ -374,3 +381,108 @@ func (p *Proxy) Get(key string) *vfs.VFS { entry := value.(cacheEntry) return entry.vfs } + +// Shutdown shuts down all cached VFS instances +func (p *Proxy) Shutdown() { + if p != nil && p.vfsCache != nil { + p.vfsCache.Clear() + } +} + +// Provider hands out VFS instances, either a fixed one or per-user via an auth proxy. +type Provider struct { + vfs *vfs.VFS // set if not using an auth proxy + proxy *Proxy // set if using an auth proxy +} + +// NewProvider creates a Provider. If proxyOpt.AuthProxy is set it creates an +// auth proxy; otherwise it creates a fixed VFS from f and vfsOpt. +func NewProvider(ctx context.Context, f fs.Fs, vfsOpt *vfscommon.Options, proxyOpt *Options) *Provider { + p := &Provider{} + if proxyOpt != nil && proxyOpt.AuthProxy != "" { + p.proxy = New(ctx, proxyOpt, vfsOpt) + } else { + p.vfs = vfs.New(ctx, f, vfsOpt) + } + return p +} + +// Get returns the VFS for the current request context. +// For fixed-VFS providers it returns the single VFS. +// For proxy providers it reads the VFS from the request's auth context. +func (p *Provider) Get(ctx context.Context) (*vfs.VFS, error) { + if p.vfs != nil { + return p.vfs, nil + } + value := libhttp.CtxGetAuth(ctx) + if value == nil { + return nil, errors.New("no VFS found in context") + } + VFS, ok := value.(*vfs.VFS) + if !ok { + return nil, fmt.Errorf("context value is not VFS: %#v", value) + } + return VFS, nil +} + +// VFS returns the fixed VFS, or nil if using an auth proxy. +func (p *Provider) VFS() *vfs.VFS { + if p == nil { + return nil + } + return p.vfs +} + +// Proxy returns the Proxy instance, or nil if not using an auth proxy. +func (p *Provider) Proxy() *Proxy { + if p == nil { + return nil + } + return p.proxy +} + +// IsProxy returns true if using an auth proxy. +func (p *Provider) IsProxy() bool { + return p != nil && p.proxy != nil +} + +// Shutdown tears down the provider: shuts down the fixed VFS or flushes all proxy cache entries. +func (p *Provider) Shutdown() { + if p == nil { + return + } + if p.vfs != nil { + p.vfs.Shutdown() + } + if p.proxy != nil { + p.proxy.Shutdown() + } +} + +// Pin pins the cache entry for key so it won't be evicted by expire +func (p *Proxy) Pin(key string) { + if p != nil && p.vfsCache != nil { + p.vfsCache.Pin(key) + } +} + +// Unpin unpins the cache entry for key +func (p *Proxy) Unpin(key string) { + if p != nil && p.vfsCache != nil { + p.vfsCache.Unpin(key) + } +} + +// Pin pins the cache entry for key if using an auth proxy +func (p *Provider) Pin(key string) { + if p != nil && p.proxy != nil { + p.proxy.Pin(key) + } +} + +// Unpin unpins the cache entry for key if using an auth proxy +func (p *Provider) Unpin(key string) { + if p != nil && p.proxy != nil { + p.proxy.Unpin(key) + } +} diff --git a/cmd/serve/s3/server.go b/cmd/serve/s3/server.go index 9043d7716..1f10839cc 100644 --- a/cmd/serve/s3/server.go +++ b/cmd/serve/s3/server.go @@ -35,11 +35,10 @@ type Server struct { server *httplib.Server opt Options f fs.Fs - _vfs *vfs.VFS // don't use directly, use getVFS + provider *proxy.Provider faker *gofakes3.GoFakeS3 backend *s3Backend handler http.Handler - proxy *proxy.Proxy ctx context.Context // for global config s3Secret string etagHashType hash.Type @@ -51,8 +50,14 @@ func newServer(ctx context.Context, f fs.Fs, opt *Options, vfsOpt *vfscommon.Opt f: f, ctx: ctx, opt: *opt, + provider: proxy.NewProvider(ctx, f, vfsOpt, proxyOpt), etagHashType: hash.None, } + defer func() { + if err != nil { + w.provider.Shutdown() + } + }() if w.opt.EtagHash == "auto" { w.etagHashType = f.Hashes().GetOne() @@ -95,17 +100,12 @@ func newServer(ctx context.Context, f fs.Fs, opt *Options, vfsOpt *vfscommon.Opt w.handler = w.faker.Server() - if proxy.Opt.AuthProxy != "" { - w.proxy = proxy.New(ctx, proxyOpt, vfsOpt) + if w.provider.IsProxy() { // proxy auth middleware w.handler = proxyAuthMiddleware(w.handler, w) w.handler = authPairMiddleware(w.handler, w) - } else { - w._vfs = vfs.New(ctx, f, vfsOpt) - - if len(opt.AuthKey) > 0 { - w.faker.AddAuthKeys(authList) - } + } else if len(opt.AuthKey) > 0 { + w.faker.AddAuthKeys(authList) } w.server, err = httplib.NewServer(ctx, @@ -123,8 +123,8 @@ func newServer(ctx context.Context, f fs.Fs, opt *Options, vfsOpt *vfscommon.Opt } func (w *Server) getVFS(ctx context.Context) (VFS *vfs.VFS, err error) { - if w._vfs != nil { - return w._vfs, nil + if w.provider.VFS() != nil { + return w.provider.VFS(), nil } value := ctx.Value(ctxKeyID) @@ -141,7 +141,7 @@ func (w *Server) getVFS(ctx context.Context) (VFS *vfs.VFS, err error) { // auth does proxy authorization func (w *Server) auth(r *http.Request, accessKeyID string) (value any, err error) { - VFS, _, err := w.proxy.Call(stringToMd5Hash(accessKeyID), accessKeyID, false, r.RemoteAddr) + VFS, _, err := w.provider.Proxy().Call(stringToMd5Hash(accessKeyID), accessKeyID, false, r.RemoteAddr) if err != nil { return nil, err } @@ -168,8 +168,9 @@ func (w *Server) Addr() net.Addr { // Shutdown the server func (w *Server) Shutdown() error { - w.backend.stopReaper() - return w.server.Shutdown() + err := w.server.Shutdown() + w.provider.Shutdown() + return err } func authPairMiddleware(next http.Handler, ws *Server) http.Handler { diff --git a/cmd/serve/sftp/server.go b/cmd/serve/sftp/server.go index b207ce2ce..7640fe261 100644 --- a/cmd/serve/sftp/server.go +++ b/cmd/serve/sftp/server.go @@ -37,28 +37,24 @@ import ( type server struct { f fs.Fs opt Options - vfs *vfs.VFS + provider *proxy.Provider ctx context.Context // for global config config *ssh.ServerConfig listener net.Listener stopped chan struct{} // for waiting on the listener to stop - proxy *proxy.Proxy } func newServer(ctx context.Context, f fs.Fs, opt *Options, vfsOpt *vfscommon.Options, proxyOpt *proxy.Options) (*server, error) { s := &server{ - f: f, - ctx: ctx, - opt: *opt, - stopped: make(chan struct{}), - } - if proxy.Opt.AuthProxy != "" { - s.proxy = proxy.New(ctx, proxyOpt, vfsOpt) - } else { - s.vfs = vfs.New(ctx, f, vfsOpt) + f: f, + ctx: ctx, + opt: *opt, + provider: proxy.NewProvider(ctx, f, vfsOpt, proxyOpt), + stopped: make(chan struct{}), } err := s.configure() if err != nil { + s.provider.Shutdown() return nil, fmt.Errorf("sftp configuration failed: %w", err) } return s, nil @@ -66,8 +62,8 @@ func newServer(ctx context.Context, f fs.Fs, opt *Options, vfsOpt *vfscommon.Opt // getVFS gets the vfs from s or the proxy func (s *server) getVFS(what string, sshConn *ssh.ServerConn) (VFS *vfs.VFS) { - if s.proxy == nil { - return s.vfs + if !s.provider.IsProxy() { + return s.provider.VFS() } if sshConn.Permissions == nil || sshConn.Permissions.Extensions == nil { fs.Infof(what, "SSH Permissions Extensions not found") @@ -78,7 +74,7 @@ func (s *server) getVFS(what string, sshConn *ssh.ServerConn) (VFS *vfs.VFS) { fs.Infof(what, "VFS key not found") return nil } - VFS = s.proxy.Get(key) + VFS = s.provider.Proxy().Get(key) if VFS == nil { fs.Infof(what, "failed to read VFS from cache") return nil @@ -160,7 +156,7 @@ func (s *server) configure() (err error) { fs.Logf(nil, "Loaded %d authorized keys from %q", len(authorizedKeysMap), authKeysFile) } - if !s.opt.NoAuth && len(authorizedKeysMap) == 0 && s.opt.User == "" && s.opt.Pass == "" && s.proxy == nil { + if !s.opt.NoAuth && len(authorizedKeysMap) == 0 && s.opt.User == "" && s.opt.Pass == "" && !s.provider.IsProxy() { return errors.New("no authorization found, use --user/--pass or --authorized-keys or --no-auth or --auth-proxy") } @@ -170,9 +166,9 @@ func (s *server) configure() (err error) { ServerVersion: "SSH-2.0-" + fs.GetConfig(s.ctx).UserAgent, PasswordCallback: func(c ssh.ConnMetadata, pass []byte) (*ssh.Permissions, error) { fs.Debugf(describeConn(c), "Password login attempt for %s", c.User()) - if s.proxy != nil { + if s.provider.IsProxy() { // query the proxy for the config - _, vfsKey, err := s.proxy.Call(c.User(), string(pass), false, c.RemoteAddr().String()) + _, vfsKey, err := s.provider.Proxy().Call(c.User(), string(pass), false, c.RemoteAddr().String()) if err != nil { return nil, err } @@ -193,9 +189,9 @@ func (s *server) configure() (err error) { }, PublicKeyCallback: func(c ssh.ConnMetadata, pubKey ssh.PublicKey) (*ssh.Permissions, error) { fs.Debugf(describeConn(c), "Public key login attempt for %s", c.User()) - if s.proxy != nil { + if s.provider.IsProxy() { //query the proxy for the config - _, vfsKey, err := s.proxy.Call( + _, vfsKey, err := s.provider.Proxy().Call( c.User(), base64.StdEncoding.EncodeToString(pubKey.Marshal()), true, @@ -327,6 +323,7 @@ func (s *server) Shutdown() error { if errors.Is(err, io.ErrUnexpectedEOF) { err = nil } + s.provider.Shutdown() s.Wait() return err } diff --git a/cmd/serve/webdav/webdav.go b/cmd/serve/webdav/webdav.go index 7011bef05..d3420bb40 100644 --- a/cmd/serve/webdav/webdav.go +++ b/cmd/serve/webdav/webdav.go @@ -4,7 +4,6 @@ package webdav import ( "context" "encoding/xml" - "errors" "fmt" "mime" "net" @@ -235,9 +234,8 @@ type WebDAV struct { server *libhttp.Server opt Options f fs.Fs - _vfs *vfs.VFS // don't use directly, use getVFS + provider *proxy.Provider webdavhandler *webdav.Handler - proxy *proxy.Proxy ctx context.Context // for global config etagHashType hash.Type } @@ -265,8 +263,15 @@ func newWebDAV(ctx context.Context, f fs.Fs, opt *Options, vfsOpt *vfscommon.Opt f: f, ctx: ctx, opt: *opt, + provider: proxy.NewProvider(ctx, f, vfsOpt, proxyOpt), etagHashType: hash.None, } + defer func() { + if err != nil { + w.provider.Shutdown() + } + }() + if opt.EtagHash == "auto" { w.etagHashType = f.Hashes().GetOne() } else if opt.EtagHash != "" { @@ -278,12 +283,9 @@ func newWebDAV(ctx context.Context, f fs.Fs, opt *Options, vfsOpt *vfscommon.Opt if w.etagHashType != hash.None { fs.Debugf(f, "Using hash %v for ETag", w.etagHashType) } - if proxyOpt.AuthProxy != "" { - w.proxy = proxy.New(ctx, proxyOpt, vfsOpt) - // override auth + + if w.provider.IsProxy() { w.opt.Auth.CustomAuthFn = w.auth - } else { - w._vfs = vfs.New(ctx, f, vfsOpt) } w.server, err = libhttp.NewServer(ctx, @@ -335,23 +337,12 @@ func newWebDAV(ctx context.Context, f fs.Fs, opt *Options, vfsOpt *vfscommon.Opt // Gets the VFS in use for this request func (w *WebDAV) getVFS(ctx context.Context) (VFS *vfs.VFS, err error) { - if w._vfs != nil { - return w._vfs, nil - } - value := libhttp.CtxGetAuth(ctx) - if value == nil { - return nil, errors.New("no VFS found in context") - } - VFS, ok := value.(*vfs.VFS) - if !ok { - return nil, fmt.Errorf("context value is not VFS: %#v", value) - } - return VFS, nil + return w.provider.Get(ctx) } // auth does proxy authorization func (w *WebDAV) auth(r *http.Request, user, pass string) (value any, err error) { - VFS, _, err := w.proxy.Call(user, pass, false, r.RemoteAddr) + VFS, _, err := w.provider.Proxy().Call(user, pass, false, r.RemoteAddr) if err != nil { return nil, err } @@ -513,7 +504,9 @@ func (w *WebDAV) Addr() net.Addr { // Shutdown the server func (w *WebDAV) Shutdown() error { - return w.server.Shutdown() + err := w.server.Shutdown() + w.provider.Shutdown() + return err } // logRequest is called by the webdav module on every request