From 179c17a337ac3418ba043f1392c2aab78bed6fcc Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 19:12:44 +0900 Subject: [PATCH] Recover expired FTP control sessions --- Cargo.lock | 59 +++++++++++++++ Cargo.toml | 2 +- README.md | 4 +- sdk/wit/filesystem.wit | 21 +++--- src/control.rs | 3 + src/lib.rs | 94 ++++++++++++++++++++---- src/session_tests.rs | 162 +++++++++++++++++++++++++++++++++++++++++ src/test_host.rs | 50 +++++++++++++ 8 files changed, 365 insertions(+), 30 deletions(-) create mode 100644 src/session_tests.rs create mode 100644 src/test_host.rs diff --git a/Cargo.lock b/Cargo.lock index f5c4452..1ac10fb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14,6 +14,12 @@ version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + [[package]] name = "equivalent" version = "1.0.2" @@ -71,6 +77,12 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + [[package]] name = "log" version = "0.4.34" @@ -109,6 +121,17 @@ dependencies = [ "wit-bindgen", ] +[[package]] +name = "mio" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -191,6 +214,16 @@ dependencies = [ "zmij", ] +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys", +] + [[package]] name = "syn" version = "3.0.6" @@ -208,8 +241,13 @@ version = "1.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" dependencies = [ + "bytes", + "libc", + "mio", "pin-project-lite", + "socket2", "tokio-macros", + "windows-sys", ] [[package]] @@ -229,6 +267,12 @@ version = "1.0.26" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + [[package]] name = "wasm-encoder" version = "0.259.0" @@ -263,6 +307,21 @@ dependencies = [ "semver", ] +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + [[package]] name = "wit-bindgen" version = "0.62.0" diff --git a/Cargo.toml b/Cargo.toml index 77a50dc..f665f24 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,7 +11,7 @@ crate-type = ["cdylib", "rlib"] memona-plugin-sdk = { path = "sdk" } [dev-dependencies] -tokio = { version = "1", features = ["macros", "rt"] } +tokio = { version = "1", features = ["macros", "rt", "net", "io-util", "sync", "time"] } [lints] workspace = true diff --git a/README.md b/README.md index fb6a6b5..befc6f4 100644 --- a/README.md +++ b/README.md @@ -16,8 +16,8 @@ One provider, `ftp`, for the `ftp` and `ftps` schemes. A connection is configured with a server name, an optional port, a TLS toggle for FTPS, and an optional username and password. An empty username signs in anonymously. -For localhost or a server on a private network, turn on **Private network** in -the connection settings. Save and reconnect after changing these settings. +Public, LAN, VPN, link-local and localhost addresses work directly from the +server name; there is no separate network permission setting. The TLS toggle currently starts TLS on the control connection only. Protected data connections and explicit `AUTH TLS` negotiation are not implemented, so diff --git a/sdk/wit/filesystem.wit b/sdk/wit/filesystem.wit index b5bcdf8..824db1d 100644 --- a/sdk/wit/filesystem.wit +++ b/sdk/wit/filesystem.wit @@ -35,11 +35,10 @@ interface host { /// `respond`. An empty result means the body is complete. pull: func(limit: u32) -> result, host-error>; } - /// Begins a request to any public HTTP or HTTPS host. Each redirect and - /// DNS result is checked, and DNS is pinned. Private/LAN addresses require - /// the connection's private-network grant. Cross-origin redirects remove - /// request headers; HTTPS cannot - /// downgrade. The host bounds total bytes per exchange, not buffer size. + /// Begins a request to any HTTP or HTTPS host, including LAN and loopback + /// addresses. Each redirect and DNS result is checked, and DNS is pinned. + /// Cross-origin redirects remove request headers; HTTPS cannot downgrade. + /// The host bounds total bytes per exchange, not buffer size. http: func(request: request) -> result; /// One open TCP connection. Dropping the handle closes the socket, and the @@ -51,10 +50,10 @@ interface host { /// Writes at most 1,048,576 bytes and returns how many were accepted. write: func(data: list) -> result; } - /// Opens a socket to any public host on a package-declared port. - /// The host resolves and pins DNS, applies the same private-address rule - /// as `http`, and starts TLS when asked. Ports - /// negotiated at runtime, such as an FTP data channel, must be declared. + /// Opens a socket to any host, including LAN and loopback addresses, on a + /// package-declared port. The host resolves and pins DNS and starts TLS + /// when asked. Ports negotiated at runtime, such as an FTP data channel, + /// must be declared. connect: func(host: string, port: u16, tls: bool) -> result; /// One datagram socket connected to a single peer. @@ -65,8 +64,8 @@ interface host { /// means none arrived inside the host's wait, not that the peer left. receive: func(limit: u32) -> result, host-error>; } - /// Connects a datagram socket under the same port and private-network rules - /// as `connect`, for the few filesystem protocols that still need one. + /// Connects a datagram socket under the same port rules as `connect`, for + /// the few filesystem protocols that still need one. bind: func(host: string, port: u16) -> result; /// A configuration value, typed as the package's manifest declared it. diff --git a/src/control.rs b/src/control.rs index 30ee037..fb0c3b8 100644 --- a/src/control.rs +++ b/src/control.rs @@ -1,6 +1,9 @@ //! The FTP control channel: one socket, line replies, and passive transfers. +#[cfg(test)] +use crate::test_host as host; use memona_plugin_sdk::ErrorCode; +#[cfg(not(test))] use memona_plugin_sdk::memona::filesystem::host; const MAX_LINE: usize = 8192; diff --git a/src/lib.rs b/src/lib.rs index 3eb4ad2..416f012 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,8 +4,14 @@ mod control; mod listing; mod transfer; +#[cfg(test)] +mod session_tests; +#[cfg(test)] +mod test_host; + use control::{Control, outcome, preliminary, refusal, transport}; use listing::{machine_info, page}; +#[cfg(not(test))] use memona_plugin_sdk::memona::filesystem::host; use memona_plugin_sdk::{ Capabilities, ChangePage, EntryKind, ErrorCode, FileInfo, Guest, GuestReader, @@ -14,6 +20,8 @@ use memona_plugin_sdk::{ use std::cell::{Cell, RefCell}; use std::collections::HashMap; use std::rc::Rc; +#[cfg(test)] +use test_host as host; use transfer::Buffered; const MAX_CHUNK: u32 = 1024 * 1024; @@ -22,6 +30,7 @@ const MAX_LISTING_BYTES: usize = 8 * 1024 * 1024; thread_local! { static CONTROL: RefCell> = const { RefCell::new(None) }; + static SESSION: RefCell> = const { RefCell::new(None) }; /// Raw MLSD listings by path, so later pages of a folder are parsed from /// memory instead of downloaded and parsed whole again. static LISTINGS: RefCell>> = RefCell::new(HashMap::new()); @@ -29,6 +38,31 @@ thread_local! { struct Ftp; +#[derive(Clone)] +struct Session { + host: String, + port: u16, + tls: bool, + user: String, + password: String, +} + +impl Session { + fn connect(&self) -> Result { + let mut control = Control::open(self.host.clone(), self.port, self.tls)?; + let (code, _) = control.send(&format!("USER {}", self.user))?; + if code == 331 { + let (code, _) = control.send(&format!("PASS {}", self.password))?; + outcome(code)?; + } else { + outcome(code)?; + } + // Binary throughout: text mode would rewrite line endings inside files. + control.expect("TYPE I")?; + Ok(control) + } +} + fn text(setting: Option<&Setting>) -> Option { match setting { Some(Setting::Text(value)) => { @@ -50,12 +84,32 @@ fn route(target: &Locator) -> Result { fn with_control( action: impl FnOnce(&mut Control) -> Result, ) -> Result { - CONTROL.with_borrow_mut(|slot| match slot.as_mut() { - Some(control) => action(control), - None => Err(ErrorCode::Internal), + CONTROL.with_borrow_mut(|slot| { + let result = match slot.as_mut() { + Some(control) => action(control), + None => Err(ErrorCode::Internal), + }; + if matches!(result, Err(ErrorCode::Unavailable | ErrorCode::Internal)) { + *slot = None; + } + result }) } +/// Checks a control channel only before a fresh filesystem operation. A lost +/// idle session is replaced and logged in again. Nothing inside a transfer or +/// mutation retries, because its missing reply may hide a completed change. +fn ready() -> Result<(), ErrorCode> { + if with_control(|control| control.expect("NOOP")).is_ok() { + return Ok(()); + } + let session = SESSION + .with_borrow(Clone::clone) + .ok_or(ErrorCode::Internal)?; + CONTROL.set(Some(session.connect()?)); + Ok(()) +} + /// Starts a passive transfer and leaves the data socket open for the caller, /// who owes the server one more reply once that socket is closed: either the /// completion, read by `settle`, or the answer to an `abort`. @@ -241,19 +295,16 @@ impl Guest for Ftp { config.iter().find(|(key, _)| key == "tls").map(|(_, v)| v), Some(Setting::Toggle(true)) ); - let mut control = Control::open(host_name, port, tls)?; let user = lookup("username").unwrap_or_else(|| "anonymous".to_owned()); - let (code, _) = control.send(&format!("USER {user}"))?; - if code == 331 { - let password = lookup("password").unwrap_or_default(); - let (code, _) = control.send(&format!("PASS {password}"))?; - outcome(code)?; - } else { - outcome(code)?; - } - // Binary throughout: text mode would rewrite line endings inside files. - control.expect("TYPE I")?; - CONTROL.set(Some(control)); + let session = Session { + host: host_name, + port, + tls, + user, + password: lookup("password").unwrap_or_default(), + }; + CONTROL.set(Some(session.connect()?)); + SESSION.set(Some(session)); Ok(Capabilities { write: true, mkdir: true, @@ -274,6 +325,7 @@ impl Guest for Ftp { } *slot = None; }); + SESSION.set(None); } fn stat(target: Locator) -> Result { @@ -285,6 +337,7 @@ impl Guest for Ftp { revision: None, }); } + ready()?; with_control(|control| probe(control, &path)) } @@ -306,13 +359,17 @@ impl Guest for Ftp { }); let listing = match kept { Some(listing) => listing, - None => fetch_listing(&path)?, + None => { + ready()?; + fetch_listing(&path)? + } }; page(&listing, start, limit as usize) } fn open_read(target: Locator, offset: u64) -> Result { let path = route(&target)?; + ready()?; if offset > 0 { with_control(|control| control.expect(&format!("REST {offset}")))?; } @@ -332,6 +389,7 @@ impl Guest for Ftp { if !matches!(condition, WriteCondition::Overwrite) { return Err(ErrorCode::Unsupported); } + ready()?; let socket = transfer(&format!("STOR {path}"))?; Ok(Writer::new(FtpWriter { path, @@ -344,6 +402,7 @@ impl Guest for Ftp { /// the one thing the server says in words rather than in its reply code. fn mkdir(target: Locator) -> Result { let path = route(&target)?; + ready()?; with_control(|control| { let (code, text) = control.send(&format!("MKD {path}"))?; refusal(code, &text) @@ -362,6 +421,7 @@ impl Guest for Ftp { destination: String, ) -> Result { let path = route(&target)?; + ready()?; with_control(|control| { match probe(control, &destination) { Ok(_) => return Err(ErrorCode::Conflict), @@ -379,6 +439,7 @@ impl Guest for Ftp { /// holds children is refused as a conflict, not as an absence. fn remove(target: Locator) -> Result<(), ErrorCode> { let path = route(&target)?; + ready()?; with_control(|control| { let (code, _) = control.send(&format!("DELE {path}"))?; if (200..300).contains(&code) { @@ -394,6 +455,7 @@ impl Guest for Ftp { } } +#[cfg(not(test))] memona_plugin_sdk::export!(Ftp); #[cfg(test)] diff --git a/src/session_tests.rs b/src/session_tests.rs new file mode 100644 index 0000000..824e8f8 --- /dev/null +++ b/src/session_tests.rs @@ -0,0 +1,162 @@ +//! Real FTP commands over a native test socket, with an async local server. +//! This verifies guest recovery; WASM host integration is exercised separately. + +use super::Ftp; +use memona_plugin_sdk::{ErrorCode, Guest, Locator, Setting}; +use std::time::Duration; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::oneshot; + +type TestResult = Result>; +type Channel = BufReader; + +fn guest(value: Result) -> TestResult { + value.map_err(|error| format!("guest operation failed: {error:?}").into()) +} + +fn config(port: u16) -> Vec<(String, Setting)> { + vec![ + ("host".into(), Setting::Text("127.0.0.1".into())), + ("port".into(), Setting::Text(port.to_string())), + ("username".into(), Setting::Text("fixture".into())), + ( + "password".into(), + Setting::Secret(" fixture secret ".into()), + ), + ] +} + +async fn reply(channel: &mut Channel, text: &str) -> TestResult { + channel.get_mut().write_all(text.as_bytes()).await?; + Ok(()) +} + +async fn expect(channel: &mut Channel, expected: &str) -> TestResult { + loop { + let mut line = String::new(); + channel.read_line(&mut line).await?; + if line == "NOOP\r\n" { + reply(channel, "200 Still connected\r\n").await?; + continue; + } + assert_eq!(line, format!("{expected}\r\n")); + return Ok(()); + } +} + +async fn login(listener: &TcpListener) -> TestResult { + let (stream, _) = listener.accept().await?; + let mut channel = BufReader::new(stream); + reply(&mut channel, "220 Fixture\r\n").await?; + expect(&mut channel, "USER fixture").await?; + reply(&mut channel, "331 Password required\r\n").await?; + expect(&mut channel, "PASS fixture secret ").await?; + reply(&mut channel, "230 Signed in\r\n").await?; + expect(&mut channel, "TYPE I").await?; + reply(&mut channel, "200 Binary\r\n").await?; + Ok(channel) +} + +async fn listing(channel: &mut Channel) -> TestResult { + expect(channel, "EPSV").await?; + let listener = TcpListener::bind("127.0.0.1:0").await?; + reply( + channel, + &format!("229 Passive (|||{}|)\r\n", listener.local_addr()?.port()), + ) + .await?; + expect(channel, "MLSD /").await?; + reply(channel, "150 Listing\r\n").await?; + let (mut data, _) = listener.accept().await?; + data.write_all(b"type=file;size=3; report.md \r\n").await?; + data.shutdown().await?; + drop(data); + reply(channel, "226 Complete\r\n").await +} + +#[tokio::test] +async fn a_listing_reconnects_after_the_server_expires_the_control_session() +-> TestResult { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let port = listener.local_addr()?.port(); + let (expired, wait_expired) = oneshot::channel(); + let server = tokio::spawn(async move { + tokio::time::timeout(Duration::from_secs(5), async { + let mut channel = login(&listener).await?; + listing(&mut channel).await?; + reply(&mut channel, "421 Control connection timed out\r\n").await?; + channel.get_mut().shutdown().await?; + drop(channel); + expired.send(()).map_err(|_| "guest disappeared")?; + let mut channel = login(&listener).await?; + listing(&mut channel).await?; + expect(&mut channel, "QUIT").await?; + reply(&mut channel, "221 Bye\r\n").await + }) + .await? + }); + let client = tokio::task::spawn_blocking(move || -> TestResult { + guest(Ftp::open("ftp".into(), config(port)))?; + let first = guest(Ftp::list(Locator::Path("/".into()), None, 10))?; + assert_eq!(first.entries[0].name, " report.md "); + wait_expired.blocking_recv()?; + let next = guest(Ftp::list(Locator::Path("/".into()), None, 10))?; + assert_eq!(next.entries[0].name, " report.md "); + Ftp::close(); + Ok(()) + }) + .await?; + if client.is_err() { + server.abort(); + } + client?; + server.await??; + Ok(()) +} + +#[tokio::test] +async fn recovery_before_a_mutation_never_replays_a_lost_mutation_reply() +-> TestResult { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let port = listener.local_addr()?.port(); + let (expired, wait_expired) = oneshot::channel(); + let server = tokio::spawn(async move { + tokio::time::timeout(Duration::from_secs(5), async { + let mut channel = login(&listener).await?; + channel.get_mut().shutdown().await?; + drop(channel); + expired.send(()).map_err(|_| "guest disappeared")?; + let mut channel = login(&listener).await?; + expect(&mut channel, "MKD /new").await?; + // The mutation may have succeeded; lose its reply deliberately. + channel.get_mut().shutdown().await?; + drop(channel); + Ok::<_, Box>(listener) + }) + .await? + }); + let client = tokio::task::spawn_blocking(move || -> TestResult { + guest(Ftp::open("ftp".into(), config(port)))?; + wait_expired.blocking_recv()?; + assert!(matches!( + Ftp::mkdir(Locator::Path("/new".into())), + Err(ErrorCode::Unavailable) + )); + Ftp::close(); + Ok(()) + }) + .await?; + if client.is_err() { + server.abort(); + } + client?; + let listener = server.await??; + assert!( + tokio::time::timeout(Duration::from_millis(50), listener.accept()) + .await + .is_err(), + "a mutation must not be replayed on a third control connection" + ); + Ok(()) +} diff --git a/src/test_host.rs b/src/test_host.rs new file mode 100644 index 0000000..d0751e5 --- /dev/null +++ b/src/test_host.rs @@ -0,0 +1,50 @@ +//! Native test transport. Guest calls run only inside `spawn_blocking`; +//! production WASM always uses the host's sockets and network policy. + +use std::cell::RefCell; +use std::io::{Read, Write}; +use std::net::{IpAddr, SocketAddr, TcpStream}; +use std::time::Duration; + +pub use memona_plugin_sdk::HostError; + +pub struct Tcp(RefCell); + +pub fn connect(name: &str, port: u16, tls: bool) -> Result { + if tls { + return Err(HostError::InvalidInput); + } + let ip: IpAddr = name.parse().map_err(|_| HostError::InvalidInput)?; + let timeout = Duration::from_secs(2); + let stream = TcpStream::connect_timeout(&SocketAddr::new(ip, port), timeout) + .map_err(|_| HostError::Unavailable)?; + stream + .set_read_timeout(Some(timeout)) + .map_err(|_| HostError::Unavailable)?; + stream + .set_write_timeout(Some(timeout)) + .map_err(|_| HostError::Unavailable)?; + Ok(Tcp(RefCell::new(stream))) +} + +impl Tcp { + pub fn read(&self, limit: u32) -> Result, HostError> { + let mut bytes = vec![0; limit as usize]; + let count = self + .0 + .borrow_mut() + .read(&mut bytes) + .map_err(|_| HostError::Unavailable)?; + bytes.truncate(count); + Ok(bytes) + } + + pub fn write(&self, bytes: &[u8]) -> Result { + self + .0 + .borrow_mut() + .write(bytes) + .map(|count| count as u32) + .map_err(|_| HostError::Unavailable) + } +}