feat(sftp): add macOS and Windows platform support (#3372)

The session watchdog now selects its detection method per
platform. It previously probed kernel TCP state through a
Linux-only procfs path, so on macOS every healthy idle session
was killed within a minute, and on Windows the watchdog never
spawned at all, leaving wedged sessions with no cleanup. Linux
keeps its fast-kill watchdog unchanged. Other platforms get a
silence-only backstop that kills a session only at the
documented 30-minute idle ceiling.

The host-key loader now has a Windows arm. It loads OpenSSH
format host keys from the configured directory and logs a
one-time warning to restrict NTFS ACLs on the key directory,
the same operator-managed approach FTPS, WebDAV, KMS, and IAM
already use on Windows. Startup previously aborted with
UnsupportedPlatform because the Unix mode-bit permission check
has no Windows equivalent. tokio's io-uring feature is now
enabled only in Linux builds. io-uring is a Linux kernel
interface and enabling it unconditionally broke the Windows
build.

Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
escapecode
2026-06-12 11:08:23 +01:00
committed by GitHub
parent e80b72ae79
commit 7a8514bdfa
17 changed files with 513 additions and 159 deletions
+3 -3
View File
@@ -52,8 +52,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Path canonicalisation rejects paths containing `\0`, `\r`, or `\n` and resolves traversal via `path::clean()` before any backend dispatch
- Cipher / KEX / MAC / host-key algorithm allowlists are hardcoded with no environment override. Strict-KEX (CVE-2023-48795 / Terrapin) marker presence asserted by unit test
- Per-session handle cap (default 64, configurable 8 to 1024) with UUID-generated handle ids
- Crate-level `#![deny(unsafe_code)]` is in force across `crates/protocols`. Socket fd duplication for the watchdog uses the safe `AsFd::try_clone_to_owned` path (Linux/Unix); non-Unix falls back to the inactivity ceiling
- `cfg(unix)` gating around platform-specific imports (`std::os::fd::AsFd`, `std::os::unix::fs::PermissionsExt`); non-Unix targets fail SFTP at config-load with `SftpInitError::UnsupportedPlatform`
- Crate-level `#![deny(unsafe_code)]` is in force across `crates/protocols`. Socket fd duplication for the watchdog uses the safe `AsFd::try_clone_to_owned` path (Linux). Non-Linux targets use the inactivity-ceiling watchdog
- Platform-specific imports are cfg-gated. Unix enforces owner-only host-key mode bits (no group or other permission bits). Windows loads host keys without a mode check and trusts operator-managed NTFS ACLs. Targets that are neither Unix nor Windows fail SFTP at config-load with SftpInitError::UnsupportedPlatform
### Documentation
- Updated `crates/keystone/README.md` with complete integration architecture and workflow
@@ -76,7 +76,7 @@ New environment variables:
- `RUSTFS_KEYSTONE_VERIFY_SSL` - Verify SSL certificates (default: true)
- `RUSTFS_SFTP_ENABLE` - Enable/disable SFTP (default: false)
- `RUSTFS_SFTP_ADDRESS` - Listen address (default: 0.0.0.0:2222)
- `RUSTFS_SFTP_HOST_KEY_DIR` - Directory containing host key files (must exist; each file must be 0o600 or 0o400)
- `RUSTFS_SFTP_HOST_KEY_DIR` - Directory containing host key files (must exist). On Unix each file must grant no group or other permission bits (owner access only). On Windows the files load without a mode check and rustfs trusts the directory NTFS ACL
- `RUSTFS_SFTP_IDLE_TIMEOUT` - Session idle timeout in seconds (default: 600)
- `RUSTFS_SFTP_PART_SIZE` - Multipart part size in bytes (default: 16 MiB)
- `RUSTFS_SFTP_READ_ONLY` - Reject write packets at the protocol layer (default: false)
@@ -123,7 +123,6 @@ use futures::stream::{FuturesUnordered, StreamExt};
use russh::client;
use russh_sftp::client::{Config, SftpSession};
use russh_sftp::protocol::{FileAttributes, OpenFlags, StatusCode};
#[cfg(target_os = "linux")]
use rustfs_config::ENV_SFTP_IDLE_TIMEOUT;
use rustfs_config::{
ENV_CONSOLE_ENABLE, ENV_RUSTFS_ADDRESS, ENV_SFTP_ADDRESS, ENV_SFTP_ENABLE, ENV_SFTP_HOST_KEY_DIR, ENV_SFTP_PART_SIZE,
@@ -134,7 +133,6 @@ use std::path::PathBuf;
use std::pin::Pin;
use std::process::Stdio;
use std::sync::Arc;
#[cfg(target_os = "linux")]
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::{AtomicBool, Ordering};
use std::task::{Context, Poll};
@@ -143,7 +141,6 @@ use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncSeekExt, AsyncWri
use tokio::net::TcpStream;
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::process::{Child, Command};
#[cfg(target_os = "linux")]
use tokio::time::sleep;
use tokio::time::timeout;
use tracing::info;
@@ -412,13 +409,11 @@ fn capture_server_stdout(child: &mut Child) -> Arc<tokio::sync::Mutex<Vec<String
/// lifecycle cases (CMPTST-24, CMPTST-25, CMPTST-26) read both fields
/// to assert the watchdog killed silent sessions on the expected path.
#[derive(Default)]
#[cfg(target_os = "linux")]
struct SessionCounters {
entered: AtomicUsize,
finished: AtomicUsize,
}
#[cfg(target_os = "linux")]
impl SessionCounters {
fn new() -> Arc<Self> {
Arc::new(Self {
@@ -432,7 +427,6 @@ impl SessionCounters {
/// increments the matching counter for every server-side session
/// lifecycle event. The task ends when stdout closes (i.e. when the
/// child is killed at teardown).
#[cfg(target_os = "linux")]
fn watch_session_lifecycle_events(child: &mut Child, counters: Arc<SessionCounters>) {
let Some(stdout) = child.stdout.take() else {
return;
@@ -2093,7 +2087,6 @@ pub(crate) mod cmptst_25 {
}
// CMPTST-26: healthy idle session past the watchdog fast-kill threshold stays alive.
#[cfg(target_os = "linux")]
pub(crate) mod cmptst_26 {
use super::*;
@@ -2108,20 +2101,21 @@ pub(crate) mod cmptst_26 {
// Idle timeout for the spawned server. 300 s sits well above the
// wait window so russh's own inactivity_timeout cannot kill during
// the test. The contract: a healthy idle session past the
// watchdog's fast-kill threshold MUST stay alive because the procfs
// probe sees ESTABLISHED and the decision function returns
// Decision::Quiet.
// watchdog's fast-kill threshold MUST stay alive. On Linux the
// procfs probe sees ESTABLISHED and the decision function returns
// Decision::Quiet. On non-Linux targets the fallback watchdog only
// cancels at the far larger fallback ceiling, so the session
// survives the wait window.
const IDLE_TIMEOUT_SECS: u64 = 300;
// Wait window. Must exceed
// WEDGE_FAST_KILL_SILENCE_SECS (30) + WEDGE_WATCHDOG_TICK_SECS (15)
// = 45 s of worst-case watchdog detection latency, plus a 15 s
// grace. Sits well below WEDGE_FALLBACK_KILL_SILENCE_SECS (1800)
// so the fallback path does not fire either.
// 90 s instead of 60 s. The case asserts the watchdog does NOT
// false-kill, so a longer wait strengthens the assertion: if the
// procfs ESTABLISHED discriminator is broken, more wait windows
// give it more chances to fire.
// Wait window. Must exceed the worst-case watchdog detection
// latency on Linux: WEDGE_FAST_KILL_SILENCE_SECS (30) plus one to
// two WEDGE_WATCHDOG_TICK_SECS ticks (15 each), so 45 to 60 s. Sits
// well below WEDGE_FALLBACK_KILL_SILENCE_SECS (1800) so the fallback
// path does not fire either. 90 s instead of 60 s: the case asserts
// the watchdog does NOT false-kill, so a longer wait strengthens the
// assertion. On Linux, if the procfs ESTABLISHED discriminator is
// broken, more wait windows give it more chances to fire.
const IDLE_WAIT_SECS: u64 = 90;
pub(crate) async fn run_healthy_idle_session_above_fast_threshold() -> Result<()> {
@@ -2229,7 +2223,10 @@ pub(crate) mod cmptst_26 {
result
}
#[cfg(target_os = "linux")]
// Excluded on Windows. Ambient operating-system traffic can inflate
// the session counter and false-positive this healthy-idle assertion.
// Linux and macOS run it.
#[cfg(not(target_os = "windows"))]
#[tokio::test]
async fn regression() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
crate::common::init_logging();
+5 -1
View File
@@ -96,7 +96,7 @@ hyper-util.workspace = true
hyper-rustls.workspace = true
rustls.workspace = true
rustls-pki-types.workspace = true
tokio = { workspace = true, features = ["io-util", "sync", "signal","io-uring"] }
tokio = { workspace = true, features = ["io-util", "sync", "signal"] }
tonic.workspace = true
xxhash-rust = { workspace = true, features = ["xxh64", "xxh3"] }
tower.workspace = true
@@ -162,3 +162,7 @@ harness = false
[lib]
doctest = false
# io-uring is Linux-only. Scope the feature to Linux targets.
[target.'cfg(target_os = "linux")'.dependencies]
tokio = { workspace = true, features = ["io-uring"] }
+5 -1
View File
@@ -30,10 +30,14 @@ workspace = true
[dependencies]
bytes = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true, features = ["io-util", "fs", "rt", "sync","io-uring"] }
tokio = { workspace = true, features = ["io-util", "fs", "rt", "sync"] }
memmap2 = { workspace = true }
rustfs-io-metrics = { workspace = true }
# io-uring is Linux-only. Scope the feature to Linux targets.
[target.'cfg(target_os = "linux")'.dependencies]
tokio = { workspace = true, features = ["io-uring"] }
[dev-dependencies]
tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+5 -1
View File
@@ -30,7 +30,7 @@ workspace = true
[dependencies]
# Core dependencies
async-trait = { workspace = true }
tokio = { workspace = true, features = ["full","io-uring"] }
tokio = { workspace = true, features = ["full"] }
uuid = { workspace = true, features = ["serde"] }
jiff = { workspace = true }
serde = { workspace = true, features = ["derive"] }
@@ -69,3 +69,7 @@ temp-env = { workspace = true }
[features]
default = []
# io-uring is Linux-only. Scope the feature to Linux targets.
[target.'cfg(target_os = "linux")'.dependencies]
tokio = { workspace = true, features = ["io-uring"] }
+6 -1
View File
@@ -67,7 +67,7 @@ rustfs-storage-api = { workspace = true }
rustfs-tls-runtime = { workspace = true, optional = true }
# Async dependencies
tokio = { workspace = true, features = ["fs", "io-util", "sync", "time","io-uring"] }
tokio = { workspace = true, features = ["fs", "io-util", "sync", "time"] }
tracing = { workspace = true }
futures-util = { workspace = true }
@@ -131,7 +131,12 @@ socket2 = { workspace = true, optional = true }
tempfile = { workspace = true }
proptest = "1"
tracing-subscriber = { workspace = true }
tokio = { workspace = true, features = ["test-util", "macros", "rt"] }
[package.metadata.docs.rs]
all-features = true
rustdoc-args = ["--cfg", "docsrs"]
# io-uring is Linux-only. Scope the feature to Linux targets.
[target.'cfg(target_os = "linux")'.dependencies]
tokio = { workspace = true, features = ["io-uring"] }
+88 -43
View File
@@ -28,7 +28,7 @@ use super::constants::limits::{
READ_CACHE_TOTAL_MEM_MIN, READ_CACHE_WINDOW_DEFAULT, READ_CACHE_WINDOW_MAX, READ_CACHE_WINDOW_MIN, S3_MAX_PART_SIZE,
S3_MIN_PART_SIZE,
};
#[cfg(unix)]
#[cfg(any(unix, windows))]
use russh::keys::PublicKeyBase64;
use std::net::SocketAddr;
#[cfg(unix)]
@@ -39,7 +39,6 @@ use thiserror::Error;
/// Upper bound on file size accepted as a candidate host key (1 MiB).
/// Guards against accidentally reading huge non-key files in the host
/// key directory. Real keys are well under 10 KiB.
#[cfg(unix)]
const MAX_HOST_KEY_FILE_SIZE: u64 = 1024 * 1024;
/// PEM pre-encapsulation boundary marker prefix per RFC 7468 section 3.
@@ -47,7 +46,6 @@ const MAX_HOST_KEY_FILE_SIZE: u64 = 1024 * 1024;
/// space, the label, and five more hyphens. Used to distinguish a file
/// that looks like a private key but failed to decode (passphrase, corrupt)
/// from a file that is genuinely something else (a .pub key, a README).
#[cfg(unix)]
const PEM_BEGIN_MARKER: &str = "-----BEGIN";
/// Errors that can occur during SFTP server initialization.
@@ -64,15 +62,15 @@ pub enum SftpInitError {
#[error("host key directory does not exist or is not readable: {path}: {source}")]
HostKeyDirUnreadable { path: PathBuf, source: std::io::Error },
/// A host-key file in the directory has world-readable or
/// group-readable bits set. Mode must be 0o600 or 0o400 so a
/// A host-key file in the directory has group or other permission
/// bits set. The mode must grant no access beyond the owner so a
/// local non-root user cannot impersonate the SFTP server.
#[error("host key file has insecure permissions {mode:#o}: {path} (must be 0o600 or 0o400)")]
#[error("host key file has insecure permissions {mode:#o}: {path} (group and other permission bits must be unset)")]
InsecureHostKeyPermissions { path: PathBuf, mode: u32 },
/// The host-key directory contained no decodable private keys.
/// Operators must place at least one ed25519 / ECDSA / RSA-SHA256
/// private key with mode 0o600 in the directory before startup.
/// private key with owner-only permissions in the directory before startup.
#[error("no valid host keys found in {path}")]
NoHostKeysFound { path: PathBuf },
@@ -88,7 +86,7 @@ pub enum SftpInitError {
Server(String),
/// The host running the binary is not a Unix-family target. The
/// host-key permission enforcement (mode 0o600 / 0o400 check)
/// host-key permission enforcement (the owner-only mode-bit check)
/// requires Unix mode bits and has no equivalent on this platform,
/// so SFTP refuses to start rather than load host keys with weaker
/// guarantees.
@@ -196,7 +194,7 @@ impl SftpConfig {
/// None passes through unchanged. Some(n) is returned unchanged
/// when n is in the inclusive range
/// HANDLES_PER_SESSION_MIN..=HANDLES_PER_SESSION_MAX. Out-of-range
/// inputs return None and emit a warn log naming the requested
/// inputs return None and log a warning naming the requested
/// value and the bounds. The driver applies
/// DEFAULT_HANDLES_PER_SESSION when the value is None.
pub fn resolve_handles_per_session(raw: Option<usize>) -> Option<usize> {
@@ -220,7 +218,7 @@ impl SftpConfig {
/// read. None passes through unchanged. Some(n) is returned
/// unchanged when n is in the inclusive range
/// BACKEND_OP_TIMEOUT_MIN_SECS..=BACKEND_OP_TIMEOUT_MAX_SECS.
/// Out-of-range inputs return None and emit a warn log naming the
/// Out-of-range inputs return None and log a warning naming the
/// requested value and the bounds. The driver applies
/// DEFAULT_BACKEND_OP_TIMEOUT_SECS when the value is None.
pub fn resolve_backend_op_timeout_secs(raw: Option<u64>) -> Option<u64> {
@@ -246,7 +244,7 @@ impl SftpConfig {
/// populate path so reads do not retain any buffer between
/// FXP_READs. Some(n) where n is in the inclusive range
/// READ_CACHE_WINDOW_MIN..=READ_CACHE_WINDOW_MAX is returned
/// unchanged. Other values return None and emit a warn log
/// unchanged. Other values return None and log a warning
/// naming the requested value and the bounds. The driver applies
/// READ_CACHE_WINDOW_DEFAULT when the value is None.
pub fn resolve_read_cache_window_bytes(raw: Option<u64>) -> Option<u64> {
@@ -270,7 +268,7 @@ impl SftpConfig {
/// Resolve the read_cache_total_mem_bytes value from a raw env-var
/// read. None passes through unchanged. Some(n) is returned
/// unchanged when n is at or above READ_CACHE_TOTAL_MEM_MIN.
/// Below-min inputs return None and emit a warn log naming the
/// Below-min inputs return None and log a warning naming the
/// requested value and the bound. The driver applies
/// READ_CACHE_TOTAL_MEM_DEFAULT when the value is None.
pub fn resolve_read_cache_total_mem_bytes(raw: Option<u64>) -> Option<u64> {
@@ -310,19 +308,16 @@ impl SftpConfig {
/// matches the secret handling in the existing S3 and FTPS auth
/// paths.
///
/// Returns SftpInitError::UnsupportedPlatform when built for a
/// non-Unix target. The mode-bit permission enforcement has no
/// portable equivalent off Unix, and starting SFTP without it
/// would silently weaken host-key protection.
#[cfg(not(unix))]
pub async fn load_host_keys(_host_key_dir: &Path) -> Result<Vec<russh::keys::PrivateKey>, SftpInitError> {
Err(SftpInitError::UnsupportedPlatform {
os: std::env::consts::OS.to_string(),
})
}
#[cfg(unix)]
/// On Unix the loader enforces owner-only mode bits (group and other
/// must have no permission). On Windows it loads without a mode check
/// and trusts operator-managed NTFS ACLs, matching the convention the
/// other secret-bearing subsystems use. Targets that are neither Unix
/// nor Windows return SftpInitError::UnsupportedPlatform.
#[cfg(any(unix, windows))]
pub async fn load_host_keys(host_key_dir: &Path) -> Result<Vec<russh::keys::PrivateKey>, SftpInitError> {
#[cfg(windows)]
warn_once_windows(host_key_dir);
let mut entries = tokio::fs::read_dir(host_key_dir)
.await
.map_err(|e| SftpInitError::HostKeyDirUnreadable {
@@ -365,13 +360,10 @@ impl SftpConfig {
continue;
}
// Permission check: hard error on insecure permissions.
// A world-readable private key lets any local user impersonate
// the SFTP server. OpenSSH enforces the same restriction.
let mode = metadata.permissions().mode() & 0o777;
if mode & 0o077 != 0 {
return Err(SftpInitError::InsecureHostKeyPermissions { path, mode });
}
// A world-readable private key lets any local user impersonate the
// SFTP server, the same restriction OpenSSH enforces. Platform
// behaviour is documented on check_host_key_permissions.
check_host_key_permissions(&path, &metadata)?;
let data = match tokio::fs::read_to_string(&path).await {
Ok(d) => d,
@@ -455,6 +447,55 @@ impl SftpConfig {
Ok(keys)
}
/// Targets that are neither Unix nor Windows are unsupported. Returns
/// SftpInitError::UnsupportedPlatform without reading the directory.
#[cfg(not(any(unix, windows)))]
pub async fn load_host_keys(_host_key_dir: &Path) -> Result<Vec<russh::keys::PrivateKey>, SftpInitError> {
Err(SftpInitError::UnsupportedPlatform {
os: std::env::consts::OS.to_string(),
})
}
}
/// Log the Windows operator-facing host-key warning at most once per
/// process. rustfs trusts operator-managed NTFS ACLs on Windows, the same
/// convention the other secret-bearing subsystems use.
#[cfg(windows)]
fn warn_once_windows(host_key_dir: &Path) {
use std::sync::Once;
static WARN: Once = Once::new();
WARN.call_once(|| {
tracing::warn!(
target: "rustfs_protocols::sftp::host_key",
host_key_dir = %host_key_dir.display(),
"SFTP host key file permission enforcement is not active on \
Windows. rustfs trusts operator-managed NTFS ACLs on the \
host-key directory. Restrict read access to the running rustfs \
service account, NT AUTHORITY\\SYSTEM, and \
BUILTIN\\Administrators. Default ProgramData inheritance grants \
BUILTIN\\Users read access."
);
});
}
/// Per-file host-key permission check. Unix enforces owner-only mode bits
/// (group and other must have no permission). Windows performs no check and
/// trusts operator-managed NTFS ACLs.
#[cfg(any(unix, windows))]
#[cfg_attr(not(unix), allow(unused_variables))]
fn check_host_key_permissions(path: &Path, metadata: &std::fs::Metadata) -> Result<(), SftpInitError> {
#[cfg(unix)]
{
let mode = metadata.permissions().mode() & 0o777;
if mode & 0o077 != 0 {
return Err(SftpInitError::InsecureHostKeyPermissions {
path: path.to_path_buf(),
mode,
});
}
}
Ok(())
}
#[cfg(test)]
@@ -466,23 +507,23 @@ mod tests {
// PEM boundary markers (RFC 7468 five-hyphen / BEGIN-or-END /
// label / five-hyphen) are composed at runtime by build_pem_block
// so the source file emits no contiguous private-key marker that
// so the source file contains no contiguous private-key marker that
// secret scanners would flag. Throwaway test-vector keys.
#[cfg(unix)]
#[cfg(any(unix, windows))]
const PEM_BOUNDARY_DASHES: &str = "-----";
#[cfg(unix)]
#[cfg(any(unix, windows))]
const PEM_OPENSSH_LABEL: &str = "OPENSSH PRIVATE KEY";
/// Wrap a base64 body in the OpenSSH-format PEM boundary markers.
/// The boundary string is composed at runtime from PEM_BOUNDARY_DASHES
/// and PEM_OPENSSH_LABEL so the source file does not contain the full
/// marker as a contiguous literal.
#[cfg(unix)]
#[cfg(any(unix, windows))]
fn build_pem_block(body: &str) -> String {
format!("{d}BEGIN {l}{d}\n{body}\n{d}END {l}{d}\n", d = PEM_BOUNDARY_DASHES, l = PEM_OPENSSH_LABEL,)
}
#[cfg(unix)]
#[cfg(any(unix, windows))]
fn test_ed25519_pem() -> String {
// Throwaway Ed25519 private key, no passphrase.
build_pem_block(
@@ -609,7 +650,7 @@ mod tests {
}
#[tokio::test]
#[cfg(unix)]
#[cfg(any(unix, windows))]
async fn load_host_keys_fails_when_dir_missing() {
let path = PathBuf::from("/this/path/does/not/exist/sftp-host-keys");
let err = SftpConfig::load_host_keys(&path).await.expect_err("missing dir must error");
@@ -617,7 +658,7 @@ mod tests {
}
#[tokio::test]
#[cfg(unix)]
#[cfg(any(unix, windows))]
async fn load_host_keys_fails_when_dir_empty() {
let dir = TempDir::new().expect("tempdir");
let err = SftpConfig::load_host_keys(dir.path())
@@ -627,13 +668,17 @@ mod tests {
}
#[tokio::test]
#[cfg(not(unix))]
async fn load_host_keys_rejects_non_unix_platform() {
#[cfg(windows)]
async fn load_host_keys_loads_valid_ed25519_on_windows() {
let dir = TempDir::new().expect("tempdir");
let err = SftpConfig::load_host_keys(dir.path())
let key_path = dir.path().join("ssh_host_ed25519_key");
// Windows does not enforce mode bits. The host key loads
// regardless of the inherited tempdir ACL.
std::fs::write(&key_path, test_ed25519_pem()).expect("write key");
let keys = SftpConfig::load_host_keys(dir.path())
.await
.expect_err("non-Unix platforms must be rejected");
assert!(matches!(err, SftpInitError::UnsupportedPlatform { .. }));
.expect("valid ed25519 key must load on Windows");
assert_eq!(keys.len(), 1);
}
#[tokio::test]
+20 -16
View File
@@ -40,14 +40,14 @@ pub mod s3_error_codes {
/// AWS S3 error code returned by HeadBucket when the bucket does
/// not exist.
pub const NO_SUCH_BUCKET: &str = "NoSuchBucket";
/// Generic "not found" string emitted by S3-compatible backends
/// Generic "not found" string returned by S3-compatible backends
/// (MinIO, Wasabi, ecstore) that do not always use the AWS
/// NoSuchKey / NoSuchBucket vocabulary on every miss.
pub const NOT_FOUND: &str = "NotFound";
/// AWS error code returned when an IAM policy denies the requested
/// action on the resource.
pub const ACCESS_DENIED: &str = "AccessDenied";
/// Generic forbidden string emitted by S3-compatible backends that
/// Generic forbidden string returned by S3-compatible backends that
/// do not always use the AWS AccessDenied vocabulary.
pub const FORBIDDEN: &str = "Forbidden";
/// Returned by AbortMultipartUpload when the upload_id is no
@@ -89,7 +89,7 @@ pub mod posix {
/// POSIX file-type mask (S_IFMT). Isolates the four high bits of a
/// mode value so the file-type field can be compared against
/// S_IFDIR, S_IFREG, S_IFLNK, and the other POSIX type constants.
/// Compiled in test builds only; the runtime path reads the full
/// Compiled in test builds only. The runtime path reads the full
/// mode from POSIX_DIR_MODE / POSIX_FILE_MODE.
#[cfg(test)]
pub const POSIX_TYPE_MASK: u32 = 0o170000;
@@ -177,29 +177,33 @@ pub mod limits {
/// inside the post-handshake session loop.
pub const HANDSHAKE_DEADLINE_SECS: u64 = 30;
/// Tick interval for the per-session wedge watchdog. Worst-case
/// detection latency is WEDGE_FAST_KILL_SILENCE_SECS + one tick.
/// Tick interval for the per-session watchdog on every platform.
/// On Linux worst-case detection latency is
/// WEDGE_FAST_KILL_SILENCE_SECS plus one to two ticks. On non-Linux
/// targets the fallback watchdog uses the same tick to check silence
/// against WEDGE_FALLBACK_KILL_SILENCE_SECS.
pub const WEDGE_WATCHDOG_TICK_SECS: u64 = 15;
/// Silence threshold at which a session whose underlying TCP socket
/// is in CLOSE_WAIT is force-cancelled by the watchdog.
///
/// A healthy session is never simultaneously silent at the SFTP
/// handler AND in CLOSE_WAIT: peer FIN normally surfaces as Ok(0)
/// handler AND in CLOSE_WAIT: peer FIN normally appears as Ok(0)
/// on the SSH library read poll within milliseconds. 30 s leaves
/// room for two keepalive intervals (15 s each) before the
/// watchdog overrides, so a transient scheduler stall does not
/// trip it.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub const WEDGE_FAST_KILL_SILENCE_SECS: u64 = 30;
/// Fallback silence threshold. The only kill path on non-Linux
/// targets, where /proc/net/tcp is unavailable and the watchdog's
/// CLOSE_WAIT probe always returns None. On Linux it is the
/// backstop for cases where /proc/net/tcp is unreadable for some
/// other reason (filesystem permissions, namespace tricks) or
/// where the wedge surfaces in a state other than CLOSE_WAIT.
/// 1800 s sits above russh's default inactivity_timeout (600 s)
/// so russh's own inactivity close fires first on a healthy idle session.
/// Fallback silence threshold. On Linux it is the backstop for
/// cases where /proc/net/tcp is unreadable (filesystem permissions,
/// namespace tricks) or where the wedge is in a state other than
/// CLOSE_WAIT. On non-Linux targets it is the only kill path from
/// the fallback watchdog module, because /proc/net/tcp does not
/// exist and the CLOSE_WAIT probe is unavailable. 1800 s sits above
/// russh's default inactivity_timeout (600 s) so russh's own
/// inactivity close fires first on a healthy idle session.
pub const WEDGE_FALLBACK_KILL_SILENCE_SECS: u64 = 1800;
// The three constants below override russh defaults for the SSH
@@ -250,7 +254,7 @@ pub mod limits {
/// carrying a body larger than this is rejected with EntityTooLarge.
/// Mirrors the MAX_PART_SIZE constant in ecstore but cannot be
/// imported from there. AWS sets S3_COPY_OBJECT_MAX_SIZE and
/// S3_MAX_PART_SIZE independently to 5 GiB; the values are not
/// S3_MAX_PART_SIZE independently to 5 GiB. The values are not
/// coupled. Future S3 versions could move them apart, so they
/// remain separate constants.
pub const S3_MAX_PART_SIZE: u64 = 5 * 1024 * 1024 * 1024;
@@ -289,7 +293,7 @@ pub mod limits {
/// Default per-call deadline applied to every StorageBackend
/// invocation issued by the SFTP driver. A backend that does not
/// respond within this many seconds returns Failure to the client
/// and emits a warn log naming the backend method. Used when
/// and logs a warning naming the backend method. Used when
/// RUSTFS_SFTP_BACKEND_OP_TIMEOUT_SECS is unset or out of range.
/// The keepalive timer (KEEPALIVE_INTERVAL_SECS times KEEPALIVE_MAX,
/// approximately 45 s) closes a stuck SSH transport but cannot detect
@@ -0,0 +1,170 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Silence-only liveness backstop for non-Linux targets.
//!
//! Non-Linux targets cannot read /proc/net/tcp, so the CLOSE_WAIT
//! probe used by the Linux watchdog is unavailable. fallback_watchdog runs
//! one tokio task per session that cancels the session once it has
//! been silent at the SFTP handler layer at or past
//! WEDGE_FALLBACK_KILL_SILENCE_SECS. There is no socket introspection
//! and no fast-kill path. A session silent below that threshold keeps
//! running, so healthy idle sessions are left alone.
use super::constants::limits::{WEDGE_FALLBACK_KILL_SILENCE_SECS, WEDGE_WATCHDOG_TICK_SECS};
use super::lifecycle::SessionDiag;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio_util::sync::CancellationToken;
/// Spawn a per-session silence backstop tick task.
///
/// The task holds a clone of the per-session CancellationToken and an
/// Arc to the SessionDiag it reads the activity stamp from. It exits
/// when it cancels the session itself or when the outer session task
/// cancels the token after a clean session end.
pub(super) fn spawn_for_session(session_diag: Arc<SessionDiag>, cancel_token: CancellationToken) {
tokio::spawn(async move {
let session_id = session_diag.session_id;
let peer = session_diag.peer;
let mut tick = tokio::time::interval(Duration::from_secs(WEDGE_WATCHDOG_TICK_SECS));
tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
// First tick fires immediately. Skip it so the backstop never
// makes a decision before one full silence window has elapsed.
tick.tick().await;
loop {
tokio::select! {
_ = cancel_token.cancelled() => break,
_ = tick.tick() => {
let silence_secs = silence_secs(&session_diag);
if fallback_threshold_reached(silence_secs) {
tracing::warn!(
target: "rustfs_protocols::sftp::watchdog",
session_id,
peer = %peer,
silence_secs,
reason = "fallback_silence",
"fallback watchdog cancelling session: silence exceeded fallback threshold",
);
cancel_token.cancel();
break;
}
}
}
}
});
}
/// Seconds since the session last stamped activity at the SFTP handler
/// layer.
fn silence_secs(session_diag: &SessionDiag) -> u64 {
let now_ms = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
let last_ms = session_diag.last_activity_ms.load(Ordering::Relaxed);
now_ms.saturating_sub(last_ms) / 1000
}
/// True once silence has reached WEDGE_FALLBACK_KILL_SILENCE_SECS, the
/// point at which the fallback backstop cancels the session.
fn fallback_threshold_reached(silence_secs: u64) -> bool {
silence_secs >= WEDGE_FALLBACK_KILL_SILENCE_SECS
}
#[cfg(test)]
mod tests {
use super::*;
use std::net::SocketAddr;
fn loopback() -> SocketAddr {
"127.0.0.1:2222".parse().expect("static loopback address parses")
}
fn now_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock is after the unix epoch")
.as_millis() as u64
}
#[test]
fn threshold_reached_at_and_above_fallback() {
assert!(fallback_threshold_reached(WEDGE_FALLBACK_KILL_SILENCE_SECS));
assert!(fallback_threshold_reached(WEDGE_FALLBACK_KILL_SILENCE_SECS + 1));
}
#[test]
fn threshold_not_reached_below_fallback() {
assert!(!fallback_threshold_reached(WEDGE_FALLBACK_KILL_SILENCE_SECS - 1));
assert!(!fallback_threshold_reached(0));
}
#[tokio::test(start_paused = true)]
async fn cancels_session_silent_at_or_past_fallback() {
let session_diag = Arc::new(SessionDiag::new(loopback(), loopback()));
// Stamp activity far enough in the past that the first decision
// tick already sees silence at or past the fallback threshold.
// silence_secs reads the wall clock, which the paused tokio
// timer does not move, so the stamp stays stale while virtual
// time auto-advances the tick interval.
let stale_ms = now_ms() - (WEDGE_FALLBACK_KILL_SILENCE_SECS + 60) * 1000;
session_diag.last_activity_ms.store(stale_ms, Ordering::Relaxed);
let cancel_token = CancellationToken::new();
spawn_for_session(session_diag, cancel_token.clone());
tokio::time::timeout(Duration::from_secs(WEDGE_WATCHDOG_TICK_SECS * 3), cancel_token.cancelled())
.await
.expect("a session silent at or past the fallback threshold must be cancelled");
assert!(cancel_token.is_cancelled());
}
#[tokio::test]
async fn cancel_token_short_circuits_cleanly() {
let session_diag = Arc::new(SessionDiag::new(loopback(), loopback()));
session_diag.last_activity_ms.store(now_ms(), Ordering::Relaxed);
let observer = Arc::downgrade(&session_diag);
let cancel_token = CancellationToken::new();
// Cancel before the task observes its first decision tick. The
// cancelled arm must win and the task must drop its Arc on the
// way out rather than waiting a full tick interval.
cancel_token.cancel();
spawn_for_session(session_diag, cancel_token.clone());
// Sleep rather than yield between checks so the runtime parks
// and its time driver fires the interval's immediate first
// tick. A busy yield loop would starve that timer and the task
// would stay parked before it reaches the cancelled arm.
for _ in 0..200 {
if observer.strong_count() == 0 {
return;
}
tokio::time::sleep(Duration::from_millis(5)).await;
}
panic!("task must release its SessionDiag after a clean cancel");
}
#[test]
fn silence_secs_grows_from_a_stale_stamp() {
let session_diag = SessionDiag::new(loopback(), loopback());
session_diag.last_activity_ms.store(now_ms() - 5_000, Ordering::Relaxed);
// Lower bound only. The exact value depends on wall-clock elapsed
// during the test, which a contended runner can stretch past a
// fixed upper bound.
assert!(silence_secs(&session_diag) >= 5);
}
}
+22 -9
View File
@@ -16,8 +16,9 @@
//!
//! Holds the per-session activity stamp and the weak-ref registry the
//! accept loop walks. Both are load-bearing infrastructure for the
//! per-session wedge watchdog (wedge_watchdog.rs): the watchdog uses
//! the activity stamp to decide whether a session is silent, and the
//! per-session liveness watchdog (wedge_watchdog.rs on Linux,
//! fallback_watchdog.rs elsewhere): the watchdog uses the activity
//! stamp to decide whether a session is silent, and the
//! TCP-state probe to disambiguate slow operations from CLOSE_WAIT.
//!
//! Activity stamps are written from every SFTP handler entry/exit and
@@ -26,10 +27,11 @@
//!
//! The TCP-state probe parses /proc/net/tcp and /proc/net/tcp6, looks
//! up the row matching the (local, peer) tuple, and returns the kernel
//! TCP state. Only Linux exposes the procfs files. On other targets
//! the probe returns None and the watchdog falls back to its absolute
//! silence threshold. Live ports are hex'd in the kernel's
//! per-architecture byte order (little-endian within each 4-byte chunk).
//! TCP state. Only Linux exposes the procfs files. On non-Linux
//! targets the probe is not called. The session watchdog
//! (fallback_watchdog) decides from the SFTP-handler activity stamp
//! alone. Live ports are hex'd in the kernel's per-architecture byte
//! order (little-endian within each 4-byte chunk).
use std::fmt::Write as _;
use std::net::{IpAddr, SocketAddr};
@@ -45,29 +47,36 @@ use std::time::{Instant, SystemTime, UNIX_EPOCH};
/// Length of an IPv6 address in bytes.
const IPV6_BYTES: usize = 16;
/// Length of an IPv4 address in bytes.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const IPV4_BYTES: usize = 4;
/// Hex characters used to render one byte in the procfs format
/// (matches the {:02X} format spec at the call sites).
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const HEX_CHARS_PER_BYTE: usize = 2;
/// Hex characters used to render the 16-bit port in the procfs format
/// (matches the {:04X} format spec at the call sites).
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const PORT_HEX_CHARS: usize = 4;
/// Number of bytes per chunk in the IPv6 procfs format. Bytes inside
/// each chunk are emitted in reverse (little-endian within the chunk).
/// each chunk are ordered in reverse (little-endian within the chunk).
const TCP6_CHUNK_BYTES: usize = 4;
/// Number of 4-byte chunks the IPv6 procfs format renders. The
/// const_assert below pins this against IPV6_BYTES so any future drift
/// surfaces at compile time.
/// appears at compile time.
const TCP6_CHUNK_COUNT: usize = IPV6_BYTES / TCP6_CHUNK_BYTES;
const _: () = assert!(TCP6_CHUNK_COUNT * TCP6_CHUNK_BYTES == IPV6_BYTES);
/// First line of /proc/net/tcp[6] is the column header. Data rows
/// follow.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const PROC_NET_TCP_HEADER_LINES: usize = 1;
/// Linux TCP_ESTABLISHED state value (include/uapi/linux/tcp.h).
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const TCP_STATE_ESTABLISHED: u8 = 0x01;
/// Linux TCP_CLOSE_WAIT state value (include/uapi/linux/tcp.h).
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const TCP_STATE_CLOSE_WAIT: u8 = 0x08;
/// Procfs renders the TCP state as a hexadecimal byte.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
const TCP_STATE_RADIX: u32 = 16;
/// Per-session activity record. Constructed once per accepted SSH
@@ -118,6 +127,7 @@ pub(super) fn new_session_registry() -> SessionRegistry {
/// Kernel TCP state for one connection, as reported by /proc/net/tcp[6].
/// Values follow the Linux TCP state numbering used in the procfs files.
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(super) enum TcpState {
/// 0x01. Connection is open and exchanging data.
Established,
@@ -141,6 +151,7 @@ pub(super) enum TcpState {
/// connection has been finalised by the kernel and removed from the
/// table, or one or both addresses do not have a renderable form
/// for the relevant procfs file.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub(super) fn probe_tcp_state(local: SocketAddr, peer: SocketAddr) -> Option<TcpState> {
if let Ok(content) = std::fs::read_to_string("/proc/net/tcp")
&& let Some(state) = lookup_tcp_state(&content, local, peer, false)
@@ -159,6 +170,7 @@ pub(super) fn probe_tcp_state(local: SocketAddr, peer: SocketAddr) -> Option<Tcp
/// ipv6_file flag selects the address-rendering convention. tcp6
/// uses 32-character hex strings and tcp uses 8-character, both with
/// little-endian byte order within each 4-byte chunk.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn lookup_tcp_state(content: &str, local: SocketAddr, peer: SocketAddr, ipv6_file: bool) -> Option<TcpState> {
let local_hex = render_proc_net_tcp_addr(local, ipv6_file)?;
let peer_hex = render_proc_net_tcp_addr(peer, ipv6_file)?;
@@ -198,6 +210,7 @@ fn lookup_tcp_state(content: &str, local: SocketAddr, peer: SocketAddr, ipv6_fil
/// IPv4 SocketAddrs presented to tcp6 are mapped via ::ffff:a.b.c.d
/// before rendering. IPv4-mapped IPv6 SocketAddrs presented to tcp
/// are unwrapped before rendering. Mismatches return None.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
fn render_proc_net_tcp_addr(addr: SocketAddr, ipv6_file: bool) -> Option<String> {
// Rendered length: address bytes encoded as 2 hex chars each + ':'
// separator + 4 hex port digits. Same shape for tcp and tcp6;
@@ -283,7 +296,7 @@ mod tests {
fn render_distinct_ipv6_bytes_for_tcp6_file() {
// Bytes 00..0F, one distinct value per octet, exercise every
// index in the chunk-and-reverse loop. Each 4-byte chunk is
// emitted little-endian-within-chunk, so chunk 0 (bytes
// ordered little-endian-within-chunk, so chunk 0 (bytes
// 00 01 02 03) renders as "03020100" and so on through chunk 3.
let addr = SocketAddr::V6(SocketAddrV6::new(
Ipv6Addr::from([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0xA, 0xB, 0xC, 0xD, 0xE, 0xF]),
+15 -10
View File
@@ -31,8 +31,11 @@
//! or more S3 calls on the supplied storage backend.
//! - lifecycle: per-session activity record, the registry the accept loop
//! walks, and the kernel TCP-state probe used by the watchdog.
//! - wedge_watchdog: per-session liveness watchdog that observes both the
//! SFTP-handler activity stamp and the TCP socket state.
//! - wedge_watchdog and fallback_watchdog: the per-session liveness
//! watchdog. On target_os = "linux" wedge_watchdog observes both the
//! SFTP-handler activity stamp and the TCP socket state. On other
//! targets fallback_watchdog provides a silence-only backstop without
//! the TCP-state probe.
//! - read_cache: per-handle in-memory read-ahead cache with a process-wide
//! memory ceiling.
//!
@@ -49,14 +52,13 @@
//! and read throughput:
//!
//! - Session-liveness watchdog. Every accepted connection runs under a
//! per-session watchdog that observes the SFTP-handler activity stamp
//! and the kernel TCP state for the connection. Sessions that fall
//! silent at the SFTP layer while the kernel reports CLOSE_WAIT are
//! canceled on a bounded schedule. The watchdog backstops resource
//! accumulation regardless of which layer stalled. On Linux the
//! detection latency is on the order of 45 seconds; on non-Linux
//! targets the watchdog falls back to an inactivity ceiling on the
//! order of 30 minutes.
//! per-session watchdog. On Linux the watchdog observes both the
//! SFTP-handler activity stamp and the kernel TCP state for the
//! connection. Sessions silent at the SFTP layer while the kernel
//! reports CLOSE_WAIT are cancelled in approximately 45 to 60
//! seconds. On non-Linux targets the watchdog observes only the
//! activity stamp and cancels at an inactivity ceiling on the order
//! of 30 minutes.
//!
//! - Per-handle read cache. Each open File handle holds an in-memory
//! buffer. On a cache miss the driver fetches a configurable byte
@@ -95,11 +97,14 @@ mod attrs;
mod dir;
mod driver;
mod errors;
#[cfg(not(target_os = "linux"))]
mod fallback_watchdog;
mod lifecycle;
mod paths;
mod read;
mod read_cache;
mod state;
#[cfg(target_os = "linux")]
mod wedge_watchdog;
mod write;
+51 -27
View File
@@ -28,7 +28,10 @@ use super::constants::limits::{
SSH_MAXIMUM_PACKET_SIZE,
};
use super::constants::protocol::SFTP_SUBSYSTEM_NAME;
#[cfg(not(target_os = "linux"))]
use super::fallback_watchdog;
use super::lifecycle::{SessionDiag, SessionRegistry, new_session_registry};
#[cfg(target_os = "linux")]
use super::wedge_watchdog;
use crate::common::client::s3::StorageBackend;
use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext};
@@ -281,7 +284,7 @@ pub struct SftpServer<S: StorageBackend> {
ssh_config: Arc<SshConfigHolder>,
storage: S,
/// Weak refs to live per-session activity records. Walked by the
/// per-session wedge watchdog and by external observers that
/// per-session liveness watchdog and by external observers that
/// enumerate live sessions.
session_registry: Arc<SessionRegistry>,
/// Process-wide accumulator of live read cache memory in bytes,
@@ -355,12 +358,14 @@ where
let mut sessions: JoinSet<()> = JoinSet::new();
// Parent cancellation token for the lifetime of this listener.
// Each per-session cancel_token is a child via child_token(),
// and the wedge watchdog selects on the same child. On
// and the per-session watchdog selects on the same child. On
// shutdown_rx fire below this token is cancelled before the
// accept loop breaks, which cascades to every live session
// and every watchdog: the watchdog tasks shut their dup'd
// sockets so russh's inner tasks unblock at the next read,
// and the session tasks drop the RunningSession futures and
// and every watchdog. On Linux the watchdog tasks shut their
// dup'd sockets so russh's inner tasks unblock at the next
// read. On non-Linux targets there is no dup'd socket and the
// session tasks observe the cancellation directly. Either way
// the session tasks drop the RunningSession futures and
// return. drain_sessions then catches up much faster than
// the SHUTDOWN_DRAIN_TIMEOUT_SECS ceiling because no session
// has to wait for the watchdog's natural tick to fire.
@@ -386,7 +391,7 @@ where
// its watchdog before the drain loop runs. Wedged
// sessions need their watchdog to call shutdown on
// the dup'd socket so russh's inner task can end
// by EOF; otherwise drain_sessions would block on
// by EOF. Otherwise drain_sessions would block on
// those sessions for the full
// SHUTDOWN_DRAIN_TIMEOUT_SECS.
server_shutdown_token.cancel();
@@ -461,15 +466,20 @@ where
// shut the socket down on wedge detection without racing
// russh for the original fd. The wedge probe itself reads
// /proc/net/tcp[6] in lifecycle::probe_tcp_state and does
// not touch this dup.
let watchdog_socket = wedge_watchdog::dup_socket(&stream);
if watchdog_socket.is_none() {
tracing::warn!(
peer = %peer_addr,
session_id = session_diag.session_id,
"wedge watchdog: dup_socket failed, session has no wedge protection (rare; usually fd exhaustion)",
);
}
// not touch this dup. Non-Linux targets run the silence-only
// fallback watchdog, which needs no socket dup.
#[cfg(target_os = "linux")]
let watchdog_socket = {
let socket = wedge_watchdog::dup_socket(&stream);
if socket.is_none() {
tracing::warn!(
peer = %peer_addr,
session_id = session_diag.session_id,
"wedge watchdog: dup_socket failed, session has no wedge protection (rare; usually fd exhaustion)",
);
}
socket
};
// Per-session cancellation token. Cascades from the
// listener-wide server_shutdown_token so a graceful server
// shutdown ends every live session promptly without waiting
@@ -498,6 +508,7 @@ where
session_id = session_diag.session_id,
"SFTP accept: spawning session task",
);
#[cfg(target_os = "linux")]
sessions.spawn(run_session(
ssh_config,
stream,
@@ -507,6 +518,15 @@ where
session_shutdown_token,
peer_addr,
));
#[cfg(not(target_os = "linux"))]
sessions.spawn(run_session(
ssh_config,
stream,
handler,
watchdog_session_diag,
session_shutdown_token,
peer_addr,
));
}
}
@@ -643,7 +663,7 @@ async fn run_session<S>(
ssh_config: Arc<russh::server::Config>,
stream: tokio::net::TcpStream,
handler: SshSessionHandler<S>,
watchdog_socket: Option<socket2::Socket>,
#[cfg(target_os = "linux")] watchdog_socket: Option<socket2::Socket>,
watchdog_session_diag: Arc<SessionDiag>,
cancel_token: CancellationToken,
peer_addr: SocketAddr,
@@ -674,19 +694,23 @@ async fn run_session<S>(
}
};
tracing::debug!(peer = %peer_addr, "SFTP session run_stream returned; awaiting session loop");
// Spawn the per-session wedge watchdog. The watchdog observes
// the SFTP-handler activity stamp and a non-blocking peek on
// the duplicated socket. On wedge detection it shuts the
// socket down (so russh's inner task unwedges via EOF
// propagation) and cancels the shared CancellationToken (so
// this task drops RunningSession and ends).
// Spawn the per-session watchdog. On Linux the wedge watchdog
// observes the SFTP-handler activity stamp and the kernel TCP
// state. On wedge detection it shuts the duplicated socket down
// (so russh's inner task unwedges via EOF propagation) and
// cancels the shared CancellationToken (so this task drops
// RunningSession and ends). On non-Linux targets the fallback
// watchdog cancels only after silence past the fallback
// threshold, with no socket introspection.
#[cfg(target_os = "linux")]
if let Some(socket) = watchdog_socket {
wedge_watchdog::spawn_for_session(watchdog_session_diag, socket, cancel_token.clone());
}
#[cfg(not(target_os = "linux"))]
fallback_watchdog::spawn_for_session(watchdog_session_diag, cancel_token.clone());
// Await the RunningSession until the client disconnects, until
// russh's inactivity_timeout and keepalive layers (set in
// build_ssh_config) close a wedged peer, or until the watchdog
// (or the listener-wide shutdown cascade) cancels.
// russh's inactivity_timeout closes a healthy idle peer, or until
// the watchdog (or the listener-wide shutdown cascade) cancels.
let session_result = tokio::select! {
res = session => Some(res),
_ = cancel_token.cancelled() => None,
@@ -698,7 +722,7 @@ async fn run_session<S>(
let session_result = match session_result {
Some(r) => r,
None => {
tracing::warn!(peer = %peer_addr, "SFTP session aborted by wedge watchdog");
tracing::warn!(peer = %peer_addr, "SFTP session cancelled (watchdog or server shutdown)");
return;
}
};
@@ -1088,7 +1112,7 @@ impl<S: StorageBackend + Send + Sync + 'static> russh::server::Handler for SshSe
}
}
// Signal requests carry want_reply = false on the wire (RFC 4254
// Signal requests carry want_reply = false (RFC 4254
// section 6.9), so there is no channel_failure to send. The
// override exists to log probe attempts and to keep the SFTP
// server SFTP-only by code rather than by relying on the russh
+17 -24
View File
@@ -34,15 +34,17 @@
//! backend operation that pipelines into a still-ESTABLISHED socket
//! cannot be misdiagnosed as a wedge.
//!
//! Platform-conditional detection latency. On Linux the procfs probe
//! gives a fast-kill window of WEDGE_FAST_KILL_SILENCE_SECS plus one
//! tick (approximately 45 s) from the moment a session enters
//! CLOSE_WAIT. On macOS, Windows, and other non-Linux targets the
//! /proc/net/tcp files are unavailable, the read returns Err, the
//! probe returns None, and the watchdog falls back to
//! WEDGE_FALLBACK_KILL_SILENCE_SECS (approximately 30 minutes).
//! Server-side resource accumulation is bounded in both cases. The
//! recommended deployment platform is Linux.
//! Compiled on Linux only. Non-Linux targets use fallback_watchdog, a
//! silence-only backstop at WEDGE_FALLBACK_KILL_SILENCE_SECS without the
//! procfs CLOSE_WAIT probe.
//!
//! Worst-case fast-kill latency on Linux is WEDGE_FAST_KILL_SILENCE_SECS
//! plus one to two ticks (approximately 45 to 60 s) from the moment a
//! session is both silent past WEDGE_FAST_KILL_SILENCE_SECS and the
//! kernel reports CLOSE_WAIT. The WEDGE_FALLBACK_KILL_SILENCE_SECS
//! ceiling is the backstop for the procfs probe when /proc/net/tcp is
//! unreadable (filesystem permissions, namespace tricks) or when the
//! wedge is in a TCP state other than CLOSE_WAIT.
//!
//! On cancel the watchdog calls shutdown(Both) on the duplicated
//! socket so russh's inner select unwedges via EOF propagation,
@@ -53,7 +55,6 @@ use super::constants::limits::{WEDGE_FALLBACK_KILL_SILENCE_SECS, WEDGE_FAST_KILL
use super::lifecycle::{SessionDiag, TcpState, probe_tcp_state};
use socket2::Socket;
use std::net::Shutdown;
#[cfg(unix)]
use std::os::fd::AsFd;
use std::sync::Arc;
use std::sync::atomic::Ordering;
@@ -61,8 +62,8 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::net::TcpStream;
use tokio_util::sync::CancellationToken;
/// Reason a watchdog cancelled its session. Surfaced in the warn log
/// the watchdog emits at cancel time so operators can correlate the
/// Reason a watchdog cancelled its session. Included in the warn log
/// the watchdog writes at cancel time so operators can correlate the
/// cancel with the upstream client behaviour.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WedgeReason {
@@ -80,7 +81,7 @@ enum WedgeReason {
ProbeFailedConfirmed,
/// Silence past WEDGE_FALLBACK_KILL_SILENCE_SECS regardless of
/// the TCP_STATE probe result. Backstop for the case where the
/// wedge surfaces in a state other than CLOSE_WAIT and probes
/// wedge is in a state other than CLOSE_WAIT and probes
/// kept returning healthy or non-decisive.
FallbackSilence,
}
@@ -101,19 +102,11 @@ impl WedgeReason {
/// racing russh for the original fd. Returns None when the dup fails.
/// Callers should treat None as "no watchdog this session, accept-loop
/// continues".
#[cfg(unix)]
pub(super) fn dup_socket(stream: &TcpStream) -> Option<Socket> {
let cloned = stream.as_fd().try_clone_to_owned().ok()?;
Some(Socket::from(cloned))
}
/// Non-Unix stub: AsFd on TcpStream is Unix-only. Returns None so the
/// caller falls back to WEDGE_FALLBACK_KILL_SILENCE_SECS.
#[cfg(not(unix))]
pub(super) fn dup_socket(_stream: &TcpStream) -> Option<Socket> {
None
}
/// Spawn a per-session watchdog tick task.
///
/// The task owns the duplicated socket (closed on task end via
@@ -127,7 +120,7 @@ pub(super) fn spawn_for_session(session_diag: Arc<SessionDiag>, socket: Socket,
let peer = session_diag.peer;
let mut tick = tokio::time::interval(Duration::from_secs(WEDGE_WATCHDOG_TICK_SECS));
tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
// First tick fires immediately; skip it so the watchdog never
// First tick fires immediately. Skip it so the watchdog never
// makes a decision before one full silence window has elapsed.
tick.tick().await;
let mut wedge_suspected = false;
@@ -171,7 +164,7 @@ pub(super) fn spawn_for_session(session_diag: Arc<SessionDiag>, socket: Socket,
// here makes the next I/O on the original fd return EOF,
// which propagates through russh-sftp and drops the mpsc
// receiver. In the clean-end case russh has already returned
// and dropped its half of the fd; this call sends a final
// and dropped its half of the fd. This call sends a final
// FIN on the still-open dup, which the peer's stack
// tolerates.
let _ = socket.shutdown(Shutdown::Both);
@@ -180,7 +173,7 @@ pub(super) fn spawn_for_session(session_diag: Arc<SessionDiag>, socket: Socket,
#[derive(Debug, PartialEq, Eq)]
enum Decision {
/// No wedge signal this tick; reset any suspected state.
/// No wedge signal this tick. Reset any suspected state.
Quiet,
/// First tick to observe silence past the fast threshold AND a
/// non-healthy probe result. Hold suspected state for one more
+5 -1
View File
@@ -33,7 +33,7 @@ workspace = true
rustfs-config = { workspace = true, features = ["server-config-model"] }
rustfs-common = { workspace = true }
rustfs-utils = { workspace = true }
tokio = { workspace = true, features = ["fs","sync","rt","time","io-uring","macros"] }
tokio = { workspace = true, features = ["fs","sync","rt","time","macros"] }
tracing = { workspace = true }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
@@ -62,3 +62,7 @@ tokio = { workspace = true, features = ["test-util"] }
[lib]
doctest = false
# io-uring is Linux-only. Scope the feature to Linux targets.
[target.'cfg(target_os = "linux")'.dependencies]
tokio = { workspace = true, features = ["io-uring"] }
+5 -1
View File
@@ -41,7 +41,7 @@ async-compression = { workspace = true, features = [
"zstd",
"xz",
] }
tokio = { workspace = true, features = ["io-uring","fs","io-util","macros"] }
tokio = { workspace = true, features = ["fs","io-util","macros"] }
tokio-stream = { workspace = true }
astral-tokio-tar = { workspace = true }
thiserror = { workspace = true }
@@ -54,3 +54,7 @@ tempfile = { workspace = true }
[lints]
workspace = true
# io-uring is Linux-only. Scope the feature to Linux targets.
[target.'cfg(target_os = "linux")'.dependencies]
tokio = { workspace = true, features = ["io-uring"] }
+3 -1
View File
@@ -112,7 +112,7 @@ http-body.workspace = true
http-body-util.workspace = true
reqwest = { workspace = true }
socket2 = { workspace = true }
tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "signal", "process", "io-util", "io-uring"] }
tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "signal", "process", "io-util"] }
tokio-rustls = { workspace = true }
aws-sdk-s3 = { workspace = true }
tokio-stream.workspace = true
@@ -177,6 +177,8 @@ hashbrown = { workspace = true }
[target.'cfg(target_os = "linux")'.dependencies]
libsystemd.workspace = true
# io-uring is Linux-only. Scope the feature to Linux targets.
tokio = { workspace = true, features = ["io-uring"] }
[target.'cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))'.dependencies]
mimalloc = { workspace = true }
+76
View File
@@ -0,0 +1,76 @@
#!/usr/bin/env bash
# Confirms rustfs.exe with --features sftp binds an SFTP listener on Windows.
# Checks the listener bind and the warn-once host-key log line. S3 traffic is out of scope.
set -euo pipefail
TMP="$(mktemp -d -t rustfs-sftp-smoke-XXXXXX)"
trap 'rm -rf "$TMP" 2>/dev/null || true' EXIT
# Generate an Ed25519 host key in OpenSSH PEM format.
ssh-keygen -t ed25519 -f "$TMP/ssh_host_ed25519_key" -N "" -q
export RUSTFS_SFTP_ENABLE=true
export RUSTFS_SFTP_ADDRESS=127.0.0.1:0
export RUSTFS_SFTP_HOST_KEY_DIR="$TMP"
export RUSTFS_VOLUMES="$TMP/data"
export RUSTFS_ACCESS_KEY=smoketest
export RUSTFS_SECRET_KEY=smoketestsecret
# Bind S3 to an ephemeral port and skip the console so the smoke does not
# depend on fixed ports being free on the runner.
export RUSTFS_ADDRESS=127.0.0.1:0
export RUSTFS_CONSOLE_ENABLE=false
# Raise the log level so the smoke can grep the server log for the SFTP
# listener bind line and the warn-once host-key line. rustfs defaults to
# the error level, which would suppress both (they log at info and warn).
export RUST_LOG=info
mkdir -p "$RUSTFS_VOLUMES"
# Launch the server in the background. The CI step that runs this script
# has already built the rustfs.exe binary with the sftp feature. The
# RUSTFS_BIN override lets the CI step point at a non-default location.
BIN="${RUSTFS_BIN:-target/release/rustfs.exe}"
if [ ! -x "$BIN" ]; then
echo "FAIL: $BIN is not an executable binary" >&2
exit 1
fi
"$BIN" server "$TMP/data" >"$TMP/server.log" 2>&1 &
PID=$!
trap 'kill "$PID" 2>/dev/null || true; rm -rf "$TMP" 2>/dev/null || true' EXIT
# Discover the bound ephemeral port from the server log.
# rustfs logs "SFTP server listening" with bind_addr 127.0.0.1:<port>
# at startup. Grepping the log avoids the PID mismatch between the
# MSYS-internal $! and netstat's native Windows PID under Git Bash.
PORT=""
for _ in $(seq 1 60); do
sleep 1
LINE="$(grep "SFTP server listening" "$TMP/server.log" 2>/dev/null | head -1 || true)"
if [ -n "$LINE" ]; then
PORT="$(echo "$LINE" | grep -oE "127\.0\.0\.1:[0-9]+" | head -1 | awk -F: '{print $NF}' || true)"
if [ -n "$PORT" ]; then
break
fi
fi
done
if [ -z "$PORT" ]; then
echo "FAIL: rustfs.exe did not bind any 127.0.0.1 LISTENING port within 60s" >&2
echo "--- server.log tail ---" >&2
tail -50 "$TMP/server.log" >&2 || true
exit 1
fi
echo "PASS: SFTP listener bound at 127.0.0.1:${PORT} (PID ${PID})"
# Confirm the warn-once log line appeared in the server log.
if ! grep -q "host key file permission enforcement is not active on Windows" \
"$TMP/server.log"; then
echo "FAIL: expected warn-once log line not found in server log" >&2
echo "--- server.log tail ---" >&2
tail -50 "$TMP/server.log" >&2 || true
exit 1
fi
echo "PASS: warn-once Windows host-key log line present"
exit 0