commons/memona-plugin-ftp
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

FTP connections preserve passwords and filenames, with CI checking every change (#1)
CI / checks (push) Successful in 53s
Details

Browse Source 
Co-authored-by: Danny Kim <temeddix@gmail.com>
Co-committed-by: Danny Kim <temeddix@gmail.com>
This commit was merged in pull request #1.
This commit is contained in:
Danny Kim
2026-09-30 13:10:40 +00:00
committed by temeddix
parent d412eb1d13
commit 90a54034f7
12 changed files with 1004 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitea/workflows/agents.yml
+49
View File
@@ -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 }}
.gitea/workflows/ci.yml
+51
View File
@@ -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
'
Cargo.lock
Generated
+76 -2
View File
@@ -14,6 +14,12 @@ version = "2.13.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06"
[[package]]
name = "cfg-if"
version = "1.0.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600"
[[package]]
name = "equivalent"
version = "1.0.2"
@@ -26,6 +32,18 @@ version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb"
[[package]]
name = "getrandom"
version = "0.3.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
dependencies = [
"cfg-if",
"libc",
"r-efi",
"wasip2",
]
[[package]]
name = "hashbrown"
version = "0.17.1"
@@ -71,6 +89,12 @@ version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2"
[[package]]
name = "libc"
version = "0.2.189"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
[[package]]
name = "log"
version = "0.4.34"
@@ -96,18 +120,26 @@ checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
[[package]]
name = "memona-plugin-ftp"
version = "0.1.2"
version = "0.1.3"
dependencies = [
"getrandom",
"memona-plugin-sdk",
"tokio",
]
[[package]]
name = "memona-plugin-sdk"
version = "0.1.0"
dependencies = [
"wit-bindgen",
"wit-bindgen 0.62.0",
]
[[package]]
name = "pin-project-lite"
version = "0.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
[[package]]
name = "prettyplease"
version = "0.3.0"
@@ -136,6 +168,12 @@ dependencies = [
"proc-macro2",
]
[[package]]
name = "r-efi"
version = "5.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
[[package]]
name = "semver"
version = "1.0.28"
@@ -195,12 +233,42 @@ dependencies = [
"unicode-ident",
]
[[package]]
name = "tokio"
version = "1.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
dependencies = [
"pin-project-lite",
"tokio-macros",
]
[[package]]
name = "tokio-macros"
version = "2.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "unicode-ident"
version = "1.0.26"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954"
[[package]]
name = "wasip2"
version = "1.0.1+wasi-0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0562428422c63773dad2c345a1882263bbf4d65cf3f42e90921f787ef5ad58e7"
dependencies = [
"wit-bindgen 0.46.0",
]
[[package]]
name = "wasm-encoder"
version = "0.259.0"
@@ -235,6 +303,12 @@ dependencies = [
"semver",
]
[[package]]
name = "wit-bindgen"
version = "0.46.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f17a85883d4e6d00e8a97c586de764dabcc06133f7f1d55dce5cdc070ad7fe59"
[[package]]
name = "wit-bindgen"
version = "0.62.0"
Cargo.toml
+5 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "memona-plugin-ftp"
version = "0.1.2"
version = "0.1.3"
edition = "2024"
license = "MIT"
@@ -8,8 +8,12 @@ license = "MIT"
crate-type = ["cdylib", "rlib"]
[dependencies]
getrandom = "0.3"
memona-plugin-sdk = { path = "sdk" }
[dev-dependencies]
tokio = { version = "1", features = ["macros", "rt"] }
[lints]
workspace = true
README.md
+14 -6
View File
@@ -8,12 +8,22 @@ The plugin is a WebAssembly component built against the Memona filesystem
contract, `memona:filesystem@0.1.0`. It uses only the host's sockets; it has no
ambient filesystem, environment or network access of its own.
This package requires a Memona version that supports plugin manifest schema 2.
## What it registers
One provider, `ftp`, for the `ftp` and `ftps` schemes. A connection is
configured with a server name, an optional port, a TLS toggle for FTPS, and an
optional username and password. An empty username signs in anonymously.
With the compatible schema 2 Memona version required by this package, public,
LAN, VPN, link-local and localhost addresses work directly from the server
name; that API has no separate connection network grant.
The TLS toggle currently starts TLS on the control connection only. Protected
data connections and explicit `AUTH TLS` negotiation are not implemented, so
the plugin does not yet provide a complete FTPS connection.
Listing and stat need a server that answers `MLSD` and `MLST`, the
machine-readable listing commands. A server without them reports the folder as
unsupported rather than guessing at the older, human-formatted listing or
@@ -23,15 +33,13 @@ so a conditional write is reported as unsupported rather than simulated.
## Passive mode and the port range
Transfers are passive: the plugin asks the server for a data port with `EPSV`
or `PASV` and connects out to it. The server picks a fresh port for every
and connects out to it; `PASV` fallback is not implemented. The server picks a fresh port for every
transfer and announces it only as the transfer begins, so the range that has to
be reachable is wide.
`manifest.json` declares `networkPorts` as 21 and 1024-65535, and `networkHosts`
as `*` because a server can be anywhere. Declaring is not granting: a
connection reaches only the host and ports the person grants it, and a grant
never accepts a wildcard host. Grant the control port and the span the server
uses for its data channel, usually `21` and `1024-65535`.
The package declares socket ports 21 and 1024-65535, which Memona applies
automatically to each connection. The control port and passive data ports
must be within these ranges.
## Build
manifest.json
+2 -3
View File
@@ -1,13 +1,12 @@
{
"schemaVersion": 1,
"schemaVersion": 2,
"id": "memona/ftp",
"name": "FTP",
"version": "0.1.2",
"version": "0.1.3",
"apiVersion": "0.1.0",
"description": "Open an FTP server as a folder, and let agents work in it.",
"license": "MIT",
"entry": "plugin.wasm",
"networkHosts": ["*"],
"networkPorts": [
{ "from": 21, "to": 21 },
{ "from": 1024, "to": 65535 }
sdk/wit/filesystem.wit
+14 -13
View File
@@ -35,11 +35,11 @@ interface host {
/// `respond`. An empty result means the body is complete.
pull: func(limit: u32) -> result<list<u8>, host-error>;
}
/// Begins a request to a host both declared by the package and granted on
/// the connection. Each redirect and DNS result is checked, and DNS is
/// pinned. Private/LAN addresses additionally require the private-network
/// grant. Cross-origin redirects remove request headers; HTTPS cannot
/// downgrade. The host bounds total bytes per exchange, not buffer size.
/// Begins a request to any HTTP or HTTPS host, including LAN and loopback
/// addresses. This API has no per-connection network grant. Each redirect
/// and DNS result is checked, and DNS is pinned. Cross-origin redirects
/// remove request headers; HTTPS cannot downgrade. The host bounds total
/// bytes per exchange, not buffer size.
http: func(request: request) -> result<exchange, host-error>;
/// One open TCP connection. Dropping the handle closes the socket, and the
@@ -51,13 +51,13 @@ interface host {
/// Writes at most 1,048,576 bytes and returns how many were accepted.
write: func(data: list<u8>) -> result<u32, host-error>;
}
/// Opens a socket to a host and port both declared by the package and
/// granted on the connection. The host resolves and pins DNS, applies the
/// same private-address rule as `http`, and starts TLS when asked. Ports
/// negotiated at runtime, such as an FTP data channel, still need a grant.
/// Opens a socket to any host, including LAN and loopback addresses, on a
/// package-declared port. This API has no per-connection network grant.
/// The host resolves and pins DNS and starts TLS when asked. Ports
/// negotiated at runtime, such as an FTP data channel, must be declared.
connect: func(host: string, port: u16, tls: bool) -> result<tcp, host-error>;
/// 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<u8>) -> result<u32, host-error>;
@@ -65,8 +65,8 @@ interface host {
/// means none arrived inside the host's wait, not that the peer left.
receive: func(limit: u32) -> result<list<u8>, host-error>;
}
/// Connects a datagram socket under the same declaration and grant rules
/// as `connect`, for the few filesystem protocols that still need one.
/// Connects a datagram socket under the same address and port rules as
/// `connect`, for the few filesystem protocols that still need one.
bind: func(host: string, port: u16) -> result<udp, host-error>;
/// A configuration value, typed as the package's manifest declared it.
@@ -81,7 +81,8 @@ interface host {
/// Filesystem operations for one connection. Paths are absolute within its
/// root, starting with `/`; they are never local OS paths or resource URIs.
/// Dot segments, backslashes, percent escapes, colons, and controls are invalid.
/// Colons and percent signs are literal; paths must never be URL-decoded.
/// Dot segments, backslashes, and controls are invalid.
/// Every invocation is bounded by host fuel, memory, time, and cancellation.
interface provider {
use host.{setting};
src/control.rs
+58 -11
View File
@@ -1,6 +1,9 @@
//! The FTP control channel: one socket, line replies, and passive transfers.
#[cfg(test)]
use crate::test_host as host;
use memona_plugin_sdk::ErrorCode;
#[cfg(not(test))]
use memona_plugin_sdk::memona::filesystem::host;
const MAX_LINE: usize = 8192;
@@ -64,7 +67,9 @@ fn names_a_conflict(text: &str) -> bool {
.collect();
words.contains(&"exists")
|| words.contains(&"nonempty")
|| words.windows(2).any(|pair| matches!(pair, ["not" | "non", "empty"]))
|| words
.windows(2)
.any(|pair| matches!(pair, ["not" | "non", "empty"]))
}
/// What the first reply to a transfer command means. A transfer answers with
@@ -86,6 +91,11 @@ pub struct Control {
buffer: Vec<u8>,
}
struct DispatchError {
code: ErrorCode,
dispatched: bool,
}
impl Control {
pub fn open(
host_name: String,
@@ -150,27 +160,61 @@ impl Control {
Err(ErrorCode::Unavailable)
}
/// Sends one command. A newline in an argument would forge a second command,
/// so it is refused rather than escaped.
pub fn send(&mut self, command: &str) -> Result<(u16, String), ErrorCode> {
/// Writes one complete command and records whether any of it reached the
/// socket. A newline in an argument would forge a second command, so it is
/// refused before dispatch rather than escaped.
fn dispatch(&mut self, command: &str) -> Result<(), DispatchError> {
if command.contains(['\r', '\n']) || command.len() > MAX_LINE {
return Err(ErrorCode::InvalidInput);
return Err(DispatchError {
code: ErrorCode::InvalidInput,
dispatched: false,
});
}
let line = format!("{command}\r\n");
let mut sent = 0;
while sent < line.len() {
let written = self
.socket
.write(&line.as_bytes()[sent..])
.map_err(transport)? as usize;
let written =
self
.socket
.write(&line.as_bytes()[sent..])
.map_err(|error| DispatchError {
code: transport(error),
dispatched: sent > 0,
})? as usize;
if written == 0 {
return Err(ErrorCode::Unavailable);
return Err(DispatchError {
code: ErrorCode::Unavailable,
dispatched: sent > 0,
});
}
sent += written;
}
Ok(())
}
/// Sends one command and reads its reply.
pub fn send(&mut self, command: &str) -> Result<(u16, String), ErrorCode> {
self.dispatch(command).map_err(|error| error.code)?;
self.reply()
}
/// Sends a command that changes remote state. Once any command byte was
/// dispatched, a write interruption or missing reply makes its outcome
/// unknowable even when the underlying transport reports cancellation.
pub fn send_mutation(
&mut self,
command: &str,
) -> Result<(u16, String), ErrorCode> {
self.dispatch(command).map_err(|error| {
if error.dispatched {
ErrorCode::OutcomeUnknown
} else {
error.code
}
})?;
self.reply().map_err(|_| ErrorCode::OutcomeUnknown)
}
pub fn expect(&mut self, command: &str) -> Result<String, ErrorCode> {
let (code, text) = self.send(command)?;
outcome(code)?;
@@ -255,7 +299,10 @@ mod tests {
));
// The same two conditions as other servers phrase them.
assert!(matches!(
refusal(550, "550 Cannot create a file when that file already exists."),
refusal(
550,
"550 Cannot create a file when that file already exists."
),
Err(ErrorCode::Conflict)
));
assert!(matches!(
src/lib.rs
+269 -47
View File
@@ -4,8 +4,14 @@ mod control;
mod listing;
mod transfer;
#[cfg(test)]
mod session_tests;
#[cfg(test)]
mod test_host;
use control::{Control, outcome, preliminary, refusal, transport};
use listing::{machine_info, page};
#[cfg(not(test))]
use memona_plugin_sdk::memona::filesystem::host;
use memona_plugin_sdk::{
Capabilities, ChangePage, EntryKind, ErrorCode, FileInfo, Guest, GuestReader,
@@ -14,14 +20,18 @@ use memona_plugin_sdk::{
use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::rc::Rc;
#[cfg(test)]
use test_host as host;
use transfer::Buffered;
const MAX_CHUNK: u32 = 1024 * 1024;
/// One listing may be this large, and so may every listing kept together.
const MAX_LISTING_BYTES: usize = 8 * 1024 * 1024;
const MAX_STAGE_ATTEMPTS: usize = 64;
thread_local! {
static CONTROL: RefCell<Option<Control>> = const { RefCell::new(None) };
static SESSION: RefCell<Option<Session>> = 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<HashMap<String, Rc<[u8]>>> = RefCell::new(HashMap::new());
@@ -29,12 +39,38 @@ thread_local! {
struct Ftp;
#[derive(Clone)]
struct Session {
host: String,
port: u16,
tls: bool,
user: String,
password: String,
}
impl Session {
fn connect(&self) -> Result<Control, ErrorCode> {
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<String> {
match setting {
Some(Setting::Text(value)) | Some(Setting::Secret(value)) => {
Some(Setting::Text(value)) => {
let value = value.trim().to_owned();
(!value.is_empty()).then_some(value)
}
Some(Setting::Secret(value)) => Some(value.clone()),
Some(Setting::Toggle(_)) | None => None,
}
}
@@ -49,12 +85,43 @@ fn route(target: &Locator) -> Result<String, ErrorCode> {
fn with_control<T>(
action: impl FnOnce(&mut Control) -> Result<T, ErrorCode>,
) -> Result<T, ErrorCode> {
CONTROL.with_borrow_mut(|slot| match slot.as_mut() {
Some(control) => action(control),
None => Err(ErrorCode::Internal),
CONTROL.with_borrow_mut(|slot| {
let result = match slot.as_mut() {
Some(control) => action(control),
None => Err(ErrorCode::Internal),
};
if matches!(
result,
Err(
ErrorCode::Unavailable
| ErrorCode::OutcomeUnknown
| ErrorCode::Internal
)
) {
*slot = None;
}
result
})
}
/// Checks a control channel only before a fresh filesystem operation. A lost
/// idle session is replaced and logged in again. Nothing inside a transfer or
/// mutation retries, because its missing reply may hide a completed change.
fn ready() -> Result<(), ErrorCode> {
if !CONTROL.with_borrow(Option::is_none) {
match with_control(|control| control.expect("NOOP")) {
Ok(_) => return Ok(()),
Err(ErrorCode::Unavailable) => {}
Err(error) => return Err(error),
}
}
let session = SESSION
.with_borrow(Clone::clone)
.ok_or(ErrorCode::Internal)?;
CONTROL.set(Some(session.connect()?));
Ok(())
}
/// Starts a passive transfer and leaves the data socket open for the caller,
/// who owes the server one more reply once that socket is closed: either the
/// completion, read by `settle`, or the answer to an `abort`.
@@ -67,6 +134,17 @@ fn transfer(command: &str) -> Result<host::Tcp, ErrorCode> {
})
}
/// 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<host::Tcp, ErrorCode> {
with_control(|control| {
let data = control.passive()?;
let (code, _) = control.send_mutation(command)?;
preliminary(code)?;
Ok(data)
})
}
/// Consumes the completion reply of a transfer whose data socket is closed.
fn settle() -> Result<(), ErrorCode> {
with_control(|control| {
@@ -75,6 +153,16 @@ fn settle() -> Result<(), ErrorCode> {
})
}
/// Consumes an upload's completion reply. The data and command are already
/// dispatched, so an interrupted reply cannot say whether the file committed.
fn settle_mutation() -> Result<(), ErrorCode> {
let (code, _) = with_control(|control| {
control.reply().map_err(|_| ErrorCode::OutcomeUnknown)
})
.map_err(|_| ErrorCode::OutcomeUnknown)?;
outcome(code)
}
/// Ends a transfer early, after its data socket is closed.
fn abort() -> Result<(), ErrorCode> {
with_control(Control::abort)
@@ -167,20 +255,62 @@ impl Drop for FtpReader {
/// Writes one file to a passive data socket.
struct FtpWriter {
path: String,
target: String,
staged: RefCell<Option<String>>,
socket: RefCell<Option<host::Tcp>>,
written: Cell<u64>,
}
impl FtpWriter {
/// Removes the private staging file when possible. A missing stage means a
/// prior promotion or cleanup already completed.
fn cleanup_stage(&self) {
let Some(staged) = self.staged.borrow().clone() else {
return;
};
if ready().is_err() {
return;
}
let removed = with_control(|control| {
let (code, _) = control.send_mutation(&format!("DELE {staged}"))?;
outcome(code)
});
if matches!(removed, Ok(()) | Err(ErrorCode::NotFound)) {
self.staged.take();
}
}
/// A failed data write can leave both the remote file and its pending
/// completion reply in an unknown state. Close both channels, then make a
/// best-effort cleanup of the private stage over a fresh session.
fn discard_uncertain_upload(&self) -> ErrorCode {
drop(self.socket.take());
CONTROL.set(None);
self.cleanup_stage();
ErrorCode::OutcomeUnknown
}
}
impl GuestWriter for FtpWriter {
fn push(&self, data: Vec<u8>) -> Result<(), ErrorCode> {
let socket = self.socket.borrow();
let socket = socket.as_ref().ok_or(ErrorCode::InvalidInput)?;
if self.socket.borrow().is_none() {
return Err(ErrorCode::InvalidInput);
}
let mut sent = 0;
while sent < data.len() {
let written = socket.write(&data[sent..]).map_err(transport)? as usize;
let result = {
let socket = self.socket.borrow();
let Some(socket) = socket.as_ref() else {
return Err(ErrorCode::InvalidInput);
};
socket.write(&data[sent..])
};
let written = match result {
Ok(written) => written as usize,
Err(_) => return Err(self.discard_uncertain_upload()),
};
if written == 0 {
return Err(ErrorCode::Unavailable);
return Err(self.discard_uncertain_upload());
}
sent += written;
}
@@ -188,10 +318,30 @@ impl GuestWriter for FtpWriter {
Ok(())
}
/// Closing the data socket is what ends an FTP upload; only then does the
/// server answer, so the socket is given up before the reply is read.
/// server answer. A confirmed stage is promoted only after that reply, so an
/// interrupted transfer never exposes a partial destination.
fn finish(&self) -> Result<Option<FileInfo>, ErrorCode> {
drop(self.socket.take().ok_or(ErrorCode::InvalidInput)?);
settle()?;
if let Err(error) = settle_mutation() {
self.cleanup_stage();
return Err(error);
}
let staged = self
.staged
.borrow()
.clone()
.ok_or(ErrorCode::InvalidInput)?;
let promoted = with_control(|control| {
control.expect(&format!("RNFR {staged}"))?;
let (code, _) =
control.send_mutation(&format!("RNTO {}", self.target))?;
outcome(code)
});
if let Err(error) = promoted {
self.cleanup_stage();
return Err(error);
}
self.staged.take();
Ok(Some(FileInfo {
kind: EntryKind::File,
size: self.written.get(),
@@ -201,19 +351,64 @@ impl GuestWriter for FtpWriter {
}
impl Drop for FtpWriter {
/// An upload abandoned before it finished must leave no partial file. The
/// server takes the closed socket as the end of the upload, so what it
/// stored is deleted once the channel is back in sync.
/// An abandoned upload closes its data channel, resynchronizes when
/// possible, and removes only its private stage. The destination was never
/// exposed unless `finish` confirmed the promotion.
fn drop(&mut self) {
let Some(socket) = self.socket.take() else {
return;
};
drop(socket);
if abort().is_ok() {
let _ =
with_control(|control| control.expect(&format!("DELE {}", self.path)));
if let Some(socket) = self.socket.take() {
drop(socket);
if abort().is_err() {
CONTROL.set(None);
}
}
self.cleanup_stage();
}
}
fn random_stage_token() -> Result<u128, ErrorCode> {
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<u128, ErrorCode>,
) -> Result<String, ErrorCode> {
let (parent, name) =
target.rsplit_once('/').ok_or(ErrorCode::InvalidInput)?;
if name.is_empty() {
return Err(ErrorCode::InvalidInput);
}
for _ in 0..MAX_STAGE_ATTEMPTS {
let token = next_token()?;
let staged = format!("{parent}/.memona-upload-{token:032x}");
match with_control(|control| probe(control, &staged)) {
Err(ErrorCode::NotFound) => return Ok(staged),
Ok(_) => {}
Err(error) => return Err(error),
}
}
Err(ErrorCode::Conflict)
}
fn stage_path(target: &str) -> Result<String, ErrorCode> {
stage_path_with(target, random_stage_token)
}
fn start_writer(target: String) -> Result<FtpWriter, ErrorCode> {
ready()?;
let staged = stage_path(&target)?;
let socket = mutation_transfer(&format!("STOR {staged}"))?;
Ok(FtpWriter {
target,
staged: RefCell::new(Some(staged)),
socket: RefCell::new(Some(socket)),
written: Cell::new(0),
})
}
impl Guest for Ftp {
@@ -240,19 +435,16 @@ impl Guest for Ftp {
config.iter().find(|(key, _)| key == "tls").map(|(_, v)| v),
Some(Setting::Toggle(true))
);
let mut control = Control::open(host_name, port, tls)?;
let user = lookup("username").unwrap_or_else(|| "anonymous".to_owned());
let (code, _) = control.send(&format!("USER {user}"))?;
if code == 331 {
let password = lookup("password").unwrap_or_default();
let (code, _) = control.send(&format!("PASS {password}"))?;
outcome(code)?;
} else {
outcome(code)?;
}
// Binary throughout: text mode would rewrite line endings inside files.
control.expect("TYPE I")?;
CONTROL.set(Some(control));
let session = Session {
host: host_name,
port,
tls,
user,
password: lookup("password").unwrap_or_default(),
};
CONTROL.set(Some(session.connect()?));
SESSION.set(Some(session));
Ok(Capabilities {
write: true,
mkdir: true,
@@ -273,6 +465,7 @@ impl Guest for Ftp {
}
*slot = None;
});
SESSION.set(None);
}
fn stat(target: Locator) -> Result<FileInfo, ErrorCode> {
@@ -284,6 +477,7 @@ impl Guest for Ftp {
revision: None,
});
}
ready()?;
with_control(|control| probe(control, &path))
}
@@ -305,13 +499,17 @@ impl Guest for Ftp {
});
let listing = match kept {
Some(listing) => listing,
None => fetch_listing(&path)?,
None => {
ready()?;
fetch_listing(&path)?
}
};
page(&listing, start, limit as usize)
}
fn open_read(target: Locator, offset: u64) -> Result<Reader, ErrorCode> {
let path = route(&target)?;
ready()?;
if offset > 0 {
with_control(|control| control.expect(&format!("REST {offset}")))?;
}
@@ -331,20 +529,16 @@ impl Guest for Ftp {
if !matches!(condition, WriteCondition::Overwrite) {
return Err(ErrorCode::Unsupported);
}
let socket = transfer(&format!("STOR {path}"))?;
Ok(Writer::new(FtpWriter {
path,
socket: RefCell::new(Some(socket)),
written: Cell::new(0),
}))
Ok(Writer::new(start_writer(path)?))
}
/// A refused create is a conflict when the path is already taken, which is
/// the one thing the server says in words rather than in its reply code.
fn mkdir(target: Locator) -> Result<FileInfo, ErrorCode> {
let path = route(&target)?;
ready()?;
with_control(|control| {
let (code, text) = control.send(&format!("MKD {path}"))?;
let (code, text) = control.send_mutation(&format!("MKD {path}"))?;
refusal(code, &text)
})?;
Ok(FileInfo {
@@ -361,15 +555,18 @@ impl Guest for Ftp {
destination: String,
) -> Result<FileInfo, ErrorCode> {
let path = route(&target)?;
ready()?;
with_control(|control| {
let info = probe(control, &path)?;
match probe(control, &destination) {
Ok(_) => return Err(ErrorCode::Conflict),
Err(ErrorCode::NotFound) => {}
Err(error) => return Err(error),
}
control.expect(&format!("RNFR {path}"))?;
control.expect(&format!("RNTO {destination}"))?;
probe(control, &destination)
let (code, _) = control.send_mutation(&format!("RNTO {destination}"))?;
outcome(code)?;
Ok(info)
})
}
@@ -378,12 +575,15 @@ impl Guest for Ftp {
/// holds children is refused as a conflict, not as an absence.
fn remove(target: Locator) -> Result<(), ErrorCode> {
let path = route(&target)?;
ready()?;
with_control(|control| {
let (code, _) = control.send(&format!("DELE {path}"))?;
if (200..300).contains(&code) {
return Ok(());
let (code, _) = control.send_mutation(&format!("DELE {path}"))?;
match outcome(code) {
Ok(()) => return Ok(()),
Err(ErrorCode::NotFound) => {}
Err(error) => return Err(error),
}
let (code, text) = control.send(&format!("RMD {path}"))?;
let (code, text) = control.send_mutation(&format!("RMD {path}"))?;
refusal(code, &text)
})
}
@@ -393,4 +593,26 @@ impl Guest for Ftp {
}
}
#[cfg(not(test))]
memona_plugin_sdk::export!(Ftp);
#[cfg(test)]
mod tests {
use super::text;
use memona_plugin_sdk::Setting;
#[tokio::test]
async fn secret_credentials_preserve_significant_whitespace() {
for password in [" ftp fixture secret ", " ", ""] {
assert_eq!(
text(Some(&Setting::Secret(password.to_owned()))),
Some(password.to_owned())
);
}
assert_eq!(
text(Some(&Setting::Text(" server.example ".to_owned()))),
Some("server.example".to_owned())
);
assert_eq!(text(Some(&Setting::Text(" ".to_owned()))), None);
}
}
src/listing.rs
+15 -1
View File
@@ -12,7 +12,6 @@ struct Facts<'a> {
fn facts(line: &str) -> Option<Facts<'_>> {
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> {
src/session_tests.rs
+173
View File
@@ -0,0 +1,173 @@
//! FTP command recovery against an in-memory protocol transport. Live QA
//! separately exercises the compiled WASM component over real host sockets.
use super::{Ftp, stage_path_with, start_writer, test_host};
use memona_plugin_sdk::{ErrorCode, Guest, GuestWriter, Locator, Setting};
type TestResult<T = ()> = Result<T, Box<dyn std::error::Error + Send + Sync>>;
fn guest<T>(value: Result<T, ErrorCode>) -> TestResult<T> {
value.map_err(|error| format!("guest operation failed: {error:?}").into())
}
fn config() -> Vec<(String, Setting)> {
vec![
("host".into(), Setting::Text("127.0.0.1".into())),
(
"port".into(),
Setting::Text(test_host::CONTROL_PORT.to_string()),
),
("username".into(), Setting::Text("fixture".into())),
(
"password".into(),
Setting::Secret(" fixture secret ".into()),
),
]
}
#[tokio::test]
async fn a_listing_reconnects_after_the_server_expires_the_control_session()
-> TestResult {
let fixture = test_host::install(test_host::Scenario::ExpiredListing);
guest(Ftp::open("ftp".into(), config()))?;
let first = guest(Ftp::list(Locator::Path("/".into()), None, 10))?;
assert_eq!(first.entries[0].name, " report.md ");
let next = guest(Ftp::list(Locator::Path("/".into()), None, 10))?;
assert_eq!(next.entries[0].name, " report.md ");
Ftp::close();
let summary = fixture.summary();
assert_eq!(summary.control_connections, 2);
assert_eq!(summary.listings, 2);
assert_eq!(summary.mkdir_commands, 0);
Ok(())
}
#[tokio::test]
async fn recovery_before_a_mutation_never_replays_a_lost_mutation_reply()
-> TestResult {
let fixture = test_host::install(test_host::Scenario::LostMutationReply);
guest(Ftp::open("ftp".into(), config()))?;
assert!(matches!(
Ftp::mkdir(Locator::Path("/new".into())),
Err(ErrorCode::OutcomeUnknown)
));
Ftp::close();
let summary = fixture.summary();
assert_eq!(summary.control_connections, 2);
assert_eq!(summary.listings, 0);
assert_eq!(summary.mkdir_commands, 1);
Ok(())
}
#[tokio::test]
async fn cancelled_preflight_never_reconnects_or_executes_the_mutation()
-> TestResult {
let fixture = test_host::install(test_host::Scenario::CancelledPreflight);
guest(Ftp::open("ftp".into(), config()))?;
assert!(matches!(
Ftp::mkdir(Locator::Path("/cancelled".into())),
Err(ErrorCode::Cancelled)
));
Ftp::close();
let summary = fixture.summary();
assert_eq!(summary.control_connections, 1);
assert_eq!(summary.listings, 0);
assert_eq!(summary.mkdir_commands, 0);
Ok(())
}
#[tokio::test]
async fn a_lost_upload_completion_reports_an_unknown_outcome() -> TestResult {
let fixture = test_host::install(test_host::Scenario::LostUploadCompletion);
guest(Ftp::open("ftp".into(), config()))?;
let writer = guest(start_writer("/target".into()))?;
guest(writer.push(b"body".to_vec()))?;
assert!(matches!(writer.finish(), Err(ErrorCode::OutcomeUnknown)));
drop(writer);
Ftp::close();
let summary = fixture.summary();
assert_eq!(summary.control_connections, 2);
assert_eq!(summary.upload_commands, 1);
assert_eq!(summary.target_upload_commands, 0);
assert_eq!(summary.uploaded_bytes, 4);
assert_eq!(summary.stage_deletes, 1);
assert_eq!(summary.promotions, 0);
Ok(())
}
#[tokio::test]
async fn an_uncertain_upload_write_discards_the_control_session() -> TestResult
{
let fixture = test_host::install(test_host::Scenario::LostUploadData);
guest(Ftp::open("ftp".into(), config()))?;
let writer = guest(start_writer("/target".into()))?;
assert!(matches!(
writer.push(b"body".to_vec()),
Err(ErrorCode::OutcomeUnknown)
));
let page = guest(Ftp::list(Locator::Path("/".into()), None, 10))?;
assert_eq!(page.entries[0].name, " report.md ");
drop(writer);
Ftp::close();
let summary = fixture.summary();
assert_eq!(summary.control_connections, 2);
assert_eq!(summary.upload_commands, 1);
assert_eq!(summary.target_upload_commands, 0);
assert_eq!(summary.uploaded_bytes, 0);
assert_eq!(summary.stage_deletes, 1);
assert_eq!(summary.promotions, 0);
assert_eq!(summary.listings, 1);
Ok(())
}
#[tokio::test]
async fn a_complete_upload_promotes_its_private_stage() -> TestResult {
let fixture = test_host::install(test_host::Scenario::SuccessfulUpload);
guest(Ftp::open("ftp".into(), config()))?;
let writer = guest(start_writer("/target".into()))?;
guest(writer.push(b"body".to_vec()))?;
let info = guest(writer.finish())?.ok_or("missing upload result")?;
assert_eq!(info.size, 4);
drop(writer);
Ftp::close();
let summary = fixture.summary();
assert_eq!(summary.control_connections, 1);
assert_eq!(summary.upload_commands, 1);
assert_eq!(summary.target_upload_commands, 0);
assert_eq!(summary.uploaded_bytes, 4);
assert_eq!(summary.stage_deletes, 0);
assert_eq!(summary.promotions, 1);
Ok(())
}
#[tokio::test]
async fn dropping_an_upload_removes_only_its_private_stage() -> TestResult {
let fixture = test_host::install(test_host::Scenario::AbandonedUpload);
guest(Ftp::open("ftp".into(), config()))?;
let writer = guest(start_writer("/target".into()))?;
guest(writer.push(b"partial".to_vec()))?;
drop(writer);
Ftp::close();
let summary = fixture.summary();
assert_eq!(summary.control_connections, 1);
assert_eq!(summary.upload_commands, 1);
assert_eq!(summary.target_upload_commands, 0);
assert_eq!(summary.uploaded_bytes, 7);
assert_eq!(summary.stage_deletes, 1);
assert_eq!(summary.promotions, 0);
Ok(())
}
#[tokio::test]
async fn a_stage_name_collision_draws_a_fresh_random_token() -> TestResult {
let fixture = test_host::install(test_host::Scenario::StageCollision);
guest(Ftp::open("ftp".into(), config()))?;
let mut tokens = [1, 2].into_iter();
let staged = guest(stage_path_with("/target", || {
tokens.next().ok_or(ErrorCode::Internal)
}))?;
assert_eq!(staged, "/.memona-upload-00000000000000000000000000000002");
Ftp::close();
assert_eq!(fixture.summary().stage_probes, 2);
Ok(())
}
src/test_host.rs
+278
View File
@@ -0,0 +1,278 @@
//! In-memory FTP host for the generated synchronous `Guest` API. Production
//! WASM uses Memona's real async socket host; live QA covers that boundary.
use std::cell::RefCell;
use std::collections::VecDeque;
use std::rc::Rc;
pub use memona_plugin_sdk::HostError;
pub const CONTROL_PORT: u16 = 21_210;
const DATA_PORT: u16 = 31_310;
const LISTING: &[u8] = b"type=file;size=3; report.md \r\n";
#[derive(Clone, Copy)]
pub enum Scenario {
AbandonedUpload,
CancelledPreflight,
ExpiredListing,
LostUploadData,
LostMutationReply,
LostUploadCompletion,
StageCollision,
SuccessfulUpload,
}
struct State {
scenario: Scenario,
control_connections: usize,
listings: usize,
mkdir_commands: usize,
upload_commands: usize,
target_upload_commands: usize,
uploaded_bytes: usize,
stage_deletes: usize,
stage_probes: usize,
promotions: usize,
}
thread_local! {
static ACTIVE: RefCell<Option<Rc<RefCell<State>>>> = const { RefCell::new(None) };
}
pub struct Fixture(Rc<RefCell<State>>);
pub struct Summary {
pub control_connections: usize,
pub listings: usize,
pub mkdir_commands: usize,
pub upload_commands: usize,
pub target_upload_commands: usize,
pub uploaded_bytes: usize,
pub stage_deletes: usize,
pub stage_probes: usize,
pub promotions: usize,
}
pub fn install(scenario: Scenario) -> Fixture {
let state = Rc::new(RefCell::new(State {
scenario,
control_connections: 0,
listings: 0,
mkdir_commands: 0,
upload_commands: 0,
target_upload_commands: 0,
uploaded_bytes: 0,
stage_deletes: 0,
stage_probes: 0,
promotions: 0,
}));
ACTIVE.set(Some(state.clone()));
Fixture(state)
}
impl Fixture {
pub fn summary(&self) -> Summary {
let state = self.0.borrow();
Summary {
control_connections: state.control_connections,
listings: state.listings,
mkdir_commands: state.mkdir_commands,
upload_commands: state.upload_commands,
target_upload_commands: state.target_upload_commands,
uploaded_bytes: state.uploaded_bytes,
stage_deletes: state.stage_deletes,
stage_probes: state.stage_probes,
promotions: state.promotions,
}
}
}
impl Drop for Fixture {
fn drop(&mut self) {
ACTIVE.with_borrow_mut(|active| {
if active
.as_ref()
.is_some_and(|state| Rc::ptr_eq(state, &self.0))
{
*active = None;
}
});
}
}
enum Kind {
Control(usize),
Data,
}
pub struct Tcp {
state: Rc<RefCell<State>>,
kind: Kind,
input: RefCell<VecDeque<u8>>,
}
pub fn connect(name: &str, port: u16, tls: bool) -> Result<Tcp, HostError> {
if tls || name != "127.0.0.1" {
return Err(HostError::InvalidInput);
}
let state = ACTIVE
.with_borrow(Clone::clone)
.ok_or(HostError::Internal)?;
match port {
CONTROL_PORT => {
let index = {
let mut state = state.borrow_mut();
let index = state.control_connections;
state.control_connections += 1;
index
};
Ok(Tcp {
state,
kind: Kind::Control(index),
input: RefCell::new(VecDeque::from(b"220 Fixture\r\n".to_vec())),
})
}
DATA_PORT => Ok(Tcp {
state,
kind: Kind::Data,
input: RefCell::new(VecDeque::from(LISTING.to_vec())),
}),
_ => Err(HostError::Denied),
}
}
impl Tcp {
pub fn read(&self, limit: u32) -> Result<Vec<u8>, HostError> {
let mut input = self.input.borrow_mut();
if input.is_empty() && matches!(self.kind, Kind::Control(_)) {
return Err(HostError::Unavailable);
}
let count = usize::min(limit as usize, input.len());
Ok(input.drain(..count).collect())
}
pub fn write(&self, bytes: &[u8]) -> Result<u32, HostError> {
if matches!(self.kind, Kind::Data) {
let mut state = self.state.borrow_mut();
if matches!(
state.scenario,
Scenario::AbandonedUpload
| Scenario::LostUploadCompletion
| Scenario::SuccessfulUpload
) {
state.uploaded_bytes += bytes.len();
return Ok(bytes.len() as u32);
}
return Err(HostError::InvalidInput);
}
let Kind::Control(index) = self.kind else {
unreachable!();
};
let command = std::str::from_utf8(bytes)
.map_err(|_| HostError::InvalidInput)?
.strip_suffix("\r\n")
.ok_or(HostError::InvalidInput)?;
let response = {
let mut state = self.state.borrow_mut();
match command {
"USER fixture" => "331 Password required\r\n",
"PASS fixture secret " => "230 Signed in\r\n",
"TYPE I" => "200 Binary\r\n",
"QUIT" => "221 Bye\r\n",
"EPSV" => "229 Passive (|||31310|)\r\n",
"MLSD /"
if matches!(
state.scenario,
Scenario::ExpiredListing | Scenario::LostUploadData
) =>
{
state.listings += 1;
"150 Listing\r\n226 Complete\r\n"
}
"NOOP" if matches!(state.scenario, Scenario::CancelledPreflight) => {
return Err(HostError::Cancelled);
}
"NOOP"
if index == 0
&& (matches!(state.scenario, Scenario::LostMutationReply)
|| state.listings > 0) =>
{
return Err(HostError::Unavailable);
}
"NOOP" => "200 Still connected\r\n",
"MLST /.memona-upload-00000000000000000000000000000001"
if matches!(state.scenario, Scenario::StageCollision) =>
{
state.stage_probes += 1;
"250-Begin\r\n type=file;size=3; /.memona-upload-00000000000000000000000000000001\r\n250 End\r\n"
}
command if command.starts_with("MLST /.memona-upload-") => {
state.stage_probes += 1;
"550 Missing\r\n"
}
command
if command.starts_with("STOR /.memona-upload-")
&& matches!(
state.scenario,
Scenario::AbandonedUpload | Scenario::LostUploadCompletion
) =>
{
state.upload_commands += 1;
"150 Upload started\r\n"
}
command
if command.starts_with("STOR /.memona-upload-")
&& matches!(state.scenario, Scenario::LostUploadData) =>
{
state.upload_commands += 1;
"150 Upload started\r\n"
}
command
if command.starts_with("STOR /.memona-upload-")
&& matches!(state.scenario, Scenario::SuccessfulUpload) =>
{
state.upload_commands += 1;
"150 Upload started\r\n226 Upload complete\r\n"
}
command if command.starts_with("STOR ") => {
state.target_upload_commands += 1;
"150 Target upload started\r\n"
}
command if command.starts_with("DELE /.memona-upload-") => {
state.stage_deletes += 1;
"250 Stage deleted\r\n"
}
command if command.starts_with("RNFR /.memona-upload-") => {
"350 Ready for destination\r\n"
}
"RNTO /target"
if matches!(state.scenario, Scenario::SuccessfulUpload) =>
{
state.promotions += 1;
"250 Promoted\r\n"
}
"ABOR" if matches!(state.scenario, Scenario::AbandonedUpload) => {
"426 Upload abandoned\r\n226 Abort complete\r\n"
}
"MKD /new"
if index == 1
&& matches!(state.scenario, Scenario::LostMutationReply) =>
{
state.mkdir_commands += 1;
return Ok(bytes.len() as u32);
}
"MKD /cancelled"
if index == 1
&& matches!(state.scenario, Scenario::CancelledPreflight) =>
{
state.mkdir_commands += 1;
return Err(HostError::Unavailable);
}
_ => return Err(HostError::InvalidInput),
}
};
self.input.borrow_mut().extend(response.as_bytes());
Ok(bytes.len() as u32)
}
}