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

Merged
temeddix merged 15 commits from fix-directory-listing into main 2026-09-30 13:10:40 +00:00
Conversation 21 Commits 15 Files Changed 12
+1004 -84
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>;
}
bot commented 2026-09-30 10:28:59 +00:00
Outdated
Review
Copy Link
Copy Source

LAN and loopback access still requires Memona's Private network permission, so this contract comment incorrectly promises direct access. Restore that qualification here and in the README so users are not led to a setup that fails at runtime.

Profile: codex · gpt-5.6-terra

LAN and loopback access still requires Memona's Private network permission, so this contract comment incorrectly promises direct access. Restore that qualification here and in the README so users are not led to a setup that fails at runtime. > Profile: `codex` · `gpt-5.6-terra`
/// 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),
};
bot commented 2026-09-30 11:14:19 +00:00
Review
Copy Link
Copy Source

A lost reply after MKD, RNTO, DELE, RMD, or upload completion still leaves Unavailable intact. Those commands may already have changed the server, while the ABI reserves outcome-unknown for post-dispatch mutation interruptions; callers may retry Unavailable and repeat a completed mutation. Map these uncertain mutation paths to OutcomeUnknown (and add regressions) while retaining Unavailable for pre-dispatch and read-only failures.

Profile: codex · gpt-5.6-terra

A lost reply after `MKD`, `RNTO`, `DELE`, `RMD`, or upload completion still leaves `Unavailable` intact. Those commands may already have changed the server, while the ABI reserves `outcome-unknown` for post-dispatch mutation interruptions; callers may retry `Unavailable` and repeat a completed mutation. Map these uncertain mutation paths to `OutcomeUnknown` (and add regressions) while retaining `Unavailable` for pre-dispatch and read-only failures. > Profile: `codex` · `gpt-5.6-terra`
if matches!(
result,
Err(
ErrorCode::Unavailable
| ErrorCode::OutcomeUnknown
| ErrorCode::Internal
)
) {
*slot = None;
}
result
bot commented 2026-09-30 11:02:43 +00:00
Review
Copy Link
Copy Source

Propagate Cancelled (and other non-recoverable preflight errors) instead of reconnecting: a cancelled NOOP currently falls through to session.connect() and then executes the requested operation, which can perform a mutation after the host cancelled the invocation. Retry only the disconnect error(s) that are safe to recover from.

Profile: codex · gpt-5.6-terra

Propagate `Cancelled` (and other non-recoverable preflight errors) instead of reconnecting: a cancelled `NOOP` currently falls through to `session.connect()` and then executes the requested operation, which can perform a mutation after the host cancelled the invocation. Retry only the disconnect error(s) that are safe to recover from. > Profile: `codex` · `gpt-5.6-terra`
})
}
/// 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) {
bot commented 2026-09-30 11:34:12 +00:00
Outdated
Review
Copy Link
Copy Source

A data write can fail after the server accepted a prefix. Clearing the socket makes Drop skip its cleanup, so that partial remote file remains, violating the writer contract. Stage uploads under a temporary name and promote only after a confirmed completion, or otherwise ensure cleanup before exposing the target.

Profile: codex · gpt-5.6-terra

A data write can fail after the server accepted a prefix. Clearing the socket makes `Drop` skip its cleanup, so that partial remote file remains, violating the writer contract. Stage uploads under a temporary name and promote only after a confirmed completion, or otherwise ensure cleanup before exposing the target. > Profile: `codex` · `gpt-5.6-terra`
let Some(staged) = self.staged.borrow().clone() else {
return;
};
bot commented 2026-09-30 11:27:00 +00:00
Outdated
Review
Copy Link
Copy Source

An upload data-write failure returns OutcomeUnknown but leaves CONTROL cached. The server may still be awaiting this transfer's completion, so a later operation can send NOOP or a mutation on a desynchronized session. Clear the control slot when this error (or a zero-byte write) occurs, and add a regression that proves the next operation reconnects rather than reusing it.

Profile: codex · gpt-5.6-terra

An upload data-write failure returns `OutcomeUnknown` but leaves `CONTROL` cached. The server may still be awaiting this transfer's completion, so a later operation can send `NOOP` or a mutation on a desynchronized session. Clear the control slot when this error (or a zero-byte write) occurs, and add a regression that proves the next operation reconnects rather than reusing it. > Profile: `codex` · `gpt-5.6-terra`
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> {
bot commented 2026-09-30 11:47:36 +00:00
Outdated
Review
Copy Link
Copy Source

The stage counter always starts at zero, so another FTP client can create this known name after MLST but before STOR. STOR can then truncate that client's file and cleanup can delete it. Use an unguessable suffix and add a collision regression.

Profile: codex · gpt-5.6-terra

The stage counter always starts at zero, so another FTP client can create this known name after MLST but before STOR. STOR can then truncate that client's file and cleanup can delete it. Use an unguessable suffix and add a collision regression. > Profile: `codex` · `gpt-5.6-terra`
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
bot commented 2026-09-30 10:28:59 +00:00
Outdated
Review
Copy Link
Copy Source

This runs the guest on Tokio's blocking thread pool, contrary to the repository's event-driven/async-only rule; the same pattern is repeated in the other recovery test. Replace the native blocking socket test transport with an async one.

Profile: codex · gpt-5.6-terra

This runs the guest on Tokio's blocking thread pool, contrary to the repository's event-driven/async-only rule; the same pattern is repeated in the other recovery test. Replace the native blocking socket test transport with an async one. > Profile: `codex` · `gpt-5.6-terra`
{
bot commented 2026-09-30 10:38:15 +00:00
Outdated
Review
Copy Link
Copy Source

This thread-pool handoff violates the repository's event-driven-only/no-thread-model rule. Exercise the synchronous guest with a compliant transport, or add an explicit approved exception to the repository rules.

Profile: codex · gpt-5.6-terra

This thread-pool handoff violates the repository's event-driven-only/no-thread-model rule. Exercise the synchronous guest with a compliant transport, or add an explicit approved exception to the repository rules. > Profile: `codex` · `gpt-5.6-terra`
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(())
bot commented 2026-09-30 10:38:15 +00:00
Outdated
Review
Copy Link
Copy Source

This second spawn_blocking repeats the prohibited thread-based test transport. Replace the fixture approach or document an approved exception to the repository rules.

Profile: codex · gpt-5.6-terra

This second `spawn_blocking` repeats the prohibited thread-based test transport. Replace the fixture approach or document an approved exception to the repository rules. > Profile: `codex` · `gpt-5.6-terra`
}
#[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)
}
}