// Copyright 2026 PaiGack
// Licensed under the Apache License, Version 2.0.
// Ported from jlaffaye/ftp (ISC License), see LICENSE-THIRD-PARTY.
// Layer: IO — uses the `moonbitlang/async` stack from the root moon.pkg.
// The data channel: everything between "a transfer was requested" and "the
// socket is open".
//
// This file merges what used to be three files (`transport_pasv`,
// `transport_epsv`, `transport_dataconn`). They were always one concern —
// opening the passive data connection — and were only split by which RFC the
// command came from, which is a poor reason to make a reader jump between
// files. `get_data_port` below is the seam that ties `PASV` and `EPSV`
// together, and `open_data_conn` / `cmd_data_conn_from` are its only callers.
///|
/// Send `PASV` and return the `(ip, port)` pair the server advertises.
///
/// `227 Entering Passive Mode (127,0,0,1,196,6)`
pub async fn pasv(session : Session) -> (String, Int) {
let response = cmd_expect(session.control(), "PASV", [
status_enter_passive_mode,
])
parse_pasv(response.message())
}
///|
/// Parse the six comma separated numbers of a `PASV` reply into an IP and a
/// port (`port = p1 * 256 + p2`). Fewer than six numbers is an error.
pub fn parse_pasv(line : String) -> (String, Int) raise FtpError {
let open = match line.find("(") {
Some(open) => open
None => raise FtpError::ParseError(msg="invalid PASV response: \{line}")
}
let close = match line.find(")") {
Some(close) => close
None => raise FtpError::ParseError(msg="invalid PASV response: \{line}")
}
guard close > open + 1 else {
raise FtpError::ParseError(msg="invalid PASV response: \{line}")
}
let inner = line[open + 1:close].to_owned()
let parts = inner.split(",").collect()
guard parts.length() >= 6 else {
raise FtpError::ParseError(msg="invalid PASV response: \{line}")
}
let numbers : Array[Int] = []
for i = 0; i < 6; i = i + 1 {
let text = parts[i].trim().to_owned()
guard is_all_digits(text) else {
raise FtpError::ParseError(msg="invalid PASV response: \{line}")
}
let value = match parse_octet(text) {
Some(value) => value
None => raise FtpError::ParseError(msg="invalid PASV response: \{line}")
}
numbers.push(value)
}
let ip = "\{numbers[0]}.\{numbers[1]}.\{numbers[2]}.\{numbers[3]}"
let port = numbers[4] * 256 + numbers[5]
(ip, port)
}
///|
/// Decide whether the IP returned by `PASV` may be used as the data
/// destination.
///
/// Upstream's rule, kept verbatim: the address is bogus when it is multicast,
/// or when its "privateness" differs from the control connection, or when its
/// loopback-ness differs. This is what makes a server that answers
/// `PASV ... (10,0,0,1,...)` from a public IP unusable unless the caller opts
/// in with `trust_pasv_ip`.
pub fn is_bogus_data_ip(cmd_ip : String, data_ip : String) -> Bool {
if is_multicast(data_ip) {
return true
}
if is_private(data_ip) != is_private(cmd_ip) {
return true
}
if is_loopback(data_ip) != is_loopback(cmd_ip) {
return true
}
false
}
///|
/// Parse a decimal byte value (0..255), returning `None` otherwise.
pub fn parse_octet(text : String) -> Int? {
guard text != "" && text.length() <= 3 else { return None }
let mut value = 0
for i = 0; i < text.length(); i = i + 1 {
let c = text.unsafe_get(i).to_int()
guard c >= 48 && c <= 57 else { return None }
value = value * 10 + (c - 48)
}
guard value <= 255 else { return None }
Some(value)
}
///|
/// Whether `ip` is a private (RFC 1918 / link local / ULA) address.
pub fn is_private(ip : String) -> Bool {
match ip_segments(ip) {
Some(s) =>
s[0] == 10 ||
(s[0] == 172 && s[1] >= 16 && s[1] <= 31) ||
(s[0] == 192 && s[1] == 168) ||
(s[0] == 169 && s[1] == 254) ||
(s[0] == 100 && s[1] >= 64 && s[1] <= 127)
None => ip.to_lower().has_prefix("fc") || ip.to_lower().has_prefix("fd")
}
}
///|
/// Whether `ip` is a loopback address (`127.0.0.0/8` or `::1`).
pub fn is_loopback(ip : String) -> Bool {
match ip_segments(ip) {
Some(s) => s[0] == 127
None => ip == "::1" || ip.to_lower() == "::1"
}
}
///|
/// Whether `ip` is a multicast address (`224.0.0.0/4` or `ff00::/8`).
pub fn is_multicast(ip : String) -> Bool {
match ip_segments(ip) {
Some(s) => s[0] >= 224 && s[0] <= 239
None => ip.to_lower().has_prefix("ff")
}
}
///|
/// Split a dotted quad into four integers, returning `None` for IPv6 or a
/// malformed input.
fn ip_segments(ip : String) -> Array[Int]? {
let parts = ip.split(".").collect()
guard parts.length() == 4 else { return None }
let out : Array[Int] = []
for part in parts {
let part = part.to_owned()
guard is_all_digits(part) else { return None }
let value = match parse_octet(part) {
Some(value) => value
None => return None
}
out.push(value)
}
Some(out)
}
///|
/// Send `EPSV` (RFC 2428) and return the data port the server opened.
///
/// `229 Entering Extended Passive Mode (|||6446|)`
pub async fn epsv(session : Session) -> Int {
let response = cmd_expect(session.control(), "EPSV", [
status_enter_extended_passive_mode,
])
parse_epsv(response.message())
}
///|
/// Extract the port from an `EPSV` reply body.
///
/// The body is expected to contain `|||` followed by the port and a closing
/// `|`. A malformed reply raises `ParseError` instead of silently using port
/// 0, which would otherwise hang the connect.
pub fn parse_epsv(line : String) -> Int raise FtpError {
let marker = match line.find("|||") {
Some(marker) => marker
None => raise FtpError::ParseError(msg="invalid EPSV response: \{line}")
}
let start = marker + 3
let tail = line[start:].to_owned()
let end = match tail.find("|") {
Some(end) => end
None => raise FtpError::ParseError(msg="invalid EPSV response: \{line}")
}
guard end > 0 else {
raise FtpError::ParseError(msg="invalid EPSV response: \{line}")
}
let digits = tail[0:end].to_owned()
guard is_all_digits(digits) else {
raise FtpError::ParseError(msg="invalid EPSV response: \{line}")
}
match parse_port(digits) {
Some(port) => port
None => raise FtpError::ParseError(msg="invalid EPSV response: \{line}")
}
}
///|
/// Whether every byte of `src` is an ASCII digit.
pub fn is_all_digits(src : String) -> Bool {
guard src != "" else { return false }
for i = 0; i < src.length(); i = i + 1 {
let c = src.unsafe_get(i).to_int()
if c < 48 || c > 57 {
return false
}
}
true
}
///|
/// Parse a non-empty decimal string into an `Int`, returning `None` on any
/// non-digit byte or on overflow.
pub fn parse_port(digits : String) -> Int? {
guard digits != "" else { return None }
let mut value = 0
for i = 0; i < digits.length(); i = i + 1 {
let c = digits.unsafe_get(i).to_int()
if c < 48 || c > 57 {
return None
}
value = value * 10 + (c - 48)
if value > 65535 {
return None
}
}
Some(value)
}
///|
/// The data connection used for one transfer.
pub struct DataConn {
/// Reads the payload coming from the server.
reader : &@io.Reader
/// Writes the payload going to the server.
writer : &@io.Writer
/// Closes the socket, keeping the raw connection alive for the 226 read.
closer : () -> Unit
}
///|
/// The readable end of the data connection.
pub fn DataConn::reader(self : DataConn) -> &@io.Reader {
self.reader
}
///|
/// The writable end of the data connection.
pub fn DataConn::writer(self : DataConn) -> &@io.Writer {
self.writer
}
///|
/// Close the data connection without touching the control channel.
pub fn DataConn::close(self : DataConn) -> Unit {
(self.closer)()
}
///|
/// Get the port to connect to, preferring `EPSV` and permanently falling back
/// to `PASV` after the first failure.
///
/// The "EPSV failed once, never try again" behaviour is a deliberate upstream
/// design choice: some servers answer `EPSV` with an error but also with a
/// malformed `PASV` reply, and retrying `EPSV` on every transfer turns a fast
/// failure into a timeout per file.
pub async fn get_data_port(session : Session) -> (String, Int) {
if session.options().disable_epsv() == false && session.skip_epsv() == false {
let port = epsv(session) catch {
_ => {
session.set_skip_epsv(true)
return pasv(session)
}
}
return (session.control().host(), port)
}
let (ip, port) = pasv(session)
// SSRF guard: use the control connection IP unless the caller explicitly
// trusts the server's answer *and* the answer is not obviously bogus.
if session.options().trust_pasv_ip() &&
is_bogus_data_ip(session.control().host(), ip) == false {
return (ip, port)
}
(session.control().host(), port)
}
///|
/// Open the passive data connection.
///
/// With TLS the socket is connected but the handshake is deferred to the first
/// read/write: ProFTPD and PureFTPD both refuse a data connection that starts
/// its handshake before the transfer command was answered.
pub async fn open_data_conn(session : Session) -> DataConn {
let (ip, port) = get_data_port(session)
let addr = @socket.Addr::parse("\{ip}:\{port}") catch {
_ =>
raise FtpError::ParseError(
msg="invalid data connection address: \{ip}:\{port}",
)
}
let tcp = @async.with_timeout(session.options().timeout_ms(), async fn() {
@socket.Tcp::connect(addr)
})
if session.options().tls() {
let tls = @tls.Tls::client(
tcp,
host=session.control().host(),
trust=session.options().trust(),
)
return { reader: tls, writer: tls, closer: () => tls.close(), }
}
{ reader: tcp, writer: tcp, closer: () => tcp.close(), }
}
///|
/// Open a data connection and start a transfer command on it.
///
/// This is the heart of the data channel and follows upstream
/// `cmdDataConnFrom` step by step:
///
/// 1. `PRET ` first when the server advertised `PRET`,
/// 2. open the data connection,
/// 3. `REST ` when a resume offset was given, expecting `350`,
/// 4. send the transfer command, expecting `125` or `150`,
/// 5. on a non 2xx answer, close the data connection before raising.
pub async fn cmd_data_conn_from(
session : Session,
offset : Int64,
cmd : String,
args : Array[String],
) -> DataConn {
if session.use_pret() {
let _ = try {
ignore(
cmd_expect(session.control(), "PRET \{cmd}", [
status_cmd_ok, status_file_action_pending_further_info,
]),
)
true
} catch {
_ => false
}
}
let conn = open_data_conn(session)
if offset != 0 {
let ok = try {
ignore(
cmd_format(session.control(), "REST {}", ["\{offset}"], [
status_file_action_pending_further_info,
]),
)
true
} catch {
_ => false
}
guard ok else {
conn.close()
raise FtpError::ParseError(msg="REST command failed")
}
}
let response = cmd_format(session.control(), cmd, args, [
status_data_conn_already_in_use, 150,
]) catch {
_ => {
conn.close()
raise FtpError::ParseError(msg="transfer command failed")
}
}
// The reply that starts a transfer is a *positive intermediate* (`125` or
// `150`), not a completion: `2xx` only arrives at the end, on `226`. Testing
// `is_positive_completion` here rejected every real transfer.
guard is_positive_intermediate(response.code()) else {
conn.close()
raise FtpError::ServerError(code=response.code(), msg=response.message())
}
conn
}