added sftp client implementation

pkg/provider/interface.go

@@ -3,10 +3,13 @@
 package provider
 
 import (
+	"errors"
 	"io/fs"
 	"os"
 )
 
+var ErrFstat = errors.New("fstat not supported")
+
 type FS interface {
 	Stat(string) (fs.FileInfo, error)
 	Remove(string) error
@@ -15,6 +18,7 @@ type FS interface {
 	MkdirAll(string, os.FileMode) error
 	Root() string
 	Fstat(uintptr) (int64, error)
+	Close() error
 }
 
 type walk struct {

pkg/provider/mount/mount.go

@@ -18,8 +18,8 @@ type FS struct {
 	root string
 }
 
-func NewFS(root string) *FS {
-	return &FS{root}
+func NewFS(root string) (*FS, error) {
+	return &FS{root}, nil
 }
 
 func (fs *FS) path(p string) string {
@@ -55,3 +55,7 @@ func (fs *FS) Fstat(fd uintptr) (int64, error) {
 	err := unix.Fstat(int(fd), &stat)
 	return stat.Blocks, err
 }
+
+func (FS) Close() error {
+	return nil
+}

pkg/provider/parse/parse.go

@@ -5,10 +5,13 @@ package parse
 import (
 	"cachefs/pkg/provider"
 	"cachefs/pkg/provider/mount"
+	"cachefs/pkg/provider/sftp"
 	"errors"
 	"fmt"
 	neturl "net/url"
 	"path/filepath"
+
+	"k8s.io/klog/v2/klogr"
 )
 
 var (
@@ -26,9 +29,9 @@ func FS(url string) (provider.FS, error) {
 	}
 	switch u.Scheme {
 	case "file":
-		return mount.NewFS(u.Path), nil
-		//case "sftp":
-		//	return sftp.FS(u.Path), nil
+		return mount.NewFS(u.Path)
+	case "sftp":
+		return sftp.NewFS(u, klogr.New().WithName("sftp"))
 	}
 	return nil, fmt.Errorf("%w: %s", ErrUnsupportedScheme, u.Scheme)
 }

pkg/provider/sftp/error.go

@@ -0,0 +1,45 @@
+// Copyright (C) 2022 Marius Schellenberger
+package sftp
+
+import (
+	"errors"
+	"io/fs"
+
+	"github.com/pkg/sftp"
+)
+
+var ErrNotConnected = errors.New("sftp not connected")
+
+type connError struct{}
+
+func (connError) Remove(_ string) error {
+	return ErrNotConnected
+}
+
+func (connError) Open(_ string) (*sftp.File, error) {
+	return nil, ErrNotConnected
+}
+
+func (connError) OpenFile(_ string, _ int) (*sftp.File, error) {
+	return nil, ErrNotConnected
+}
+
+func (connError) Stat(_ string) (fs.FileInfo, error) {
+	return nil, ErrNotConnected
+}
+
+func (connError) ReadDir(_ string) ([]fs.FileInfo, error) {
+	return nil, ErrNotConnected
+}
+
+func (connError) MkdirAll(_ string) error {
+	return ErrNotConnected
+}
+
+func (connError) Wait() error {
+	return ErrNotConnected
+}
+
+func (connError) Close() error {
+	return nil
+}

pkg/provider/sftp/sftp.go

@@ -2,4 +2,241 @@
 
 package sftp
 
-// github.com/pkg/sftp
+import (
+	"cachefs/pkg/provider"
+	"context"
+	"errors"
+	"io/fs"
+	"net"
+	"net/url"
+	"os"
+	"path"
+	"path/filepath"
+	"time"
+
+	"github.com/go-logr/logr"
+	"github.com/pkg/sftp"
+	"golang.org/x/crypto/ssh"
+)
+
+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, 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)
+	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))
+			select {
+			case <-ctx.Done():
+				return
+			case err = <-fs.wait(ctx):
+				log.V(2).Info("connection lost", "err", err)
+				fs.conn.Close()
+				fs.client = connError{}
+				fs.c = nil
+			}
+		}
+		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")
+	}
+	//var hostKey ssh.PublicKey
+	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,
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
+		//HostKeyCallback: ssh.FixedHostKey(hostKey),
+	}
+	if pw, ok := u.User.Password(); ok {
+		c.Auth = []ssh.AuthMethod{ssh.Password(pw)}
+	} else {
+		q, err := url.ParseQuery(u.RawQuery)
+		if err != nil {
+			return nil, err
+		}
+
+		key, err := os.ReadFile(q.Get("keyfile"))
+		if err != nil {
+			return nil, err
+		}
+		signer, err := ssh.ParsePrivateKey(key)
+		c.Auth = []ssh.AuthMethod{ssh.PublicKeys(signer)}
+	}
+	return
+}