From e4f427fa9d88fc7d8a5bc91eacd74727eea795c3 Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 15:24:28 +0900 Subject: [PATCH 01/15] Preserve FTP password and filename whitespace --- Cargo.lock | 28 ++++++++++++++++++++++++++++ Cargo.toml | 3 +++ src/control.rs | 9 +++++++-- src/lib.rs | 24 +++++++++++++++++++++++- src/listing.rs | 16 +++++++++++++++- 5 files changed, 76 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8b97008..f94ad01 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -99,6 +99,7 @@ name = "memona-plugin-ftp" version = "0.1.2" dependencies = [ "memona-plugin-sdk", + "tokio", ] [[package]] @@ -108,6 +109,12 @@ dependencies = [ "wit-bindgen", ] +[[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" @@ -195,6 +202,27 @@ 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" diff --git a/Cargo.toml b/Cargo.toml index de78c21..12289e4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,9 @@ crate-type = ["cdylib", "rlib"] [dependencies] memona-plugin-sdk = { path = "sdk" } +[dev-dependencies] +tokio = { version = "1", features = ["macros", "rt"] } + [lints] workspace = true diff --git a/src/control.rs b/src/control.rs index 4cfbbba..30ee037 100644 --- a/src/control.rs +++ b/src/control.rs @@ -64,7 +64,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 @@ -255,7 +257,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..3eb4ad2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -31,10 +31,11 @@ struct Ftp; 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, } } @@ -394,3 +395,24 @@ impl Guest for Ftp { } 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> { -- 2.52.0 From 69c845eabeb9eec9420fed4bbbedc3992b0b11ad Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 15:30:52 +0900 Subject: [PATCH 02/15] Check plugin builds and request AI review in Gitea --- .gitea/workflows/agents.yml | 49 +++++++++++++++++++++++++++++++++++ .gitea/workflows/ci.yml | 51 +++++++++++++++++++++++++++++++++++++ 2 files changed, 100 insertions(+) create mode 100644 .gitea/workflows/agents.yml create mode 100644 .gitea/workflows/ci.yml 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 + ' -- 2.52.0 From 0993f2f36fbe70d9ddcfd7e1a8f685e1728731d0 Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 15:31:30 +0900 Subject: [PATCH 03/15] Prepare FTP 0.1.3 --- Cargo.lock | 2 +- Cargo.toml | 2 +- manifest.json | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f94ad01..f5c4452 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -96,7 +96,7 @@ checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" [[package]] name = "memona-plugin-ftp" -version = "0.1.2" +version = "0.1.3" dependencies = [ "memona-plugin-sdk", "tokio", diff --git a/Cargo.toml b/Cargo.toml index 12289e4..77a50dc 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" diff --git a/manifest.json b/manifest.json index 1a792d9..2093339 100644 --- a/manifest.json +++ b/manifest.json @@ -2,7 +2,7 @@ "schemaVersion": 1, "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", -- 2.52.0 From 7f9fd13a02259410d1d79aee11a42e3bff9efcff Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 15:40:31 +0900 Subject: [PATCH 04/15] Clarify that remote paths carry literal filesystem names --- sdk/wit/filesystem.wit | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/sdk/wit/filesystem.wit b/sdk/wit/filesystem.wit index 4529372..dc8d5c8 100644 --- a/sdk/wit/filesystem.wit +++ b/sdk/wit/filesystem.wit @@ -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}; -- 2.52.0 From 55a4f458b6dc395361b5d85ab1696368de648166 Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 15:46:45 +0900 Subject: [PATCH 05/15] Explain connection permissions and supported transports --- README.md | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index ebd7ee2..a2f126b 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,16 @@ 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. +In Memona's connection permissions, enter the server's hostname or IP in +**Allowed hosts**, without a scheme, port or path. An empty list blocks the +connection. Turn on **Private network** as well for localhost or a private +network; this switch does not grant a host by itself. Save and reconnect after +changing permissions. + +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,7 +33,7 @@ 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. -- 2.52.0 From cd4b4c5e70d71b832d14ff689d7fb2030cb7f442 Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 17:46:45 +0900 Subject: [PATCH 06/15] Remove duplicate network allowlists from FTP setup --- README.md | 17 +++++++---------- manifest.json | 3 +-- sdk/wit/filesystem.wit | 20 ++++++++++---------- 3 files changed, 18 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index a2f126b..fb6a6b5 100644 --- a/README.md +++ b/README.md @@ -8,17 +8,16 @@ 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. -In Memona's connection permissions, enter the server's hostname or IP in -**Allowed hosts**, without a scheme, port or path. An empty list blocks the -connection. Turn on **Private network** as well for localhost or a private -network; this switch does not grant a host by itself. Save and reconnect after -changing permissions. +For localhost or a server on a private network, turn on **Private network** in +the connection settings. Save and reconnect after changing these settings. The TLS toggle currently starts TLS on the control connection only. Protected data connections and explicit `AUTH TLS` negotiation are not implemented, so @@ -37,11 +36,9 @@ and connects out to it; `PASV` fallback is not implemented. The server picks a f 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 2093339..f88bda5 100644 --- a/manifest.json +++ b/manifest.json @@ -1,5 +1,5 @@ { - "schemaVersion": 1, + "schemaVersion": 2, "id": "memona/ftp", "name": "FTP", "version": "0.1.3", @@ -7,7 +7,6 @@ "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 dc8d5c8..b5bcdf8 100644 --- a/sdk/wit/filesystem.wit +++ b/sdk/wit/filesystem.wit @@ -35,10 +35,10 @@ 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 + /// 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. http: func(request: request) -> result; @@ -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 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. 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,7 +65,7 @@ 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 + /// Connects a datagram socket under the same port and private-network rules /// as `connect`, for the few filesystem protocols that still need one. bind: func(host: string, port: u16) -> result; -- 2.52.0 From 179c17a337ac3418ba043f1392c2aab78bed6fcc Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 19:12:44 +0900 Subject: [PATCH 07/15] 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) + } +} -- 2.52.0 From a3175ded3a58a99246fe01c2e4def7ea054f460c Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 19:31:54 +0900 Subject: [PATCH 08/15] Clarify network and test execution contracts --- README.md | 5 +++-- sdk/wit/filesystem.wit | 17 +++++++++-------- src/session_tests.rs | 5 +++-- src/test_host.rs | 5 +++-- 4 files changed, 18 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index befc6f4..31924d2 100644 --- a/README.md +++ b/README.md @@ -16,8 +16,9 @@ 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. -Public, LAN, VPN, link-local and localhost addresses work directly from the -server name; there is no separate network permission setting. +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 diff --git a/sdk/wit/filesystem.wit b/sdk/wit/filesystem.wit index 824db1d..624ee9d 100644 --- a/sdk/wit/filesystem.wit +++ b/sdk/wit/filesystem.wit @@ -36,9 +36,10 @@ interface host { pull: func(limit: u32) -> result, host-error>; } /// 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. + /// 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,9 +52,9 @@ interface host { write: func(data: list) -> result; } /// 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. + /// 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 peer. @@ -64,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 port 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. diff --git a/src/session_tests.rs b/src/session_tests.rs index 824e8f8..3ab133e 100644 --- a/src/session_tests.rs +++ b/src/session_tests.rs @@ -1,5 +1,6 @@ -//! Real FTP commands over a native test socket, with an async local server. -//! This verifies guest recovery; WASM host integration is exercised separately. +//! Real FTP commands against an async local server. The generated guest API is +//! synchronous, so each guest call is isolated with `spawn_blocking`; production +//! WASM host integration is exercised separately. use super::Ftp; use memona_plugin_sdk::{ErrorCode, Guest, Locator, Setting}; diff --git a/src/test_host.rs b/src/test_host.rs index d0751e5..32f870b 100644 --- a/src/test_host.rs +++ b/src/test_host.rs @@ -1,5 +1,6 @@ -//! Native test transport. Guest calls run only inside `spawn_blocking`; -//! production WASM always uses the host's sockets and network policy. +//! Native test transport for the generated synchronous `Guest` API. It cannot +//! await a Tokio socket, so tests confine it to `spawn_blocking` while their FTP +//! server remains async. Production WASM always uses the host's async sockets. use std::cell::RefCell; use std::io::{Read, Write}; -- 2.52.0 From a12e4d03f76757f1205f015c70d0d399f2f993db Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 19:39:46 +0900 Subject: [PATCH 09/15] Document synchronous guest test isolation --- src/session_tests.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/session_tests.rs b/src/session_tests.rs index 3ab133e..ddcb5e0 100644 --- a/src/session_tests.rs +++ b/src/session_tests.rs @@ -97,6 +97,8 @@ async fn a_listing_reconnects_after_the_server_expires_the_control_session() }) .await? }); + // `Guest` is generated as a synchronous trait, so the repository rule for + // unavoidable blocking I/O requires isolating this native test call here. 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))?; @@ -137,6 +139,8 @@ async fn recovery_before_a_mutation_never_replays_a_lost_mutation_reply() }) .await? }); + // `Guest` is generated as a synchronous trait, so the repository rule for + // unavoidable blocking I/O requires isolating this native test call here. let client = tokio::task::spawn_blocking(move || -> TestResult { guest(Ftp::open("ftp".into(), config(port)))?; wait_expired.blocking_recv()?; -- 2.52.0 From a1c449843f7d66bbc0bbdb9154600ec65a57f751 Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 19:57:11 +0900 Subject: [PATCH 10/15] Use an in-memory FTP recovery transport --- Cargo.lock | 59 -------------- Cargo.toml | 2 +- src/session_tests.rs | 167 +++++++-------------------------------- src/test_host.rs | 181 +++++++++++++++++++++++++++++++++++-------- 4 files changed, 179 insertions(+), 230 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1ac10fb..f5c4452 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14,12 +14,6 @@ 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" @@ -77,12 +71,6 @@ 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" @@ -121,17 +109,6 @@ 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" @@ -214,16 +191,6 @@ 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" @@ -241,13 +208,8 @@ 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]] @@ -267,12 +229,6 @@ 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" @@ -307,21 +263,6 @@ 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 f665f24..77a50dc 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", "net", "io-util", "sync", "time"] } +tokio = { version = "1", features = ["macros", "rt"] } [lints] workspace = true diff --git a/src/session_tests.rs b/src/session_tests.rs index ddcb5e0..bef5fb2 100644 --- a/src/session_tests.rs +++ b/src/session_tests.rs @@ -1,25 +1,22 @@ -//! Real FTP commands against an async local server. The generated guest API is -//! synchronous, so each guest call is isolated with `spawn_blocking`; production -//! WASM host integration is exercised separately. +//! FTP command recovery against an in-memory protocol transport. Live QA +//! separately exercises the compiled WASM component over real host sockets. -use super::Ftp; +use super::{Ftp, test_host}; 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)> { +fn config() -> Vec<(String, Setting)> { vec![ ("host".into(), Setting::Text("127.0.0.1".into())), - ("port".into(), Setting::Text(port.to_string())), + ( + "port".into(), + Setting::Text(test_host::CONTROL_PORT.to_string()), + ), ("username".into(), Setting::Text("fixture".into())), ( "password".into(), @@ -28,140 +25,36 @@ fn config(port: u16) -> Vec<(String, Setting)> { ] } -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? - }); - // `Guest` is generated as a synchronous trait, so the repository rule for - // unavoidable blocking I/O requires isolating this native test call here. - 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??; + 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 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? - }); - // `Guest` is generated as a synchronous trait, so the repository rule for - // unavoidable blocking I/O requires isolating this native test call here. - 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" - ); + 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::Unavailable) + )); + Ftp::close(); + let summary = fixture.summary(); + assert_eq!(summary.control_connections, 2); + assert_eq!(summary.listings, 0); + assert_eq!(summary.mkdir_commands, 1); Ok(()) } diff --git a/src/test_host.rs b/src/test_host.rs index 32f870b..9be6f4e 100644 --- a/src/test_host.rs +++ b/src/test_host.rs @@ -1,51 +1,166 @@ -//! Native test transport for the generated synchronous `Guest` API. It cannot -//! await a Tokio socket, so tests confine it to `spawn_blocking` while their FTP -//! server remains async. Production WASM always uses the host's async sockets. +//! 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::io::{Read, Write}; -use std::net::{IpAddr, SocketAddr, TcpStream}; -use std::time::Duration; +use std::collections::VecDeque; +use std::rc::Rc; pub use memona_plugin_sdk::HostError; -pub struct Tcp(RefCell); +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 { + ExpiredListing, + LostMutationReply, +} + +struct State { + scenario: Scenario, + control_connections: usize, + listings: usize, + mkdir_commands: 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 fn install(scenario: Scenario) -> Fixture { + let state = Rc::new(RefCell::new(State { + scenario, + control_connections: 0, + listings: 0, + mkdir_commands: 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, + } + } +} + +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 { + if tls || name != "127.0.0.1" { 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))) + 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 bytes = vec![0; limit as usize]; - let count = self - .0 - .borrow_mut() - .read(&mut bytes) - .map_err(|_| HostError::Unavailable)?; - bytes.truncate(count); - Ok(bytes) + 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 { - self - .0 - .borrow_mut() - .write(bytes) - .map(|count| count as u32) - .map_err(|_| HostError::Unavailable) + let Kind::Control(index) = self.kind else { + return Err(HostError::InvalidInput); + }; + 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) => { + state.listings += 1; + "150 Listing\r\n226 Complete\r\n" + } + "NOOP" + if index == 0 + && (matches!(state.scenario, Scenario::LostMutationReply) + || state.listings > 0) => + { + return Err(HostError::Unavailable); + } + "NOOP" => "200 Still connected\r\n", + "MKD /new" + if index == 1 + && matches!(state.scenario, Scenario::LostMutationReply) => + { + state.mkdir_commands += 1; + return Ok(bytes.len() as u32); + } + _ => return Err(HostError::InvalidInput), + } + }; + self.input.borrow_mut().extend(response.as_bytes()); + Ok(bytes.len() as u32) } } -- 2.52.0 From 25d572aa3365674fe0d611b50f3bae4ea59a035b Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 20:06:43 +0900 Subject: [PATCH 11/15] Propagate FTP preflight cancellation --- src/lib.rs | 8 ++++++-- src/session_tests.rs | 17 +++++++++++++++++ src/test_host.rs | 11 +++++++++++ 3 files changed, 34 insertions(+), 2 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 416f012..cbc279b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -100,8 +100,12 @@ fn with_control( /// 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(()); + 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) diff --git a/src/session_tests.rs b/src/session_tests.rs index bef5fb2..8e059fa 100644 --- a/src/session_tests.rs +++ b/src/session_tests.rs @@ -58,3 +58,20 @@ async fn recovery_before_a_mutation_never_replays_a_lost_mutation_reply() 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(()) +} diff --git a/src/test_host.rs b/src/test_host.rs index 9be6f4e..cff73ec 100644 --- a/src/test_host.rs +++ b/src/test_host.rs @@ -13,6 +13,7 @@ const LISTING: &[u8] = b"type=file;size=3; report.md \r\n"; #[derive(Clone, Copy)] pub enum Scenario { + CancelledPreflight, ExpiredListing, LostMutationReply, } @@ -142,6 +143,9 @@ impl Tcp { 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) @@ -157,6 +161,13 @@ impl Tcp { 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), } }; -- 2.52.0 From 70b0cf146bb0f8e00e7a4f024a31b6d1c278874d Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 20:22:22 +0900 Subject: [PATCH 12/15] Report uncertain FTP mutation outcomes --- src/control.rs | 57 ++++++++++++++++++++++++++++++++------ src/lib.rs | 65 ++++++++++++++++++++++++++++++++++---------- src/session_tests.rs | 20 ++++++++++++-- src/test_host.rs | 25 ++++++++++++++++- 4 files changed, 141 insertions(+), 26 deletions(-) diff --git a/src/control.rs b/src/control.rs index fb0c3b8..1cf2022 100644 --- a/src/control.rs +++ b/src/control.rs @@ -91,6 +91,11 @@ pub struct Control { buffer: Vec, } +struct DispatchError { + code: ErrorCode, + dispatched: bool, +} + impl Control { pub fn open( host_name: String, @@ -155,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)?; diff --git a/src/lib.rs b/src/lib.rs index cbc279b..4cb2613 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -89,7 +89,14 @@ fn with_control( Some(control) => action(control), None => Err(ErrorCode::Internal), }; - if matches!(result, Err(ErrorCode::Unavailable | ErrorCode::Internal)) { + if matches!( + result, + Err( + ErrorCode::Unavailable + | ErrorCode::OutcomeUnknown + | ErrorCode::Internal + ) + ) { *slot = None; } result @@ -126,6 +133,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| { @@ -134,6 +152,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) @@ -237,9 +265,11 @@ impl GuestWriter for FtpWriter { let socket = socket.as_ref().ok_or(ErrorCode::InvalidInput)?; let mut sent = 0; while sent < data.len() { - let written = socket.write(&data[sent..]).map_err(transport)? as usize; + let written = socket + .write(&data[sent..]) + .map_err(|_| ErrorCode::OutcomeUnknown)? as usize; if written == 0 { - return Err(ErrorCode::Unavailable); + return Err(ErrorCode::OutcomeUnknown); } sent += written; } @@ -250,7 +280,7 @@ impl GuestWriter for FtpWriter { /// server answer, so the socket is given up before the reply is read. fn finish(&self) -> Result, ErrorCode> { drop(self.socket.take().ok_or(ErrorCode::InvalidInput)?); - settle()?; + settle_mutation()?; Ok(Some(FileInfo { kind: EntryKind::File, size: self.written.get(), @@ -269,8 +299,11 @@ impl Drop for FtpWriter { }; drop(socket); if abort().is_ok() { - let _ = - with_control(|control| control.expect(&format!("DELE {}", self.path))); + let _ = with_control(|control| { + let (code, _) = + control.send_mutation(&format!("DELE {}", self.path))?; + outcome(code) + }); } } } @@ -394,7 +427,7 @@ impl Guest for Ftp { return Err(ErrorCode::Unsupported); } ready()?; - let socket = transfer(&format!("STOR {path}"))?; + let socket = mutation_transfer(&format!("STOR {path}"))?; Ok(Writer::new(FtpWriter { path, socket: RefCell::new(Some(socket)), @@ -408,7 +441,7 @@ impl Guest for Ftp { 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 { @@ -427,14 +460,16 @@ impl Guest for Ftp { 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) }) } @@ -445,11 +480,13 @@ impl Guest for Ftp { 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) }) } diff --git a/src/session_tests.rs b/src/session_tests.rs index 8e059fa..611b1a1 100644 --- a/src/session_tests.rs +++ b/src/session_tests.rs @@ -1,7 +1,7 @@ //! FTP command recovery against an in-memory protocol transport. Live QA //! separately exercises the compiled WASM component over real host sockets. -use super::{Ftp, test_host}; +use super::{Ftp, mutation_transfer, settle_mutation, test_host}; use memona_plugin_sdk::{ErrorCode, Guest, Locator, Setting}; type TestResult = Result>; @@ -49,7 +49,7 @@ async fn recovery_before_a_mutation_never_replays_a_lost_mutation_reply() guest(Ftp::open("ftp".into(), config()))?; assert!(matches!( Ftp::mkdir(Locator::Path("/new".into())), - Err(ErrorCode::Unavailable) + Err(ErrorCode::OutcomeUnknown) )); Ftp::close(); let summary = fixture.summary(); @@ -75,3 +75,19 @@ async fn cancelled_preflight_never_reconnects_or_executes_the_mutation() 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 socket = guest(mutation_transfer("STOR /upload"))?; + assert!(matches!(socket.write(b"body"), Ok(4))); + drop(socket); + assert!(matches!(settle_mutation(), Err(ErrorCode::OutcomeUnknown))); + Ftp::close(); + let summary = fixture.summary(); + assert_eq!(summary.control_connections, 1); + assert_eq!(summary.upload_commands, 1); + assert_eq!(summary.uploaded_bytes, 4); + Ok(()) +} diff --git a/src/test_host.rs b/src/test_host.rs index cff73ec..98a1bf7 100644 --- a/src/test_host.rs +++ b/src/test_host.rs @@ -16,6 +16,7 @@ pub enum Scenario { CancelledPreflight, ExpiredListing, LostMutationReply, + LostUploadCompletion, } struct State { @@ -23,6 +24,8 @@ struct State { control_connections: usize, listings: usize, mkdir_commands: usize, + upload_commands: usize, + uploaded_bytes: usize, } thread_local! { @@ -35,6 +38,8 @@ pub struct Summary { pub control_connections: usize, pub listings: usize, pub mkdir_commands: usize, + pub upload_commands: usize, + pub uploaded_bytes: usize, } pub fn install(scenario: Scenario) -> Fixture { @@ -43,6 +48,8 @@ pub fn install(scenario: Scenario) -> Fixture { control_connections: 0, listings: 0, mkdir_commands: 0, + upload_commands: 0, + uploaded_bytes: 0, })); ACTIVE.set(Some(state.clone())); Fixture(state) @@ -55,6 +62,8 @@ impl Fixture { control_connections: state.control_connections, listings: state.listings, mkdir_commands: state.mkdir_commands, + upload_commands: state.upload_commands, + uploaded_bytes: state.uploaded_bytes, } } } @@ -124,8 +133,16 @@ impl Tcp { } pub fn write(&self, bytes: &[u8]) -> Result { - let Kind::Control(index) = self.kind else { + if matches!(self.kind, Kind::Data) { + let mut state = self.state.borrow_mut(); + if matches!(state.scenario, Scenario::LostUploadCompletion) { + 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)? @@ -154,6 +171,12 @@ impl Tcp { return Err(HostError::Unavailable); } "NOOP" => "200 Still connected\r\n", + "STOR /upload" + if matches!(state.scenario, Scenario::LostUploadCompletion) => + { + state.upload_commands += 1; + "150 Upload started\r\n" + } "MKD /new" if index == 1 && matches!(state.scenario, Scenario::LostMutationReply) => -- 2.52.0 From e789a2ee6da9adfc4a6e183309567e26930491d7 Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 20:29:52 +0900 Subject: [PATCH 13/15] Discard uncertain FTP upload sessions --- src/lib.rs | 32 ++++++++++++++++++++++++++------ src/session_tests.rs | 31 +++++++++++++++++++++++++++++-- src/test_host.rs | 14 +++++++++++++- 3 files changed, 68 insertions(+), 9 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 4cb2613..39d386a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -259,17 +259,37 @@ struct FtpWriter { written: Cell, } +impl FtpWriter { + /// A failed data write can leave both the remote file and its pending + /// completion reply in an unknown state. Close both channels so the next + /// filesystem operation must start a fresh session. + fn discard_uncertain_upload(&self) -> ErrorCode { + drop(self.socket.take()); + CONTROL.set(None); + 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(|_| ErrorCode::OutcomeUnknown)? 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::OutcomeUnknown); + return Err(self.discard_uncertain_upload()); } sent += written; } diff --git a/src/session_tests.rs b/src/session_tests.rs index 611b1a1..d0a3f25 100644 --- a/src/session_tests.rs +++ b/src/session_tests.rs @@ -1,8 +1,9 @@ //! FTP command recovery against an in-memory protocol transport. Live QA //! separately exercises the compiled WASM component over real host sockets. -use super::{Ftp, mutation_transfer, settle_mutation, test_host}; -use memona_plugin_sdk::{ErrorCode, Guest, Locator, Setting}; +use super::{Ftp, FtpWriter, mutation_transfer, settle_mutation, test_host}; +use memona_plugin_sdk::{ErrorCode, Guest, GuestWriter, Locator, Setting}; +use std::cell::{Cell, RefCell}; type TestResult = Result>; @@ -91,3 +92,29 @@ async fn a_lost_upload_completion_reports_an_unknown_outcome() -> TestResult { assert_eq!(summary.uploaded_bytes, 4); 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 = FtpWriter { + path: "/broken".into(), + socket: RefCell::new(Some(guest(mutation_transfer("STOR /broken"))?)), + written: Cell::new(0), + }; + 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.uploaded_bytes, 0); + assert_eq!(summary.listings, 1); + Ok(()) +} diff --git a/src/test_host.rs b/src/test_host.rs index 98a1bf7..6a22113 100644 --- a/src/test_host.rs +++ b/src/test_host.rs @@ -15,6 +15,7 @@ const LISTING: &[u8] = b"type=file;size=3; report.md \r\n"; pub enum Scenario { CancelledPreflight, ExpiredListing, + LostUploadData, LostMutationReply, LostUploadCompletion, } @@ -156,7 +157,12 @@ impl Tcp { "TYPE I" => "200 Binary\r\n", "QUIT" => "221 Bye\r\n", "EPSV" => "229 Passive (|||31310|)\r\n", - "MLSD /" if matches!(state.scenario, Scenario::ExpiredListing) => { + "MLSD /" + if matches!( + state.scenario, + Scenario::ExpiredListing | Scenario::LostUploadData + ) => + { state.listings += 1; "150 Listing\r\n226 Complete\r\n" } @@ -177,6 +183,12 @@ impl Tcp { state.upload_commands += 1; "150 Upload started\r\n" } + "STOR /broken" + if matches!(state.scenario, Scenario::LostUploadData) => + { + state.upload_commands += 1; + "150 Upload started\r\n" + } "MKD /new" if index == 1 && matches!(state.scenario, Scenario::LostMutationReply) => -- 2.52.0 From bc6a4953610a002875dfaedda2d1b1dc440a53a1 Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 20:42:29 +0900 Subject: [PATCH 14/15] Stage FTP uploads before promotion --- src/lib.rs | 115 +++++++++++++++++++++++++++++++++---------- src/session_tests.rs | 63 +++++++++++++++++++----- src/test_host.rs | 64 ++++++++++++++++++++++-- 3 files changed, 200 insertions(+), 42 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 39d386a..e02b80c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -27,10 +27,12 @@ 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) }; + static NEXT_STAGE: Cell = const { Cell::new(0) }; /// 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()); @@ -254,18 +256,38 @@ 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 so the next - /// filesystem operation must start a fresh session. + /// 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 } } @@ -297,10 +319,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_mutation()?; + 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(), @@ -310,22 +352,51 @@ 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| { - let (code, _) = - control.send_mutation(&format!("DELE {}", self.path))?; - outcome(code) - }); + if let Some(socket) = self.socket.take() { + drop(socket); + if abort().is_err() { + CONTROL.set(None); + } + } + self.cleanup_stage(); + } +} + +/// Chooses an unused private sibling for one upload. The sequence is local to +/// the component, and the remote probe prevents reuse after an app restart. +fn stage_path(target: &str) -> 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 sequence = NEXT_STAGE.get(); + NEXT_STAGE.set(sequence.wrapping_add(1)); + let staged = format!("{parent}/.memona-upload-{sequence:016x}"); + match with_control(|control| probe(control, &staged)) { + Err(ErrorCode::NotFound) => return Ok(staged), + Ok(_) => {} + Err(error) => return Err(error), } } + Err(ErrorCode::Conflict) +} + +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 { @@ -446,13 +517,7 @@ impl Guest for Ftp { if !matches!(condition, WriteCondition::Overwrite) { return Err(ErrorCode::Unsupported); } - ready()?; - let socket = mutation_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 diff --git a/src/session_tests.rs b/src/session_tests.rs index d0a3f25..a594aa9 100644 --- a/src/session_tests.rs +++ b/src/session_tests.rs @@ -1,9 +1,8 @@ //! FTP command recovery against an in-memory protocol transport. Live QA //! separately exercises the compiled WASM component over real host sockets. -use super::{Ftp, FtpWriter, mutation_transfer, settle_mutation, test_host}; +use super::{Ftp, start_writer, test_host}; use memona_plugin_sdk::{ErrorCode, Guest, GuestWriter, Locator, Setting}; -use std::cell::{Cell, RefCell}; type TestResult = Result>; @@ -81,15 +80,18 @@ async fn cancelled_preflight_never_reconnects_or_executes_the_mutation() 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 socket = guest(mutation_transfer("STOR /upload"))?; - assert!(matches!(socket.write(b"body"), Ok(4))); - drop(socket); - assert!(matches!(settle_mutation(), Err(ErrorCode::OutcomeUnknown))); + 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, 1); + 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(()) } @@ -98,11 +100,7 @@ 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 = FtpWriter { - path: "/broken".into(), - socket: RefCell::new(Some(guest(mutation_transfer("STOR /broken"))?)), - written: Cell::new(0), - }; + let writer = guest(start_writer("/target".into()))?; assert!(matches!( writer.push(b"body".to_vec()), Err(ErrorCode::OutcomeUnknown) @@ -114,7 +112,48 @@ async fn an_uncertain_upload_write_discards_the_control_session() -> TestResult 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(()) +} diff --git a/src/test_host.rs b/src/test_host.rs index 6a22113..a83444a 100644 --- a/src/test_host.rs +++ b/src/test_host.rs @@ -13,11 +13,13 @@ const LISTING: &[u8] = b"type=file;size=3; report.md \r\n"; #[derive(Clone, Copy)] pub enum Scenario { + AbandonedUpload, CancelledPreflight, ExpiredListing, LostUploadData, LostMutationReply, LostUploadCompletion, + SuccessfulUpload, } struct State { @@ -26,7 +28,10 @@ struct State { listings: usize, mkdir_commands: usize, upload_commands: usize, + target_upload_commands: usize, uploaded_bytes: usize, + stage_deletes: usize, + promotions: usize, } thread_local! { @@ -40,7 +45,10 @@ pub struct Summary { 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 promotions: usize, } pub fn install(scenario: Scenario) -> Fixture { @@ -50,7 +58,10 @@ pub fn install(scenario: Scenario) -> Fixture { listings: 0, mkdir_commands: 0, upload_commands: 0, + target_upload_commands: 0, uploaded_bytes: 0, + stage_deletes: 0, + promotions: 0, })); ACTIVE.set(Some(state.clone())); Fixture(state) @@ -64,7 +75,10 @@ impl Fixture { 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, + promotions: state.promotions, } } } @@ -136,7 +150,12 @@ impl Tcp { 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::LostUploadCompletion) { + if matches!( + state.scenario, + Scenario::AbandonedUpload + | Scenario::LostUploadCompletion + | Scenario::SuccessfulUpload + ) { state.uploaded_bytes += bytes.len(); return Ok(bytes.len() as u32); } @@ -177,18 +196,53 @@ impl Tcp { return Err(HostError::Unavailable); } "NOOP" => "200 Still connected\r\n", - "STOR /upload" - if matches!(state.scenario, Scenario::LostUploadCompletion) => + command if command.starts_with("MLST /.memona-upload-") => { + "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" } - "STOR /broken" - if matches!(state.scenario, Scenario::LostUploadData) => + 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) => -- 2.52.0 From 3e7c4f2f032af5d4ea38ba253996531db5d4fbd2 Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 20:51:45 +0900 Subject: [PATCH 15/15] Randomize FTP upload stages --- Cargo.lock | 48 +++++++++++++++++++++++++++++++++++++++++++- Cargo.toml | 1 + src/lib.rs | 26 +++++++++++++++++------- src/session_tests.rs | 16 ++++++++++++++- src/test_host.rs | 12 +++++++++++ 5 files changed, 94 insertions(+), 9 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f5c4452..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" @@ -98,6 +122,7 @@ checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" name = "memona-plugin-ftp" version = "0.1.3" dependencies = [ + "getrandom", "memona-plugin-sdk", "tokio", ] @@ -106,7 +131,7 @@ dependencies = [ name = "memona-plugin-sdk" version = "0.1.0" dependencies = [ - "wit-bindgen", + "wit-bindgen 0.62.0", ] [[package]] @@ -143,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" @@ -229,6 +260,15 @@ 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" @@ -263,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 77a50dc..2d3a3ea 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ license = "MIT" crate-type = ["cdylib", "rlib"] [dependencies] +getrandom = "0.3" memona-plugin-sdk = { path = "sdk" } [dev-dependencies] diff --git a/src/lib.rs b/src/lib.rs index e02b80c..944ec76 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -32,7 +32,6 @@ const MAX_STAGE_ATTEMPTS: usize = 64; thread_local! { static CONTROL: RefCell> = const { RefCell::new(None) }; static SESSION: RefCell> = const { RefCell::new(None) }; - static NEXT_STAGE: Cell = const { Cell::new(0) }; /// 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()); @@ -366,18 +365,27 @@ impl Drop for FtpWriter { } } -/// Chooses an unused private sibling for one upload. The sequence is local to -/// the component, and the remote probe prevents reuse after an app restart. -fn stage_path(target: &str) -> Result { +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 sequence = NEXT_STAGE.get(); - NEXT_STAGE.set(sequence.wrapping_add(1)); - let staged = format!("{parent}/.memona-upload-{sequence:016x}"); + 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(_) => {} @@ -387,6 +395,10 @@ fn stage_path(target: &str) -> Result { 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)?; diff --git a/src/session_tests.rs b/src/session_tests.rs index a594aa9..37f78be 100644 --- a/src/session_tests.rs +++ b/src/session_tests.rs @@ -1,7 +1,7 @@ //! FTP command recovery against an in-memory protocol transport. Live QA //! separately exercises the compiled WASM component over real host sockets. -use super::{Ftp, start_writer, test_host}; +use super::{Ftp, stage_path_with, start_writer, test_host}; use memona_plugin_sdk::{ErrorCode, Guest, GuestWriter, Locator, Setting}; type TestResult = Result>; @@ -157,3 +157,17 @@ async fn dropping_an_upload_removes_only_its_private_stage() -> TestResult { 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 index a83444a..1fabc45 100644 --- a/src/test_host.rs +++ b/src/test_host.rs @@ -19,6 +19,7 @@ pub enum Scenario { LostUploadData, LostMutationReply, LostUploadCompletion, + StageCollision, SuccessfulUpload, } @@ -31,6 +32,7 @@ struct State { target_upload_commands: usize, uploaded_bytes: usize, stage_deletes: usize, + stage_probes: usize, promotions: usize, } @@ -48,6 +50,7 @@ pub struct Summary { pub target_upload_commands: usize, pub uploaded_bytes: usize, pub stage_deletes: usize, + pub stage_probes: usize, pub promotions: usize, } @@ -61,6 +64,7 @@ pub fn install(scenario: Scenario) -> Fixture { target_upload_commands: 0, uploaded_bytes: 0, stage_deletes: 0, + stage_probes: 0, promotions: 0, })); ACTIVE.set(Some(state.clone())); @@ -78,6 +82,7 @@ impl Fixture { 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, } } @@ -196,7 +201,14 @@ impl Tcp { 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 -- 2.52.0