+
diff --git a/templates/users.html b/templates/users.html
index d8b13e74..85d6fe08 100644
--- a/templates/users.html
+++ b/templates/users.html
@@ -260,7 +260,7 @@
}
],
"scrollX": false,
- "scrollY": "50vh",
+ "scrollY": false,
"responsive": true,
"order": [[1, 'asc']]
});
diff --git a/vfs/azblobfs.go b/vfs/azblobfs.go
index f8be3fa7..e3dd74e3 100644
--- a/vfs/azblobfs.go
+++ b/vfs/azblobfs.go
@@ -473,8 +473,8 @@ func (fs *AzureBlobFs) ReadDir(dirname string) ([]os.FileInfo, error) {
return result, nil
}
-// IsUploadResumeSupported returns true if upload resume is supported.
-// Upload Resume is not supported on Azure Blob
+// IsUploadResumeSupported returns true if resuming uploads is supported.
+// Resuming uploads is not supported on Azure Blob
func (*AzureBlobFs) IsUploadResumeSupported() bool {
return false
}
diff --git a/vfs/cryptfs.go b/vfs/cryptfs.go
index fb5bde1b..01309e02 100644
--- a/vfs/cryptfs.go
+++ b/vfs/cryptfs.go
@@ -172,7 +172,10 @@ func (fs *CryptFs) Create(name string, flag int) (File, *PipeWriter, func(), err
go func() {
n, err := sio.Encrypt(f, r, fs.getSIOConfig(key))
- f.Close()
+ errClose := f.Close()
+ if err == nil && errClose != nil {
+ err = errClose
+ }
r.CloseWithError(err) //nolint:errcheck
p.Done(err)
fsLog(fs, logger.LevelDebug, "upload completed, path: %#v, readed bytes: %v, err: %v", name, n, err)
diff --git a/vfs/filesystem.go b/vfs/filesystem.go
index cda9bcbb..5388ee3a 100644
--- a/vfs/filesystem.go
+++ b/vfs/filesystem.go
@@ -139,6 +139,7 @@ func (f *Filesystem) GetACopy() Filesystem {
PrivateKey: f.SFTPConfig.PrivateKey.Clone(),
Prefix: f.SFTPConfig.Prefix,
DisableCouncurrentReads: f.SFTPConfig.DisableCouncurrentReads,
+ BufferSize: f.SFTPConfig.BufferSize,
},
}
if len(f.SFTPConfig.Fingerprints) > 0 {
diff --git a/vfs/folder.go b/vfs/folder.go
index 54b6b815..c443b35c 100644
--- a/vfs/folder.go
+++ b/vfs/folder.go
@@ -183,7 +183,7 @@ func (v *VirtualFolder) GetFilesystem(connectionID string, forbiddenSelfUsers []
case CryptedFilesystemProvider:
return NewCryptFs(connectionID, v.MappedPath, v.VirtualPath, v.FsConfig.CryptConfig)
case SFTPFilesystemProvider:
- return NewSFTPFs(connectionID, v.VirtualPath, forbiddenSelfUsers, v.FsConfig.SFTPConfig)
+ return NewSFTPFs(connectionID, v.VirtualPath, v.MappedPath, forbiddenSelfUsers, v.FsConfig.SFTPConfig)
default:
return NewOsFs(connectionID, v.MappedPath, v.VirtualPath), nil
}
diff --git a/vfs/gcsfs.go b/vfs/gcsfs.go
index c4288546..e06bddeb 100644
--- a/vfs/gcsfs.go
+++ b/vfs/gcsfs.go
@@ -396,8 +396,8 @@ func (fs *GCSFs) ReadDir(dirname string) ([]os.FileInfo, error) {
return result, nil
}
-// IsUploadResumeSupported returns true if upload resume is supported.
-// SFTP Resume is not supported on S3
+// IsUploadResumeSupported returns true if resuming uploads is supported.
+// Resuming uploads is not supported on GCS
func (*GCSFs) IsUploadResumeSupported() bool {
return false
}
diff --git a/vfs/osfs.go b/vfs/osfs.go
index 2af7ba69..7ed75626 100644
--- a/vfs/osfs.go
+++ b/vfs/osfs.go
@@ -151,7 +151,7 @@ func (*OsFs) ReadDir(dirname string) ([]os.FileInfo, error) {
return list, nil
}
-// IsUploadResumeSupported returns true if upload resume is supported
+// IsUploadResumeSupported returns true if resuming uploads is supported
func (*OsFs) IsUploadResumeSupported() bool {
return true
}
diff --git a/vfs/s3fs.go b/vfs/s3fs.go
index 53701dad..4feb09da 100644
--- a/vfs/s3fs.go
+++ b/vfs/s3fs.go
@@ -415,8 +415,8 @@ func (fs *S3Fs) ReadDir(dirname string) ([]os.FileInfo, error) {
return result, err
}
-// IsUploadResumeSupported returns true if upload resume is supported.
-// SFTP Resume is not supported on S3
+// IsUploadResumeSupported returns true if resuming uploads is supported.
+// Resuming uploads is not supported on S3
func (*S3Fs) IsUploadResumeSupported() bool {
return false
}
diff --git a/vfs/sftpfs.go b/vfs/sftpfs.go
index e7feb456..44efbb80 100644
--- a/vfs/sftpfs.go
+++ b/vfs/sftpfs.go
@@ -1,6 +1,7 @@
package vfs
import (
+ "bufio"
"errors"
"fmt"
"io"
@@ -43,8 +44,14 @@ type SFTPFsConfig struct {
// Concurrent reads are safe to use and disabling them will degrade performance.
// Some servers automatically delete files once they are downloaded.
// Using concurrent reads is problematic with such servers.
- DisableCouncurrentReads bool `json:"disable_concurrent_reads,omitempty"`
- forbiddenSelfUsernames []string `json:"-"`
+ DisableCouncurrentReads bool `json:"disable_concurrent_reads,omitempty"`
+ // The buffer size (in MB) to use for transfers.
+ // Buffering could improve performance for high latency networks.
+ // With buffering enabled upload resume is not supported and a file
+ // cannot be opened for both reading and writing at the same time
+ // 0 means disabled.
+ BufferSize int64 `json:"buffer_size,omitempty"`
+ forbiddenSelfUsernames []string `json:"-"`
}
func (c *SFTPFsConfig) isEqual(other *SFTPFsConfig) bool {
@@ -60,6 +67,9 @@ func (c *SFTPFsConfig) isEqual(other *SFTPFsConfig) bool {
if c.DisableCouncurrentReads != other.DisableCouncurrentReads {
return false
}
+ if c.BufferSize != other.BufferSize {
+ return false
+ }
if len(c.Fingerprints) != len(other.Fingerprints) {
return false
}
@@ -98,6 +108,21 @@ func (c *SFTPFsConfig) Validate() error {
if c.Username == "" {
return errors.New("username cannot be empty")
}
+ if c.BufferSize < 0 || c.BufferSize > 64 {
+ return errors.New("invalid buffer_size, valid range is 0-64")
+ }
+ if err := c.validateCredentials(); err != nil {
+ return err
+ }
+ if c.Prefix != "" {
+ c.Prefix = utils.CleanPath(c.Prefix)
+ } else {
+ c.Prefix = "/"
+ }
+ return nil
+}
+
+func (c *SFTPFsConfig) validateCredentials() error {
if c.Password.IsEmpty() && c.PrivateKey.IsEmpty() {
return errors.New("credentials cannot be empty")
}
@@ -113,11 +138,6 @@ func (c *SFTPFsConfig) Validate() error {
if !c.PrivateKey.IsEmpty() && !c.PrivateKey.IsValidInput() {
return errors.New("invalid private key")
}
- if c.Prefix != "" {
- c.Prefix = utils.CleanPath(c.Prefix)
- } else {
- c.Prefix = "/"
- }
return nil
}
@@ -143,15 +163,19 @@ type SFTPFs struct {
sync.Mutex
connectionID string
// if not empty this fs is mouted as virtual folder in the specified path
- mountPath string
- config *SFTPFsConfig
- sshClient *ssh.Client
- sftpClient *sftp.Client
- err chan error
+ mountPath string
+ localTempDir string
+ config *SFTPFsConfig
+ sshClient *ssh.Client
+ sftpClient *sftp.Client
+ err chan error
}
// NewSFTPFs returns an SFTPFa object that allows to interact with an SFTP server
-func NewSFTPFs(connectionID, mountPath string, forbiddenSelfUsernames []string, config SFTPFsConfig) (Fs, error) {
+func NewSFTPFs(connectionID, mountPath, localTempDir string, forbiddenSelfUsernames []string, config SFTPFsConfig) (Fs, error) {
+ if localTempDir == "" {
+ localTempDir = filepath.Clean(os.TempDir())
+ }
if err := config.Validate(); err != nil {
return nil, err
}
@@ -169,6 +193,7 @@ func NewSFTPFs(connectionID, mountPath string, forbiddenSelfUsernames []string,
sftpFs := &SFTPFs{
connectionID: connectionID,
mountPath: mountPath,
+ localTempDir: localTempDir,
config: &config,
err: make(chan error, 1),
}
@@ -220,7 +245,32 @@ func (fs *SFTPFs) Open(name string, offset int64) (File, *pipeat.PipeReaderAt, f
return nil, nil, nil, err
}
f, err := fs.sftpClient.Open(name)
- return f, nil, nil, err
+ if fs.config.BufferSize == 0 {
+ return f, nil, nil, err
+ }
+ if offset > 0 {
+ _, err = f.Seek(offset, io.SeekStart)
+ if err != nil {
+ f.Close()
+ return nil, nil, nil, err
+ }
+ }
+ r, w, err := pipeat.PipeInDir(fs.localTempDir)
+ if err != nil {
+ f.Close()
+ return nil, nil, nil, err
+ }
+ go func() {
+ br := bufio.NewReaderSize(f, int(fs.config.BufferSize)*1024*1024)
+ // we don't use io.Copy since bufio.Reader implements io.ReadFrom and
+ // so it calls the sftp.File ReadFrom method without buffering
+ n, err := fs.copy(w, br)
+ w.CloseWithError(err) //nolint:errcheck
+ f.Close()
+ fsLog(fs, logger.LevelDebug, "download completed, path: %#v size: %v, err: %v", name, n, err)
+ }()
+
+ return nil, r, nil, nil
}
// Create creates or opens the named file for writing
@@ -229,13 +279,51 @@ func (fs *SFTPFs) Create(name string, flag int) (File, *PipeWriter, func(), erro
if err != nil {
return nil, nil, nil, err
}
- var f File
- if flag == 0 {
- f, err = fs.sftpClient.Create(name)
- } else {
- f, err = fs.sftpClient.OpenFile(name, flag)
+ if fs.config.BufferSize == 0 {
+ var f File
+ if flag == 0 {
+ f, err = fs.sftpClient.Create(name)
+ } else {
+ f, err = fs.sftpClient.OpenFile(name, flag)
+ }
+ return f, nil, nil, err
}
- return f, nil, nil, err
+ // buffering is enabled
+ f, err := fs.sftpClient.OpenFile(name, os.O_WRONLY|os.O_CREATE|os.O_TRUNC)
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ r, w, err := pipeat.PipeInDir(fs.localTempDir)
+ if err != nil {
+ f.Close()
+ return nil, nil, nil, err
+ }
+ p := NewPipeWriter(w)
+
+ go func() {
+ bw := bufio.NewWriterSize(f, int(fs.config.BufferSize)*1024*1024)
+ // we don't use io.Copy since bufio.Writer implements io.WriterTo and
+ // so it calls the sftp.File WriteTo method without buffering
+ n, err := fs.copy(bw, r)
+ errFlush := bw.Flush()
+ if err == nil && errFlush != nil {
+ err = errFlush
+ }
+ errClose := f.Close()
+ if err == nil && errClose != nil {
+ err = errClose
+ }
+ r.CloseWithError(err) //nolint:errcheck
+ var errTruncate error
+ if err != nil {
+ errTruncate = f.Truncate(n)
+ }
+ p.Done(err)
+ fsLog(fs, logger.LevelDebug, "upload completed, path: %#v, readed bytes: %v, err: %v err truncate: %v",
+ name, n, err, errTruncate)
+ }()
+
+ return nil, p, nil, nil
}
// Rename renames (moves) source to target.
@@ -340,14 +428,14 @@ func (fs *SFTPFs) ReadDir(dirname string) ([]os.FileInfo, error) {
return result, nil
}
-// IsUploadResumeSupported returns true if upload resume is supported.
-func (*SFTPFs) IsUploadResumeSupported() bool {
- return true
+// IsUploadResumeSupported returns true if resuming uploads is supported.
+func (fs *SFTPFs) IsUploadResumeSupported() bool {
+ return fs.config.BufferSize == 0
}
// IsAtomicUploadSupported returns true if atomic upload is supported.
-func (*SFTPFs) IsAtomicUploadSupported() bool {
- return true
+func (fs *SFTPFs) IsAtomicUploadSupported() bool {
+ return fs.config.BufferSize == 0
}
// IsNotExist returns a boolean indicating whether the error is known to
@@ -372,6 +460,11 @@ func (*SFTPFs) IsNotSupported(err error) bool {
// CheckRootPath creates the specified local root directory if it does not exists
func (fs *SFTPFs) CheckRootPath(username string, uid int, gid int) bool {
+ if fs.config.BufferSize > 0 {
+ // we need a local directory for temporary files
+ osFs := NewOsFs(fs.ConnectionID(), fs.localTempDir, "")
+ osFs.CheckRootPath(username, uid, gid)
+ }
if fs.config.Prefix == "/" {
return true
}
@@ -595,6 +688,38 @@ func (fs *SFTPFs) Close() error {
return sshErr
}
+func (fs *SFTPFs) copy(dst io.Writer, src io.Reader) (written int64, err error) {
+ buf := make([]byte, 32768)
+ for {
+ nr, er := src.Read(buf)
+ if nr > 0 {
+ nw, ew := dst.Write(buf[0:nr])
+ if nw < 0 || nr < nw {
+ nw = 0
+ if ew == nil {
+ ew = errors.New("invalid write")
+ }
+ }
+ written += int64(nw)
+ if ew != nil {
+ err = ew
+ break
+ }
+ if nr != nw {
+ err = io.ErrShortWrite
+ break
+ }
+ }
+ if er != nil {
+ if er != io.EOF {
+ err = er
+ }
+ break
+ }
+ }
+ return written, err
+}
+
func (fs *SFTPFs) checkConnection() error {
err := fs.closed()
if err == nil {
@@ -659,6 +784,11 @@ func (fs *SFTPFs) createConnection() error {
opt := sftp.UseConcurrentReads(false)
opt(fs.sftpClient) //nolint:errcheck
}
+ if fs.config.BufferSize > 0 {
+ fsLog(fs, logger.LevelDebug, "enabling concurrent writes")
+ opt := sftp.UseConcurrentWrites(true)
+ opt(fs.sftpClient) //nolint:errcheck
+ }
go fs.wait()
return nil
}
diff --git a/vfs/vfs.go b/vfs/vfs.go
index 91e4d035..9a7920a8 100644
--- a/vfs/vfs.go
+++ b/vfs/vfs.go
@@ -553,11 +553,42 @@ func IsSFTPFs(fs Fs) bool {
return strings.HasPrefix(fs.Name(), sftpFsName)
}
+// IsBufferedSFTPFs returns true if this is a buffered SFTP filesystem
+func IsBufferedSFTPFs(fs Fs) bool {
+ if !IsSFTPFs(fs) {
+ return false
+ }
+ return !fs.IsUploadResumeSupported()
+}
+
+// IsLocalOrUnbufferedSFTPFs returns true if fs is local or SFTP with no buffer
+func IsLocalOrUnbufferedSFTPFs(fs Fs) bool {
+ if IsLocalOsFs(fs) {
+ return true
+ }
+ if IsSFTPFs(fs) {
+ return fs.IsUploadResumeSupported()
+ }
+ return false
+}
+
// IsLocalOrSFTPFs returns true if fs is local or SFTP
func IsLocalOrSFTPFs(fs Fs) bool {
return IsLocalOsFs(fs) || IsSFTPFs(fs)
}
+// HasOpenRWSupport returns true if the fs can open a file
+// for reading and writing at the same time
+func HasOpenRWSupport(fs Fs) bool {
+ if IsLocalOsFs(fs) {
+ return true
+ }
+ if IsSFTPFs(fs) && fs.IsUploadResumeSupported() {
+ return true
+ }
+ return false
+}
+
// IsLocalOrCryptoFs returns true if fs is local or local encrypted
func IsLocalOrCryptoFs(fs Fs) bool {
return IsLocalOsFs(fs) || IsCryptOsFs(fs)
diff --git a/webdavd/handler.go b/webdavd/handler.go
index f8995bf9..68eb9af5 100644
--- a/webdavd/handler.go
+++ b/webdavd/handler.go
@@ -141,7 +141,7 @@ func (c *Connection) getFile(fs vfs.Fs, fsPath, virtualPath string) (webdav.File
// for cloud fs we open the file when we receive the first read to avoid to download the first part of
// the file if it was opened only to do a stat or a readdir and so it is not a real download
- if vfs.IsLocalOrSFTPFs(fs) {
+ if vfs.IsLocalOrUnbufferedSFTPFs(fs) {
file, r, cancelFn, err = fs.Open(fsPath, 0)
if err != nil {
c.Log(logger.LevelWarn, "could not open file %#v for reading: %+v", fsPath, err)
diff --git a/webdavd/internal_test.go b/webdavd/internal_test.go
index 569625c2..fd8626e9 100644
--- a/webdavd/internal_test.go
+++ b/webdavd/internal_test.go
@@ -278,7 +278,7 @@ func (fs *MockOsFs) Open(name string, offset int64) (vfs.File, *pipeat.PipeReade
return nil, fs.reader, nil, nil
}
-// IsUploadResumeSupported returns true if upload resume is supported
+// IsUploadResumeSupported returns true if resuming uploads is supported
func (*MockOsFs) IsUploadResumeSupported() bool {
return false
}
diff --git a/webdavd/webdavd_test.go b/webdavd/webdavd_test.go
index d4adffd6..85ddeda8 100644
--- a/webdavd/webdavd_test.go
+++ b/webdavd/webdavd_test.go
@@ -1114,9 +1114,9 @@ func TestQuotaLimits(t *testing.T) {
client := getWebDavClient(user, true, nil)
// test quota files
err = uploadFile(testFilePath, testFileName+".quota", testFileSize, client)
- assert.NoError(t, err)
+ assert.NoError(t, err, "username: %v", user.Username)
err = uploadFile(testFilePath, testFileName+".quota1", testFileSize, client)
- assert.Error(t, err)
+ assert.Error(t, err, "username: %v", user.Username)
err = client.Rename(testFileName+".quota", testFileName, false)
assert.NoError(t, err)
files, err := client.ReadDir("/")
@@ -1385,6 +1385,80 @@ func TestLoginInvalidFs(t *testing.T) {
assert.NoError(t, err)
}
+func TestSFTPBuffered(t *testing.T) {
+ u := getTestUser()
+ localUser, _, err := httpdtest.AddUser(u, http.StatusCreated)
+ assert.NoError(t, err)
+ u = getTestSFTPUser()
+ u.QuotaFiles = 1000
+ u.HomeDir = filepath.Join(os.TempDir(), u.Username)
+ u.FsConfig.SFTPConfig.BufferSize = 2
+ sftpUser, _, err := httpdtest.AddUser(u, http.StatusCreated)
+ assert.NoError(t, err)
+
+ client := getWebDavClient(sftpUser, true, nil)
+ assert.NoError(t, checkBasicFunc(client))
+ testFilePath := filepath.Join(homeBasePath, testFileName)
+ testFileSize := int64(65535)
+ expectedQuotaSize := testFileSize
+ expectedQuotaFiles := 1
+ err = createTestFile(testFilePath, testFileSize)
+ assert.NoError(t, err)
+ err = uploadFile(testFilePath, testFileName, testFileSize, client)
+ assert.NoError(t, err)
+ // overwrite an existing file
+ err = uploadFile(testFilePath, testFileName, testFileSize, client)
+ assert.NoError(t, err)
+ localDownloadPath := filepath.Join(homeBasePath, testDLFileName)
+ err = downloadFile(testFileName, localDownloadPath, testFileSize, client)
+ assert.NoError(t, err)
+
+ user, _, err := httpdtest.GetUserByUsername(sftpUser.Username, http.StatusOK)
+ assert.NoError(t, err)
+ assert.Equal(t, expectedQuotaFiles, user.UsedQuotaFiles)
+ assert.Equal(t, expectedQuotaSize, user.UsedQuotaSize)
+
+ fileContent := []byte("test file contents")
+ err = os.WriteFile(testFilePath, fileContent, os.ModePerm)
+ assert.NoError(t, err)
+ err = uploadFile(testFilePath, testFileName, int64(len(fileContent)), client)
+ assert.NoError(t, err)
+ remotePath := fmt.Sprintf("http://%v/%v", webDavServerAddr, testFileName)
+ req, err := http.NewRequest(http.MethodGet, remotePath, nil)
+ assert.NoError(t, err)
+ httpClient := httpclient.GetHTTPClient()
+ req.SetBasicAuth(user.Username, defaultPassword)
+ req.Header.Set("Range", "bytes=5-")
+ resp, err := httpClient.Do(req)
+ if assert.NoError(t, err) {
+ defer resp.Body.Close()
+ assert.Equal(t, http.StatusPartialContent, resp.StatusCode)
+ bodyBytes, err := io.ReadAll(resp.Body)
+ assert.NoError(t, err)
+ assert.Equal(t, "file contents", string(bodyBytes))
+ }
+ req.Header.Set("Range", "bytes=5-8")
+ resp, err = httpClient.Do(req)
+ if assert.NoError(t, err) {
+ defer resp.Body.Close()
+ assert.Equal(t, http.StatusPartialContent, resp.StatusCode)
+ bodyBytes, err := io.ReadAll(resp.Body)
+ assert.NoError(t, err)
+ assert.Equal(t, "file", string(bodyBytes))
+ }
+
+ err = os.Remove(testFilePath)
+ assert.NoError(t, err)
+ _, err = httpdtest.RemoveUser(sftpUser, http.StatusOK)
+ assert.NoError(t, err)
+ _, err = httpdtest.RemoveUser(localUser, http.StatusOK)
+ assert.NoError(t, err)
+ err = os.RemoveAll(localUser.GetHomeDir())
+ assert.NoError(t, err)
+ err = os.RemoveAll(sftpUser.GetHomeDir())
+ assert.NoError(t, err)
+}
+
func TestBytesRangeRequests(t *testing.T) {
u := getTestUser()
u.Username = u.Username + "1"
@@ -1429,7 +1503,6 @@ func TestBytesRangeRequests(t *testing.T) {
}
}
- assert.NoError(t, err)
err = os.Remove(testFilePath)
assert.NoError(t, err)
_, err = httpdtest.RemoveUser(user, http.StatusOK)
@@ -2125,6 +2198,8 @@ func TestSFTPLoopVirtualFolders(t *testing.T) {
assert.NoError(t, err)
err = os.RemoveAll(user2.GetHomeDir())
assert.NoError(t, err)
+ _, err = httpdtest.RemoveFolder(vfs.BaseVirtualFolder{Name: "sftp"}, http.StatusOK)
+ assert.NoError(t, err)
}
func TestNestedVirtualFolders(t *testing.T) {