put: bound the SSH connect and session operations with timeouts
TcpStream::connect and the blocking libssh2 session had no timeouts: a black-holed host hung pkh put forever, mid-resolution, mid-handshake or mid-upload. Connect attempts now get a 15 s timeout per resolved address, the session gets a 30 s API timeout for the handshake/auth phase and a 300 s per-call timeout for SFTP operations (per low-level libssh2 call, not per transfer — documented); failures name the operation and host.
This commit is contained in:
+1
-1
@@ -159,7 +159,7 @@ pub async fn put(
|
|||||||
bar.set_prefix(format!("Uploading {name}..."));
|
bar.set_prefix(format!("Uploading {name}..."));
|
||||||
|
|
||||||
let remote = format!("{}/{}", target.incoming.trim_end_matches('/'), name);
|
let remote = format!("{}/{}", target.incoming.trim_end_matches('/'), name);
|
||||||
let result = ssh::upload_file(&sftp, path, &remote, &bar);
|
let result = ssh::upload_file(&sftp, path, &remote, &host, &bar);
|
||||||
bar.finish_and_clear();
|
bar.finish_and_clear();
|
||||||
result?;
|
result?;
|
||||||
}
|
}
|
||||||
|
|||||||
+194
-12
@@ -5,14 +5,20 @@
|
|||||||
//! authentication (every ssh-agent identity first, then configured and
|
//! authentication (every ssh-agent identity first, then configured and
|
||||||
//! default key files) and chunked SFTP upload with progress reporting.
|
//! default key files) and chunked SFTP upload with progress reporting.
|
||||||
//!
|
//!
|
||||||
|
//! Every network phase is time-bounded — the TCP connect, the SSH
|
||||||
|
//! handshake/authentication, and each low-level call of the SFTP data
|
||||||
|
//! transfer (see the `*_TIMEOUT` constants) — so a black-holed or stalled
|
||||||
|
//! server fails the upload instead of hanging it forever.
|
||||||
|
//!
|
||||||
//! This replaces dput-ng's paramiko transport with the `ssh2` (libssh2)
|
//! This replaces dput-ng's paramiko transport with the `ssh2` (libssh2)
|
||||||
//! stack the rest of pkh already uses for build contexts.
|
//! stack the rest of pkh already uses for build contexts.
|
||||||
|
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::fs;
|
use std::fs;
|
||||||
use std::io::{Read, Write};
|
use std::io::{Read, Write};
|
||||||
use std::net::TcpStream;
|
use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
use indicatif::ProgressBar;
|
use indicatif::ProgressBar;
|
||||||
use lazy_static::lazy_static;
|
use lazy_static::lazy_static;
|
||||||
@@ -221,19 +227,103 @@ fn wildmatch(text: &str, pattern: &str) -> bool {
|
|||||||
pattern[p..].iter().all(|&c| c == '*')
|
pattern[p..].iter().all(|&c| c == '*')
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Bound on one TCP connection attempt to one resolved address: `connect(2)`
|
||||||
|
/// would otherwise block for minutes (or forever, behind a silent firewall)
|
||||||
|
/// per address. Generous enough for slow links to Launchpad, short enough
|
||||||
|
/// that a dead target fails in seconds.
|
||||||
|
const TCP_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
|
||||||
|
|
||||||
|
/// Timeout for the blocking libssh2 calls of the connection phase (banner
|
||||||
|
/// exchange and key exchange of the handshake, host key check,
|
||||||
|
/// authentication): when it expires the pending call fails with
|
||||||
|
/// `LIBSSH2_ERROR_TIMEOUT` (surfaced as "timed out") instead of hanging
|
||||||
|
/// forever on a stalled server. See [`connect`] for what this does and does
|
||||||
|
/// not cover.
|
||||||
|
const SSH_API_TIMEOUT: Duration = Duration::from_secs(30);
|
||||||
|
|
||||||
|
/// Timeout for the blocking libssh2 calls of the SFTP data transfer (file
|
||||||
|
/// creation, chunk writes, final close handshake). It applies *per
|
||||||
|
/// low-level libssh2 call*, not to the whole transfer: the clock restarts
|
||||||
|
/// at every API entry, so the wall-clock duration of a transfer is not
|
||||||
|
/// bounded by design — but a stalled or black-holed server fails one call
|
||||||
|
/// after this budget instead of hanging `pkh put` forever.
|
||||||
|
const SSH_TRANSFER_TIMEOUT: Duration = Duration::from_secs(300);
|
||||||
|
|
||||||
|
/// libssh2 timeouts are millisecond counts; the constants above are exact
|
||||||
|
/// multiples of a millisecond, and anything larger saturates instead of
|
||||||
|
/// truncating
|
||||||
|
fn duration_ms(timeout: Duration) -> u32 {
|
||||||
|
u32::try_from(timeout.as_millis()).unwrap_or(u32::MAX)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Resolve `host:port` and open a TCP connection, trying each resolved
|
||||||
|
/// 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<TcpStream, String> {
|
||||||
|
let addrs: Vec<SocketAddr> = (host, port)
|
||||||
|
.to_socket_addrs()
|
||||||
|
.map_err(|e| format!("cannot resolve {host}:{port}: {e}"))?
|
||||||
|
.collect();
|
||||||
|
if addrs.is_empty() {
|
||||||
|
return Err(format!(
|
||||||
|
"cannot connect to {host}:{port}: '{host}' resolved to no addresses"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut attempts: Vec<(SocketAddr, std::io::Error)> = Vec::new();
|
||||||
|
for addr in addrs {
|
||||||
|
match TcpStream::connect_timeout(&addr, TCP_CONNECT_TIMEOUT) {
|
||||||
|
Ok(stream) => return Ok(stream),
|
||||||
|
Err(e) => attempts.push((addr, e)),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(connect_failed_message(host, port, &attempts))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Message for a target none of whose resolved addresses accepted a
|
||||||
|
/// connection within [`TCP_CONNECT_TIMEOUT`]. Pure so tests can assert it
|
||||||
|
/// without any network.
|
||||||
|
fn connect_failed_message(
|
||||||
|
host: &str,
|
||||||
|
port: u16,
|
||||||
|
attempts: &[(SocketAddr, std::io::Error)],
|
||||||
|
) -> String {
|
||||||
|
let details = attempts
|
||||||
|
.iter()
|
||||||
|
.map(|(addr, e)| format!("{addr}: {e}"))
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
.join("; ");
|
||||||
|
format!(
|
||||||
|
"cannot connect to {host}:{port} (tried {} address(es)): {details}",
|
||||||
|
attempts.len()
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
/// Connect to `host:port`, verify the server host key and authenticate as
|
/// Connect to `host:port`, verify the server host key and authenticate as
|
||||||
/// `login`: every ssh-agent identity first, then the configured and default
|
/// `login`: every ssh-agent identity first, then the configured and default
|
||||||
/// identity files.
|
/// identity files. The TCP connect, handshake and authentication steps are
|
||||||
|
/// time-bounded ([`TCP_CONNECT_TIMEOUT`], [`SSH_API_TIMEOUT`]).
|
||||||
pub fn connect(
|
pub fn connect(
|
||||||
host: &str,
|
host: &str,
|
||||||
port: u16,
|
port: u16,
|
||||||
login: &str,
|
login: &str,
|
||||||
config: &SshConfig,
|
config: &SshConfig,
|
||||||
) -> Result<Session, Box<dyn std::error::Error>> {
|
) -> Result<Session, Box<dyn std::error::Error>> {
|
||||||
let tcp = TcpStream::connect((host, port))
|
let tcp = tcp_connect(host, port)?;
|
||||||
.map_err(|e| format!("cannot connect to {host}:{port}: {e}"))?;
|
|
||||||
|
|
||||||
let mut session = Session::new()?;
|
let mut session = Session::new()?;
|
||||||
|
// 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
|
||||||
|
// underlying poll()/select() wait by this timeout and fails the call
|
||||||
|
// with LIBSSH2_ERROR_TIMEOUT when it expires. Verified against
|
||||||
|
// libssh2-sys 0.3.3: the handshake (session.c), host key check,
|
||||||
|
// authentication, channel and SFTP calls all route through it, so a
|
||||||
|
// stalled server errors out instead of hanging forever. It bounds
|
||||||
|
// *each* libssh2 call, not the whole phase: a server that trickles
|
||||||
|
// bytes often enough keeps resuming every call in time.
|
||||||
|
session.set_timeout(duration_ms(SSH_API_TIMEOUT));
|
||||||
session.set_tcp_stream(tcp);
|
session.set_tcp_stream(tcp);
|
||||||
session
|
session
|
||||||
.handshake()
|
.handshake()
|
||||||
@@ -246,6 +336,10 @@ pub fn connect(
|
|||||||
|
|
||||||
authenticate(&session, host, login, config)?;
|
authenticate(&session, host, login, config)?;
|
||||||
|
|
||||||
|
// Only SFTP open/data calls remain on this session: switch from the
|
||||||
|
// connection-phase budget to the generous per-call transfer one
|
||||||
|
session.set_timeout(duration_ms(SSH_TRANSFER_TIMEOUT));
|
||||||
|
|
||||||
Ok(session)
|
Ok(session)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -426,14 +520,17 @@ fn authenticate(
|
|||||||
.into())
|
.into())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Upload `local` to the remote SFTP `path`, updating `bar` per chunk. The
|
/// Upload `local` to the remote SFTP `path` on `host`, updating `bar` per
|
||||||
/// remote file is created/truncated; Launchpad's upload queue is
|
/// chunk. The remote file is created/truncated; Launchpad's upload queue is
|
||||||
/// write-only, so failures here mean the upload failed — there is nothing
|
/// write-only, so failures here mean the upload failed — there is nothing
|
||||||
/// to inspect server-side.
|
/// to inspect server-side. Every SFTP call is bounded by the session's
|
||||||
|
/// per-call transfer timeout (see [`SSH_TRANSFER_TIMEOUT`]), so a stalled
|
||||||
|
/// server fails the upload instead of hanging it.
|
||||||
pub fn upload_file(
|
pub fn upload_file(
|
||||||
sftp: &ssh2::Sftp,
|
sftp: &ssh2::Sftp,
|
||||||
local: &Path,
|
local: &Path,
|
||||||
remote: &str,
|
remote: &str,
|
||||||
|
host: &str,
|
||||||
bar: &ProgressBar,
|
bar: &ProgressBar,
|
||||||
) -> Result<(), Box<dyn std::error::Error>> {
|
) -> Result<(), Box<dyn std::error::Error>> {
|
||||||
let mut local_file =
|
let mut local_file =
|
||||||
@@ -441,7 +538,7 @@ pub fn upload_file(
|
|||||||
|
|
||||||
let mut remote_file = sftp
|
let mut remote_file = sftp
|
||||||
.create(Path::new(remote))
|
.create(Path::new(remote))
|
||||||
.map_err(|e| format!("cannot create remote file {remote}: {e}"))?;
|
.map_err(|e| format!("cannot create remote file {remote} on {host}: {e}"))?;
|
||||||
|
|
||||||
let mut buf = [0u8; 32 * 1024];
|
let mut buf = [0u8; 32 * 1024];
|
||||||
loop {
|
loop {
|
||||||
@@ -451,7 +548,7 @@ pub fn upload_file(
|
|||||||
}
|
}
|
||||||
remote_file
|
remote_file
|
||||||
.write_all(&buf[..n])
|
.write_all(&buf[..n])
|
||||||
.map_err(|e| format!("failed uploading to {remote}: {e}"))?;
|
.map_err(|e| format!("failed uploading to {remote} on {host}: {e}"))?;
|
||||||
bar.inc(n as u64);
|
bar.inc(n as u64);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -461,9 +558,9 @@ pub fn upload_file(
|
|||||||
// truncated remote file as a successful upload. `ssh2::File::write` is
|
// truncated remote file as a successful upload. `ssh2::File::write` is
|
||||||
// unbuffered (`Write::flush` is a documented no-op) and `close`
|
// unbuffered (`Write::flush` is a documented no-op) and `close`
|
||||||
// finalizes the pending writes server-side, so no flush is needed.
|
// finalizes the pending writes server-side, so no flush is needed.
|
||||||
remote_file
|
remote_file.close().map_err(|e| {
|
||||||
.close()
|
format!("failed to close remote file '{remote}' on {host} after upload: {e}")
|
||||||
.map_err(|e| format!("failed to close remote file '{remote}' after upload: {e}"))?;
|
})?;
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
@@ -761,4 +858,89 @@ Host other
|
|||||||
apply_config_file(&mut config, content, host);
|
apply_config_file(&mut config, content, host);
|
||||||
config
|
config
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The all-addresses-failed message names the target and reports every
|
||||||
|
/// attempted address with its error. Pure: no network involved.
|
||||||
|
#[test]
|
||||||
|
fn connect_failed_message_names_target_and_attempts() {
|
||||||
|
let attempts = vec![
|
||||||
|
(
|
||||||
|
SocketAddr::new(
|
||||||
|
std::net::IpAddr::V4(std::net::Ipv4Addr::new(192, 0, 2, 10)),
|
||||||
|
22,
|
||||||
|
),
|
||||||
|
std::io::Error::new(std::io::ErrorKind::TimedOut, "connection timed out"),
|
||||||
|
),
|
||||||
|
(
|
||||||
|
SocketAddr::new(std::net::IpAddr::V6(std::net::Ipv6Addr::LOCALHOST), 22),
|
||||||
|
std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "refused"),
|
||||||
|
),
|
||||||
|
];
|
||||||
|
|
||||||
|
let message = connect_failed_message("ppa.launchpad.net", 22, &attempts);
|
||||||
|
assert!(
|
||||||
|
message.contains("cannot connect to ppa.launchpad.net:22"),
|
||||||
|
"unexpected: {message}"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
message.contains("tried 2 address(es)"),
|
||||||
|
"unexpected: {message}"
|
||||||
|
);
|
||||||
|
assert!(message.contains("192.0.2.10:22"), "unexpected: {message}");
|
||||||
|
assert!(
|
||||||
|
message.contains("connection timed out"),
|
||||||
|
"unexpected: {message}"
|
||||||
|
);
|
||||||
|
assert!(message.contains("[::1]:22"), "unexpected: {message}");
|
||||||
|
assert!(message.contains("refused"), "unexpected: {message}");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The whole `tcp_connect` failure path on a loopback port with no
|
||||||
|
/// listener (numeric literal, so no DNS; loopback only, so no external
|
||||||
|
/// network): the per-address `connect_timeout` loop produces the
|
||||||
|
/// formatted message naming the target. DNS resolution failures are
|
||||||
|
/// deliberately not tested: resolving a bogus name would touch the
|
||||||
|
/// system resolver, which is not offline-safe.
|
||||||
|
#[test]
|
||||||
|
fn tcp_connect_closed_port_fails_naming_the_target() {
|
||||||
|
// Grab a free port, then release it: connecting to the now-closed
|
||||||
|
// port fails immediately (ECONNREFUSED), never reaching the timeout
|
||||||
|
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
|
||||||
|
let addr = listener.local_addr().unwrap();
|
||||||
|
drop(listener);
|
||||||
|
|
||||||
|
let error = tcp_connect("127.0.0.1", addr.port()).unwrap_err();
|
||||||
|
assert!(
|
||||||
|
error.contains(&format!("cannot connect to 127.0.0.1:{}", addr.port())),
|
||||||
|
"unexpected: {error}"
|
||||||
|
);
|
||||||
|
assert!(error.contains("tried 1 address(es)"), "unexpected: {error}");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The libssh2 timeout wiring round-trips through the session API with
|
||||||
|
/// the configured constants (no network: a bare session suffices)
|
||||||
|
#[test]
|
||||||
|
fn session_timeouts_use_the_configured_constants() {
|
||||||
|
let session = Session::new().unwrap();
|
||||||
|
|
||||||
|
session.set_timeout(duration_ms(SSH_API_TIMEOUT));
|
||||||
|
assert_eq!(session.timeout(), duration_ms(SSH_API_TIMEOUT));
|
||||||
|
|
||||||
|
// The transfer budget replaces the connection one once authenticated
|
||||||
|
session.set_timeout(duration_ms(SSH_TRANSFER_TIMEOUT));
|
||||||
|
assert_eq!(session.timeout(), duration_ms(SSH_TRANSFER_TIMEOUT));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Millisecond conversion is exact for the constants and saturates for
|
||||||
|
/// larger values instead of truncating or panicking
|
||||||
|
#[test]
|
||||||
|
fn duration_ms_converts_and_saturates() {
|
||||||
|
assert_eq!(duration_ms(TCP_CONNECT_TIMEOUT), 15_000);
|
||||||
|
assert_eq!(duration_ms(SSH_API_TIMEOUT), 30_000);
|
||||||
|
assert_eq!(duration_ms(SSH_TRANSFER_TIMEOUT), 300_000);
|
||||||
|
assert_eq!(
|
||||||
|
duration_ms(Duration::from_millis(u64::from(u32::MAX) + 1)),
|
||||||
|
u32::MAX
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user