diff --git a/CHANGELOG.md b/CHANGELOG.md index c39be2185..f06ffa1d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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) diff --git a/crates/e2e_test/src/protocols/sftp_compliance_tests.rs b/crates/e2e_test/src/protocols/sftp_compliance_tests.rs index 593e6618d..ed6c5cc2f 100644 --- a/crates/e2e_test/src/protocols/sftp_compliance_tests.rs +++ b/crates/e2e_test/src/protocols/sftp_compliance_tests.rs @@ -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 Arc { 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) { 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> { crate::common::init_logging(); diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index e9ec0bbaa..8ae7479b1 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -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"] } diff --git a/crates/io-core/Cargo.toml b/crates/io-core/Cargo.toml index e85c7e92e..08b1cb529 100644 --- a/crates/io-core/Cargo.toml +++ b/crates/io-core/Cargo.toml @@ -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"] } diff --git a/crates/kms/Cargo.toml b/crates/kms/Cargo.toml index e1f41e731..db91879b1 100644 --- a/crates/kms/Cargo.toml +++ b/crates/kms/Cargo.toml @@ -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"] } diff --git a/crates/protocols/Cargo.toml b/crates/protocols/Cargo.toml index 4afeafccc..6b2ab319d 100644 --- a/crates/protocols/Cargo.toml +++ b/crates/protocols/Cargo.toml @@ -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"] } diff --git a/crates/protocols/src/sftp/config.rs b/crates/protocols/src/sftp/config.rs index 4fee437a3..12b0c8894 100644 --- a/crates/protocols/src/sftp/config.rs +++ b/crates/protocols/src/sftp/config.rs @@ -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) -> Option { @@ -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) -> Option { @@ -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) -> Option { @@ -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) -> Option { @@ -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, 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, 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, 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] diff --git a/crates/protocols/src/sftp/constants.rs b/crates/protocols/src/sftp/constants.rs index 811dd4b1d..fbf9dda3e 100644 --- a/crates/protocols/src/sftp/constants.rs +++ b/crates/protocols/src/sftp/constants.rs @@ -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 diff --git a/crates/protocols/src/sftp/fallback_watchdog.rs b/crates/protocols/src/sftp/fallback_watchdog.rs new file mode 100644 index 000000000..6aa84ddd2 --- /dev/null +++ b/crates/protocols/src/sftp/fallback_watchdog.rs @@ -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, 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); + } +} diff --git a/crates/protocols/src/sftp/lifecycle.rs b/crates/protocols/src/sftp/lifecycle.rs index 7518957d5..4bd3f1aad 100644 --- a/crates/protocols/src/sftp/lifecycle.rs +++ b/crates/protocols/src/sftp/lifecycle.rs @@ -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 { 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 Option { 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 { // 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]), diff --git a/crates/protocols/src/sftp/mod.rs b/crates/protocols/src/sftp/mod.rs index fcca85ccb..00822801d 100644 --- a/crates/protocols/src/sftp/mod.rs +++ b/crates/protocols/src/sftp/mod.rs @@ -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; diff --git a/crates/protocols/src/sftp/server.rs b/crates/protocols/src/sftp/server.rs index 7d7878ae9..ab038111a 100644 --- a/crates/protocols/src/sftp/server.rs +++ b/crates/protocols/src/sftp/server.rs @@ -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 { ssh_config: Arc, 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, /// 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( ssh_config: Arc, stream: tokio::net::TcpStream, handler: SshSessionHandler, - watchdog_socket: Option, + #[cfg(target_os = "linux")] watchdog_socket: Option, watchdog_session_diag: Arc, cancel_token: CancellationToken, peer_addr: SocketAddr, @@ -674,19 +694,23 @@ async fn run_session( } }; 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( 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 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 diff --git a/crates/protocols/src/sftp/wedge_watchdog.rs b/crates/protocols/src/sftp/wedge_watchdog.rs index 532aed596..3eec057ad 100644 --- a/crates/protocols/src/sftp/wedge_watchdog.rs +++ b/crates/protocols/src/sftp/wedge_watchdog.rs @@ -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 { 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 { - 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, 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, 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, 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 diff --git a/crates/scanner/Cargo.toml b/crates/scanner/Cargo.toml index 584f8b45f..95389a997 100644 --- a/crates/scanner/Cargo.toml +++ b/crates/scanner/Cargo.toml @@ -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"] } diff --git a/crates/zip/Cargo.toml b/crates/zip/Cargo.toml index ca3152537..2b18f3696 100644 --- a/crates/zip/Cargo.toml +++ b/crates/zip/Cargo.toml @@ -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"] } diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index c36cd6a13..dfaf2b1ab 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -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 } diff --git a/scripts/windows-sftp-listener-smoke.sh b/scripts/windows-sftp-listener-smoke.sh new file mode 100755 index 000000000..8f3d9e3c0 --- /dev/null +++ b/scripts/windows-sftp-listener-smoke.sh @@ -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: +# 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