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(), } }