diff --git a/README.md b/README.md index 1936205..7b6a986 100644 --- a/README.md +++ b/README.md @@ -133,6 +133,7 @@ Missing features: - [ ] Show build status - [ ] `pkh put` - [x] Upload the source package to a PPA (native SFTP, no `dput` dependency) + - [x] Degrade to the anonymous FTP queue when the SSH connection never comes up (dput's upload method) - [ ] Upload the source package to the archive - [ ] `pkh commit` - [ ] Commit the changes to git diff --git a/data/launchpad.yml b/data/launchpad.yml index cbdda76..3b1ff1a 100644 --- a/data/launchpad.yml +++ b/data/launchpad.yml @@ -13,6 +13,9 @@ ## ssh_*: the PPA upload queue, as expanded by dput-ng's ## ppa:user/ppa profile (ppa.launchpad.net:22, incoming ## ~/) +## ftp_*: the same upload queue over anonymous FTP, dput-ng's +## plain ppa: profile: the transport pkh degrades to +## when the SSH connection itself never comes up ## content_host_template: ppa.launchpadcontent.net serves PPA apt ## repositories since the 2022 move off ppa.launchpad.net ## git_web_template: Launchpad's CGit mirrors of Ubuntu source packages @@ -21,6 +24,10 @@ api_base: https://api.launchpad.net/1.0 ssh_host: ppa.launchpad.net ssh_port: 22 +## The anonymous FTP upload queue dput-ng's plain ppa: profile uses: +## pkh degrades to it when the SSH connection itself never comes up. +ftp_host: ppa.launchpad.net +ftp_port: 21 incoming_template: "~{owner}/{ppa}" content_host_template: https://ppa.launchpadcontent.net/{owner}/{ppa}/ubuntu git_web_template: https://git.launchpad.net/ubuntu/+source/{package} diff --git a/src/launchpad.rs b/src/launchpad.rs index f53f2dd..60b9ac1 100644 --- a/src/launchpad.rs +++ b/src/launchpad.rs @@ -41,6 +41,11 @@ struct LaunchpadData { ssh_host: String, /// Port of the PPA SFTP upload server ssh_port: u16, + /// Host of the PPA upload queue over anonymous FTP (the transport + /// `pkh put` degrades to when the SSH connection never comes up) + ftp_host: String, + /// Port of the anonymous FTP upload queue + ftp_port: u16, /// Upload queue incoming directory template (`{owner}`/`{ppa}`) incoming_template: String, /// PPA package-content (apt repository) URL template @@ -53,6 +58,13 @@ embed_data! { static ref LAUNCHPAD_DATA: LaunchpadData = "../data/launchpad.yml" } +/// The PPA upload queue over anonymous FTP (host, port): the transport +/// dput-ng's plain `ppa:` profile pushes over, and the one `pkh put` +/// degrades to when the SSH connection itself never comes up. +pub(crate) fn ppa_ftp_queue() -> (String, u16) { + (LAUNCHPAD_DATA.ftp_host.clone(), LAUNCHPAD_DATA.ftp_port) +} + /// Base URL of the Launchpad REST API fn api_base() -> &'static str { &LAUNCHPAD_DATA.api_base @@ -422,6 +434,13 @@ mod tests { assert_eq!(target.login, None); } + /// The anonymous FTP fallback queue resolves from the same data the + /// dput-ng `ppa:` profile uses. + #[test] + fn ppa_ftp_queue_resolves() { + assert_eq!(ppa_ftp_queue(), ("ppa.launchpad.net".to_string(), 21)); + } + #[test] fn ppa_target_rejects_missing_separator() { assert!(ppa_target("just-a-name").is_err()); diff --git a/src/put/ftp.rs b/src/put/ftp.rs new file mode 100644 index 0000000..112f7a2 --- /dev/null +++ b/src/put/ftp.rs @@ -0,0 +1,456 @@ +//! Anonymous FTP transport for the Launchpad PPA upload queue: the graceful +//! degradation of the SFTP transport when the SSH connection itself never +//! comes up (name resolution, TCP, banner or key exchange). dput-ng's plain +//! `ppa:user/ppa` profile pushes over this same queue — ppa.launchpad.net +//! over FTP, anonymous login, incoming `~user/ppa` — so it is the +//! interoperability-tested path. +//! +//! The client is [`suppaftp`]'s (plain-FTP, no TLS) blocking stream on the +//! same time-bounded sockets as the ssh2 transport: the TCP connect and +//! the control/data channel reads and writes all carry timeouts, so a +//! black-holed or stalled server fails the upload instead of hanging it +//! (suppaftp's defaults do not bound them). The upload order (payload +//! first, `.changes` last — the caller passes the files in that order) and +//! the best-effort cleanup of a failed upload (`DELE` of what was already +//! pushed, in reverse upload order) mirror the SFTP path exactly. + +use std::fs::File; +use std::io::{Read, Write}; +use std::net::{SocketAddr, TcpStream}; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use suppaftp::types::FileType; +use suppaftp::{FtpError, FtpStream}; + +use super::ssh; + +/// Bounds one TCP connection attempt, to the control channel or a passive +/// data port: `connect(2)` would otherwise block for minutes (or forever, +/// behind a silent firewall). Generous enough for slow links to Launchpad, +/// short enough that a dead target fails in seconds — the same value and +/// rationale as the SSH path's bound. +const TCP_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); + +/// Read/write timeout on the control channel: every reply is a few bytes, +/// so a stalled server has had its say by the time this expires. +const CONTROL_TIMEOUT: Duration = Duration::from_secs(60); + +/// Write timeout of one data-transfer chunk: the clock restarts at every +/// write, so the wall-clock duration of a large upload is not bounded by +/// design — but a black-holed data connection fails one chunk after this +/// budget. +const DATA_TIMEOUT: Duration = Duration::from_secs(120); + +/// Data transfers stream in chunks of this size. +const CHUNK_SIZE: usize = 32 * 1024; + +/// Upload `files` to the `incoming` queue of `host:port` over anonymous FTP, +/// in the order given (payload first, `.changes` last — the queue processor +/// must never observe a `.changes` without its payload). +/// `on_progress(name, uploaded_bytes, total_bytes)` reports progress. +/// +/// A failure mid-upload best-effort removes what was already pushed (`DELE`, +/// reverse upload order — a lingering payload in the write-only queue area +/// is only hygiene) before returning the original error, like the SFTP +/// path's [`super::cleanup_partial_upload`]. +pub fn upload_queue( + host: &str, + port: u16, + incoming: &str, + files: &[(PathBuf, String)], + on_progress: &dyn Fn(&str, u64, u64), +) -> Result<(), Box> { + let mut ftp = connect(host, port)?; + let who = std::env::var("USER").unwrap_or_else(|_| "anonymous".to_string()); + ftp.login("anonymous".to_string(), format!("{who}@pkh.invalid")) + .map_err(|e| format!("the FTP queue rejected the anonymous login: {e}"))?; + ftp.cwd(incoming) + .map_err(|e| format!("cannot enter the upload queue '{incoming}': {e}"))?; + ftp.transfer_type(FileType::Binary) + .map_err(|e| format!("the FTP queue refused binary transfers: {e}"))?; + + // Remote names pushed so far, in upload order, for the cleanup + let mut uploaded: Vec = Vec::new(); + for (path, name) in files { + if let Err(e) = store(&mut ftp, path, name, on_progress) { + // The failed file itself joins the cleanup: its `STOR` was + // accepted before the transfer failure, so a partial may be + // sitting in the queue + for leftover in super::cleanup_list(&uploaded, Some(name)) { + match ftp.rm(leftover.as_str()) { + Ok(()) => { + log::info!("Removed leftover {leftover} from the failed upload") + } + Err(e) => log::warn!( + "Could not remove the leftover {leftover} of the \ + failed upload: {e}" + ), + } + } + return Err(e); + } + uploaded.push(name.clone()); + } + + // Best effort: the queue keeps what it accepted, so a failure here must + // not fail a completed upload + if let Err(e) = ftp.quit() { + log::debug!("closing the FTP session: {e}"); + } + Ok(()) +} + +/// Connect to the queue, read its banner and set every time bound and +/// workaround the plain suppaftp stream does not carry by itself. +fn connect(host: &str, port: u16) -> Result> { + let tcp = ssh::tcp_connect(host, port)?; + let mut ftp = FtpStream::connect_with_stream(tcp)?.passive_stream_builder(data_connect); + { + let control = ftp.get_ref(); + control.set_read_timeout(Some(CONTROL_TIMEOUT))?; + control.set_write_timeout(Some(CONTROL_TIMEOUT))?; + } + // A `PASV` reply announcing an unroutable address (a server behind NAT + // that does not know its public IP) means the control connection's peer + ftp.set_passive_nat_workaround(true); + Ok(ftp) +} + +/// The passive data-channel connect: bounded like every other network call, +/// where suppaftp's default builder is a plain, unbounded +/// `TcpStream::connect`. +fn data_connect(addr: SocketAddr) -> Result { + TcpStream::connect_timeout(&addr, TCP_CONNECT_TIMEOUT).map_err(FtpError::ConnectionError) +} + +/// Upload `path` as `name` over a passive data connection, reporting +/// progress through `on_progress` per chunk. `finish` reads the transfer +/// completion reply — the only way to learn the server accepted the file. +fn store( + ftp: &mut FtpStream, + path: &Path, + name: &str, + on_progress: &dyn Fn(&str, u64, u64), +) -> Result<(), Box> { + let mut upload = ftp + .put_with_stream(name) + .map_err(|e| format!("the FTP queue rejected '{name}': {e}"))?; + if let suppaftp::DataStream::Tcp(socket) = upload.get_mut() { + socket.set_write_timeout(Some(DATA_TIMEOUT))?; + } + + let mut file = + File::open(path).map_err(|e| format!("cannot read '{}': {e}", path.display()))?; + let total = file.metadata().map(|m| m.len()).unwrap_or(0); + let mut buffer = vec![0u8; CHUNK_SIZE]; + let mut uploaded: u64 = 0; + loop { + let read = file.read(&mut buffer)?; + if read == 0 { + break; + } + upload.write_all(&buffer[..read])?; + uploaded += read as u64; + on_progress(name, uploaded, total); + } + upload + .finish() + .map_err(|e| format!("upload of '{name}' failed: {e}"))?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + use std::net::TcpListener; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + + use parking_lot::Mutex; + + /// In-process fake of the Launchpad FTP upload queue: one control + /// session speaking the protocol subset the client uses (banner, + /// USER/PASS, CWD, TYPE, PASV, STOR, DELE, QUIT). Records every + /// command, stores received bytes under the `STOR` name, and can be + /// told to reject one `STOR` (by index) to exercise the cleanup. Its + /// `PASV` replies announce `0.0.0.0`, the NAT form, so the + /// happy-path test proves the control-peer fallback too. + struct FakeQueue { + addr: SocketAddr, + commands: Arc>>, + files: Arc>>>, + } + + impl FakeQueue { + /// Serve one session on a background thread, greeting it with + /// `banner` and rejecting the `STOR` number `reject_stor`, when + /// set. + fn start(banner: &str, reject_stor: Option) -> FakeQueue { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let commands = Arc::new(Mutex::new(Vec::new())); + let files = Arc::new(Mutex::new(HashMap::new())); + let (c, f) = (Arc::clone(&commands), Arc::clone(&files)); + let banner = banner.to_string(); + std::thread::spawn(move || serve(listener, &banner, c, f, reject_stor)); + FakeQueue { + addr, + commands, + files, + } + } + } + + fn serve( + listener: TcpListener, + banner: &str, + commands: Arc>>, + files: Arc>>>, + reject_stor: Option, + ) { + use std::io::BufRead; + + let (stream, _) = listener.accept().unwrap(); + let mut reader = std::io::BufReader::new(stream.try_clone().unwrap()); + let mut writer = stream; + writer + .write_all(format!("{banner}\r\n").as_bytes()) + .unwrap(); + + let mut pending_data: Option = None; + let stor_index = AtomicUsize::new(0); + loop { + let mut line = String::new(); + if reader.read_line(&mut line).unwrap_or(0) == 0 { + break; + } + let cmd = line.trim_end().to_string(); + commands.lock().push(cmd.clone()); + let (verb, arg) = cmd.split_once(' ').unwrap_or((cmd.as_str(), "")); + match verb { + "USER" | "PASS" => { + writer + .write_all(b"230 Anonymous login ok, access restrictions apply.\r\n") + .unwrap(); + } + "CWD" => { + writer.write_all(b"250 Command successful\r\n").unwrap(); + } + "TYPE" => { + writer.write_all(b"200 Type set to I\r\n").unwrap(); + } + "PASV" => { + let data = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = data.local_addr().unwrap().port(); + writer + .write_all( + format!( + "227 Entering Passive Mode (0,0,0,0,{},{})\r\n", + port / 256, + port % 256 + ) + .as_bytes(), + ) + .unwrap(); + pending_data = Some(data); + } + "STOR" => { + let index = stor_index.fetch_add(1, Ordering::SeqCst); + if Some(index) == reject_stor { + writer.write_all(b"550 Permission denied\r\n").unwrap(); + continue; + } + writer.write_all(b"150 Ok to send data\r\n").unwrap(); + let (mut data, _) = pending_data.take().unwrap().accept().unwrap(); + let mut bytes = Vec::new(); + data.read_to_end(&mut bytes).unwrap(); + files.lock().insert(arg.to_string(), bytes); + writer.write_all(b"226 Transfer complete\r\n").unwrap(); + } + "DELE" => { + if files.lock().remove(arg).is_some() { + writer.write_all(b"250 File deleted\r\n").unwrap(); + } else { + writer.write_all(b"550 No such file\r\n").unwrap(); + } + } + "QUIT" => { + writer.write_all(b"221 Bye\r\n").unwrap(); + break; + } + other => { + writer + .write_all(format!("502 Command '{other}' not implemented\r\n").as_bytes()) + .unwrap(); + } + } + } + } + + /// A temp file with `content`, to upload. + fn upload_file(dir: &Path, name: &str, content: &[u8]) -> (PathBuf, String) { + let path = dir.join(name); + std::fs::write(&path, content).unwrap(); + (path, name.to_string()) + } + + /// The full push: anonymous login, the queue directory entered, + /// payload files uploaded in order before the `.changes`, their bytes + /// intact, and progress reported up to each file's size. + #[test] + fn upload_queue_pushes_payload_before_changes() { + let queue = FakeQueue::start("220 Launchpad upload server", None); + let dir = tempfile::tempdir().unwrap(); + let files = vec![ + upload_file(dir.path(), "pkg_1.0.orig.tar.xz", b"orig bytes"), + upload_file(dir.path(), "pkg_1.0-1.dsc", b"dsc bytes"), + upload_file(dir.path(), "pkg_1.0-1_source.changes", b"changes bytes"), + ]; + let seen = Arc::new(Mutex::new(Vec::new())); + let progress_log = Arc::clone(&seen); + + upload_queue( + "127.0.0.1", + queue.addr.port(), + "~vhaudiquet/lp2167827", + &files, + &|name, uploaded, total| { + progress_log + .lock() + .push((name.to_string(), uploaded, total)) + }, + ) + .unwrap(); + + let commands = queue.commands.lock().clone(); + assert_eq!( + commands[0], "USER anonymous", + "the anonymous login comes first" + ); + assert_eq!( + commands + .iter() + .filter(|c| c == &"CWD ~vhaudiquet/lp2167827") + .count(), + 1 + ); + let cwd = commands.iter().position(|c| c.starts_with("CWD")).unwrap(); + let first_stor = commands.iter().position(|c| c.starts_with("STOR")).unwrap(); + assert!(cwd < first_stor, "the queue directory is entered first"); + assert!( + commands.iter().any(|c| c == "TYPE I"), + "binary transfers are requested" + ); + // Upload order: the payload first, the .changes last + let stors: Vec<&String> = commands.iter().filter(|c| c.starts_with("STOR")).collect(); + assert_eq!( + stors.iter().map(|s| s.as_str()).collect::>(), + vec![ + "STOR pkg_1.0.orig.tar.xz", + "STOR pkg_1.0-1.dsc", + "STOR pkg_1.0-1_source.changes", + ] + ); + assert_eq!(*commands.last().unwrap(), "QUIT"); + + // The received bytes are intact + let stored = queue.files.lock(); + assert_eq!(stored.get("pkg_1.0.orig.tar.xz").unwrap(), b"orig bytes"); + assert_eq!(stored.get("pkg_1.0-1.dsc").unwrap(), b"dsc bytes"); + assert_eq!( + stored.get("pkg_1.0-1_source.changes").unwrap(), + b"changes bytes" + ); + + // Progress reached each file's size + let progress = seen.lock(); + for (path, name) in &files { + let size = path.metadata().unwrap().len(); + let reached = progress + .iter() + .any(|(n, uploaded, total)| n == name && *uploaded == size && *total == size); + assert!(reached, "no progress report completed for '{name}'"); + } + } + + /// A `STOR` the queue rejects fails the upload, and the client removes + /// what it already pushed — the failed file first, then the earlier + /// payloads, in reverse upload order — before returning the error. + #[test] + fn upload_queue_cleans_up_after_a_rejected_stor() { + let queue = FakeQueue::start("220 Launchpad upload server", Some(1)); + let dir = tempfile::tempdir().unwrap(); + let files = vec![ + upload_file(dir.path(), "one.dsc", b"one"), + upload_file(dir.path(), "two.tar.xz", b"two"), + upload_file(dir.path(), "three.changes", b"three"), + ]; + + let err = upload_queue( + "127.0.0.1", + queue.addr.port(), + "~user/ppa", + &files, + &|_, _, _| {}, + ) + .unwrap_err(); + assert!( + err.to_string().contains("rejected 'two.tar.xz'"), + "the error names the rejected file, got: {err}" + ); + + let commands = queue.commands.lock(); + let deles: Vec<&String> = commands.iter().filter(|c| c.starts_with("DELE")).collect(); + assert_eq!( + deles.iter().map(|d| d.as_str()).collect::>(), + vec!["DELE two.tar.xz", "DELE one.dsc"], + "the failed file is removed first, then the earlier payloads" + ); + let files = queue.files.lock(); + assert!( + !files.contains_key("one.dsc"), + "the pushed payload is removed" + ); + assert!(!files.contains_key("three.changes"), "never reached"); + } + + /// A multi-line banner (`220-first` closed by `220 last`) parses like + /// the single-line form. + #[test] + fn upload_queue_reads_multiline_replies() { + let queue = FakeQueue::start("220-Launchpad\r\n220 upload server", None); + let dir = tempfile::tempdir().unwrap(); + let files = vec![upload_file(dir.path(), "pkg.dsc", b"bytes")]; + + upload_queue( + "127.0.0.1", + queue.addr.port(), + "~user/ppa", + &files, + &|_, _, _| {}, + ) + .unwrap(); + + let stored = queue.files.lock(); + assert_eq!(stored.get("pkg.dsc").unwrap(), b"bytes"); + } + + /// Live control-channel handshake with the real Launchpad FTP queue: + /// banner, anonymous login and a `CWD` — nothing is uploaded, the + /// queue is left untouched. For deliberate ad-hoc runs + /// (`cargo test -- --ignored`), not the pre-commit pass: it hits the + /// network. + #[test] + #[ignore = "hits the network: the real Launchpad FTP queue"] + fn live_launchpad_control_channel() { + let (host, port) = crate::launchpad::ppa_ftp_queue(); + assert_eq!((host.as_str(), port), ("ppa.launchpad.net", 21)); + + let mut ftp = connect(&host, port).unwrap(); + ftp.login("anonymous", "pkh@invalid").unwrap(); + ftp.cwd("/").unwrap(); + ftp.quit().unwrap(); + } +} diff --git a/src/put/mod.rs b/src/put/mod.rs index 2676887..8ffb67b 100644 --- a/src/put/mod.rs +++ b/src/put/mod.rs @@ -1,17 +1,12 @@ -//! Native upload of built source packages (`pkh put`): the dput -//! replacement. Resolves the upload target, discovers and validates the -//! `.changes` file and its artifacts, then pushes them over SFTP with -//! host-key verification and an upload record preventing accidental -//! duplicate uploads. -//! -//! Payload files are uploaded first and the `.changes` file last, like -//! dput does, so a partially uploaded set cannot be picked up by the -//! server-side queue processors. A run that fails mid-upload removes its -//! already-uploaded files from the incoming queue (best effort), so a -//! retried upload starts from a clean queue; a failed upload is never -//! recorded in the upload log, so a re-run replays every file. +//! PPA upload: the dput replacement. Resolves the upload target, discovers +//! and validates the `.changes` file and its artifacts, then pushes them +//! over SFTP with host-key verification and an upload record preventing +//! accidental duplicate uploads. When the SSH connection itself never +//! comes up, the upload degrades to the anonymous FTP queue — the +//! transport dput's plain `ppa:` profiles use — through [`ftp`]. pub mod changes; +pub mod ftp; pub mod ssh; pub mod target; @@ -139,11 +134,6 @@ async fn put_impl(opts: &PutOptions<'_>) -> Result<(), Box) -> Result<(), Box = Vec::new(); - - for (path, name) in &uploads { - let size = match path.metadata() { - Ok(metadata) => metadata.len(), - Err(e) => { - // Nothing was attempted for this file: only what earlier - // iterations uploaded needs removing - let error: Box = - format!("cannot stat '{}': {}", path.display(), e).into(); - cleanup_partial_upload(&sftp, incoming, &uploaded, None, &host); - return Err(error); - } - }; - let remote = format!("{incoming}/{name}"); - let label = format!("Uploading {name}"); - let view = opts.view; - let on_progress = |uploaded: u64| view.progress(&label, uploaded as usize, size as usize); - let result = ssh::upload_file(&sftp, path, &remote, &host, &on_progress); - if let Err(e) = result { - // The failed file itself joins the cleanup: its remote `create` - // may have succeeded before the failure, leaving a partial — or, - // on a failed close, a truncated — file behind - cleanup_partial_upload(&sftp, incoming, &uploaded, Some(name), &host); - return Err(e); + // The upload queue is SFTP first, degrading to the anonymous FTP + // queue when the SSH connection itself never comes up (name + // resolution, TCP, banner or key exchange): dput pushes PPAs over + // that FTP queue by default, so it is the interoperability-tested + // fallback. A server that answers but refuses the upload (host key + // not accepted, no matching key) stays an error: silently switching + // transport would bypass the refusal. + opts.view + .message(&format!("Connecting to {login}@{host}:{port}...")); + let transfer = match ssh::connect(&host, port, &login, &ssh_config, opts.prompter) { + Ok(session) => sftp_transfer(&session, &uploads, incoming, &host, opts.view), + Err(ssh::ConnectFailure::Transport(e)) => { + log::warn!("SSH transport to {host}:{port} failed: {e}"); + let (ftp_host, ftp_port) = launchpad::ppa_ftp_queue(); + opts.view.message(&format!( + "Falling back to the anonymous FTP queue on {ftp_host}:{ftp_port} \ + (dput's upload method)..." + )); + ftp::upload_queue( + &ftp_host, + ftp_port, + incoming, + &uploads, + &|name, uploaded, total| { + opts.view.progress( + &format!("Uploading {name}"), + uploaded as usize, + total as usize, + ); + }, + ) } - uploaded.push(name.clone()); - } + Err(ssh::ConnectFailure::Refused(e)) => return Err(e), + }; + transfer?; // Recorded only once the whole upload succeeded: the log backs the // duplicate-upload guard, and a failed upload must not count as - // uploaded (a re-run replays every file — `sftp.create` truncates, so - // replaying is safe). + // uploaded (a re-run replays every file — `sftp.create` truncates and + // the FTP `STOR` overwrites, so replaying is safe). record_upload(&upload_log_path()?, &record)?; info!( @@ -212,6 +204,54 @@ async fn put_impl(opts: &PutOptions<'_>) -> Result<(), Box Result<(), Box> { + let sftp = ssh::sftp(session)?; + + // Remote names of the files uploaded so far, in upload order (the + // .changes last). A run failing mid-upload removes these from the + // write-only incoming queue before returning: the uploaded payloads + // would otherwise linger in the queue area forever, and a .changes + // truncated by a failed close could even be picked up by the scanner. + let mut uploaded: Vec = Vec::new(); + + for (path, name) in uploads { + let size = match path.metadata() { + Ok(metadata) => metadata.len(), + Err(e) => { + // Nothing was attempted for this file: only what earlier + // iterations uploaded needs removing + let error: Box = + format!("cannot stat '{}': {}", path.display(), e).into(); + cleanup_partial_upload(&sftp, incoming, &uploaded, None, host); + return Err(error); + } + }; + let remote = format!("{incoming}/{name}"); + let label = format!("Uploading {name}"); + let on_progress = |uploaded: u64| view.progress(&label, uploaded as usize, size as usize); + let result = ssh::upload_file(&sftp, path, &remote, host, &on_progress); + if let Err(e) = result { + // The failed file itself joins the cleanup: its remote `create` + // may have succeeded before the failure, leaving a partial — or, + // on a failed close, a truncated — file behind + cleanup_partial_upload(&sftp, incoming, &uploaded, Some(name), host); + return Err(e); + } + uploaded.push(name.clone()); + } + Ok(()) +} + /// The remote names to attempt removing after a failed upload: everything /// already uploaded plus, when set, `failed` (the file whose upload just /// failed — its remote `create` may have succeeded before the failure, @@ -219,8 +259,9 @@ async fn put_impl(opts: &PutOptions<'_>) -> Result<(), Box) -> Vec { +/// without a server; the network sides are [`cleanup_partial_upload`] and +/// the FTP transport's `DELE` loop ([`ftp`]). +pub(crate) fn cleanup_list(uploaded: &[String], failed: Option<&str>) -> Vec { let mut names: Vec = uploaded.to_vec(); if let Some(failed) = failed { names.push(failed.to_string()); diff --git a/src/put/ssh.rs b/src/put/ssh.rs index 38e8d14..7f62371 100644 --- a/src/put/ssh.rs +++ b/src/put/ssh.rs @@ -253,7 +253,9 @@ fn duration_ms(timeout: Duration) -> u32 { /// address in order (like `TcpStream::connect` does) with /// [`TCP_CONNECT_TIMEOUT`] per attempt instead of blocking indefinitely. /// Fails with a message naming the target and every per-address error. -fn tcp_connect(host: &str, port: u16) -> Result { +/// Also the TCP layer of the anonymous FTP fallback transport +/// ([`super::ftp`]), whose connection semantics are identical. +pub(crate) fn tcp_connect(host: &str, port: u16) -> Result { let addrs: Vec = (host, port) .to_socket_addrs() .map_err(|e| format!("cannot resolve {host}:{port}: {e}"))? @@ -292,6 +294,39 @@ fn connect_failed_message( attempts.len() ) } +/// Why establishing the SSH session failed. +/// +/// [`Transport`] failures mean the connection never came up (name +/// resolution, TCP, banner or key exchange): the target may still be +/// reachable over another transport, so `pkh put` degrades to the +/// anonymous FTP queue — dput's default for PPAs ([`super::ftp`]). +/// [`Refused`] failures mean the server answered but rejected the +/// upload (host key not accepted, no matching authentication): silently +/// switching to anonymous FTP would bypass a refusal, so they stay +/// errors. +#[derive(Debug)] +pub enum ConnectFailure { + /// The connection itself never came up. + Transport(Box), + /// The server answered but rejected the upload. + Refused(Box), +} + +impl std::fmt::Display for ConnectFailure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ConnectFailure::Transport(e) | ConnectFailure::Refused(e) => write!(f, "{e}"), + } + } +} + +impl std::error::Error for ConnectFailure { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + ConnectFailure::Transport(e) | ConnectFailure::Refused(e) => Some(e.as_ref()), + } + } +} /// Connect to `host:port`, verify the server host key and authenticate as /// `login`: every ssh-agent identity first, then the configured and default @@ -303,10 +338,13 @@ pub fn connect( login: &str, config: &SshConfig, prompter: &dyn Prompter, -) -> Result> { - let tcp = tcp_connect(host, port)?; +) -> Result { + use ConnectFailure::*; - let mut session = Session::new()?; + let tcp = tcp_connect(host, port).map_err(|e| Transport(e.into()))?; + + let mut session = Session::new() + .map_err(|e| Transport(format!("cannot initialize the SSH session: {e}").into()))?; // In blocking mode (the libssh2 default), a call that would block loops // in `_libssh2_wait_socket` (via the `BLOCK_ADJUST` macros of // session.h in the vendored libssh2-sys sources), which bounds the @@ -321,14 +359,14 @@ pub fn connect( session.set_tcp_stream(tcp); session .handshake() - .map_err(|e| format!("SSH handshake with {host} failed: {e}"))?; + .map_err(|e| Transport(format!("SSH handshake with {host} failed: {e}").into()))?; let (key, key_type) = session .host_key() - .ok_or_else(|| format!("{host} offered no host key"))?; - verify_host_key(host, port, key, key_type, prompter)?; + .ok_or_else(|| Transport(format!("{host} offered no host key").into()))?; + verify_host_key(host, port, key, key_type, prompter).map_err(Refused)?; - authenticate(&session, host, login, config)?; + authenticate(&session, host, login, config).map_err(Refused)?; // Only SFTP open/data calls remain on this session: switch from the // connection-phase budget to the generous per-call transfer one