feat(sftp): real SFTP via the File Transfer protocol dropdown (#16)

Selecting SFTP in the File Transfer (src.ftp) Protocol dropdown now does real
SFTP (SSH), not a broken FTP attempt. SFTP runs on russh + russh-sftp with the
ring crypto backend (default-features=false drops aws-lc-rs, so no NASM and no
C toolchain; reuses the ring stack we already ship, builds on all release
targets incl. windows-arm64). The old 'blocked on aws-lc-sys NASM' note was
outdated.

- protocol=sftp routes to SftpSourceSpec + run_sftp_source (async russh wrapped
  in block_on like the mongo / tiberius sources). Password or OpenSSH
  private-key auth (paste or key file); optional SHA256 host-fingerprint pin
  enforced in check_server_key (the Host Fingerprint ask). Lists a directory,
  glob filter, downloads up to maxFiles, emits {filename, size, content_b64,
  modified} - same row shape as FTP.
- reconcile the form field keys with the engine (user / directory, with
  username / remotePath fallbacks); add key-passphrase + host-fingerprint fields.
- README: Sources table + roadmap updated.

Verified live against a public SFTP server: connect, host fingerprint, password
auth, sftp subsystem, directory listing with metadata.

Author: SouravRoy-ETL

Cargo.lock

@@ -215,7 +215,7 @@ Duckle is not a CSV tool with extras. It reads a broad set of formats and source
 | **NoSQL and search** | MongoDB (official driver), Cassandra / ScyllaDB (CQL), Elasticsearch / OpenSearch (from+size + search_after), Redis (SCAN + GET), CouchDB (`_all_docs`), DynamoDB (HTTP + SigV4 - no AWS SDK; auto-unwraps typed attributes) | Available |
 | **Vector / AI databases** | pgvector (postgres ATTACH), Qdrant (`/points/scroll`), Weaviate (`/v1/objects`), Milvus (`/v1/vector/query`) | Available |
 | **Vector / AI databases** | Pinecone (no list-all-vectors API), Chroma, LanceDB | Preview |
-| **File transfer** | FTP / FTPS (pure-Rust `suppaftp` with glob filter and base64-content per file) | Available |
+| **File transfer** | FTP / FTPS (pure-Rust `suppaftp`) and SFTP (SSH, pure-Rust `russh` + `russh-sftp` on the ring backend; password or private-key auth, optional host-fingerprint pin) - one File Transfer component, pick the protocol. Glob filter, base64 content per file | Available |
 | **Mailbox** | IMAP (rustls TLS, `mail-parser`) - basic auth today, OAuth (gmail / o365) on the roadmap | Available |
 | **Webhook listener** | Binds `127.0.0.1:port`, collects N inbound HTTP requests with a timeout, parses JSON-object / JSON-array bodies into rows | Available |
 | **Desktop** | System clipboard (pure-Rust `arboard`, auto-detects JSON-array shape) | Available |
@@ -991,7 +991,7 @@ A complete planned-component breakdown lives in [`docs/roadmap.md`](docs/roadmap
 
 - [ ] **Multi-shard Kinesis** and **Pulsar** streaming (Pulsar blocked on `protoc` at build time)
 - [ ] **Apache ORC** read / write (blocked on the Arrow version conflict between `orc-rust` and our workspace pin)
-- [ ] **SFTP** source (blocked on the `aws-lc-sys` NASM build dep in `russh`)
+- [x] **SFTP** source (shipped - `russh` + `russh-sftp` on the ring backend, password / key auth, host-fingerprint pin)
 - [ ] **OAuth-heavy SaaS** (Google Sheets, Excel Online, full Salesforce OAuth, Gmail / O365 IMAP)
 - [ ] **Embedded Python / Rust** code stages (current code.* family: SQL, Shell, JavaScript, WebAssembly all ship)
 - [ ] **Hosted documentation site**

README.md

@@ -97,10 +97,17 @@ quick-xml = "0.36"
 lapin = { version = "2", default-features = false, features = ["rustls"] }
 # FTP / FTPS client. Pure Rust, synchronous API (fits the per-stage
 # blocking model - no block_on dance). Rustls-backed TLS keeps the
-# build self-contained on every OS. SFTP (SSH-based, different
-# protocol) is on the roadmap; russh pulls in aws-lc-sys which
-# requires NASM at build time, breaking the self-contained build.
+# build self-contained on every OS.
 suppaftp = { version = "8", default-features = false, features = ["rustls-ring"] }
+# SFTP (SSH-based) source, src.sftp. russh + russh-sftp on the `ring`
+# crypto backend (default-features=false drops aws-lc-rs, so NO NASM and
+# NO C toolchain - the same ring 0.17 stack the rest of our TLS already
+# uses, building on all release targets incl. windows-arm64). russh is
+# async; run_sftp_source wraps it in block_on like the mongo / tiberius
+# sources to stay synchronous. Host-key/fingerprint is enforced in russh's
+# check_server_key callback.
+russh = { version = "0.54", default-features = false, features = ["ring", "flate2", "rsa"] }
+russh-sftp = "2.3"
 # System clipboard reader (src.clipboard). Text-only - image-data is
 # off because we never round-trip images through pipelines. arboard
 # is pure-Rust on Windows and macOS; on Linux it talks to X11/Wayland

crates/duckdb-engine/Cargo.toml

@@ -2711,6 +2711,164 @@ impl DuckdbEngine {
         ))
     }
 
+    /// src.sftp: connect over SSH, verify the host key against an optional
+    /// SHA256 fingerprint pin, authenticate (private key or password), list
+    /// `directory`, filter by optional glob `pattern`, download up to
+    /// `max_files`. Each file becomes a row {filename, size, content_b64,
+    /// modified}. russh / russh-sftp are async (ring backend); we drive them
+    /// on a private current-thread tokio runtime so the stage stays blocking
+    /// like every other source.
+    pub(crate) fn run_sftp_source(&self, db: &Path, spec: &SftpSourceSpec) -> Result<String, EngineError> {
+        use base64::engine::general_purpose::STANDARD as B64;
+        use base64::Engine as _;
+        self.check_cancelled()?;
+
+        // Host-key verification. With a pinned fingerprint, refuse any other
+        // server key; without one, accept on trust (trust-on-first-use).
+        struct Verifier {
+            expected: Option<String>,
+        }
+        impl russh::client::Handler for Verifier {
+            type Error = russh::Error;
+            async fn check_server_key(
+                &mut self,
+                server_public_key: &russh::keys::ssh_key::PublicKey,
+            ) -> Result<bool, Self::Error> {
+                match &self.expected {
+                    None => Ok(true),
+                    Some(want) => {
+                        let got = server_public_key
+                            .fingerprint(russh::keys::HashAlg::Sha256)
+                            .to_string();
+                        // Compare case-sensitively but tolerant of the
+                        // "SHA256:" prefix on either side.
+                        let norm = |s: &str| s.trim().trim_start_matches("SHA256:").to_string();
+                        Ok(norm(&got) == norm(want))
+                    }
+                }
+            }
+        }
+
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .map_err(|e| EngineError::Query(format!("sftp: tokio rt: {}", e)))?;
+
+        let result: Result<Vec<JsonValue>, String> = rt.block_on(async {
+            use russh_sftp::client::SftpSession;
+            use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+            let config = std::sync::Arc::new(russh::client::Config::default());
+            let handler = Verifier {
+                expected: spec.host_fingerprint.clone(),
+            };
+            let mut session =
+                russh::client::connect(config, (spec.host.as_str(), spec.port), handler)
+                    .await
+                    .map_err(|e| format!("connect {}:{}: {}", spec.host, spec.port, e))?;
+
+            // Auth: a private key wins over a password if both are present.
+            let authed = if let Some(pem) = &spec.private_key {
+                let key = russh::keys::decode_secret_key(pem, spec.key_passphrase.as_deref())
+                    .map_err(|e| format!("private key: {}", e))?;
+                let with_alg = russh::keys::PrivateKeyWithHashAlg::new(
+                    std::sync::Arc::new(key),
+                    Some(russh::keys::HashAlg::Sha256),
+                );
+                session
+                    .authenticate_publickey(spec.user.as_str(), with_alg)
+                    .await
+                    .map_err(|e| format!("publickey auth: {}", e))?
+                    .success()
+            } else if let Some(pw) = &spec.password {
+                session
+                    .authenticate_password(spec.user.as_str(), pw)
+                    .await
+                    .map_err(|e| format!("password auth: {}", e))?
+                    .success()
+            } else {
+                return Err("no credentials: set a password or a private key".into());
+            };
+            if !authed {
+                return Err(format!(
+                    "authentication failed for user '{}' (check credentials / host fingerprint)",
+                    spec.user
+                ));
+            }
+
+            let channel = session
+                .channel_open_session()
+                .await
+                .map_err(|e| format!("open channel: {}", e))?;
+            channel
+                .request_subsystem(true, "sftp")
+                .await
+                .map_err(|e| format!("request sftp subsystem: {}", e))?;
+            let sftp = SftpSession::new(channel.into_stream())
+                .await
+                .map_err(|e| format!("sftp session: {}", e))?;
+
+            let entries = sftp
+                .read_dir(spec.directory.clone())
+                .await
+                .map_err(|e| format!("read_dir {}: {}", spec.directory, e))?;
+
+            let mut rows: Vec<JsonValue> = Vec::new();
+            for entry in entries {
+                if rows.len() as u64 >= spec.max_files {
+                    break;
+                }
+                if entry.file_type().is_dir() {
+                    continue;
+                }
+                let name = entry.file_name();
+                if let Some(p) = &spec.pattern {
+                    if !glob_match(p, &name) {
+                        continue;
+                    }
+                }
+                let meta = entry.metadata();
+                let size = meta.size.map(|n| n as i64);
+                let modified = meta.mtime.and_then(|t| {
+                    chrono::DateTime::<chrono::Utc>::from_timestamp(t as i64, 0)
+                        .map(|dt| dt.format("%Y-%m-%dT%H:%M:%SZ").to_string())
+                });
+                let full = entry.path();
+                let mut file = sftp
+                    .open(full.clone())
+                    .await
+                    .map_err(|e| format!("open {}: {}", full, e))?;
+                let mut bytes = Vec::new();
+                file.read_to_end(&mut bytes)
+                    .await
+                    .map_err(|e| format!("read {}: {}", full, e))?;
+                let _ = file.shutdown().await;
+
+                let mut row = serde_json::Map::new();
+                row.insert("filename".into(), JsonValue::String(name));
+                row.insert(
+                    "size".into(),
+                    size.map(JsonValue::from).unwrap_or(JsonValue::Null),
+                );
+                row.insert(
+                    "modified".into(),
+                    modified.map(JsonValue::String).unwrap_or(JsonValue::Null),
+                );
+                row.insert("content_b64".into(), JsonValue::String(B64.encode(&bytes)));
+                rows.push(JsonValue::Object(row));
+            }
+            Ok(rows)
+        });
+
+        let rows = result.map_err(EngineError::Query)?;
+        let count = rows.len();
+        materialize_jsonobjects_as_table(&self.bin, db, &spec.node_id, &rows)?;
+        Ok(format!(
+            "sftp: materialized {} file(s) from {}:{} into {}",
+            count, spec.host, spec.port, spec.node_id
+        ))
+    }
+
     /// xf.ai.embed: per-row embedding via an OpenAI-compatible API.
     /// Reads the upstream view, batches rows into groups of
     /// batch_size, sends the input_column text array to /v1/embeddings,

crates/duckdb-engine/src/connectors.rs

@@ -42,8 +42,9 @@ use plan::{
     NatsSinkSpec, NatsSourceSpec, OracleSinkSpec, OracleSourceSpec, PubSubSinkSpec,
     PubSubSourceSpec, QdrantSourceSpec, RabbitSinkSpec, RabbitSourceSpec, RedisSinkSpec,
     RedisSourceSpec, RestPagination, RestResponseFormat, RestSourceSpec, RuntimeSpec, ShellSpec,
-    SnowflakeAuth, SnowflakeSinkSpec, SnowflakeSourceSpec, SqlServerSinkSpec, SqlServerSourceSpec,
-    WasmSpec, WeaviateSourceSpec, WebhookSourceSpec, WebhookSpec, XmlSinkSpec, XmlSourceSpec,
+    SftpSourceSpec, SnowflakeAuth, SnowflakeSinkSpec, SnowflakeSourceSpec, SqlServerSinkSpec,
+    SqlServerSourceSpec, WasmSpec, WeaviateSourceSpec, WebhookSourceSpec, WebhookSpec, XmlSinkSpec,
+    XmlSourceSpec,
 };
 
 #[derive(Debug, Error)]
@@ -834,6 +835,7 @@ impl DuckdbEngine {
                     Some(RuntimeSpec::GitSource(spec)) => self.run_git_source(&db_path, spec),
                     Some(RuntimeSpec::Shell(spec)) => self.run_shell(&db_path, spec),
                     Some(RuntimeSpec::FtpSource(spec)) => self.run_ftp_source(&db_path, spec),
+                    Some(RuntimeSpec::SftpSource(spec)) => self.run_sftp_source(&db_path, spec),
                     Some(RuntimeSpec::ClipboardSource(spec)) => {
                         self.run_clipboard_source(&db_path, spec)
                     }

crates/duckdb-engine/src/lib.rs

@@ -140,6 +140,7 @@ pub enum RuntimeSpec {
     GitSource(GitSourceSpec),
     Shell(ShellSpec),
     FtpSource(FtpSourceSpec),
+    SftpSource(SftpSourceSpec),
     ClipboardSource(ClipboardSourceSpec),
     EmailSource(EmailSourceSpec),
     EmailSink(EmailSinkSpec),
@@ -510,6 +511,7 @@ fn build_stage(
     let mut git_source: Option<GitSourceSpec> = None;
     let mut shell: Option<ShellSpec> = None;
     let mut ftp_source: Option<FtpSourceSpec> = None;
+    let mut sftp_source: Option<SftpSourceSpec> = None;
     let mut clipboard_source: Option<ClipboardSourceSpec> = None;
     let mut email_source: Option<EmailSourceSpec> = None;
     let mut email_sink: Option<EmailSinkSpec> = None;
@@ -1916,39 +1918,73 @@ fn build_stage(
         });
         (String::new(), StageKind::View, None)
     } else if component_id == "src.ftp" {
-        // FTP / FTPS list+download. List files at `directory`, filter
-        // by optional glob `pattern` (* and ? wildcards), download
-        // each up to `maxFiles`. Each file becomes a row with the
-        // bytes as a base64 string in `content` (so the row is JSON-
-        // serializable and round-trips through DuckDB cleanly).
+        // File-transfer source. The Protocol dropdown selects FTP, FTPS, or
+        // SFTP. FTP / FTPS go through suppaftp; SFTP (SSH - a different
+        // protocol) goes through russh + russh-sftp (issue #16). All three
+        // list files at `directory`, filter by optional glob `pattern`,
+        // download up to `maxFiles`, and emit one row per file
+        // {filename, size, content_b64, modified}.
+        let protocol = string_prop(&props, "protocol")
+            .unwrap_or_default()
+            .to_ascii_lowercase();
         let host = string_prop(&props, "host")
             .filter(|s| !s.is_empty())
             .ok_or_else(|| EngineError::Config(format!("{}: host required", component_id)))?;
-        ftp_source = Some(FtpSourceSpec {
-            node_id: node.id.clone(),
-            host,
-            port: props
-                .get("port")
-                .and_then(|v| v.as_u64())
-                .filter(|n| *n > 0 && *n < 65536)
-                .map(|n| n as u16)
-                .unwrap_or(21),
-            user: string_prop(&props, "user").unwrap_or_else(|| "anonymous".into()),
-            password: string_prop(&props, "password").unwrap_or_else(|| "anonymous@".into()),
-            secure: props
-                .get("secure")
-                .and_then(|v| v.as_bool())
-                .unwrap_or(false),
-            directory: string_prop(&props, "directory")
-                .filter(|s| !s.is_empty())
-                .unwrap_or_else(|| "/".into()),
-            pattern: string_prop(&props, "pattern").filter(|s| !s.is_empty()),
-            max_files: props
-                .get("maxFiles")
-                .and_then(|v| v.as_u64())
-                .filter(|n| *n > 0)
-                .unwrap_or(100),
-        });
+        // The form historically wrote `username` / `remotePath`; accept those
+        // as fallbacks for the canonical `user` / `directory`.
+        let user = string_prop(&props, "user")
+            .or_else(|| string_prop(&props, "username"))
+            .filter(|s| !s.is_empty());
+        let directory = string_prop(&props, "directory")
+            .or_else(|| string_prop(&props, "remotePath"))
+            .filter(|s| !s.is_empty());
+        let pattern = string_prop(&props, "pattern").filter(|s| !s.is_empty());
+        let max_files = props
+            .get("maxFiles")
+            .and_then(|v| v.as_u64())
+            .filter(|n| *n > 0)
+            .unwrap_or(100);
+        let port = props
+            .get("port")
+            .and_then(|v| v.as_u64())
+            .filter(|n| *n > 0 && *n < 65536)
+            .map(|n| n as u16);
+        if protocol == "sftp" {
+            sftp_source = Some(SftpSourceSpec {
+                node_id: node.id.clone(),
+                host,
+                port: port.unwrap_or(22),
+                user: user.ok_or_else(|| {
+                    EngineError::Config(format!("{}: user required for SFTP", component_id))
+                })?,
+                password: string_prop(&props, "password").filter(|s| !s.is_empty()),
+                // Accept a pasted PEM (privateKey) or a key file (privateKeyPath).
+                private_key: string_prop(&props, "privateKey")
+                    .or_else(|| {
+                        string_prop(&props, "privateKeyPath")
+                            .and_then(|p| std::fs::read_to_string(&p).ok())
+                    })
+                    .filter(|s| !s.is_empty()),
+                key_passphrase: string_prop(&props, "keyPassphrase").filter(|s| !s.is_empty()),
+                directory: directory.unwrap_or_else(|| ".".into()),
+                pattern,
+                max_files,
+                host_fingerprint: string_prop(&props, "hostFingerprint").filter(|s| !s.is_empty()),
+            });
+        } else {
+            ftp_source = Some(FtpSourceSpec {
+                node_id: node.id.clone(),
+                host,
+                port: port.unwrap_or(21),
+                user: user.unwrap_or_else(|| "anonymous".into()),
+                password: string_prop(&props, "password").unwrap_or_else(|| "anonymous@".into()),
+                secure: protocol == "ftps"
+                    || props.get("secure").and_then(|v| v.as_bool()).unwrap_or(false),
+                directory: directory.unwrap_or_else(|| "/".into()),
+                pattern,
+                max_files,
+            });
+        }
         (String::new(), StageKind::View, None)
     } else if component_id == "src.xml" {
         // XML row-path source. rowPath is a slash-separated element
@@ -2972,6 +3008,7 @@ fn build_stage(
         .or_else(|| git_source.map(RuntimeSpec::GitSource))
         .or_else(|| shell.map(RuntimeSpec::Shell))
         .or_else(|| ftp_source.map(RuntimeSpec::FtpSource))
+        .or_else(|| sftp_source.map(RuntimeSpec::SftpSource))
         .or_else(|| clipboard_source.map(RuntimeSpec::ClipboardSource))
         .or_else(|| email_source.map(RuntimeSpec::EmailSource))
         .or_else(|| email_sink.map(RuntimeSpec::EmailSink))

crates/duckdb-engine/src/plan/mod.rs

@@ -519,6 +519,30 @@ pub struct FtpSourceSpec {
     pub max_files: u64,
 }
 
+/// src.sftp: download files from an SFTP (SSH) server, one row per file
+/// {filename, size, content_b64, modified}. Distinct from FTP/FTPS - SSH
+/// transport via russh + russh-sftp on the ring backend (async, wrapped in
+/// block_on by the executor). Auth by password or an OpenSSH private key;
+/// the server's host key is verified against an optional SHA256 fingerprint
+/// pin (the reporter's "Host Fingerprint" ask, issue #16).
+#[derive(Debug, Clone)]
+pub struct SftpSourceSpec {
+    pub node_id: String,
+    pub host: String,
+    pub port: u16,
+    pub user: String,
+    pub password: Option<String>,
+    pub private_key: Option<String>,
+    pub key_passphrase: Option<String>,
+    pub directory: String,
+    pub pattern: Option<String>,
+    pub max_files: u64,
+    /// Expected server host-key fingerprint, e.g. "SHA256:abc123...". When set,
+    /// the connection is refused unless the server key matches. When empty,
+    /// the key is accepted on trust (trust-on-first-use, logged).
+    pub host_fingerprint: Option<String>,
+}
+
 /// src.clipboard: read the system clipboard. If the text parses as
 /// JSON-array-of-objects, the array becomes rows directly; otherwise
 /// a single row {text, length} is emitted. Desktop-only by definition;