f-denkena/rclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

serve: refactor VFS and proxy handling into Provider

Browse Source
This commit is contained in:
Hakan İSMAİL
2026-08-18 09:03:12 +01:00
committed by Nick Craig-Wood
parent a1f98b49df
commit f425f8d466
6 changed files with 200 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/serve/ftp/ftp.go
+20 -15
View File
@@ -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)
}
cmd/serve/http/http.go
+20 -29
View File
@@ -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
cmd/serve/proxy/proxy.go
+113 -1
View File
@@ -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
+16 -15
View File
@@ -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 {
cmd/serve/sftp/server.go
+16 -19
View File
@@ -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
}
cmd/serve/webdav/webdav.go
+15 -22
View File
@@ -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