cachefs/pkg/provider/sftp/sftp.go

// Copyright (C) 2022 Marius Schellenberger

package sftp

import (
	"cachefs/pkg/provider"
	"context"
	"errors"
	"fmt"
	"io/fs"
	"net"
	"net/url"
	"os"
	"path"
	"path/filepath"
	"time"

	"github.com/go-logr/logr"
	"github.com/pkg/sftp"
	"golang.org/x/crypto/ssh"
	"golang.org/x/crypto/ssh/knownhosts"
)

var (
	_ provider.FS = (*FS)(nil)
	_ SFTP        = (*sftp.Client)(nil)
	_ SFTP        = (*connError)(nil)
)

type SFTP interface {
	Remove(string) error
	Open(string) (*sftp.File, error)
	OpenFile(string, int) (*sftp.File, error)
	Stat(string) (fs.FileInfo, error)
	ReadDir(string) ([]fs.FileInfo, error)
	MkdirAll(string) error
	Wait() error
	Close() error
}

type FS struct {
	log       logr.Logger
	root      string
	addr      string
	ctx       context.Context
	conn      net.Conn
	cfg       *ssh.ClientConfig
	c         *ssh.Client
	client    SFTP
	cancel    func()
	keepalive bool
}

func NewFS(u *url.URL, block bool, log logr.Logger) (fs *FS, err error) {
	cfg, err := sshConfig(u)
	if err != nil {
		return
	}
	fs = &FS{
		log:    log.WithValues("host", u.Host),
		root:   u.Path,
		addr:   u.Host,
		cfg:    cfg,
		client: connError{},
	}
	ctx, cancel := context.WithCancel(context.Background())
	fs.cancel = cancel
	go fs.connect(ctx)
	if block {
		for {
			if _, e := fs.Stat("."); e != ErrNotConnected {
				return
			}
			time.Sleep(time.Millisecond * 100)
		}
	}
	return
}

func (fs *FS) dial() (err error) {
	fs.conn, err = net.DialTimeout("tcp", fs.addr, fs.cfg.Timeout)
	if err != nil {
		return
	}
	c, chans, reqs, err := ssh.NewClientConn(fs.conn, fs.addr, fs.cfg)
	if err != nil {
		return
	}
	fs.c = ssh.NewClient(c, chans, reqs)
	return nil
}

func (fs *FS) connect(ctx context.Context) {
	log := fs.log
	t := time.NewTicker(time.Second * 2)
	defer t.Stop()
	for {
		log.V(2).Info("connecting")
		err := fs.dial()
		if err == nil {
			log.V(2).Info("connected")
			if !fs.keepalive {
				go fs.keepAlive(ctx)
				fs.keepalive = true
			}
			fs.client, err = sftp.NewClient(fs.c,
				sftp.UseFstat(true),
				sftp.UseConcurrentReads(true),
				//sftp.UseConcurrentWrites(true),
				//sftp.MaxPacket(23552),
			)
			select {
			case <-ctx.Done():
				return
			case e := <-fs.wait(ctx):
				log.V(2).Info("connection lost", "err", e)
				fs.conn.Close()
				fs.client = connError{}
				fs.c = nil
			}
		} else {
			log.Error(err, "error connecting")
		}
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

func (fs *FS) wait(ctx context.Context) <-chan error {
	w := make(chan error)
	go func() {
		select {
		case <-ctx.Done():
		case w <- fs.client.Wait():
		}
	}()
	return w
}

func (fs *FS) keepAlive(ctx context.Context) {
	t := time.NewTicker(time.Second * 15)
	for {
		if fs.c != nil {
			_, _, err := fs.c.SendRequest("keepalive@cachefs", true, nil)
			if err != nil {
				fs.log.Error(err, "sending keep-alive request failed")
			} else {
				fs.conn.SetReadDeadline(time.Now().Add(time.Second * 20))
			}
		}
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

func (fs *FS) Remove(p string) error {
	return fs.client.Remove(fs.path(p))
}

func (fs *FS) Open(p string) (provider.File, error) {
	f, err := fs.client.Open(fs.path(p))
	return &file{f, fs}, err
}

func (fs *FS) OpenFile(p string, flags int, _ os.FileMode) (provider.File, error) {
	f, err := fs.client.OpenFile(fs.path(p), flags)
	return &file{f, fs}, err
}

func (fs *FS) Fstat(_ uintptr) (int64, error) {
	return 0, provider.ErrFstat
}

func (fs *FS) Stat(p string) (fs.FileInfo, error) {
	return fs.client.Stat(fs.path(p))
}

func (fs *FS) Root() string {
	return fs.root
}

func (fs *FS) MkdirAll(p string, _ os.FileMode) error {
	return fs.client.MkdirAll(fs.path(p))
}

func (fs *FS) Close() error {
	fs.cancel()
	fs.client.Close()
	return fs.c.Close()
}

func (fs *FS) path(p string) string {
	return filepath.Join(fs.root, filepath.FromSlash(path.Clean("/"+p)))
}

type file struct {
	*sftp.File
	fs *FS
}

func (f *file) ReadDir(_ int) (d []fs.DirEntry, err error) {
	list, err := f.Readdir(0)
	if err != nil {
		return
	}
	d = make([]fs.DirEntry, len(list))
	for i, v := range list {
		d[i] = fs.FileInfoToDirEntry(v)
	}
	return
}

func (f *file) Readdir(_ int) ([]fs.FileInfo, error) {
	return f.fs.client.ReadDir(f.Name())
}

func (file) Fd() uintptr {
	return 0
}

func sshConfig(u *url.URL) (c *ssh.ClientConfig, err error) {
	if u.User == nil {
		return nil, errors.New("missing username")
	}
	q, err := url.ParseQuery(u.RawQuery)
	if err != nil {
		return nil, err
	}
	c = &ssh.ClientConfig{
		Config: ssh.Config{
			//KeyExchanges: []string{"curve25519-sha256"},
			Ciphers: []string{"aes128-ctr"},
			MACs:    []string{"hmac-sha2-256"},
		},
		User:    u.User.Username(),
		Timeout: time.Second * 30,
	}
	known := q.Get("known_hosts")
	if known == "" {
		home, err := os.UserHomeDir()
		if err != nil {
			return nil, fmt.Errorf("error reading user home: %w", err)
		}
		known = filepath.Join(home, ".ssh", "known_hosts")
	}
	c.HostKeyCallback, err = knownhosts.New(known)
	if err != nil {
		return nil, err
	}
	if pw, ok := u.User.Password(); ok {
		c.Auth = []ssh.AuthMethod{ssh.Password(pw)}
	} else {
		buf, err := os.ReadFile(q.Get("key"))
		if err != nil {
			return nil, err
		}
		signer, err := ssh.ParsePrivateKey(buf)
		c.Auth = []ssh.AuthMethod{ssh.PublicKeys(signer)}
	}
	return
}