@@ -461,15 +537,27 @@
if (val == '1'){
$('.form-group.row.gcs').hide();
$('.form-group.gcs').hide();
+ $('.form-group.row.azblob').hide();
+ $('.form-group.azblob').hide();
$('.form-group.row.s3').show();
} else if (val == '2'){
$('.form-group.row.gcs').show();
$('.form-group.gcs').show();
+ $('.form-group.row.azblob').hide();
+ $('.form-group.azblob').hide();
+ $('.form-group.row.s3').hide();
+ } else if (val == '3'){
+ $('.form-group.row.azblob').show();
+ $('.form-group.azblob').show();
+ $('.form-group.row.gcs').hide();
+ $('.form-group.gcs').hide();
$('.form-group.row.s3').hide();
} else {
$('.form-group.row.gcs').hide();
$('.form-group.gcs').hide();
$('.form-group.row.s3').hide();
+ $('.form-group.row.azblob').hide();
+ $('.form-group.azblob').hide();
}
}
diff --git a/vfs/azblobfs.go b/vfs/azblobfs.go
new file mode 100644
index 00000000..4828602a
--- /dev/null
+++ b/vfs/azblobfs.go
@@ -0,0 +1,724 @@
+// +build !noazblob
+
+package vfs
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "mime"
+ "net/http"
+ "net/url"
+ "os"
+ "path"
+ "path/filepath"
+ "strings"
+ "time"
+
+ "github.com/Azure/azure-storage-blob-go/azblob"
+ "github.com/eikenb/pipeat"
+
+ "github.com/drakkan/sftpgo/logger"
+ "github.com/drakkan/sftpgo/utils"
+ "github.com/drakkan/sftpgo/version"
+)
+
+const azureDefaultEndpoint = "blob.core.windows.net"
+
+// max time of an azure web request response window (whether or not data is flowing)
+// this is the same value used in rclone
+var maxTryTimeout = time.Hour * 24 * 365
+
+// AzureBlobFs is a Fs implementation for Azure Blob storage.
+type AzureBlobFs struct {
+ connectionID string
+ localTempDir string
+ config AzBlobFsConfig
+ svc *azblob.ServiceURL
+ containerURL azblob.ContainerURL
+ ctxTimeout time.Duration
+ ctxLongTimeout time.Duration
+}
+
+func init() {
+ version.AddFeature("+azblob")
+}
+
+// NewAzBlobFs returns an AzBlobFs object that allows to interact with Azure Blob storage
+func NewAzBlobFs(connectionID, localTempDir string, config AzBlobFsConfig) (Fs, error) {
+ fs := AzureBlobFs{
+ connectionID: connectionID,
+ localTempDir: localTempDir,
+ config: config,
+ ctxTimeout: 30 * time.Second,
+ ctxLongTimeout: 300 * time.Second,
+ }
+ if err := ValidateAzBlobFsConfig(&fs.config); err != nil {
+ return fs, err
+ }
+ if fs.config.AccountKey != "" {
+ accountKey, err := utils.DecryptData(fs.config.AccountKey)
+ if err != nil {
+ return fs, err
+ }
+ fs.config.AccountKey = accountKey
+ }
+ setConfigDefaults(&fs)
+
+ if fs.config.SASURL != "" {
+ u, err := url.Parse(fs.config.SASURL)
+ if err != nil {
+ return fs, fmt.Errorf("invalid credentials: %v", err)
+ }
+ pipeline := azblob.NewPipeline(azblob.NewAnonymousCredential(), azblob.PipelineOptions{
+ Retry: azblob.RetryOptions{
+ TryTimeout: maxTryTimeout,
+ },
+ Telemetry: azblob.TelemetryOptions{
+ Value: "SFTPGo",
+ },
+ })
+ // Check if we have container level SAS or account level SAS
+ parts := azblob.NewBlobURLParts(*u)
+ if parts.ContainerName != "" {
+ if fs.config.Container != "" && fs.config.Container != parts.ContainerName {
+ return fs, fmt.Errorf("Container name in SAS URL %#v and container provided %#v do not match",
+ parts.ContainerName, fs.config.Container)
+ }
+ fs.svc = nil
+ fs.containerURL = azblob.NewContainerURL(*u, pipeline)
+ } else {
+ if fs.config.Container == "" {
+ return fs, errors.New("container is required with this SAS URL")
+ }
+ serviceURL := azblob.NewServiceURL(*u, pipeline)
+ fs.svc = &serviceURL
+ fs.containerURL = fs.svc.NewContainerURL(fs.config.Container)
+ }
+ return fs, nil
+ }
+
+ credential, err := azblob.NewSharedKeyCredential(fs.config.AccountName, fs.config.AccountKey)
+ if err != nil {
+ return fs, fmt.Errorf("invalid credentials: %v", err)
+ }
+ var u *url.URL
+ if fs.config.UseEmulator {
+ // for the emulator we expect the endpoint prefixed with the protocol, for example:
+ // http://127.0.0.1:10000
+ u, err = url.Parse(fmt.Sprintf("%s/%s", fs.config.Endpoint, fs.config.AccountName))
+ } else {
+ u, err = url.Parse(fmt.Sprintf("https://%s.%s", fs.config.AccountName, fs.config.Endpoint))
+ }
+ if err != nil {
+ return fs, fmt.Errorf("invalid credentials: %v", err)
+ }
+ pipeline := azblob.NewPipeline(credential, azblob.PipelineOptions{
+ Retry: azblob.RetryOptions{
+ TryTimeout: maxTryTimeout,
+ },
+ Telemetry: azblob.TelemetryOptions{
+ Value: "SFTPGo",
+ },
+ })
+ serviceURL := azblob.NewServiceURL(*u, pipeline)
+ fs.svc = &serviceURL
+ fs.containerURL = fs.svc.NewContainerURL(fs.config.Container)
+ return fs, nil
+}
+
+func setConfigDefaults(fs *AzureBlobFs) {
+ if fs.config.Endpoint == "" {
+ fs.config.Endpoint = azureDefaultEndpoint
+ }
+ if fs.config.UploadPartSize == 0 {
+ fs.config.UploadPartSize = 4
+ }
+ fs.config.UploadPartSize *= 1024 * 1024
+ if fs.config.UploadConcurrency == 0 {
+ fs.config.UploadConcurrency = 2
+ }
+}
+
+// Name returns the name for the Fs implementation
+func (fs AzureBlobFs) Name() string {
+ if fs.config.SASURL != "" {
+ return fmt.Sprintf("Azure Blob SAS URL %#v", fs.config.Container)
+ }
+ return fmt.Sprintf("Azure Blob container %#v", fs.config.Container)
+}
+
+// ConnectionID returns the connection ID associated to this Fs implementation
+func (fs AzureBlobFs) ConnectionID() string {
+ return fs.connectionID
+}
+
+// Stat returns a FileInfo describing the named file
+func (fs AzureBlobFs) Stat(name string) (os.FileInfo, error) {
+ if name == "" || name == "." {
+ err := fs.checkIfBucketExists()
+ if err != nil {
+ return nil, err
+ }
+ return NewFileInfo(name, true, 0, time.Now(), false), nil
+ }
+ if fs.config.KeyPrefix == name+"/" {
+ return NewFileInfo(name, true, 0, time.Now(), false), nil
+ }
+ prefix := fs.getPrefixForStat(name)
+ ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
+ defer cancelFn()
+
+ for marker := (azblob.Marker{}); marker.NotDone(); {
+ listBlob, err := fs.containerURL.ListBlobsHierarchySegment(ctx, marker, "/", azblob.ListBlobsSegmentOptions{
+ Details: azblob.BlobListingDetails{
+ Copy: false,
+ Metadata: false,
+ Snapshots: false,
+ UncommittedBlobs: false,
+ Deleted: false,
+ },
+ Prefix: prefix,
+ })
+ if err != nil {
+ return nil, err
+ }
+ marker = listBlob.NextMarker
+ for _, blobPrefix := range listBlob.Segment.BlobPrefixes {
+ if fs.isEqual(blobPrefix.Name, name) {
+ return NewFileInfo(name, true, 0, time.Now(), false), nil
+ }
+ }
+ for _, blobInfo := range listBlob.Segment.BlobItems {
+ if fs.isEqual(blobInfo.Name, name) {
+ isDir := false
+ if blobInfo.Properties.ContentType != nil {
+ isDir = (*blobInfo.Properties.ContentType == dirMimeType)
+ }
+ size := int64(0)
+ if blobInfo.Properties.ContentLength != nil {
+ size = *blobInfo.Properties.ContentLength
+ }
+ return NewFileInfo(name, isDir, size, blobInfo.Properties.LastModified, false), nil
+ }
+ }
+ }
+
+ return nil, errors.New("404 no such file or directory")
+}
+
+// Lstat returns a FileInfo describing the named file
+func (fs AzureBlobFs) Lstat(name string) (os.FileInfo, error) {
+ return fs.Stat(name)
+}
+
+// Open opens the named file for reading
+func (fs AzureBlobFs) Open(name string, offset int64) (*os.File, *pipeat.PipeReaderAt, func(), error) {
+ r, w, err := pipeat.PipeInDir(fs.localTempDir)
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ blobBlockURL := fs.containerURL.NewBlockBlobURL(name)
+ ctx, cancelFn := context.WithCancel(context.Background())
+ blobDownloadResponse, err := blobBlockURL.Download(ctx, offset, azblob.CountToEnd, azblob.BlobAccessConditions{}, false)
+ if err != nil {
+ r.Close()
+ w.Close()
+ cancelFn()
+ return nil, nil, nil, err
+ }
+ body := blobDownloadResponse.Body(azblob.RetryReaderOptions{
+ MaxRetryRequests: 3,
+ })
+
+ go func() {
+ defer cancelFn()
+ defer body.Close()
+
+ n, err := io.Copy(w, body)
+ w.CloseWithError(err) //nolint:errcheck
+ fsLog(fs, logger.LevelDebug, "download completed, path: %#v size: %v, err: %v", name, n, err)
+ }()
+
+ return nil, r, cancelFn, nil
+}
+
+// Create creates or opens the named file for writing
+func (fs AzureBlobFs) Create(name string, flag int) (*os.File, *PipeWriter, func(), error) {
+ r, w, err := pipeat.PipeInDir(fs.localTempDir)
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ p := NewPipeWriter(w)
+ blobBlockURL := fs.containerURL.NewBlockBlobURL(name)
+ ctx, cancelFn := context.WithCancel(context.Background())
+
+ headers := azblob.BlobHTTPHeaders{}
+ var contentType string
+ if flag == -1 {
+ contentType = dirMimeType
+ } else {
+ contentType = mime.TypeByExtension(path.Ext(name))
+ }
+ if contentType != "" {
+ headers.ContentType = contentType
+ }
+
+ go func() {
+ defer cancelFn()
+
+ uploadOptions := azblob.UploadStreamToBlockBlobOptions{
+ BufferSize: int(fs.config.UploadPartSize),
+ BlobHTTPHeaders: headers,
+ MaxBuffers: fs.config.UploadConcurrency,
+ }
+ _, err := azblob.UploadStreamToBlockBlob(ctx, r, blobBlockURL, uploadOptions)
+ r.CloseWithError(err) //nolint:errcheck
+ p.Done(err)
+ fsLog(fs, logger.LevelDebug, "upload completed, path: %#v, readed bytes: %v, err: %v", name, r.GetReadedBytes(), err)
+ }()
+
+ return nil, p, cancelFn, nil
+}
+
+// Rename renames (moves) source to target.
+// We don't support renaming non empty directories since we should
+// rename all the contents too and this could take long time: think
+// about directories with thousands of files, for each file we should
+// execute a StartCopyFromURL call.
+func (fs AzureBlobFs) Rename(source, target string) error {
+ if source == target {
+ return nil
+ }
+ fi, err := fs.Stat(source)
+ if err != nil {
+ return err
+ }
+ if fi.IsDir() {
+ contents, err := fs.ReadDir(source)
+ if err != nil {
+ return err
+ }
+ if len(contents) > 0 {
+ return fmt.Errorf("Cannot rename non empty directory: %#v", source)
+ }
+ }
+ dstBlobURL := fs.containerURL.NewBlobURL(target)
+ srcURL := fs.containerURL.NewBlobURL(source).URL()
+
+ md := azblob.Metadata{}
+ mac := azblob.ModifiedAccessConditions{}
+ bac := azblob.BlobAccessConditions{}
+ ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
+ defer cancelFn()
+
+ resp, err := dstBlobURL.StartCopyFromURL(ctx, srcURL, md, mac, bac)
+ if err != nil {
+ return err
+ }
+ copyStatus := resp.CopyStatus()
+ nErrors := 0
+ for copyStatus == azblob.CopyStatusPending {
+ // Poll until the copy is complete.
+ time.Sleep(500 * time.Millisecond)
+ propertiesResp, err := dstBlobURL.GetProperties(ctx, azblob.BlobAccessConditions{})
+ if err != nil {
+ // A GetProperties failure may be transient, so allow a couple
+ // of them before giving up.
+ nErrors++
+ if ctx.Err() != nil || nErrors == 3 {
+ return err
+ }
+ } else {
+ copyStatus = propertiesResp.CopyStatus()
+ }
+ }
+ if copyStatus != azblob.CopyStatusSuccess {
+ return fmt.Errorf("Copy failed with status: %s", copyStatus)
+ }
+ return fs.Remove(source, fi.IsDir())
+}
+
+// Remove removes the named file or (empty) directory.
+func (fs AzureBlobFs) Remove(name string, isDir bool) error {
+ if isDir {
+ contents, err := fs.ReadDir(name)
+ if err != nil {
+ return err
+ }
+ if len(contents) > 0 {
+ return fmt.Errorf("Cannot remove non empty directory: %#v", name)
+ }
+ }
+ blobBlockURL := fs.containerURL.NewBlockBlobURL(name)
+ ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
+ defer cancelFn()
+
+ _, err := blobBlockURL.Delete(ctx, azblob.DeleteSnapshotsOptionNone, azblob.BlobAccessConditions{})
+ return err
+}
+
+// Mkdir creates a new directory with the specified name and default permissions
+func (fs AzureBlobFs) Mkdir(name string) error {
+ _, err := fs.Stat(name)
+ if !fs.IsNotExist(err) {
+ return err
+ }
+ _, w, _, err := fs.Create(name, -1)
+ if err != nil {
+ return err
+ }
+ return w.Close()
+}
+
+// Symlink creates source as a symbolic link to target.
+func (AzureBlobFs) Symlink(source, target string) error {
+ return errors.New("403 symlinks are not supported")
+}
+
+// Readlink returns the destination of the named symbolic link
+func (AzureBlobFs) Readlink(name string) (string, error) {
+ return "", errors.New("403 readlink is not supported")
+}
+
+// Chown changes the numeric uid and gid of the named file.
+// Silently ignored.
+func (AzureBlobFs) Chown(name string, uid int, gid int) error {
+ return nil
+}
+
+// Chmod changes the mode of the named file to mode.
+// Silently ignored.
+func (AzureBlobFs) Chmod(name string, mode os.FileMode) error {
+ return nil
+}
+
+// Chtimes changes the access and modification times of the named file.
+// Silently ignored.
+func (AzureBlobFs) Chtimes(name string, atime, mtime time.Time) error {
+ return errors.New("403 chtimes is not supported")
+}
+
+// Truncate changes the size of the named file.
+// Truncate by path is not supported, while truncating an opened
+// file is handled inside base transfer
+func (AzureBlobFs) Truncate(name string, size int64) error {
+ return errors.New("403 truncate is not supported")
+}
+
+// ReadDir reads the directory named by dirname and returns
+// a list of directory entries.
+func (fs AzureBlobFs) ReadDir(dirname string) ([]os.FileInfo, error) {
+ var result []os.FileInfo
+ // dirname must be already cleaned
+ prefix := ""
+ if dirname != "" && dirname != "." {
+ prefix = strings.TrimPrefix(dirname, "/")
+ if !strings.HasSuffix(prefix, "/") {
+ prefix += "/"
+ }
+ }
+
+ prefixes := make(map[string]bool)
+
+ for marker := (azblob.Marker{}); marker.NotDone(); {
+ ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
+ defer cancelFn()
+
+ listBlob, err := fs.containerURL.ListBlobsHierarchySegment(ctx, marker, "/", azblob.ListBlobsSegmentOptions{
+ Details: azblob.BlobListingDetails{
+ Copy: false,
+ Metadata: false,
+ Snapshots: false,
+ UncommittedBlobs: false,
+ Deleted: false,
+ },
+ Prefix: prefix,
+ })
+ if err != nil {
+ return nil, err
+ }
+ marker = listBlob.NextMarker
+ for _, blobPrefix := range listBlob.Segment.BlobPrefixes {
+ // we don't support prefixes == "/" this will be sent if a key starts with "/"
+ if blobPrefix.Name == "/" {
+ continue
+ }
+ name := strings.TrimPrefix(blobPrefix.Name, prefix)
+ result = append(result, NewFileInfo(name, true, 0, time.Now(), false))
+ prefixes[strings.TrimSuffix(name, "/")] = true
+ }
+ for _, blobInfo := range listBlob.Segment.BlobItems {
+ name := strings.TrimPrefix(blobInfo.Name, prefix)
+ size := int64(0)
+ if blobInfo.Properties.ContentLength != nil {
+ size = *blobInfo.Properties.ContentLength
+ }
+ isDir := false
+ if blobInfo.Properties.ContentType != nil {
+ isDir = (*blobInfo.Properties.ContentType == dirMimeType)
+ if isDir {
+ // check if the dir is already included, it will be sent as blob prefix if it contains at least one item
+ if _, ok := prefixes[name]; ok {
+ continue
+ }
+ }
+ }
+ result = append(result, NewFileInfo(name, isDir, size, blobInfo.Properties.LastModified, false))
+ }
+ }
+
+ return result, nil
+}
+
+// IsUploadResumeSupported returns true if upload resume is supported.
+// Upload Resume is not supported on Azure Blob
+func (AzureBlobFs) IsUploadResumeSupported() bool {
+ return false
+}
+
+// IsAtomicUploadSupported returns true if atomic upload is supported.
+// Azure Blob uploads are already atomic, we don't need to upload to a temporary
+// file
+func (AzureBlobFs) IsAtomicUploadSupported() bool {
+ return false
+}
+
+// IsNotExist returns a boolean indicating whether the error is known to
+// report that a file or directory does not exist
+func (AzureBlobFs) IsNotExist(err error) bool {
+ if err == nil {
+ return false
+ }
+
+ if storageErr, ok := err.(azblob.StorageError); ok {
+ if storageErr.Response().StatusCode == http.StatusNotFound { //nolint:bodyclose
+ return true
+ }
+ if storageErr.ServiceCode() == azblob.ServiceCodeContainerNotFound ||
+ storageErr.ServiceCode() == azblob.ServiceCodeBlobNotFound {
+ return true
+ }
+ }
+
+ return strings.Contains(err.Error(), "404")
+}
+
+// IsPermission returns a boolean indicating whether the error is known to
+// report that permission is denied.
+func (AzureBlobFs) IsPermission(err error) bool {
+ if err == nil {
+ return false
+ }
+ if storageErr, ok := err.(azblob.StorageError); ok {
+ code := storageErr.Response().StatusCode //nolint:bodyclose
+ if code == http.StatusForbidden || code == http.StatusUnauthorized {
+ return true
+ }
+ if storageErr.ServiceCode() == azblob.ServiceCodeInsufficientAccountPermissions ||
+ storageErr.ServiceCode() == azblob.ServiceCodeInvalidAuthenticationInfo ||
+ storageErr.ServiceCode() == azblob.ServiceCodeUnauthorizedBlobOverwrite {
+ return true
+ }
+ }
+ return strings.Contains(err.Error(), "403")
+}
+
+// CheckRootPath creates the specified local root directory if it does not exists
+func (fs AzureBlobFs) CheckRootPath(username string, uid int, gid int) bool {
+ // we need a local directory for temporary files
+ osFs := NewOsFs(fs.ConnectionID(), fs.localTempDir, nil)
+ return osFs.CheckRootPath(username, uid, gid)
+}
+
+// ScanRootDirContents returns the number of files contained in the bucket,
+// and their size
+func (fs AzureBlobFs) ScanRootDirContents() (int, int64, error) {
+ numFiles := 0
+ size := int64(0)
+
+ for marker := (azblob.Marker{}); marker.NotDone(); {
+ ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
+ defer cancelFn()
+
+ listBlob, err := fs.containerURL.ListBlobsFlatSegment(ctx, marker, azblob.ListBlobsSegmentOptions{
+ Details: azblob.BlobListingDetails{
+ Copy: false,
+ Metadata: false,
+ Snapshots: false,
+ UncommittedBlobs: false,
+ Deleted: false,
+ },
+ Prefix: fs.config.KeyPrefix,
+ })
+ if err != nil {
+ return numFiles, size, err
+ }
+ marker = listBlob.NextMarker
+ for _, blobInfo := range listBlob.Segment.BlobItems {
+ isDir := false
+ if blobInfo.Properties.ContentType != nil {
+ isDir = (*blobInfo.Properties.ContentType == dirMimeType)
+ }
+ blobSize := int64(0)
+ if blobInfo.Properties.ContentLength != nil {
+ blobSize = *blobInfo.Properties.ContentLength
+ }
+ if isDir && blobSize == 0 {
+ continue
+ }
+ numFiles++
+ size += blobSize
+ }
+ }
+
+ return numFiles, size, nil
+}
+
+// GetDirSize returns the number of files and the size for a folder
+// including any subfolders
+func (AzureBlobFs) GetDirSize(dirname string) (int, int64, error) {
+ return 0, 0, errUnsupported
+}
+
+// GetAtomicUploadPath returns the path to use for an atomic upload.
+// Azure Blob Storage uploads are already atomic, we never call this method
+func (AzureBlobFs) GetAtomicUploadPath(name string) string {
+ return ""
+}
+
+// GetRelativePath returns the path for a file relative to the user's home dir.
+// This is the path as seen by SFTPGo users
+func (fs AzureBlobFs) GetRelativePath(name string) string {
+ rel := path.Clean(name)
+ if rel == "." {
+ rel = ""
+ }
+ if !path.IsAbs(rel) {
+ rel = "/" + rel
+ }
+ if len(fs.config.KeyPrefix) > 0 {
+ if !strings.HasPrefix(rel, "/"+fs.config.KeyPrefix) {
+ rel = "/"
+ }
+ rel = path.Clean("/" + strings.TrimPrefix(rel, "/"+fs.config.KeyPrefix))
+ }
+ return rel
+}
+
+// Walk walks the file tree rooted at root, calling walkFn for each file or
+// directory in the tree, including root
+func (fs AzureBlobFs) Walk(root string, walkFn filepath.WalkFunc) error {
+ prefix := ""
+ if root != "" && root != "." {
+ prefix = strings.TrimPrefix(root, "/")
+ if !strings.HasSuffix(prefix, "/") {
+ prefix += "/"
+ }
+ }
+ for marker := (azblob.Marker{}); marker.NotDone(); {
+ ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
+ defer cancelFn()
+
+ listBlob, err := fs.containerURL.ListBlobsFlatSegment(ctx, marker, azblob.ListBlobsSegmentOptions{
+ Details: azblob.BlobListingDetails{
+ Copy: false,
+ Metadata: false,
+ Snapshots: false,
+ UncommittedBlobs: false,
+ Deleted: false,
+ },
+ Prefix: prefix,
+ })
+ if err != nil {
+ return err
+ }
+ marker = listBlob.NextMarker
+ for _, blobInfo := range listBlob.Segment.BlobItems {
+ isDir := false
+ if blobInfo.Properties.ContentType != nil {
+ isDir = (*blobInfo.Properties.ContentType == dirMimeType)
+ }
+ name := path.Clean(blobInfo.Name)
+ if len(name) == 0 {
+ continue
+ }
+ blobSize := int64(0)
+ if blobInfo.Properties.ContentLength != nil {
+ blobSize = *blobInfo.Properties.ContentLength
+ }
+ err = walkFn(blobInfo.Name, NewFileInfo(name, isDir, blobSize, blobInfo.Properties.LastModified, false), nil)
+ if err != nil {
+ return err
+ }
+ }
+ }
+
+ return walkFn(root, NewFileInfo(root, true, 0, time.Now(), false), nil)
+}
+
+// Join joins any number of path elements into a single path
+func (AzureBlobFs) Join(elem ...string) string {
+ return strings.TrimPrefix(path.Join(elem...), "/")
+}
+
+// HasVirtualFolders returns true if folders are emulated
+func (AzureBlobFs) HasVirtualFolders() bool {
+ return true
+}
+
+// ResolvePath returns the matching filesystem path for the specified sftp path
+func (fs AzureBlobFs) ResolvePath(virtualPath string) (string, error) {
+ if !path.IsAbs(virtualPath) {
+ virtualPath = path.Clean("/" + virtualPath)
+ }
+ return fs.Join(fs.config.KeyPrefix, strings.TrimPrefix(virtualPath, "/")), nil
+}
+
+// GetMimeType implements MimeTyper interface
+func (fs AzureBlobFs) GetMimeType(name string) (string, error) {
+ ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
+ defer cancelFn()
+
+ blobBlockURL := fs.containerURL.NewBlockBlobURL(name)
+ response, err := blobBlockURL.GetProperties(ctx, azblob.BlobAccessConditions{})
+ if err != nil {
+ return "", err
+ }
+ return response.ContentType(), nil
+}
+
+func (fs *AzureBlobFs) isEqual(key string, virtualName string) bool {
+ if key == virtualName {
+ return true
+ }
+ if key == virtualName+"/" {
+ return true
+ }
+ if key+"/" == virtualName {
+ return true
+ }
+ return false
+}
+
+func (fs *AzureBlobFs) checkIfBucketExists() error {
+ ctx, cancelFn := context.WithDeadline(context.Background(), time.Now().Add(fs.ctxTimeout))
+ defer cancelFn()
+
+ _, err := fs.containerURL.GetProperties(ctx, azblob.LeaseAccessConditions{})
+ return err
+}
+
+func (fs *AzureBlobFs) getPrefixForStat(name string) string {
+ prefix := path.Dir(name)
+ if prefix == "/" || prefix == "." || prefix == "" {
+ prefix = ""
+ } else {
+ prefix = strings.TrimPrefix(prefix, "/")
+ if !strings.HasSuffix(prefix, "/") {
+ prefix += "/"
+ }
+ }
+ return prefix
+}
diff --git a/vfs/azblobfs_disabled.go b/vfs/azblobfs_disabled.go
new file mode 100644
index 00000000..e262c481
--- /dev/null
+++ b/vfs/azblobfs_disabled.go
@@ -0,0 +1,18 @@
+// +build noazblob
+
+package vfs
+
+import (
+ "errors"
+
+ "github.com/drakkan/sftpgo/version"
+)
+
+func init() {
+ version.AddFeature("-azblob")
+}
+
+// NewAzBlobFs returns an error, Azure Blob storage is disabled
+func NewAzBlobFs(connectionID, localTempDir string, config AzBlobFsConfig) (Fs, error) {
+ return nil, errors.New("Azure Blob Storage disabled at build time")
+}
diff --git a/vfs/gcsfs.go b/vfs/gcsfs.go
index 64e0d403..88438f93 100644
--- a/vfs/gcsfs.go
+++ b/vfs/gcsfs.go
@@ -70,10 +70,10 @@ func NewGCSFs(connectionID, localTempDir string, config GCSFsConfig) (Fs, error)
// Name returns the name for the Fs implementation
func (fs GCSFs) Name() string {
- return fmt.Sprintf("GCSFs bucket: %#v", fs.config.Bucket)
+ return fmt.Sprintf("GCSFs bucket %#v", fs.config.Bucket)
}
-// ConnectionID returns the SSH connection ID associated to this Fs implementation
+// ConnectionID returns the connection ID associated to this Fs implementation
func (fs GCSFs) ConnectionID() string {
return fs.connectionID
}
@@ -82,7 +82,7 @@ func (fs GCSFs) ConnectionID() string {
func (fs GCSFs) Stat(name string) (os.FileInfo, error) {
var result FileInfo
var err error
- if len(name) == 0 || name == "." {
+ if name == "" || name == "." {
err := fs.checkIfBucketExists()
if err != nil {
return result, err
@@ -111,7 +111,7 @@ func (fs GCSFs) Stat(name string) (os.FileInfo, error) {
metrics.GCSListObjectsCompleted(err)
return result, err
}
- if len(attrs.Prefix) > 0 {
+ if attrs.Prefix != "" {
if fs.isEqual(attrs.Prefix, name) {
result = NewFileInfo(name, true, 0, time.Now(), false)
break
@@ -128,7 +128,7 @@ func (fs GCSFs) Stat(name string) (os.FileInfo, error) {
}
}
metrics.GCSListObjectsCompleted(nil)
- if len(result.Name()) == 0 {
+ if result.Name() == "" {
err = errors.New("404 no such file or directory")
}
return result, err
@@ -181,7 +181,12 @@ func (fs GCSFs) Create(name string, flag int) (*os.File, *PipeWriter, func(), er
obj := bkt.Object(name)
ctx, cancelFn := context.WithCancel(context.Background())
objectWriter := obj.NewWriter(ctx)
- contentType := mime.TypeByExtension(path.Ext(name))
+ var contentType string
+ if flag == -1 {
+ contentType = dirMimeType
+ } else {
+ contentType = mime.TypeByExtension(path.Ext(name))
+ }
if contentType != "" {
objectWriter.ObjectAttrs.ContentType = contentType
}
@@ -274,7 +279,7 @@ func (fs GCSFs) Mkdir(name string) error {
if !strings.HasSuffix(name, "/") {
name += "/"
}
- _, w, _, err := fs.Create(name, 0)
+ _, w, _, err := fs.Create(name, -1)
if err != nil {
return err
}
@@ -322,7 +327,7 @@ func (fs GCSFs) ReadDir(dirname string) ([]os.FileInfo, error) {
var result []os.FileInfo
// dirname must be already cleaned
prefix := ""
- if len(dirname) > 0 && dirname != "." {
+ if dirname != "" && dirname != "." {
prefix = strings.TrimPrefix(dirname, "/")
if !strings.HasSuffix(prefix, "/") {
prefix += "/"
@@ -346,7 +351,7 @@ func (fs GCSFs) ReadDir(dirname string) ([]os.FileInfo, error) {
metrics.GCSListObjectsCompleted(err)
return result, err
}
- if len(attrs.Prefix) > 0 {
+ if attrs.Prefix != "" {
name, _ := fs.resolve(attrs.Prefix, prefix)
result = append(result, NewFileInfo(name, true, 0, time.Now(), false))
} else {
@@ -442,6 +447,10 @@ func (fs GCSFs) ScanRootDirContents() (int, int64, error) {
if !attrs.Deleted.IsZero() {
continue
}
+ isDir := strings.HasSuffix(attrs.Name, "/")
+ if isDir && attrs.Size == 0 {
+ continue
+ }
numFiles++
size += attrs.Size
}
@@ -456,13 +465,13 @@ func (GCSFs) GetDirSize(dirname string) (int, int64, error) {
}
// GetAtomicUploadPath returns the path to use for an atomic upload.
-// S3 uploads are already atomic, we never call this method for S3
+// GCS uploads are already atomic, we never call this method for GCS
func (GCSFs) GetAtomicUploadPath(name string) string {
return ""
}
// GetRelativePath returns the path for a file relative to the user's home dir.
-// This is the path as seen by SFTP users
+// This is the path as seen by SFTPGo users
func (fs GCSFs) GetRelativePath(name string) string {
rel := path.Clean(name)
if rel == "." {
@@ -484,7 +493,7 @@ func (fs GCSFs) GetRelativePath(name string) string {
// directory in the tree, including root
func (fs GCSFs) Walk(root string, walkFn filepath.WalkFunc) error {
prefix := ""
- if len(root) > 0 && root != "." {
+ if root != "" && root != "." {
prefix = strings.TrimPrefix(root, "/")
if !strings.HasSuffix(prefix, "/") {
prefix += "/"
@@ -522,7 +531,7 @@ func (fs GCSFs) Walk(root string, walkFn filepath.WalkFunc) error {
}
err = walkFn(attrs.Name, NewFileInfo(name, isDir, attrs.Size, attrs.Updated, false), nil)
if err != nil {
- break
+ return err
}
}
@@ -541,12 +550,12 @@ func (GCSFs) HasVirtualFolders() bool {
return true
}
-// ResolvePath returns the matching filesystem path for the specified sftp path
-func (fs GCSFs) ResolvePath(sftpPath string) (string, error) {
- if !path.IsAbs(sftpPath) {
- sftpPath = path.Clean("/" + sftpPath)
+// ResolvePath returns the matching filesystem path for the specified virtual path
+func (fs GCSFs) ResolvePath(virtualPath string) (string, error) {
+ if !path.IsAbs(virtualPath) {
+ virtualPath = path.Clean("/" + virtualPath)
}
- return fs.Join(fs.config.KeyPrefix, strings.TrimPrefix(sftpPath, "/")), nil
+ return fs.Join(fs.config.KeyPrefix, strings.TrimPrefix(virtualPath, "/")), nil
}
func (fs *GCSFs) resolve(name string, prefix string) (string, bool) {
@@ -558,14 +567,14 @@ func (fs *GCSFs) resolve(name string, prefix string) (string, bool) {
return result, isDir
}
-func (fs *GCSFs) isEqual(key string, sftpName string) bool {
- if key == sftpName {
+func (fs *GCSFs) isEqual(key string, virtualName string) bool {
+ if key == virtualName {
return true
}
- if key == sftpName+"/" {
+ if key == virtualName+"/" {
return true
}
- if key+"/" == sftpName {
+ if key+"/" == virtualName {
return true
}
return false
@@ -582,7 +591,7 @@ func (fs *GCSFs) checkIfBucketExists() error {
func (fs *GCSFs) getPrefixForStat(name string) string {
prefix := path.Dir(name)
- if prefix == "/" || prefix == "." || len(prefix) == 0 {
+ if prefix == "/" || prefix == "." || prefix == "" {
prefix = ""
} else {
prefix = strings.TrimPrefix(prefix, "/")
diff --git a/vfs/s3fs.go b/vfs/s3fs.go
index 95bbba43..a606e599 100644
--- a/vfs/s3fs.go
+++ b/vfs/s3fs.go
@@ -6,6 +6,7 @@ import (
"context"
"errors"
"fmt"
+ "mime"
"os"
"path"
"path/filepath"
@@ -55,11 +56,11 @@ func NewS3Fs(connectionID, localTempDir string, config S3FsConfig) (Fs, error) {
}
awsConfig := aws.NewConfig()
- if len(fs.config.Region) > 0 {
+ if fs.config.Region != "" {
awsConfig.WithRegion(fs.config.Region)
}
- if len(fs.config.AccessSecret) > 0 {
+ if fs.config.AccessSecret != "" {
accessSecret, err := utils.DecryptData(fs.config.AccessSecret)
if err != nil {
return fs, err
@@ -68,7 +69,7 @@ func NewS3Fs(connectionID, localTempDir string, config S3FsConfig) (Fs, error) {
awsConfig.Credentials = credentials.NewStaticCredentials(fs.config.AccessKey, fs.config.AccessSecret, "")
}
- if len(fs.config.Endpoint) > 0 {
+ if fs.config.Endpoint != "" {
awsConfig.Endpoint = aws.String(fs.config.Endpoint)
awsConfig.S3ForcePathStyle = aws.Bool(true)
}
@@ -96,10 +97,10 @@ func NewS3Fs(connectionID, localTempDir string, config S3FsConfig) (Fs, error) {
// Name returns the name for the Fs implementation
func (fs S3Fs) Name() string {
- return fmt.Sprintf("S3Fs bucket: %#v", fs.config.Bucket)
+ return fmt.Sprintf("S3Fs bucket %#v", fs.config.Bucket)
}
-// ConnectionID returns the SSH connection ID associated to this Fs implementation
+// ConnectionID returns the connection ID associated to this Fs implementation
func (fs S3Fs) ConnectionID() string {
return fs.connectionID
}
@@ -151,7 +152,7 @@ func (fs S3Fs) Stat(name string) (os.FileInfo, error) {
return true
})
metrics.S3ListObjectsCompleted(err)
- if err == nil && len(result.Name()) == 0 {
+ if err == nil && result.Name() == "" {
err = errors.New("404 no such file or directory")
}
return result, err
@@ -201,11 +202,18 @@ func (fs S3Fs) Create(name string, flag int) (*os.File, *PipeWriter, func(), err
go func() {
defer cancelFn()
key := name
+ var contentType string
+ if flag == -1 {
+ contentType = dirMimeType
+ } else {
+ contentType = mime.TypeByExtension(path.Ext(name))
+ }
response, err := uploader.UploadWithContext(ctx, &s3manager.UploadInput{
- Bucket: aws.String(fs.config.Bucket),
- Key: aws.String(key),
- Body: r,
- StorageClass: utils.NilIfEmpty(fs.config.StorageClass),
+ Bucket: aws.String(fs.config.Bucket),
+ Key: aws.String(key),
+ Body: r,
+ StorageClass: utils.NilIfEmpty(fs.config.StorageClass),
+ ContentEncoding: utils.NilIfEmpty(contentType),
}, func(u *s3manager.Uploader) {
u.Concurrency = fs.config.UploadConcurrency
u.PartSize = fs.config.UploadPartSize
@@ -300,7 +308,7 @@ func (fs S3Fs) Mkdir(name string) error {
if !strings.HasSuffix(name, "/") {
name += "/"
}
- _, w, _, err := fs.Create(name, 0)
+ _, w, _, err := fs.Create(name, -1)
if err != nil {
return err
}
@@ -446,6 +454,10 @@ func (fs S3Fs) ScanRootDirContents() (int, int64, error) {
Prefix: aws.String(fs.config.KeyPrefix),
}, func(page *s3.ListObjectsV2Output, lastPage bool) bool {
for _, fileObject := range page.Contents {
+ isDir := strings.HasSuffix(*fileObject.Key, "/")
+ if isDir && *fileObject.Size == 0 {
+ continue
+ }
numFiles++
size += *fileObject.Size
}
@@ -468,7 +480,7 @@ func (S3Fs) GetAtomicUploadPath(name string) string {
}
// GetRelativePath returns the path for a file relative to the user's home dir.
-// This is the path as seen by SFTP users
+// This is the path as seen by SFTPGo users
func (fs S3Fs) GetRelativePath(name string) string {
rel := path.Clean(name)
if rel == "." {
@@ -533,12 +545,12 @@ func (S3Fs) HasVirtualFolders() bool {
return true
}
-// ResolvePath returns the matching filesystem path for the specified sftp path
-func (fs S3Fs) ResolvePath(sftpPath string) (string, error) {
- if !path.IsAbs(sftpPath) {
- sftpPath = path.Clean("/" + sftpPath)
+// ResolvePath returns the matching filesystem path for the specified virtual path
+func (fs S3Fs) ResolvePath(virtualPath string) (string, error) {
+ if !path.IsAbs(virtualPath) {
+ virtualPath = path.Clean("/" + virtualPath)
}
- return fs.Join("/", fs.config.KeyPrefix, sftpPath), nil
+ return fs.Join("/", fs.config.KeyPrefix, virtualPath), nil
}
func (fs *S3Fs) resolve(name *string, prefix string) (string, bool) {
@@ -555,14 +567,14 @@ func (fs *S3Fs) resolve(name *string, prefix string) (string, bool) {
return result, isDir
}
-func (fs *S3Fs) isEqual(s3Key *string, sftpName string) bool {
- if *s3Key == sftpName {
+func (fs *S3Fs) isEqual(s3Key *string, virtualName string) bool {
+ if *s3Key == virtualName {
return true
}
- if "/"+*s3Key == sftpName {
+ if "/"+*s3Key == virtualName {
return true
}
- if "/"+*s3Key == sftpName+"/" {
+ if "/"+*s3Key == virtualName+"/" {
return true
}
return false
diff --git a/vfs/vfs.go b/vfs/vfs.go
index 119fa2b1..38eec35d 100644
--- a/vfs/vfs.go
+++ b/vfs/vfs.go
@@ -4,6 +4,7 @@ package vfs
import (
"errors"
"fmt"
+ "net/url"
"os"
"path"
"path/filepath"
@@ -16,6 +17,8 @@ import (
"github.com/drakkan/sftpgo/logger"
)
+const dirMimeType = "inode/directory"
+
// Fs defines the interface for filesystem backends
type Fs interface {
Name() string
@@ -126,6 +129,41 @@ type GCSFsConfig struct {
StorageClass string `json:"storage_class,omitempty"`
}
+// AzBlobFsConfig defines the configuration for Azure Blob Storage based filesystem
+type AzBlobFsConfig struct {
+ Container string `json:"container,omitempty"`
+ // Storage Account Name, leave blank to use SAS URL
+ AccountName string `json:"account_name,omitempty"`
+ // Storage Account Key leave blank to use SAS URL.
+ // The access key is stored encrypted (AES-256-GCM)
+ AccountKey string `json:"account_key,omitempty"`
+ // Optional endpoint. Default is "blob.core.windows.net".
+ // If you use the emulator the endpoint must include the protocol,
+ // for example "http://127.0.0.1:10000"
+ Endpoint string `json:"endpoint,omitempty"`
+ // Shared access signature URL, leave blank if using account/key
+ SASURL string `json:"sas_url,omitempty"`
+ // KeyPrefix is similar to a chroot directory for local filesystem.
+ // If specified then the SFTPGo userd will only see objects that starts
+ // with this prefix and so you can restrict access to a specific
+ // folder. The prefix, if not empty, must not start with "/" and must
+ // end with "/".
+ // If empty the whole bucket contents will be available
+ KeyPrefix string `json:"key_prefix,omitempty"`
+ // The buffer size (in MB) to use for multipart uploads.
+ // If this value is set to zero, the default value (1MB) for the Azure SDK will be used.
+ // Please note that if the upload bandwidth between the SFTPGo client and SFTPGo server is
+ // greater than the upload bandwidth between SFTPGo and Azure then the SFTP client have
+ // to wait for the upload of the last parts to Azure after it ends the file upload to SFTPGo,
+ // and it may time out.
+ // Keep this in mind if you customize these parameters.
+ UploadPartSize int64 `json:"upload_part_size,omitempty"`
+ // How many parts are uploaded in parallel
+ UploadConcurrency int `json:"upload_concurrency,omitempty"`
+ // Set to true if you use an Azure emulator such as Azurite
+ UseEmulator bool `json:"use_emulator,omitempty"`
+}
+
// PipeWriter defines a wrapper for pipeat.PipeWriterAt.
type PipeWriter struct {
writer *pipeat.PipeWriterAt
@@ -194,7 +232,7 @@ func ValidateS3FsConfig(config *S3FsConfig) error {
if len(config.AccessSecret) == 0 && len(config.AccessKey) > 0 {
return errors.New("access_secret cannot be empty with access_key not empty")
}
- if len(config.KeyPrefix) > 0 {
+ if config.KeyPrefix != "" {
if strings.HasPrefix(config.KeyPrefix, "/") {
return errors.New("key_prefix cannot start with /")
}
@@ -214,10 +252,10 @@ func ValidateS3FsConfig(config *S3FsConfig) error {
// ValidateGCSFsConfig returns nil if the specified GCS config is valid, otherwise an error
func ValidateGCSFsConfig(config *GCSFsConfig, credentialsFilePath string) error {
- if len(config.Bucket) == 0 {
+ if config.Bucket == "" {
return errors.New("bucket cannot be empty")
}
- if len(config.KeyPrefix) > 0 {
+ if config.KeyPrefix != "" {
if strings.HasPrefix(config.KeyPrefix, "/") {
return errors.New("key_prefix cannot start with /")
}
@@ -238,6 +276,36 @@ func ValidateGCSFsConfig(config *GCSFsConfig, credentialsFilePath string) error
return nil
}
+// ValidateAzBlobFsConfig returns nil if the specified Azure Blob config is valid, otherwise an error
+func ValidateAzBlobFsConfig(config *AzBlobFsConfig) error {
+ if config.SASURL != "" {
+ _, err := url.Parse(config.SASURL)
+ return err
+ }
+ if config.Container == "" {
+ return errors.New("container cannot be empty")
+ }
+ if config.AccountName == "" || config.AccountKey == "" {
+ return errors.New("credentials cannot be empty")
+ }
+ if config.KeyPrefix != "" {
+ if strings.HasPrefix(config.KeyPrefix, "/") {
+ return errors.New("key_prefix cannot start with /")
+ }
+ config.KeyPrefix = path.Clean(config.KeyPrefix)
+ if !strings.HasSuffix(config.KeyPrefix, "/") {
+ config.KeyPrefix += "/"
+ }
+ }
+ if config.UploadPartSize < 0 {
+ return fmt.Errorf("invalid upload part size: %v", config.UploadPartSize)
+ }
+ if config.UploadConcurrency < 0 {
+ return fmt.Errorf("invalid upload concurrency: %v", config.UploadConcurrency)
+ }
+ return nil
+}
+
// SetPathPermissions calls fs.Chown.
// It does nothing for local filesystem on windows
func SetPathPermissions(fs Fs, path string, uid int, gid int) {
diff --git a/webdavd/file.go b/webdavd/file.go
index 2462d9cc..45629eaf 100644
--- a/webdavd/file.go
+++ b/webdavd/file.go
@@ -122,29 +122,27 @@ func (f *webDavFile) Read(p []byte) (n int, err error) {
f.TransferError(common.ErrOpUnsupported)
return 0, common.ErrOpUnsupported
}
- _, r, cancelFn, err := f.Fs.Open(f.GetFsPath(), 0)
+ _, r, cancelFn, e := f.Fs.Open(f.GetFsPath(), 0)
f.Lock()
f.reader = r
- f.ErrTransfer = err
+ f.ErrTransfer = e
f.BaseTransfer.SetCancelFn(cancelFn)
f.startOffset = 0
f.Unlock()
- if err != nil {
- return 0, err
+ if e != nil {
+ return 0, e
}
}
- var readed int
- var e error
- readed, e = f.reader.Read(p)
- atomic.AddInt64(&f.BytesSent, int64(readed))
+ n, err = f.reader.Read(p)
+ atomic.AddInt64(&f.BytesSent, int64(n))
- if e != nil && e != io.EOF {
- f.TransferError(e)
- return readed, e
+ if err != nil && err != io.EOF {
+ f.TransferError(err)
+ return
}
f.HandleThrottle()
- return readed, e
+ return
}
// Write writes the uploaded contents.
@@ -154,21 +152,19 @@ func (f *webDavFile) Write(p []byte) (n int, err error) {
}
f.Connection.UpdateLastActivity()
- var written int
- var e error
- written, e = f.writer.Write(p)
- atomic.AddInt64(&f.BytesReceived, int64(written))
+ n, err = f.writer.Write(p)
+ atomic.AddInt64(&f.BytesReceived, int64(n))
- if f.MaxWriteSize > 0 && e == nil && atomic.LoadInt64(&f.BytesReceived) > f.MaxWriteSize {
- e = common.ErrQuotaExceeded
+ if f.MaxWriteSize > 0 && err == nil && atomic.LoadInt64(&f.BytesReceived) > f.MaxWriteSize {
+ err = common.ErrQuotaExceeded
}
- if e != nil {
- f.TransferError(e)
- return written, e
+ if err != nil {
+ f.TransferError(err)
+ return
}
f.HandleThrottle()
- return written, e
+ return
}
// Seek sets the offset for the next Read or Write on the writer to offset,