522 lines
15 KiB
Go
522 lines
15 KiB
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 := b.client.Rename(writePath, nativePath); err != nil {
|
|
return storage.Entry{}, b.translateError(storage.OpWriteFrom, logicalPath, err)
|
|
}
|
|
cleanup = false
|
|
}
|
|
return b.Stat(ctx, logicalPath)
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
visited := 0
|
|
emit := func(entry storage.Entry) error {
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
if opts.Limit > 0 && visited >= opts.Limit {
|
|
return storage.ErrStopWalk
|
|
}
|
|
visited++
|
|
if err := fn(entry); err != nil {
|
|
if errors.Is(err, storage.ErrStopWalk) {
|
|
return storage.ErrStopWalk
|
|
}
|
|
return storage.NewError(storage.OpWalk, BackendName, entry.Path, storage.ErrUnknown, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
if !info.IsDir() {
|
|
if err := emit(entryFromInfo(prefix, info)); errors.Is(err, storage.ErrStopWalk) {
|
|
return nil
|
|
} else if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
if err := b.walkDirectory(ctx, prefix, nativePrefix, opts, emit); errors.Is(err, storage.ErrStopWalk) {
|
|
return nil
|
|
} else if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b *Backend) HasAny(ctx context.Context, prefix string) (bool, error) {
|
|
found := false
|
|
err := b.Walk(ctx, prefix, storage.WalkOptions{Recursive: false, Limit: 1}, func(storage.Entry) error {
|
|
found = true
|
|
return storage.ErrStopWalk
|
|
})
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
return found, nil
|
|
}
|
|
|
|
func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error {
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
targets, err := storage.ManagedBundleTargets(bundlePath, managedOutputPaths)
|
|
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(storage.OpDeleteManagedBundle, BackendName, target, storage.ErrInvalidPath, nil)
|
|
}
|
|
info, err := b.client.Lstat(nativePath)
|
|
if err != nil {
|
|
if opts.IgnoreMissing && isNotExist(err) {
|
|
continue
|
|
}
|
|
return b.translateError(storage.OpDeleteManagedBundle, target, err)
|
|
}
|
|
if info.IsDir() {
|
|
return storage.NewError(storage.OpDeleteManagedBundle, BackendName, target, storage.ErrUnsupported, nil)
|
|
}
|
|
if err := b.client.Remove(nativePath); err != nil {
|
|
return b.translateError(storage.OpDeleteManagedBundle, 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 (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(),
|
|
}
|
|
}
|