PilotVault/API Server/internal/plugins/builtin/filetransfer/filetransfer.go

// Package filetransfer is a built-in plugin that connects to a file-transfer
// server over FTP, FTPS (explicit TLS), or SFTP (SSH). It demonstrates a
// stateful third-party integration behind the plugin contract: one descriptor
// with a protocol switch, and a small protocol-agnostic `conn` abstraction that
// HealthCheck and Invoke drive without caring which wire protocol is in use.
//
// Connections are opened per operation rather than pooled: FTP/SFTP sessions are
// stateful and idle-timeout aggressively, so dialling on demand is both simpler
// and more robust than keeping a long-lived connection healthy. Init only stores
// the resolved config; nothing connects until HealthCheck or Invoke runs.
//
//   - FTP  : github.com/jlaffaye/ftp
//   - FTPS : github.com/jlaffaye/ftp with explicit TLS (AUTH TLS)
//   - SFTP : golang.org/x/crypto/ssh + github.com/pkg/sftp
package filetransfer

import (
	"context"
	"crypto/tls"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"path"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/jlaffaye/ftp"
	"github.com/pkg/sftp"
	"golang.org/x/crypto/ssh"

	"pilotvault/apiserver/internal/plugins"
)

const (
	protoSFTP = "sftp"
	protoFTP  = "ftp"
	protoFTPS = "ftps"

	dialTimeout = 12 * time.Second
	// maxReadBytes caps a download so a huge remote file can't exhaust memory;
	// the health probe and Invoke both honour it.
	maxReadBytes = 32 << 20 // 32 MiB
)

func init() {
	plugins.Register("filetransfer", func() plugins.Plugin { return &Plugin{} })
}

// Plugin is the FTP/FTPS/SFTP connector. All fields are guarded by mu because
// Init may run concurrently with a HealthCheck/Invoke from another request.
type Plugin struct {
	mu         sync.Mutex
	protocol   string
	host       string
	port       int
	username   string
	password   string
	privateKey string // PEM-encoded SSH private key (sftp only)
	keyPass    string // passphrase for the private key
	basePath   string
	// hostKeyFP, when set, pins the SFTP server's SHA256 host-key fingerprint
	// ("SHA256:…"); empty means accept any host key (trust-on-first-use, no
	// verification — flagged as degraded by the health probe).
	hostKeyFP string
	// insecureTLS skips FTPS certificate verification when true.
	insecureTLS bool
}

func (p *Plugin) Descriptor() plugins.Descriptor {
	return plugins.Descriptor{
		Name:     "filetransfer",
		Provider: "FTP / SFTP",
		Version:  "1.0.0",
		Kind:     plugins.KindBuiltin,
		Category: plugins.CategoryDrivesExternal,
		AuthType: plugins.AuthBasic,
		Capabilities: []plugins.Capability{
			{ID: "list", Method: "GET", Endpoint: "/", Description: "List a remote directory. params: {path}"},
			{ID: "stat", Method: "GET", Endpoint: "/", Description: "Stat one remote path. params: {path}"},
			{ID: "download", Method: "GET", Endpoint: "/", Description: "Read a remote file (base64, ≤32 MiB). params: {path}"},
			{ID: "upload", Method: "PUT", Endpoint: "/", Description: "Write a remote file. params: {path, contentBase64}"},
			{ID: "delete", Method: "DELETE", Endpoint: "/", Description: "Delete a remote file. params: {path}"},
			{ID: "mkdir", Method: "PUT", Endpoint: "/", Description: "Create a remote directory. params: {path}"},
		},
		ConfigFields: []plugins.ConfigField{
			// No field is Required: the plugin can be enabled as a master switch with
			// an empty global config, leaving each organization or user to supply
			// their own connection through the cascade (mirrors OpenSky). A missing
			// host is reported gracefully by the health probe.
			{Key: "protocol", Label: "Protocol", Type: "select", Default: protoSFTP,
				Options: []plugins.SelectOption{
					{Value: protoSFTP, Label: "SFTP — file transfer over SSH (recommended)"},
					{Value: protoFTPS, Label: "FTPS — FTP with explicit TLS (AUTH TLS)"},
					{Value: protoFTP, Label: "FTP — plaintext (insecure)"},
				},
				Help: "SFTP runs over SSH (port 22); FTP/FTPS use port 21 by default."},
			{Key: "host", Label: "Host", Type: "text", Help: "Server hostname or IP, e.g. files.example.com"},
			{Key: "port", Label: "Port", Type: "number", Help: "Leave blank for the protocol default (22 for SFTP, 21 for FTP/FTPS)."},
			{Key: "username", Label: "Username", Type: "text"},
			{Key: "password", Label: "Password", Type: "password", Secret: true,
				Help: "Password for FTP/FTPS, or SFTP password auth. Leave blank to use an SFTP private key."},
			{Key: "privateKey", Label: "SSH private key (SFTP)", Type: "password", Secret: true,
				Help: "PEM-encoded private key for SFTP key auth. Used instead of, or alongside, a password."},
			{Key: "keyPassphrase", Label: "Private key passphrase", Type: "password", Secret: true,
				Help: "Passphrase protecting the SSH private key, if any."},
			{Key: "basePath", Label: "Base path", Type: "text", Default: ".",
				Help: "Directory used as the working root and probed by the health check, e.g. /uploads. Relative capability paths are resolved under it."},
			{Key: "hostKeyFingerprint", Label: "SFTP host key fingerprint", Type: "text",
				Help: "Optional SHA256:… fingerprint to pin the SFTP server's host key. Leave blank to accept any key (no verification)."},
			{Key: "insecureSkipVerify", Label: "FTPS TLS verification", Type: "select", Default: "false",
				Options: []plugins.SelectOption{
					{Value: "false", Label: "Verify certificate (recommended)"},
					{Value: "true", Label: "Skip verification — accept any certificate"},
				},
				Help: "Only affects FTPS. Skip verification only for self-signed test servers."},
		},
	}
}

func (p *Plugin) Init(_ context.Context, config map[string]string) error {
	p.mu.Lock()
	defer p.mu.Unlock()

	p.protocol = strings.ToLower(strings.TrimSpace(config["protocol"]))
	if p.protocol == "" {
		p.protocol = protoSFTP
	}
	p.host = strings.TrimSpace(config["host"])
	p.port = 0
	if raw := strings.TrimSpace(config["port"]); raw != "" {
		if n, err := strconv.Atoi(raw); err == nil {
			p.port = n
		}
	}
	p.username = strings.TrimSpace(config["username"])
	p.password = config["password"]
	p.privateKey = config["privateKey"]
	p.keyPass = config["keyPassphrase"]
	p.basePath = strings.TrimSpace(config["basePath"])
	if p.basePath == "" {
		p.basePath = "."
	}
	p.hostKeyFP = strings.TrimSpace(config["hostKeyFingerprint"])
	p.insecureTLS = strings.EqualFold(strings.TrimSpace(config["insecureSkipVerify"]), "true")
	return nil
}

// effectivePort returns the configured port or the protocol default.
func (p *Plugin) effectivePort() int {
	if p.port > 0 {
		return p.port
	}
	if p.protocol == protoSFTP {
		return 22
	}
	return 21
}

// resolve joins a caller-supplied path against the base path. An absolute path
// is used as-is; an empty path becomes the base path itself.
func (p *Plugin) resolve(rel string) string {
	rel = strings.TrimSpace(rel)
	if rel == "" {
		return p.basePath
	}
	if strings.HasPrefix(rel, "/") || p.basePath == "" || p.basePath == "." {
		return rel
	}
	return path.Join(p.basePath, rel)
}

// HealthCheck dials, authenticates, and lists the base path, classifying the
// outcome. A missing/unverified SFTP host key downgrades OK to degraded.
func (p *Plugin) HealthCheck(ctx context.Context) plugins.Health {
	start := time.Now()

	p.mu.Lock()
	proto, host, hostKeyFP := p.protocol, p.host, p.hostKeyFP
	base := p.basePath
	p.mu.Unlock()

	if host == "" {
		return plugins.Health{Status: plugins.StatusDown, Detail: "no host configured"}
	}

	c, err := p.dial(ctx)
	if err != nil {
		lat := time.Since(start).Milliseconds()
		return plugins.Health{Status: classifyDialErr(err), LatencyMs: lat, Detail: err.Error()}
	}
	defer c.close()

	entries, err := c.list(base)
	lat := time.Since(start).Milliseconds()
	if err != nil {
		return plugins.Health{Status: plugins.StatusDegraded, LatencyMs: lat,
			Detail: fmt.Sprintf("connected (%s) but listing %q failed: %v", proto, base, err)}
	}

	detail := fmt.Sprintf("%s reachable — %d entr%s under %q", strings.ToUpper(proto), len(entries), plural(len(entries)), base)
	status := plugins.StatusOK
	if proto == protoSFTP && hostKeyFP == "" {
		status = plugins.StatusDegraded
		detail += " · host key not verified (no fingerprint pinned)"
	}
	if proto == protoFTP {
		detail += " · plaintext (no encryption)"
	}
	return plugins.Health{Status: status, LatencyMs: lat, Detail: detail}
}

// Invoke runs one capability against a freshly-dialled connection.
func (p *Plugin) Invoke(ctx context.Context, action string, params json.RawMessage) (json.RawMessage, error) {
	c, err := p.dial(ctx)
	if err != nil {
		return nil, err
	}
	defer c.close()

	switch action {
	case "list":
		var in pathParams
		_ = json.Unmarshal(params, &in)
		entries, err := c.list(p.resolve(in.Path))
		if err != nil {
			return nil, err
		}
		return json.Marshal(map[string]any{"path": p.resolve(in.Path), "entries": entries})

	case "stat":
		var in pathParams
		_ = json.Unmarshal(params, &in)
		fi, err := c.stat(p.resolve(in.Path))
		if err != nil {
			return nil, err
		}
		return json.Marshal(fi)

	case "download":
		var in pathParams
		_ = json.Unmarshal(params, &in)
		data, err := c.read(p.resolve(in.Path))
		if err != nil {
			return nil, err
		}
		return json.Marshal(map[string]any{
			"path":          p.resolve(in.Path),
			"size":          len(data),
			"contentBase64": base64.StdEncoding.EncodeToString(data),
		})

	case "upload":
		var in writeParams
		if err := json.Unmarshal(params, &in); err != nil {
			return nil, fmt.Errorf("invalid params: %w", err)
		}
		data, err := base64.StdEncoding.DecodeString(in.ContentBase64)
		if err != nil {
			return nil, fmt.Errorf("contentBase64 is not valid base64: %w", err)
		}
		if err := c.write(p.resolve(in.Path), data); err != nil {
			return nil, err
		}
		return json.Marshal(map[string]any{"path": p.resolve(in.Path), "size": len(data), "ok": true})

	case "delete":
		var in pathParams
		_ = json.Unmarshal(params, &in)
		if err := c.remove(p.resolve(in.Path)); err != nil {
			return nil, err
		}
		return json.Marshal(map[string]any{"path": p.resolve(in.Path), "ok": true})

	case "mkdir":
		var in pathParams
		_ = json.Unmarshal(params, &in)
		if err := c.mkdir(p.resolve(in.Path)); err != nil {
			return nil, err
		}
		return json.Marshal(map[string]any{"path": p.resolve(in.Path), "ok": true})

	default:
		return nil, errors.New("unknown action: " + action)
	}
}

func (p *Plugin) Shutdown(context.Context) error { return nil }

// pathParams / writeParams are the Invoke request shapes.
type pathParams struct {
	Path string `json:"path"`
}
type writeParams struct {
	Path          string `json:"path"`
	ContentBase64 string `json:"contentBase64"`
}

// fileInfo is the normalized directory-entry shape returned by list/stat.
type fileInfo struct {
	Name    string `json:"name"`
	Size    int64  `json:"size"`
	IsDir   bool   `json:"isDir"`
	ModTime string `json:"modTime,omitempty"`
}

// conn is the protocol-agnostic surface HealthCheck and Invoke drive. Both the
// FTP and SFTP implementations satisfy it.
type conn interface {
	list(path string) ([]fileInfo, error)
	stat(path string) (fileInfo, error)
	read(path string) ([]byte, error)
	write(path string, data []byte) error
	remove(path string) error
	mkdir(path string) error
	close() error
}

// dial builds an authenticated connection for the configured protocol.
func (p *Plugin) dial(ctx context.Context) (conn, error) {
	p.mu.Lock()
	proto := p.protocol
	p.mu.Unlock()

	switch proto {
	case protoSFTP:
		return p.dialSFTP(ctx)
	case protoFTP, protoFTPS:
		return p.dialFTP(ctx)
	default:
		return nil, errors.New("unsupported protocol: " + proto)
	}
}

// classifyDialErr maps a dial/auth failure to a health status: an auth rejection
// is degraded (server reachable, credentials wrong); anything else is down.
func classifyDialErr(err error) string {
	msg := strings.ToLower(err.Error())
	switch {
	case strings.Contains(msg, "unable to authenticate"),
		strings.Contains(msg, "auth"),
		strings.Contains(msg, "password"),
		strings.Contains(msg, "login"),
		strings.Contains(msg, "530"), // FTP: not logged in
		strings.Contains(msg, "permission denied"):
		return plugins.StatusDegraded
	default:
		return plugins.StatusDown
	}
}

func plural(n int) string {
	if n == 1 {
		return "y"
	}
	return "ies"
}

// ---------------------------------------------------------------------------
// SFTP implementation
// ---------------------------------------------------------------------------

type sftpConn struct {
	ssh *ssh.Client
	cli *sftp.Client
}

func (p *Plugin) dialSFTP(ctx context.Context) (conn, error) {
	p.mu.Lock()
	host, user, pass := p.host, p.username, p.password
	key, keyPass, hostKeyFP := p.privateKey, p.keyPass, p.hostKeyFP
	addr := net.JoinHostPort(host, strconv.Itoa(p.effectivePort()))
	p.mu.Unlock()

	var auth []ssh.AuthMethod
	if strings.TrimSpace(key) != "" {
		signer, err := parseSigner(key, keyPass)
		if err != nil {
			return nil, fmt.Errorf("private key: %w", err)
		}
		auth = append(auth, ssh.PublicKeys(signer))
	}
	if pass != "" {
		auth = append(auth, ssh.Password(pass))
	}
	if len(auth) == 0 {
		return nil, errors.New("SFTP requires a password or a private key")
	}

	hostKeyCallback, err := hostKeyChecker(hostKeyFP)
	if err != nil {
		return nil, err
	}
	cfg := &ssh.ClientConfig{
		User:            user,
		Auth:            auth,
		HostKeyCallback: hostKeyCallback,
		Timeout:         dialTimeout,
	}

	// ssh.Dial has no context form; dial the TCP conn with the context, then
	// run the SSH handshake over it.
	d := net.Dialer{Timeout: dialTimeout}
	tcp, err := d.DialContext(ctx, "tcp", addr)
	if err != nil {
		return nil, err
	}
	sshConn, chans, reqs, err := ssh.NewClientConn(tcp, addr, cfg)
	if err != nil {
		_ = tcp.Close()
		return nil, err
	}
	client := ssh.NewClient(sshConn, chans, reqs)
	sc, err := sftp.NewClient(client)
	if err != nil {
		_ = client.Close()
		return nil, err
	}
	return &sftpConn{ssh: client, cli: sc}, nil
}

// parseSigner parses a PEM private key, with or without a passphrase.
func parseSigner(pem, passphrase string) (ssh.Signer, error) {
	if strings.TrimSpace(passphrase) != "" {
		return ssh.ParsePrivateKeyWithPassphrase([]byte(pem), []byte(passphrase))
	}
	return ssh.ParsePrivateKey([]byte(pem))
}

// hostKeyChecker returns a HostKeyCallback that pins the given SHA256:…
// fingerprint, or accepts any key when the fingerprint is empty.
func hostKeyChecker(fingerprint string) (ssh.HostKeyCallback, error) {
	if fingerprint == "" {
		return ssh.InsecureIgnoreHostKey(), nil //nolint:gosec // opt-in: no fingerprint pinned
	}
	want := strings.TrimSpace(fingerprint)
	return func(_ string, _ net.Addr, key ssh.PublicKey) error {
		got := ssh.FingerprintSHA256(key)
		if got != want {
			return fmt.Errorf("host key mismatch: server presented %s, expected %s", got, want)
		}
		return nil
	}, nil
}

func (c *sftpConn) list(p string) ([]fileInfo, error) {
	infos, err := c.cli.ReadDir(p)
	if err != nil {
		return nil, err
	}
	out := make([]fileInfo, 0, len(infos))
	for _, fi := range infos {
		out = append(out, fileInfo{
			Name:    fi.Name(),
			Size:    fi.Size(),
			IsDir:   fi.IsDir(),
			ModTime: fi.ModTime().UTC().Format(time.RFC3339),
		})
	}
	return out, nil
}

func (c *sftpConn) stat(p string) (fileInfo, error) {
	fi, err := c.cli.Stat(p)
	if err != nil {
		return fileInfo{}, err
	}
	return fileInfo{
		Name:    fi.Name(),
		Size:    fi.Size(),
		IsDir:   fi.IsDir(),
		ModTime: fi.ModTime().UTC().Format(time.RFC3339),
	}, nil
}

func (c *sftpConn) read(p string) ([]byte, error) {
	f, err := c.cli.Open(p)
	if err != nil {
		return nil, err
	}
	defer f.Close()
	return io.ReadAll(io.LimitReader(f, maxReadBytes))
}

func (c *sftpConn) write(p string, data []byte) error {
	f, err := c.cli.Create(p)
	if err != nil {
		return err
	}
	defer f.Close()
	_, err = f.Write(data)
	return err
}

func (c *sftpConn) remove(p string) error { return c.cli.Remove(p) }
func (c *sftpConn) mkdir(p string) error  { return c.cli.MkdirAll(p) }

func (c *sftpConn) close() error {
	err := c.cli.Close()
	if c.ssh != nil {
		_ = c.ssh.Close()
	}
	return err
}

// ---------------------------------------------------------------------------
// FTP / FTPS implementation
// ---------------------------------------------------------------------------

type ftpConn struct {
	c *ftp.ServerConn
}

func (p *Plugin) dialFTP(ctx context.Context) (conn, error) {
	p.mu.Lock()
	host, user, pass, proto := p.host, p.username, p.password, p.protocol
	insecure := p.insecureTLS
	addr := net.JoinHostPort(host, strconv.Itoa(p.effectivePort()))
	p.mu.Unlock()

	opts := []ftp.DialOption{ftp.DialWithContext(ctx), ftp.DialWithTimeout(dialTimeout)}
	if proto == protoFTPS {
		opts = append(opts, ftp.DialWithExplicitTLS(&tls.Config{
			ServerName:         host,
			InsecureSkipVerify: insecure, //nolint:gosec // opt-in for self-signed test servers
		}))
	}

	sc, err := ftp.Dial(addr, opts...)
	if err != nil {
		return nil, err
	}
	if err := sc.Login(user, pass); err != nil {
		_ = sc.Quit()
		return nil, err
	}
	return &ftpConn{c: sc}, nil
}

func (c *ftpConn) list(p string) ([]fileInfo, error) {
	entries, err := c.c.List(p)
	if err != nil {
		return nil, err
	}
	out := make([]fileInfo, 0, len(entries))
	for _, e := range entries {
		if e.Name == "." || e.Name == ".." {
			continue
		}
		out = append(out, entryToInfo(e))
	}
	return out, nil
}

func (c *ftpConn) stat(p string) (fileInfo, error) {
	// FTP has no portable stat; MLST via GetEntry works on servers that support
	// it, otherwise fall back to listing the parent and matching the name.
	if e, err := c.c.GetEntry(p); err == nil && e != nil {
		return entryToInfo(e), nil
	}
	dir, base := path.Split(strings.TrimRight(p, "/"))
	if dir == "" {
		dir = "."
	}
	entries, err := c.c.List(dir)
	if err != nil {
		return fileInfo{}, err
	}
	for _, e := range entries {
		if e.Name == base {
			return entryToInfo(e), nil
		}
	}
	return fileInfo{}, fmt.Errorf("not found: %s", p)
}

func (c *ftpConn) read(p string) ([]byte, error) {
	resp, err := c.c.Retr(p)
	if err != nil {
		return nil, err
	}
	defer resp.Close()
	return io.ReadAll(io.LimitReader(resp, maxReadBytes))
}

func (c *ftpConn) write(p string, data []byte) error {
	return c.c.Stor(p, strings.NewReader(string(data)))
}

func (c *ftpConn) remove(p string) error { return c.c.Delete(p) }
func (c *ftpConn) mkdir(p string) error  { return c.c.MakeDir(p) }

func (c *ftpConn) close() error { return c.c.Quit() }

// entryToInfo normalizes a jlaffaye/ftp entry.
func entryToInfo(e *ftp.Entry) fileInfo {
	fi := fileInfo{
		Name:  e.Name,
		Size:  int64(e.Size),
		IsDir: e.Type == ftp.EntryTypeFolder,
	}
	if !e.Time.IsZero() {
		fi.ModTime = e.Time.UTC().Format(time.RFC3339)
	}
	return fi
}