duplicacy/src/duplicacy_sftpstorage.go

// Copyright (c) Acrosync LLC. All rights reserved.
// Free for personal use and commercial trial
// Commercial use requires per-user licenses available from https://duplicacy.com

package duplicacy

import (
	"fmt"
	"io"
	"math/rand"
	"net"
	"os"
	"path"
	"runtime"
	"strings"
	"time"

	"github.com/pkg/sftp"
	"golang.org/x/crypto/ssh"
)

type SFTPStorage struct {
	StorageBase

	client          *sftp.Client
	minimumNesting  int // The minimum level of directories to dive into before searching for the chunk file.
	storageDir      string
	numberOfThreads int
}

func CreateSFTPStorageWithPassword(server string, port int, username string, storageDir string,
	minimumNesting int, password string, threads int) (storage *SFTPStorage, err error) {

	authMethods := []ssh.AuthMethod{ssh.Password(password)}

	hostKeyCallback := func(hostname string, remote net.Addr,
		key ssh.PublicKey) error {
		return nil
	}

	return CreateSFTPStorage(server, port, username, storageDir, minimumNesting, authMethods, hostKeyCallback, threads)
}

func CreateSFTPStorage(server string, port int, username string, storageDir string, minimumNesting int,
	authMethods []ssh.AuthMethod,
	hostKeyCallback func(hostname string, remote net.Addr,
		key ssh.PublicKey) error, threads int) (storage *SFTPStorage, err error) {

	sftpConfig := &ssh.ClientConfig{
		User:            username,
		Auth:            authMethods,
		HostKeyCallback: hostKeyCallback,
	}

	if server == "sftp.hidrive.strato.com" {
		sftpConfig.Ciphers = []string{"aes128-cbc", "aes128-ctr", "aes256-ctr"}
	}

	serverAddress := fmt.Sprintf("%s:%d", server, port)
	connection, err := ssh.Dial("tcp", serverAddress, sftpConfig)
	if err != nil {
		return nil, err
	}

	client, err := sftp.NewClient(connection)
	if err != nil {
		connection.Close()
		return nil, err
	}

	for storageDir[len(storageDir)-1] == '/' {
		storageDir = storageDir[:len(storageDir)-1]
	}

	fileInfo, err := client.Stat(storageDir)
	if err != nil {
		return nil, fmt.Errorf("Can't access the storage path %s: %v", storageDir, err)
	}

	if !fileInfo.IsDir() {
		return nil, fmt.Errorf("The storage path %s is not a directory", storageDir)
	}

	storage = &SFTPStorage{
		client:          client,
		storageDir:      storageDir,
		minimumNesting:  minimumNesting,
		numberOfThreads: threads,
	}

	// Random number fo generating the temporary chunk file suffix.
	rand.Seed(time.Now().UnixNano())

	runtime.SetFinalizer(storage, CloseSFTPStorage)

	storage.DerivedStorage = storage
	storage.SetDefaultNestingLevels([]int{2, 3}, 2)
	return storage, nil
}

func CloseSFTPStorage(storage *SFTPStorage) {
	storage.client.Close()
}

// ListFiles return the list of files and subdirectories under 'file' (non-recursively)
func (storage *SFTPStorage) ListFiles(threadIndex int, dirPath string) (files []string, sizes []int64, err error) {

	entries, err := storage.client.ReadDir(path.Join(storage.storageDir, dirPath))
	if err != nil {
		return nil, nil, err
	}

	for _, entry := range entries {
		name := entry.Name()
		if entry.IsDir() && name[len(name)-1] != '/' {
			name += "/"
		}

		files = append(files, name)
		sizes = append(sizes, entry.Size())
	}

	return files, sizes, nil
}

// DeleteFile deletes the file or directory at 'filePath'.
func (storage *SFTPStorage) DeleteFile(threadIndex int, filePath string) (err error) {
	fullPath := path.Join(storage.storageDir, filePath)
	fileInfo, err := storage.client.Stat(fullPath)
	if err != nil {
		if os.IsNotExist(err) {
			LOG_TRACE("SFTP_STORAGE", "File %s has disappeared before deletion", filePath)
			return nil
		}
		return err
	}
	if fileInfo == nil {
		return nil
	}
	return storage.client.Remove(path.Join(storage.storageDir, filePath))
}

// MoveFile renames the file.
func (storage *SFTPStorage) MoveFile(threadIndex int, from string, to string) (err error) {
	toPath := path.Join(storage.storageDir, to)
	fileInfo, err := storage.client.Stat(toPath)
	if fileInfo != nil {
		return fmt.Errorf("The destination file %s already exists", toPath)
	}
	return storage.client.Rename(path.Join(storage.storageDir, from),
		path.Join(storage.storageDir, to))
}

// CreateDirectory creates a new directory.
func (storage *SFTPStorage) CreateDirectory(threadIndex int, dirPath string) (err error) {
	fullPath := path.Join(storage.storageDir, dirPath)
	fileInfo, err := storage.client.Stat(fullPath)
	if fileInfo != nil && fileInfo.IsDir() {
		return nil
	}
	return storage.client.Mkdir(path.Join(storage.storageDir, dirPath))
}

// GetFileInfo returns the information about the file or directory at 'filePath'.
func (storage *SFTPStorage) GetFileInfo(threadIndex int, filePath string) (exist bool, isDir bool, size int64, err error) {
	fileInfo, err := storage.client.Stat(path.Join(storage.storageDir, filePath))
	if err != nil {
		if os.IsNotExist(err) {
			return false, false, 0, nil
		} else {
			return false, false, 0, err
		}
	}

	if fileInfo == nil {
		return false, false, 0, nil
	}

	return true, fileInfo.IsDir(), fileInfo.Size(), nil
}

// DownloadFile reads the file at 'filePath' into the chunk.
func (storage *SFTPStorage) DownloadFile(threadIndex int, filePath string, chunk *Chunk) (err error) {
	file, err := storage.client.Open(path.Join(storage.storageDir, filePath))

	if err != nil {
		return err
	}

	defer file.Close()
	if _, err = RateLimitedCopy(chunk, file, storage.DownloadRateLimit/storage.numberOfThreads); err != nil {
		return err
	}

	return nil
}

// UploadFile writes 'content' to the file at 'filePath'.
func (storage *SFTPStorage) UploadFile(threadIndex int, filePath string, content []byte) (err error) {

	fullPath := path.Join(storage.storageDir, filePath)

	dirs := strings.Split(filePath, "/")
	if len(dirs) > 1 {
		fullDir := path.Dir(fullPath)
		_, err := storage.client.Stat(fullDir)
		if err != nil {
			// The error may be caused by a non-existent fullDir, or a broken connection.  In either case,
			// we just assume it is the former because there isn't a way to tell which is the case.
			for i, _ := range dirs[1 : len(dirs)-1] {
				subDir := path.Join(storage.storageDir, path.Join(dirs[0:i+2]...))
				// We don't check the error; just keep going blindly but always store the last err
				err = storage.client.Mkdir(subDir)
			}

			// If there is an error creating the dirs, we check fullDir one more time, because another thread
			// may happen to create the same fullDir ahead of this thread
			if err != nil {
				_, err := storage.client.Stat(fullDir)
				if err != nil {
					return err
				}
			}
		}
	}

	letters := "abcdefghijklmnopqrstuvwxyz"
	suffix := make([]byte, 8)
	for i := range suffix {
		suffix[i] = letters[rand.Intn(len(letters))]
	}

	temporaryFile := fullPath + "." + string(suffix) + ".tmp"

	file, err := storage.client.OpenFile(temporaryFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC)
	if err != nil {
		return err
	}

	reader := CreateRateLimitedReader(content, storage.UploadRateLimit/storage.numberOfThreads)
	_, err = io.Copy(file, reader)
	if err != nil {
		file.Close()
		return err
	}
	file.Close()

	err = storage.client.Rename(temporaryFile, fullPath)
	if err != nil {

		if _, err = storage.client.Stat(fullPath); err == nil {
			storage.client.Remove(temporaryFile)
			return nil
		} else {
			return fmt.Errorf("Uploaded file but failed to store it at %s: %v", fullPath, err)
		}
	}

	return nil
}

// If a local snapshot cache is needed for the storage to avoid downloading/uploading chunks too often when
// managing snapshots.
func (storage *SFTPStorage) IsCacheNeeded() bool { return true }

// If the 'MoveFile' method is implemented.
func (storage *SFTPStorage) IsMoveFileImplemented() bool { return true }

// If the storage can guarantee strong consistency.
func (storage *SFTPStorage) IsStrongConsistent() bool { return true }

// If the storage supports fast listing of files names.
func (storage *SFTPStorage) IsFastListing() bool {
	for _, level := range storage.readLevels {
		if level > 1 {
			return false
		}
	}
	return true
}

// Enable the test mode.
func (storage *SFTPStorage) EnableTestMode() {}