distributor/internal/adapters/ssh/backend.go

package ssh

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"os"
	"path"
	"sort"
	"strings"
	"time"

	"gitea.maximumdirect.net/eric/distributor/internal/storage"
	"github.com/pkg/sftp"
	cryptossh "golang.org/x/crypto/ssh"
)

type Backend struct {
	client    *sftp.Client
	sshClient *cryptossh.Client
	root      string
}

func New(ctx context.Context, options Options) (*Backend, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	options, err := options.normalized()
	if err != nil {
		return nil, storage.NewError(storage.OpOpenBackend, BackendName, options.Root, storage.ErrInvalidPath, err)
	}
	hostKeyCallback, err := hostKeyCallback(options)
	if err != nil {
		return nil, storage.NewError(storage.OpOpenBackend, BackendName, options.KnownHosts, storage.ErrInvalidPath, err)
	}
	auth, cleanupAuth, err := authMethods(options.KeyFile)
	if err != nil {
		return nil, storage.NewError(storage.OpOpenBackend, BackendName, options.KeyFile, storage.ErrPermission, err)
	}
	defer cleanupAuth()

	sshClient, err := cryptossh.Dial("tcp", options.address(), &cryptossh.ClientConfig{
		User:            options.User,
		Auth:            auth,
		HostKeyCallback: hostKeyCallback,
		Timeout:         30 * time.Second,
	})
	if err != nil {
		return nil, storage.NewError(storage.OpOpenBackend, BackendName, options.address(), storage.ErrUnknown, err)
	}
	client, err := sftp.NewClient(sshClient)
	if err != nil {
		_ = sshClient.Close()
		return nil, storage.NewError(storage.OpOpenBackend, BackendName, options.address(), storage.ErrUnknown, err)
	}
	return &Backend{client: client, sshClient: sshClient, root: options.Root}, nil
}

func (b *Backend) Close() error {
	var err error
	if b.client != nil {
		err = b.client.Close()
	}
	if b.sshClient != nil {
		if closeErr := b.sshClient.Close(); err == nil {
			err = closeErr
		}
	}
	return err
}

func (b *Backend) ReadFile(ctx context.Context, logicalPath string) ([]byte, error) {
	reader, err := b.OpenReader(ctx, logicalPath)
	if err != nil {
		return nil, err
	}
	defer reader.Close()
	data, err := io.ReadAll(reader)
	if err != nil {
		return nil, storage.NewError(storage.OpReadFile, BackendName, logicalPath, storage.ErrUnknown, err)
	}
	return data, nil
}

func (b *Backend) OpenReader(ctx context.Context, logicalPath string) (io.ReadCloser, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	nativePath, err := b.nativePath(logicalPath, false)
	if err != nil {
		return nil, err
	}
	if err := b.rejectSymlinkAncestors(ctx, logicalPath, true); err != nil {
		return nil, err
	}
	info, err := b.client.Lstat(nativePath)
	if err != nil {
		return nil, b.translateError(storage.OpOpenReader, logicalPath, err)
	}
	if !info.Mode().IsRegular() {
		return nil, storage.NewError(storage.OpOpenReader, BackendName, logicalPath, storage.ErrUnsupported, nil)
	}
	file, err := b.client.Open(nativePath)
	if err != nil {
		return nil, b.translateError(storage.OpOpenReader, logicalPath, err)
	}
	return file, nil
}

func (b *Backend) WriteFile(ctx context.Context, logicalPath string, data []byte, opts storage.WriteOptions) (storage.Entry, error) {
	opts.Size = int64(len(data))
	opts.SizeKnown = true
	return b.WriteFrom(ctx, logicalPath, bytes.NewReader(data), opts)
}

func (b *Backend) WriteFrom(ctx context.Context, logicalPath string, r io.Reader, opts storage.WriteOptions) (storage.Entry, error) {
	if err := ctx.Err(); err != nil {
		return storage.Entry{}, err
	}
	nativePath, err := b.nativePath(logicalPath, false)
	if err != nil {
		return storage.Entry{}, err
	}
	if err := b.rejectSymlinkAncestors(ctx, parentOf(logicalPath), true); err != nil {
		return storage.Entry{}, err
	}
	if info, err := b.client.Lstat(nativePath); err == nil {
		if !opts.Overwrite {
			return storage.Entry{}, storage.NewError(storage.OpWriteFrom, BackendName, logicalPath, storage.ErrAlreadyExist, nil)
		}
		if !info.Mode().IsRegular() {
			return storage.Entry{}, storage.NewError(storage.OpWriteFrom, BackendName, logicalPath, storage.ErrConflict, nil)
		}
	} else if !isNotExist(err) {
		return storage.Entry{}, b.translateError(storage.OpWriteFrom, logicalPath, err)
	}

	parentNative := path.Dir(nativePath)
	if err := b.client.MkdirAll(parentNative); err != nil {
		return storage.Entry{}, b.translateError(storage.OpWriteFrom, logicalPath, err)
	}

	writePath := nativePath
	if opts.PreferAtomic {
		writePath = path.Join(parentNative, fmt.Sprintf(".distributor-write-%d", time.Now().UnixNano()))
	}
	file, err := b.client.Create(writePath)
	if err != nil {
		return storage.Entry{}, b.translateError(storage.OpWriteFrom, logicalPath, err)
	}
	cleanup := opts.PreferAtomic
	defer func() {
		if cleanup {
			_ = b.client.Remove(writePath)
		}
	}()

	written, copyErr := io.Copy(file, r)
	closeErr := file.Close()
	if copyErr != nil {
		return storage.Entry{}, storage.NewError(storage.OpWriteFrom, BackendName, logicalPath, storage.ErrUnknown, copyErr)
	}
	if closeErr != nil {
		return storage.Entry{}, storage.NewError(storage.OpWriteFrom, BackendName, logicalPath, storage.ErrUnknown, closeErr)
	}
	if opts.SizeKnown && written != opts.Size {
		return storage.Entry{}, storage.NewError(storage.OpWriteFrom, BackendName, logicalPath, storage.ErrConflict, fmt.Errorf("stream size %d does not match expected size %d", written, opts.Size))
	}
	if opts.PreferAtomic {
		if err := renamePromotedFile(b.client, writePath, nativePath, opts.Overwrite); err != nil {
			return storage.Entry{}, b.translateError(storage.OpWriteFrom, logicalPath, err)
		}
		cleanup = false
	}
	return b.Stat(ctx, logicalPath)
}

type sftpRenamer interface {
	PosixRename(oldname, newname string) error
	Rename(oldname, newname string) error
	Remove(path string) error
}

func renamePromotedFile(client sftpRenamer, oldname, newname string, overwrite bool) error {
	if !overwrite {
		return client.Rename(oldname, newname)
	}
	if err := client.PosixRename(oldname, newname); err == nil {
		return nil
	} else if !isReplaceRenameFallbackError(err) {
		return err
	}
	if err := client.Remove(newname); err != nil && !isNotExist(err) {
		return err
	}
	return client.Rename(oldname, newname)
}

func (b *Backend) Stat(ctx context.Context, logicalPath string) (storage.Entry, error) {
	if err := ctx.Err(); err != nil {
		return storage.Entry{}, err
	}
	nativePath, err := b.nativePath(logicalPath, true)
	if err != nil {
		return storage.Entry{}, err
	}
	info, err := b.client.Lstat(nativePath)
	if err != nil {
		return storage.Entry{}, b.translateError(storage.OpStat, logicalPath, err)
	}
	return entryFromInfo(logicalPath, info), nil
}

func (b *Backend) Walk(ctx context.Context, prefix string, opts storage.WalkOptions, fn storage.WalkFunc) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	nativePrefix, err := b.nativePath(prefix, true)
	if err != nil {
		return err
	}
	info, err := b.client.Lstat(nativePrefix)
	if err != nil {
		if isNotExist(err) {
			return nil
		}
		return b.translateError(storage.OpWalk, prefix, err)
	}

	emitter := storage.NewWalkEmitter(ctx, BackendName, opts, fn)

	if !info.IsDir() {
		return storage.FinishWalk(emitter.Emit(entryFromInfo(prefix, info)))
	}

	return storage.FinishWalk(b.walkDirectory(ctx, prefix, nativePrefix, opts, emitter.Emit))
}

func (b *Backend) HasAny(ctx context.Context, prefix string) (bool, error) {
	return storage.HasAny(ctx, b, prefix)
}

func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error {
	return b.deleteManagedTargets(ctx, storage.OpDeleteManagedBundle, func() ([]string, error) {
		return storage.ManagedBundleTargets(bundlePath, managedOutputPaths)
	}, opts)
}

func (b *Backend) DeleteManagedOutputs(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error {
	return b.deleteManagedTargets(ctx, storage.OpDeleteManagedOutputs, func() ([]string, error) {
		return storage.ManagedOutputTargets(bundlePath, managedOutputPaths)
	}, opts)
}

func (b *Backend) deleteManagedTargets(ctx context.Context, op string, targetsFunc func() ([]string, error), opts storage.DeleteOptions) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	targets, err := targetsFunc()
	if err != nil {
		return err
	}
	for _, target := range targets {
		nativePath, err := b.nativePath(target, false)
		if err != nil {
			return err
		}
		if nativePath == b.root {
			return storage.NewError(op, BackendName, target, storage.ErrInvalidPath, nil)
		}
		info, err := b.client.Lstat(nativePath)
		if err != nil {
			if opts.IgnoreMissing && isNotExist(err) {
				continue
			}
			return b.translateError(op, target, err)
		}
		if info.IsDir() {
			return storage.NewError(op, BackendName, target, storage.ErrUnsupported, nil)
		}
		if err := b.client.Remove(nativePath); err != nil {
			return b.translateError(op, target, err)
		}
		if opts.PruneEmptyDirs {
			b.pruneEmptyParents(parentOf(target))
		}
	}
	return nil
}

func (b *Backend) DeletePrefix(ctx context.Context, prefix string, opts storage.DeleteOptions) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	if err := storage.ValidatePrefix(prefix); err != nil {
		return err
	}
	var entries []storage.Entry
	if prefix != "" {
		entry, err := b.Stat(ctx, prefix)
		if err != nil {
			if opts.IgnoreMissing && storage.IsNotFound(err) {
				return nil
			}
			return err
		}
		if entry.Type != storage.EntryTypeDirectory {
			return b.deleteEntry(ctx, entry, opts)
		}
		entries = append(entries, entry)
	}
	if err := b.Walk(ctx, prefix, storage.WalkOptions{Recursive: true}, func(entry storage.Entry) error {
		entries = append(entries, entry)
		return nil
	}); err != nil {
		return err
	}
	if prefix != "" && len(entries) == 1 {
		if err := b.deleteEntry(ctx, entries[0], opts); err != nil {
			return err
		}
		if opts.PruneEmptyDirs {
			b.pruneEmptyParents(parentOf(prefix))
		}
		return nil
	}
	sort.Slice(entries, func(i, j int) bool {
		return strings.Count(entries[i].Path, "/") > strings.Count(entries[j].Path, "/")
	})
	for _, entry := range entries {
		if entry.Path == "" {
			continue
		}
		if err := b.deleteEntry(ctx, entry, storage.DeleteOptions{IgnoreMissing: true}); err != nil {
			return err
		}
	}
	if opts.PruneEmptyDirs {
		b.pruneEmptyParents(parentOf(prefix))
	}
	return nil
}

func (b *Backend) deleteEntry(ctx context.Context, entry storage.Entry, opts storage.DeleteOptions) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	nativePath, err := b.nativePath(entry.Path, false)
	if err != nil {
		return err
	}
	var removeErr error
	if entry.Type == storage.EntryTypeDirectory {
		removeErr = b.client.RemoveDirectory(nativePath)
	} else {
		removeErr = b.client.Remove(nativePath)
	}
	if removeErr != nil {
		if opts.IgnoreMissing && isNotExist(removeErr) {
			return nil
		}
		return b.translateError(storage.OpDeletePrefix, entry.Path, removeErr)
	}
	return nil
}

func (b *Backend) walkDirectory(ctx context.Context, logicalPrefix, nativePrefix string, opts storage.WalkOptions, emit func(storage.Entry) error) error {
	entries, err := b.client.ReadDir(nativePrefix)
	if err != nil {
		return b.translateError(storage.OpWalk, logicalPrefix, err)
	}
	sort.Slice(entries, func(i, j int) bool { return entries[i].Name() < entries[j].Name() })
	for _, info := range entries {
		if err := ctx.Err(); err != nil {
			return err
		}
		logicalPath := info.Name()
		if logicalPrefix != "" {
			logicalPath = logicalPrefix + "/" + info.Name()
		}
		if err := emit(entryFromInfo(logicalPath, info)); err != nil {
			return err
		}
		if opts.Recursive && info.IsDir() {
			if err := b.walkDirectory(ctx, logicalPath, path.Join(nativePrefix, info.Name()), opts, emit); err != nil {
				return err
			}
		}
	}
	return nil
}

func (b *Backend) nativePath(logicalPath string, allowEmpty bool) (string, error) {
	if logicalPath == "" {
		if !allowEmpty {
			return "", storage.NewError(storage.OpValidatePath, BackendName, logicalPath, storage.ErrInvalidPath, nil)
		}
		return b.root, nil
	}
	if err := storage.ValidatePath(logicalPath); err != nil {
		return "", err
	}
	nativePath := path.Clean(path.Join(b.root, logicalPath))
	if !withinRoot(b.root, nativePath) {
		return "", storage.NewError(storage.OpValidatePath, BackendName, logicalPath, storage.ErrInvalidPath, nil)
	}
	return nativePath, nil
}

func (b *Backend) rejectSymlinkAncestors(ctx context.Context, logicalPath string, includeFinal bool) error {
	if logicalPath == "" {
		return nil
	}
	if err := storage.ValidatePath(logicalPath); err != nil {
		return err
	}
	segments := strings.Split(logicalPath, "/")
	limit := len(segments)
	if !includeFinal {
		limit--
	}
	current := ""
	for index := 0; index < limit; index++ {
		if err := ctx.Err(); err != nil {
			return err
		}
		if current == "" {
			current = segments[index]
		} else {
			current += "/" + segments[index]
		}
		nativePath, err := b.nativePath(current, false)
		if err != nil {
			return err
		}
		info, err := b.client.Lstat(nativePath)
		if err != nil {
			if isNotExist(err) {
				return nil
			}
			return b.translateError(storage.OpStat, current, err)
		}
		if info.Mode()&os.ModeSymlink != 0 {
			return storage.NewError(storage.OpStat, BackendName, current, storage.ErrUnsupported, nil)
		}
	}
	return nil
}

func (b *Backend) pruneEmptyParents(logicalPath string) {
	for logicalPath != "" {
		nativePath, err := b.nativePath(logicalPath, false)
		if err != nil || nativePath == b.root {
			return
		}
		if err := b.client.RemoveDirectory(nativePath); err != nil {
			return
		}
		logicalPath = parentOf(logicalPath)
	}
}

func withinRoot(root, candidate string) bool {
	if candidate == root {
		return true
	}
	if root == "/" {
		return strings.HasPrefix(candidate, "/")
	}
	return strings.HasPrefix(candidate, strings.TrimSuffix(root, "/")+"/")
}

func parentOf(logicalPath string) string {
	index := strings.LastIndex(logicalPath, "/")
	if index == -1 {
		return ""
	}
	return logicalPath[:index]
}

func isNotExist(err error) bool {
	return errors.Is(err, fs.ErrNotExist) || errors.Is(err, os.ErrNotExist) || errors.Is(err, sftp.ErrSSHFxNoSuchFile)
}

func isReplaceRenameFallbackError(err error) bool {
	if errors.Is(err, sftp.ErrSSHFxFailure) || errors.Is(err, sftp.ErrSSHFxOpUnsupported) {
		return true
	}
	var statusErr *sftp.StatusError
	return errors.As(err, &statusErr) && (statusErr.FxCode() == sftp.ErrSSHFxFailure || statusErr.FxCode() == sftp.ErrSSHFxOpUnsupported)
}

func (b *Backend) translateError(op, logicalPath string, err error) error {
	kind := storage.ErrUnknown
	switch {
	case isNotExist(err):
		kind = storage.ErrNotFound
	case errors.Is(err, fs.ErrExist), errors.Is(err, os.ErrExist):
		kind = storage.ErrAlreadyExist
	case errors.Is(err, fs.ErrPermission), errors.Is(err, os.ErrPermission), errors.Is(err, sftp.ErrSSHFxPermissionDenied):
		kind = storage.ErrPermission
	case errors.Is(err, sftp.ErrSSHFxOpUnsupported):
		kind = storage.ErrUnsupported
	case errors.Is(err, sftp.ErrSSHFxNoConnection), errors.Is(err, sftp.ErrSSHFxConnectionLost):
		kind = storage.ErrTemporary
	}
	return storage.NewError(op, BackendName, logicalPath, kind, err)
}

func entryFromInfo(logicalPath string, info fs.FileInfo) storage.Entry {
	entryType := storage.EntryTypeOther
	switch {
	case info.Mode()&os.ModeSymlink != 0:
		entryType = storage.EntryTypeSymlink
	case info.Mode().IsRegular():
		entryType = storage.EntryTypeFile
	case info.IsDir():
		entryType = storage.EntryTypeDirectory
	}
	return storage.Entry{
		Path: logicalPath,
		Type: entryType,
		Size: info.Size(),
	}
}