serve: refactor VFS and proxy handling into Provider

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
+	provider   *proxy.Provider
-	proxy      *proxy.Proxy // may be nil if not in use
 	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,
+		f:        f,
-		ctx: ctx,
+		ctx:      ctx,
-		opt: *opt,
+		opt:      *opt,
+		provider: proxy.NewProvider(ctx, f, vfsOpt, proxyOpt),
 	}
-	if proxy.Opt.AuthProxy != "" {
+	defer func() {
-		d.proxy = proxy.New(ctx, proxyOpt, vfsOpt)
+		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 {
+	if d.provider.IsProxy() {
-		_, _, err = d.proxy.Call(user, pass, false, sctx.Sess.RemoteAddr().String())
+		_, _, 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)
 	}

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
+	f        fs.Fs
-	_vfs   *vfs.VFS // don't use directly, use getVFS
+	provider *proxy.Provider
-	server *libhttp.Server
+	server   *libhttp.Server
-	opt    Options
+	opt      Options
-	proxy  *proxy.Proxy
+	ctx      context.Context // for global config
-	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.provider.Get(ctx)
-		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
 }
 
 // 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,
+		f:        f,
-		ctx: ctx,
+		ctx:      ctx,
-		opt: *opt,
+		opt:      *opt,
+		provider: proxy.NewProvider(ctx, f, vfsOpt, proxyOpt),
 	}
+	defer func() {
+		if err != nil {
+			s.provider.Shutdown()
+		}
+	}()
 
-	if proxyOpt.AuthProxy != "" {
+	if s.provider.IsProxy() {
-		s.proxy = proxy.New(ctx, proxyOpt, vfsOpt)
-		// override auth
 		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

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)
+	}
+}

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 != "" {
+	if w.provider.IsProxy() {
-		w.proxy = proxy.New(ctx, proxyOpt, vfsOpt)
 		// proxy auth middleware
 		w.handler = proxyAuthMiddleware(w.handler, w)
 		w.handler = authPairMiddleware(w.handler, w)
-	} else {
+	} else if len(opt.AuthKey) > 0 {
-		w._vfs = vfs.New(ctx, f, vfsOpt)
+		w.faker.AddAuthKeys(authList)
-
-		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 {
+	if w.provider.VFS() != nil {
-		return w._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()
+	err := w.server.Shutdown()
-	return w.server.Shutdown()
+	w.provider.Shutdown()
+	return err
 }
 
 func authPairMiddleware(next http.Handler, ws *Server) http.Handler {

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,
+		f:        f,
-		ctx:     ctx,
+		ctx:      ctx,
-		opt:     *opt,
+		opt:      *opt,
-		stopped: make(chan struct{}),
+		provider: proxy.NewProvider(ctx, f, vfsOpt, proxyOpt),
-	}
+		stopped:  make(chan struct{}),
-	if proxy.Opt.AuthProxy != "" {
-		s.proxy = proxy.New(ctx, proxyOpt, vfsOpt)
-	} else {
-		s.vfs = vfs.New(ctx, f, vfsOpt)
 	}
 	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 {
+	if !s.provider.IsProxy() {
-		return s.vfs
+		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
 }

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)
+	if w.provider.IsProxy() {
-		// override auth
 		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.provider.Get(ctx)
-		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
 }
 
 // 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