From 90a54034f75a61a91daabd157b4dc6bacef9691b Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 13:10:40 +0000 Subject: [PATCH] FTP connections preserve passwords and filenames, with CI checking every change (#1) Co-authored-by: Danny Kim Co-committed-by: Danny Kim --- .gitea/workflows/agents.yml | 49 ++++++ .gitea/workflows/ci.yml | 51 ++++++ Cargo.lock | 78 ++++++++- Cargo.toml | 6 +- README.md | 20 ++- manifest.json | 5 +- sdk/wit/filesystem.wit | 27 +-- src/control.rs | 69 ++++++-- src/lib.rs | 316 ++++++++++++++++++++++++++++++------ src/listing.rs | 16 +- src/session_tests.rs | 173 ++++++++++++++++++++ src/test_host.rs | 278 +++++++++++++++++++++++++++++++ 12 files changed, 1004 insertions(+), 84 deletions(-) create mode 100644 .gitea/workflows/agents.yml create mode 100644 .gitea/workflows/ci.yml create mode 100644 src/session_tests.rs create mode 100644 src/test_host.rs diff --git a/.gitea/workflows/agents.yml b/.gitea/workflows/agents.yml new file mode 100644 index 0000000..7f49767 --- /dev/null +++ b/.gitea/workflows/agents.yml @@ -0,0 +1,49 @@ +name: Agents +run-name: >- + AI bot at + #${{ gitea.event.issue.number || gitea.event.pull_request.number }} + +on: + issues: + types: [opened] + pull_request: + types: [opened, review_requested] + issue_comment: + types: [created] + pull_request_review_comment: + types: [created] + +# New human events replace stale runs for the same work item. AI-bot comments +# get unique groups so their skipped workflows cannot cancel an active login. +concurrency: + group: >- + agents-${{ gitea.event.sender.login == 'bot' && gitea.run_id || + gitea.event.issue.number || gitea.event.pull_request.number }} + cancel-in-progress: true + +jobs: + bot: + # Nothing the bot account posts can start a run, so it cannot trigger itself. + if: > + gitea.event.sender.login != 'bot' && + ((gitea.event_name == 'issues' && + contains(gitea.event.issue.body, '@bot')) || + (gitea.event_name == 'pull_request' && + ((gitea.event.action == 'opened' && + contains(gitea.event.pull_request.body, '@bot')) || + gitea.event.requested_reviewer.login == 'bot')) || + (gitea.event_name == 'issue_comment' && + contains(gitea.event.comment.body, '@bot')) || + (gitea.event_name == 'pull_request_review_comment' && + contains(gitea.event.review.content, '@bot'))) + runs-on: self-hosted + container: + volumes: + - bot-home:/root + steps: + - name: Run AI bot + # Intentionally follow protected main so centrally managed profiles and + # quota fallbacks update without per-repository credential changes. + uses: https://git.capsulizers.com/commons/bot-agents@main + with: + master-key: ${{ secrets.BOT_AGENTS_MASTER_KEY }} diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..32da818 --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,51 @@ +name: CI + +on: + pull_request: + branches: [main] + push: + branches: [main] + +concurrency: + group: ${{ gitea.workflow }}-${{ gitea.ref }} + cancel-in-progress: true + +jobs: + checks: + runs-on: self-hosted + container: + volumes: + - memona-plugin-target:/cargo-target + env: + CARGO_BUILD_JOBS: 2 + CARGO_TARGET_DIR: /cargo-target/serialized + CARGO_HOME: /cargo-target/cargo-home + RUSTUP_HOME: /cargo-target/rustup-home + RUSTUP_TOOLCHAIN: stable + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Install system packages + run: >- + apt-get update && apt-get install -y --no-install-recommends + build-essential ca-certificates curl util-linux + + # The same lock protects the shared toolchain and every Cargo command. + # Keep its target and lock paths in sync with Memona's plugin checks. + - name: Check and build plugin + run: | + flock --no-fork "$CARGO_TARGET_DIR.lock" sh -ec ' + export PATH="$CARGO_HOME/bin:$PATH" + if [ ! -x "$CARGO_HOME/bin/rustup" ]; then + mkdir -p "$CARGO_HOME" + curl --proto =https --tlsv1.2 -fsSL https://sh.rustup.rs -o "$CARGO_HOME/rustup-init.sh" + sh "$CARGO_HOME/rustup-init.sh" -y --no-modify-path --default-toolchain none + rm "$CARGO_HOME/rustup-init.sh" + fi + rustup toolchain install stable --profile minimal --component clippy,rustfmt --target wasm32-wasip2 --no-self-update + cargo fmt --check + cargo clippy --locked --all-targets -- -D warnings + cargo test --locked --lib + cargo build --locked --release --target wasm32-wasip2 + ' diff --git a/Cargo.lock b/Cargo.lock index 8b97008..7a94fab 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 = "cfg-if" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" + [[package]] name = "equivalent" version = "1.0.2" @@ -26,6 +32,18 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "wasip2", +] + [[package]] name = "hashbrown" version = "0.17.1" @@ -71,6 +89,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" @@ -96,18 +120,26 @@ checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" [[package]] name = "memona-plugin-ftp" -version = "0.1.2" +version = "0.1.3" dependencies = [ + "getrandom", "memona-plugin-sdk", + "tokio", ] [[package]] name = "memona-plugin-sdk" version = "0.1.0" dependencies = [ - "wit-bindgen", + "wit-bindgen 0.62.0", ] +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + [[package]] name = "prettyplease" version = "0.3.0" @@ -136,6 +168,12 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + [[package]] name = "semver" version = "1.0.28" @@ -195,12 +233,42 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "pin-project-lite", + "tokio-macros", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "unicode-ident" version = "1.0.26" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" +[[package]] +name = "wasip2" +version = "1.0.1+wasi-0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0562428422c63773dad2c345a1882263bbf4d65cf3f42e90921f787ef5ad58e7" +dependencies = [ + "wit-bindgen 0.46.0", +] + [[package]] name = "wasm-encoder" version = "0.259.0" @@ -235,6 +303,12 @@ dependencies = [ "semver", ] +[[package]] +name = "wit-bindgen" +version = "0.46.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f17a85883d4e6d00e8a97c586de764dabcc06133f7f1d55dce5cdc070ad7fe59" + [[package]] name = "wit-bindgen" version = "0.62.0" diff --git a/Cargo.toml b/Cargo.toml index de78c21..2d3a3ea 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "memona-plugin-ftp" -version = "0.1.2" +version = "0.1.3" edition = "2024" license = "MIT" @@ -8,8 +8,12 @@ license = "MIT" crate-type = ["cdylib", "rlib"] [dependencies] +getrandom = "0.3" memona-plugin-sdk = { path = "sdk" } +[dev-dependencies] +tokio = { version = "1", features = ["macros", "rt"] } + [lints] workspace = true diff --git a/README.md b/README.md index ebd7ee2..31924d2 100644 --- a/README.md +++ b/README.md @@ -8,12 +8,22 @@ The plugin is a WebAssembly component built against the Memona filesystem contract, `memona:filesystem@0.1.0`. It uses only the host's sockets; it has no ambient filesystem, environment or network access of its own. +This package requires a Memona version that supports plugin manifest schema 2. + ## What it registers 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. +With the compatible schema 2 Memona version required by this package, public, +LAN, VPN, link-local and localhost addresses work directly from the server +name; that API has no separate connection network grant. + +The TLS toggle currently starts TLS on the control connection only. Protected +data connections and explicit `AUTH TLS` negotiation are not implemented, so +the plugin does not yet provide a complete FTPS connection. + Listing and stat need a server that answers `MLSD` and `MLST`, the machine-readable listing commands. A server without them reports the folder as unsupported rather than guessing at the older, human-formatted listing or @@ -23,15 +33,13 @@ so a conditional write is reported as unsupported rather than simulated. ## Passive mode and the port range Transfers are passive: the plugin asks the server for a data port with `EPSV` -or `PASV` and connects out to it. The server picks a fresh port for every +and connects out to it; `PASV` fallback is not implemented. The server picks a fresh port for every transfer and announces it only as the transfer begins, so the range that has to be reachable is wide. -`manifest.json` declares `networkPorts` as 21 and 1024-65535, and `networkHosts` -as `*` because a server can be anywhere. Declaring is not granting: a -connection reaches only the host and ports the person grants it, and a grant -never accepts a wildcard host. Grant the control port and the span the server -uses for its data channel, usually `21` and `1024-65535`. +The package declares socket ports 21 and 1024-65535, which Memona applies +automatically to each connection. The control port and passive data ports +must be within these ranges. ## Build diff --git a/manifest.json b/manifest.json index 1a792d9..f88bda5 100644 --- a/manifest.json +++ b/manifest.json @@ -1,13 +1,12 @@ { - "schemaVersion": 1, + "schemaVersion": 2, "id": "memona/ftp", "name": "FTP", - "version": "0.1.2", + "version": "0.1.3", "apiVersion": "0.1.0", "description": "Open an FTP server as a folder, and let agents work in it.", "license": "MIT", "entry": "plugin.wasm", - "networkHosts": ["*"], "networkPorts": [ { "from": 21, "to": 21 }, { "from": 1024, "to": 65535 } diff --git a/sdk/wit/filesystem.wit b/sdk/wit/filesystem.wit index 4529372..624ee9d 100644 --- a/sdk/wit/filesystem.wit +++ b/sdk/wit/filesystem.wit @@ -35,11 +35,11 @@ interface host { /// `respond`. An empty result means the body is complete. pull: func(limit: u32) -> result, host-error>; } - /// Begins a request to a host both declared by the package and granted on - /// the connection. Each redirect and DNS result is checked, and DNS is - /// pinned. Private/LAN addresses additionally require the 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. This API has no per-connection network grant. 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,13 +51,13 @@ interface host { /// Writes at most 1,048,576 bytes and returns how many were accepted. write: func(data: list) -> result; } - /// Opens a socket to a host and port both declared by the package and - /// granted on the connection. 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, still need a grant. + /// Opens a socket to any host, including LAN and loopback addresses, on a + /// package-declared port. This API has no per-connection network grant. + /// 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 granted peer. + /// One datagram socket connected to a single peer. resource udp { /// Sends one datagram of at most 65,507 bytes. send: func(data: list) -> result; @@ -65,8 +65,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 declaration and grant rules - /// as `connect`, for the few filesystem protocols that still need one. + /// Connects a datagram socket under the same address and 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. @@ -81,7 +81,8 @@ interface host { /// Filesystem operations for one connection. Paths are absolute within its /// root, starting with `/`; they are never local OS paths or resource URIs. -/// Dot segments, backslashes, percent escapes, colons, and controls are invalid. +/// Colons and percent signs are literal; paths must never be URL-decoded. +/// Dot segments, backslashes, and controls are invalid. /// Every invocation is bounded by host fuel, memory, time, and cancellation. interface provider { use host.{setting}; diff --git a/src/control.rs b/src/control.rs index 4cfbbba..1cf2022 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; @@ -64,7 +67,9 @@ fn names_a_conflict(text: &str) -> bool { .collect(); words.contains(&"exists") || words.contains(&"nonempty") - || words.windows(2).any(|pair| matches!(pair, ["not" | "non", "empty"])) + || words + .windows(2) + .any(|pair| matches!(pair, ["not" | "non", "empty"])) } /// What the first reply to a transfer command means. A transfer answers with @@ -86,6 +91,11 @@ pub struct Control { buffer: Vec, } +struct DispatchError { + code: ErrorCode, + dispatched: bool, +} + impl Control { pub fn open( host_name: String, @@ -150,27 +160,61 @@ impl Control { Err(ErrorCode::Unavailable) } - /// Sends one command. A newline in an argument would forge a second command, - /// so it is refused rather than escaped. - pub fn send(&mut self, command: &str) -> Result<(u16, String), ErrorCode> { + /// Writes one complete command and records whether any of it reached the + /// socket. A newline in an argument would forge a second command, so it is + /// refused before dispatch rather than escaped. + fn dispatch(&mut self, command: &str) -> Result<(), DispatchError> { if command.contains(['\r', '\n']) || command.len() > MAX_LINE { - return Err(ErrorCode::InvalidInput); + return Err(DispatchError { + code: ErrorCode::InvalidInput, + dispatched: false, + }); } let line = format!("{command}\r\n"); let mut sent = 0; while sent < line.len() { - let written = self - .socket - .write(&line.as_bytes()[sent..]) - .map_err(transport)? as usize; + let written = + self + .socket + .write(&line.as_bytes()[sent..]) + .map_err(|error| DispatchError { + code: transport(error), + dispatched: sent > 0, + })? as usize; if written == 0 { - return Err(ErrorCode::Unavailable); + return Err(DispatchError { + code: ErrorCode::Unavailable, + dispatched: sent > 0, + }); } sent += written; } + Ok(()) + } + + /// Sends one command and reads its reply. + pub fn send(&mut self, command: &str) -> Result<(u16, String), ErrorCode> { + self.dispatch(command).map_err(|error| error.code)?; self.reply() } + /// Sends a command that changes remote state. Once any command byte was + /// dispatched, a write interruption or missing reply makes its outcome + /// unknowable even when the underlying transport reports cancellation. + pub fn send_mutation( + &mut self, + command: &str, + ) -> Result<(u16, String), ErrorCode> { + self.dispatch(command).map_err(|error| { + if error.dispatched { + ErrorCode::OutcomeUnknown + } else { + error.code + } + })?; + self.reply().map_err(|_| ErrorCode::OutcomeUnknown) + } + pub fn expect(&mut self, command: &str) -> Result { let (code, text) = self.send(command)?; outcome(code)?; @@ -255,7 +299,10 @@ mod tests { )); // The same two conditions as other servers phrase them. assert!(matches!( - refusal(550, "550 Cannot create a file when that file already exists."), + refusal( + 550, + "550 Cannot create a file when that file already exists." + ), Err(ErrorCode::Conflict) )); assert!(matches!( diff --git a/src/lib.rs b/src/lib.rs index 0d3be11..944ec76 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,14 +20,18 @@ 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; /// One listing may be this large, and so may every listing kept together. const MAX_LISTING_BYTES: usize = 8 * 1024 * 1024; +const MAX_STAGE_ATTEMPTS: usize = 64; 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,12 +39,38 @@ 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)) | Some(Setting::Secret(value)) => { + Some(Setting::Text(value)) => { let value = value.trim().to_owned(); (!value.is_empty()).then_some(value) } + Some(Setting::Secret(value)) => Some(value.clone()), Some(Setting::Toggle(_)) | None => None, } } @@ -49,12 +85,43 @@ 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::OutcomeUnknown + | 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 !CONTROL.with_borrow(Option::is_none) { + match with_control(|control| control.expect("NOOP")) { + Ok(_) => return Ok(()), + Err(ErrorCode::Unavailable) => {} + Err(error) => return Err(error), + } + } + 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`. @@ -67,6 +134,17 @@ fn transfer(command: &str) -> Result { }) } +/// Starts a passive upload. Losing the command reply is ambiguous because an +/// FTP server may have created or truncated the destination on receipt. +fn mutation_transfer(command: &str) -> Result { + with_control(|control| { + let data = control.passive()?; + let (code, _) = control.send_mutation(command)?; + preliminary(code)?; + Ok(data) + }) +} + /// Consumes the completion reply of a transfer whose data socket is closed. fn settle() -> Result<(), ErrorCode> { with_control(|control| { @@ -75,6 +153,16 @@ fn settle() -> Result<(), ErrorCode> { }) } +/// Consumes an upload's completion reply. The data and command are already +/// dispatched, so an interrupted reply cannot say whether the file committed. +fn settle_mutation() -> Result<(), ErrorCode> { + let (code, _) = with_control(|control| { + control.reply().map_err(|_| ErrorCode::OutcomeUnknown) + }) + .map_err(|_| ErrorCode::OutcomeUnknown)?; + outcome(code) +} + /// Ends a transfer early, after its data socket is closed. fn abort() -> Result<(), ErrorCode> { with_control(Control::abort) @@ -167,20 +255,62 @@ impl Drop for FtpReader { /// Writes one file to a passive data socket. struct FtpWriter { - path: String, + target: String, + staged: RefCell>, socket: RefCell>, written: Cell, } +impl FtpWriter { + /// Removes the private staging file when possible. A missing stage means a + /// prior promotion or cleanup already completed. + fn cleanup_stage(&self) { + let Some(staged) = self.staged.borrow().clone() else { + return; + }; + if ready().is_err() { + return; + } + let removed = with_control(|control| { + let (code, _) = control.send_mutation(&format!("DELE {staged}"))?; + outcome(code) + }); + if matches!(removed, Ok(()) | Err(ErrorCode::NotFound)) { + self.staged.take(); + } + } + + /// A failed data write can leave both the remote file and its pending + /// completion reply in an unknown state. Close both channels, then make a + /// best-effort cleanup of the private stage over a fresh session. + fn discard_uncertain_upload(&self) -> ErrorCode { + drop(self.socket.take()); + CONTROL.set(None); + self.cleanup_stage(); + ErrorCode::OutcomeUnknown + } +} + impl GuestWriter for FtpWriter { fn push(&self, data: Vec) -> Result<(), ErrorCode> { - let socket = self.socket.borrow(); - let socket = socket.as_ref().ok_or(ErrorCode::InvalidInput)?; + if self.socket.borrow().is_none() { + return Err(ErrorCode::InvalidInput); + } let mut sent = 0; while sent < data.len() { - let written = socket.write(&data[sent..]).map_err(transport)? as usize; + let result = { + let socket = self.socket.borrow(); + let Some(socket) = socket.as_ref() else { + return Err(ErrorCode::InvalidInput); + }; + socket.write(&data[sent..]) + }; + let written = match result { + Ok(written) => written as usize, + Err(_) => return Err(self.discard_uncertain_upload()), + }; if written == 0 { - return Err(ErrorCode::Unavailable); + return Err(self.discard_uncertain_upload()); } sent += written; } @@ -188,10 +318,30 @@ impl GuestWriter for FtpWriter { Ok(()) } /// Closing the data socket is what ends an FTP upload; only then does the - /// server answer, so the socket is given up before the reply is read. + /// server answer. A confirmed stage is promoted only after that reply, so an + /// interrupted transfer never exposes a partial destination. fn finish(&self) -> Result, ErrorCode> { drop(self.socket.take().ok_or(ErrorCode::InvalidInput)?); - settle()?; + if let Err(error) = settle_mutation() { + self.cleanup_stage(); + return Err(error); + } + let staged = self + .staged + .borrow() + .clone() + .ok_or(ErrorCode::InvalidInput)?; + let promoted = with_control(|control| { + control.expect(&format!("RNFR {staged}"))?; + let (code, _) = + control.send_mutation(&format!("RNTO {}", self.target))?; + outcome(code) + }); + if let Err(error) = promoted { + self.cleanup_stage(); + return Err(error); + } + self.staged.take(); Ok(Some(FileInfo { kind: EntryKind::File, size: self.written.get(), @@ -201,19 +351,64 @@ impl GuestWriter for FtpWriter { } impl Drop for FtpWriter { - /// An upload abandoned before it finished must leave no partial file. The - /// server takes the closed socket as the end of the upload, so what it - /// stored is deleted once the channel is back in sync. + /// An abandoned upload closes its data channel, resynchronizes when + /// possible, and removes only its private stage. The destination was never + /// exposed unless `finish` confirmed the promotion. fn drop(&mut self) { - let Some(socket) = self.socket.take() else { - return; - }; - drop(socket); - if abort().is_ok() { - let _ = - with_control(|control| control.expect(&format!("DELE {}", self.path))); + if let Some(socket) = self.socket.take() { + drop(socket); + if abort().is_err() { + CONTROL.set(None); + } + } + self.cleanup_stage(); + } +} + +fn random_stage_token() -> Result { + let mut bytes = [0; 16]; + getrandom::fill(&mut bytes).map_err(|_| ErrorCode::Internal)?; + Ok(u128::from_be_bytes(bytes)) +} + +/// Chooses an unused, unguessable private sibling for one upload. The remote +/// probe remains a collision check, while 128 random bits prevent another +/// client from racing a predictable name between that probe and `STOR`. +fn stage_path_with( + target: &str, + mut next_token: impl FnMut() -> Result, +) -> Result { + let (parent, name) = + target.rsplit_once('/').ok_or(ErrorCode::InvalidInput)?; + if name.is_empty() { + return Err(ErrorCode::InvalidInput); + } + for _ in 0..MAX_STAGE_ATTEMPTS { + let token = next_token()?; + let staged = format!("{parent}/.memona-upload-{token:032x}"); + match with_control(|control| probe(control, &staged)) { + Err(ErrorCode::NotFound) => return Ok(staged), + Ok(_) => {} + Err(error) => return Err(error), } } + Err(ErrorCode::Conflict) +} + +fn stage_path(target: &str) -> Result { + stage_path_with(target, random_stage_token) +} + +fn start_writer(target: String) -> Result { + ready()?; + let staged = stage_path(&target)?; + let socket = mutation_transfer(&format!("STOR {staged}"))?; + Ok(FtpWriter { + target, + staged: RefCell::new(Some(staged)), + socket: RefCell::new(Some(socket)), + written: Cell::new(0), + }) } impl Guest for Ftp { @@ -240,19 +435,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, @@ -273,6 +465,7 @@ impl Guest for Ftp { } *slot = None; }); + SESSION.set(None); } fn stat(target: Locator) -> Result { @@ -284,6 +477,7 @@ impl Guest for Ftp { revision: None, }); } + ready()?; with_control(|control| probe(control, &path)) } @@ -305,13 +499,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}")))?; } @@ -331,20 +529,16 @@ impl Guest for Ftp { if !matches!(condition, WriteCondition::Overwrite) { return Err(ErrorCode::Unsupported); } - let socket = transfer(&format!("STOR {path}"))?; - Ok(Writer::new(FtpWriter { - path, - socket: RefCell::new(Some(socket)), - written: Cell::new(0), - })) + Ok(Writer::new(start_writer(path)?)) } /// A refused create is a conflict when the path is already taken, which is /// 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}"))?; + let (code, text) = control.send_mutation(&format!("MKD {path}"))?; refusal(code, &text) })?; Ok(FileInfo { @@ -361,15 +555,18 @@ impl Guest for Ftp { destination: String, ) -> Result { let path = route(&target)?; + ready()?; with_control(|control| { + let info = probe(control, &path)?; match probe(control, &destination) { Ok(_) => return Err(ErrorCode::Conflict), Err(ErrorCode::NotFound) => {} Err(error) => return Err(error), } control.expect(&format!("RNFR {path}"))?; - control.expect(&format!("RNTO {destination}"))?; - probe(control, &destination) + let (code, _) = control.send_mutation(&format!("RNTO {destination}"))?; + outcome(code)?; + Ok(info) }) } @@ -378,12 +575,15 @@ 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) { - return Ok(()); + let (code, _) = control.send_mutation(&format!("DELE {path}"))?; + match outcome(code) { + Ok(()) => return Ok(()), + Err(ErrorCode::NotFound) => {} + Err(error) => return Err(error), } - let (code, text) = control.send(&format!("RMD {path}"))?; + let (code, text) = control.send_mutation(&format!("RMD {path}"))?; refusal(code, &text) }) } @@ -393,4 +593,26 @@ impl Guest for Ftp { } } +#[cfg(not(test))] memona_plugin_sdk::export!(Ftp); + +#[cfg(test)] +mod tests { + use super::text; + use memona_plugin_sdk::Setting; + + #[tokio::test] + async fn secret_credentials_preserve_significant_whitespace() { + for password in [" ftp fixture secret ", " ", ""] { + assert_eq!( + text(Some(&Setting::Secret(password.to_owned()))), + Some(password.to_owned()) + ); + } + assert_eq!( + text(Some(&Setting::Text(" server.example ".to_owned()))), + Some("server.example".to_owned()) + ); + assert_eq!(text(Some(&Setting::Text(" ".to_owned()))), None); + } +} diff --git a/src/listing.rs b/src/listing.rs index 607dc82..badf192 100644 --- a/src/listing.rs +++ b/src/listing.rs @@ -12,7 +12,6 @@ struct Facts<'a> { fn facts(line: &str) -> Option> { let (facts, name) = line.split_once(' ')?; - let name = name.trim(); if name.is_empty() { return None; } @@ -106,6 +105,21 @@ mod tests { use super::{machine_entry, machine_info, page}; use memona_plugin_sdk::{EntryKind, ErrorCode}; + #[tokio::test] + async fn listing_names_preserve_significant_whitespace() + -> Result<(), ErrorCode> { + let listing = b"type=file;size=1; report.md \r\ntype=dir; spaced folder \r\ntype=file;size=2; \r\n"; + let listed = page(listing, 0, 10)?; + let names: Vec<&str> = listed + .entries + .iter() + .map(|entry| entry.name.as_str()) + .collect(); + assert_eq!(names, [" report.md ", " spaced folder ", " "]); + assert!(listed.cursor.is_none()); + Ok(()) + } + #[test] fn reads_kind_and_size_and_skips_the_directory_itself() -> Result<(), &'static str> { diff --git a/src/session_tests.rs b/src/session_tests.rs new file mode 100644 index 0000000..37f78be --- /dev/null +++ b/src/session_tests.rs @@ -0,0 +1,173 @@ +//! FTP command recovery against an in-memory protocol transport. Live QA +//! separately exercises the compiled WASM component over real host sockets. + +use super::{Ftp, stage_path_with, start_writer, test_host}; +use memona_plugin_sdk::{ErrorCode, Guest, GuestWriter, Locator, Setting}; + +type TestResult = Result>; + +fn guest(value: Result) -> TestResult { + value.map_err(|error| format!("guest operation failed: {error:?}").into()) +} + +fn config() -> Vec<(String, Setting)> { + vec![ + ("host".into(), Setting::Text("127.0.0.1".into())), + ( + "port".into(), + Setting::Text(test_host::CONTROL_PORT.to_string()), + ), + ("username".into(), Setting::Text("fixture".into())), + ( + "password".into(), + Setting::Secret(" fixture secret ".into()), + ), + ] +} + +#[tokio::test] +async fn a_listing_reconnects_after_the_server_expires_the_control_session() +-> TestResult { + let fixture = test_host::install(test_host::Scenario::ExpiredListing); + guest(Ftp::open("ftp".into(), config()))?; + let first = guest(Ftp::list(Locator::Path("/".into()), None, 10))?; + assert_eq!(first.entries[0].name, " report.md "); + let next = guest(Ftp::list(Locator::Path("/".into()), None, 10))?; + assert_eq!(next.entries[0].name, " report.md "); + Ftp::close(); + let summary = fixture.summary(); + assert_eq!(summary.control_connections, 2); + assert_eq!(summary.listings, 2); + assert_eq!(summary.mkdir_commands, 0); + Ok(()) +} + +#[tokio::test] +async fn recovery_before_a_mutation_never_replays_a_lost_mutation_reply() +-> TestResult { + let fixture = test_host::install(test_host::Scenario::LostMutationReply); + guest(Ftp::open("ftp".into(), config()))?; + assert!(matches!( + Ftp::mkdir(Locator::Path("/new".into())), + Err(ErrorCode::OutcomeUnknown) + )); + Ftp::close(); + let summary = fixture.summary(); + assert_eq!(summary.control_connections, 2); + assert_eq!(summary.listings, 0); + assert_eq!(summary.mkdir_commands, 1); + Ok(()) +} + +#[tokio::test] +async fn cancelled_preflight_never_reconnects_or_executes_the_mutation() +-> TestResult { + let fixture = test_host::install(test_host::Scenario::CancelledPreflight); + guest(Ftp::open("ftp".into(), config()))?; + assert!(matches!( + Ftp::mkdir(Locator::Path("/cancelled".into())), + Err(ErrorCode::Cancelled) + )); + Ftp::close(); + let summary = fixture.summary(); + assert_eq!(summary.control_connections, 1); + assert_eq!(summary.listings, 0); + assert_eq!(summary.mkdir_commands, 0); + Ok(()) +} + +#[tokio::test] +async fn a_lost_upload_completion_reports_an_unknown_outcome() -> TestResult { + let fixture = test_host::install(test_host::Scenario::LostUploadCompletion); + guest(Ftp::open("ftp".into(), config()))?; + let writer = guest(start_writer("/target".into()))?; + guest(writer.push(b"body".to_vec()))?; + assert!(matches!(writer.finish(), Err(ErrorCode::OutcomeUnknown))); + drop(writer); + Ftp::close(); + let summary = fixture.summary(); + assert_eq!(summary.control_connections, 2); + assert_eq!(summary.upload_commands, 1); + assert_eq!(summary.target_upload_commands, 0); + assert_eq!(summary.uploaded_bytes, 4); + assert_eq!(summary.stage_deletes, 1); + assert_eq!(summary.promotions, 0); + Ok(()) +} + +#[tokio::test] +async fn an_uncertain_upload_write_discards_the_control_session() -> TestResult +{ + let fixture = test_host::install(test_host::Scenario::LostUploadData); + guest(Ftp::open("ftp".into(), config()))?; + let writer = guest(start_writer("/target".into()))?; + assert!(matches!( + writer.push(b"body".to_vec()), + Err(ErrorCode::OutcomeUnknown) + )); + let page = guest(Ftp::list(Locator::Path("/".into()), None, 10))?; + assert_eq!(page.entries[0].name, " report.md "); + drop(writer); + Ftp::close(); + let summary = fixture.summary(); + assert_eq!(summary.control_connections, 2); + assert_eq!(summary.upload_commands, 1); + assert_eq!(summary.target_upload_commands, 0); + assert_eq!(summary.uploaded_bytes, 0); + assert_eq!(summary.stage_deletes, 1); + assert_eq!(summary.promotions, 0); + assert_eq!(summary.listings, 1); + Ok(()) +} + +#[tokio::test] +async fn a_complete_upload_promotes_its_private_stage() -> TestResult { + let fixture = test_host::install(test_host::Scenario::SuccessfulUpload); + guest(Ftp::open("ftp".into(), config()))?; + let writer = guest(start_writer("/target".into()))?; + guest(writer.push(b"body".to_vec()))?; + let info = guest(writer.finish())?.ok_or("missing upload result")?; + assert_eq!(info.size, 4); + drop(writer); + Ftp::close(); + let summary = fixture.summary(); + assert_eq!(summary.control_connections, 1); + assert_eq!(summary.upload_commands, 1); + assert_eq!(summary.target_upload_commands, 0); + assert_eq!(summary.uploaded_bytes, 4); + assert_eq!(summary.stage_deletes, 0); + assert_eq!(summary.promotions, 1); + Ok(()) +} + +#[tokio::test] +async fn dropping_an_upload_removes_only_its_private_stage() -> TestResult { + let fixture = test_host::install(test_host::Scenario::AbandonedUpload); + guest(Ftp::open("ftp".into(), config()))?; + let writer = guest(start_writer("/target".into()))?; + guest(writer.push(b"partial".to_vec()))?; + drop(writer); + Ftp::close(); + let summary = fixture.summary(); + assert_eq!(summary.control_connections, 1); + assert_eq!(summary.upload_commands, 1); + assert_eq!(summary.target_upload_commands, 0); + assert_eq!(summary.uploaded_bytes, 7); + assert_eq!(summary.stage_deletes, 1); + assert_eq!(summary.promotions, 0); + Ok(()) +} + +#[tokio::test] +async fn a_stage_name_collision_draws_a_fresh_random_token() -> TestResult { + let fixture = test_host::install(test_host::Scenario::StageCollision); + guest(Ftp::open("ftp".into(), config()))?; + let mut tokens = [1, 2].into_iter(); + let staged = guest(stage_path_with("/target", || { + tokens.next().ok_or(ErrorCode::Internal) + }))?; + assert_eq!(staged, "/.memona-upload-00000000000000000000000000000002"); + Ftp::close(); + assert_eq!(fixture.summary().stage_probes, 2); + Ok(()) +} diff --git a/src/test_host.rs b/src/test_host.rs new file mode 100644 index 0000000..1fabc45 --- /dev/null +++ b/src/test_host.rs @@ -0,0 +1,278 @@ +//! In-memory FTP host for the generated synchronous `Guest` API. Production +//! WASM uses Memona's real async socket host; live QA covers that boundary. + +use std::cell::RefCell; +use std::collections::VecDeque; +use std::rc::Rc; + +pub use memona_plugin_sdk::HostError; + +pub const CONTROL_PORT: u16 = 21_210; +const DATA_PORT: u16 = 31_310; +const LISTING: &[u8] = b"type=file;size=3; report.md \r\n"; + +#[derive(Clone, Copy)] +pub enum Scenario { + AbandonedUpload, + CancelledPreflight, + ExpiredListing, + LostUploadData, + LostMutationReply, + LostUploadCompletion, + StageCollision, + SuccessfulUpload, +} + +struct State { + scenario: Scenario, + control_connections: usize, + listings: usize, + mkdir_commands: usize, + upload_commands: usize, + target_upload_commands: usize, + uploaded_bytes: usize, + stage_deletes: usize, + stage_probes: usize, + promotions: usize, +} + +thread_local! { + static ACTIVE: RefCell>>> = const { RefCell::new(None) }; +} + +pub struct Fixture(Rc>); + +pub struct Summary { + pub control_connections: usize, + pub listings: usize, + pub mkdir_commands: usize, + pub upload_commands: usize, + pub target_upload_commands: usize, + pub uploaded_bytes: usize, + pub stage_deletes: usize, + pub stage_probes: usize, + pub promotions: usize, +} + +pub fn install(scenario: Scenario) -> Fixture { + let state = Rc::new(RefCell::new(State { + scenario, + control_connections: 0, + listings: 0, + mkdir_commands: 0, + upload_commands: 0, + target_upload_commands: 0, + uploaded_bytes: 0, + stage_deletes: 0, + stage_probes: 0, + promotions: 0, + })); + ACTIVE.set(Some(state.clone())); + Fixture(state) +} + +impl Fixture { + pub fn summary(&self) -> Summary { + let state = self.0.borrow(); + Summary { + control_connections: state.control_connections, + listings: state.listings, + mkdir_commands: state.mkdir_commands, + upload_commands: state.upload_commands, + target_upload_commands: state.target_upload_commands, + uploaded_bytes: state.uploaded_bytes, + stage_deletes: state.stage_deletes, + stage_probes: state.stage_probes, + promotions: state.promotions, + } + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + ACTIVE.with_borrow_mut(|active| { + if active + .as_ref() + .is_some_and(|state| Rc::ptr_eq(state, &self.0)) + { + *active = None; + } + }); + } +} + +enum Kind { + Control(usize), + Data, +} + +pub struct Tcp { + state: Rc>, + kind: Kind, + input: RefCell>, +} + +pub fn connect(name: &str, port: u16, tls: bool) -> Result { + if tls || name != "127.0.0.1" { + return Err(HostError::InvalidInput); + } + let state = ACTIVE + .with_borrow(Clone::clone) + .ok_or(HostError::Internal)?; + match port { + CONTROL_PORT => { + let index = { + let mut state = state.borrow_mut(); + let index = state.control_connections; + state.control_connections += 1; + index + }; + Ok(Tcp { + state, + kind: Kind::Control(index), + input: RefCell::new(VecDeque::from(b"220 Fixture\r\n".to_vec())), + }) + } + DATA_PORT => Ok(Tcp { + state, + kind: Kind::Data, + input: RefCell::new(VecDeque::from(LISTING.to_vec())), + }), + _ => Err(HostError::Denied), + } +} + +impl Tcp { + pub fn read(&self, limit: u32) -> Result, HostError> { + let mut input = self.input.borrow_mut(); + if input.is_empty() && matches!(self.kind, Kind::Control(_)) { + return Err(HostError::Unavailable); + } + let count = usize::min(limit as usize, input.len()); + Ok(input.drain(..count).collect()) + } + + pub fn write(&self, bytes: &[u8]) -> Result { + if matches!(self.kind, Kind::Data) { + let mut state = self.state.borrow_mut(); + if matches!( + state.scenario, + Scenario::AbandonedUpload + | Scenario::LostUploadCompletion + | Scenario::SuccessfulUpload + ) { + state.uploaded_bytes += bytes.len(); + return Ok(bytes.len() as u32); + } + return Err(HostError::InvalidInput); + } + let Kind::Control(index) = self.kind else { + unreachable!(); + }; + let command = std::str::from_utf8(bytes) + .map_err(|_| HostError::InvalidInput)? + .strip_suffix("\r\n") + .ok_or(HostError::InvalidInput)?; + let response = { + let mut state = self.state.borrow_mut(); + match command { + "USER fixture" => "331 Password required\r\n", + "PASS fixture secret " => "230 Signed in\r\n", + "TYPE I" => "200 Binary\r\n", + "QUIT" => "221 Bye\r\n", + "EPSV" => "229 Passive (|||31310|)\r\n", + "MLSD /" + if matches!( + state.scenario, + Scenario::ExpiredListing | Scenario::LostUploadData + ) => + { + state.listings += 1; + "150 Listing\r\n226 Complete\r\n" + } + "NOOP" if matches!(state.scenario, Scenario::CancelledPreflight) => { + return Err(HostError::Cancelled); + } + "NOOP" + if index == 0 + && (matches!(state.scenario, Scenario::LostMutationReply) + || state.listings > 0) => + { + return Err(HostError::Unavailable); + } + "NOOP" => "200 Still connected\r\n", + "MLST /.memona-upload-00000000000000000000000000000001" + if matches!(state.scenario, Scenario::StageCollision) => + { + state.stage_probes += 1; + "250-Begin\r\n type=file;size=3; /.memona-upload-00000000000000000000000000000001\r\n250 End\r\n" + } + command if command.starts_with("MLST /.memona-upload-") => { + state.stage_probes += 1; + "550 Missing\r\n" + } + command + if command.starts_with("STOR /.memona-upload-") + && matches!( + state.scenario, + Scenario::AbandonedUpload | Scenario::LostUploadCompletion + ) => + { + state.upload_commands += 1; + "150 Upload started\r\n" + } + command + if command.starts_with("STOR /.memona-upload-") + && matches!(state.scenario, Scenario::LostUploadData) => + { + state.upload_commands += 1; + "150 Upload started\r\n" + } + command + if command.starts_with("STOR /.memona-upload-") + && matches!(state.scenario, Scenario::SuccessfulUpload) => + { + state.upload_commands += 1; + "150 Upload started\r\n226 Upload complete\r\n" + } + command if command.starts_with("STOR ") => { + state.target_upload_commands += 1; + "150 Target upload started\r\n" + } + command if command.starts_with("DELE /.memona-upload-") => { + state.stage_deletes += 1; + "250 Stage deleted\r\n" + } + command if command.starts_with("RNFR /.memona-upload-") => { + "350 Ready for destination\r\n" + } + "RNTO /target" + if matches!(state.scenario, Scenario::SuccessfulUpload) => + { + state.promotions += 1; + "250 Promoted\r\n" + } + "ABOR" if matches!(state.scenario, Scenario::AbandonedUpload) => { + "426 Upload abandoned\r\n226 Abort complete\r\n" + } + "MKD /new" + if index == 1 + && matches!(state.scenario, Scenario::LostMutationReply) => + { + state.mkdir_commands += 1; + return Ok(bytes.len() as u32); + } + "MKD /cancelled" + if index == 1 + && matches!(state.scenario, Scenario::CancelledPreflight) => + { + state.mkdir_commands += 1; + return Err(HostError::Unavailable); + } + _ => return Err(HostError::InvalidInput), + } + }; + self.input.borrow_mut().extend(response.as_bytes()); + Ok(bytes.len() as u32) + } +}