// 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 }