test(sftp): cover init negotiation and platform gating (#2896)

crates/e2e_test/src/protocols/sftp_compliance.rs

@@ -39,8 +39,10 @@
 use crate::protocols::sftp_compliance_tests::{
     cmptst_01, cmptst_02, cmptst_03, cmptst_04, cmptst_05, cmptst_06, cmptst_07, cmptst_08, cmptst_09, cmptst_10, cmptst_11,
     cmptst_12, cmptst_13, cmptst_14, cmptst_15, cmptst_16, cmptst_17, cmptst_18, cmptst_19, cmptst_20, cmptst_21, cmptst_22,
-    cmptst_23, cmptst_24, cmptst_25, cmptst_26, cmptst_27, cmptst_28, cmptst_29, cmptst_32, cmptst_33, spawn_compliance_rustfs,
+    cmptst_23, cmptst_27, cmptst_28, cmptst_29, cmptst_32, cmptst_33, spawn_compliance_rustfs,
 };
+#[cfg(target_os = "linux")]
+use crate::protocols::sftp_compliance_tests::{cmptst_24, cmptst_25, cmptst_26};
 use crate::protocols::sftp_helpers::{build_test_s3_client, connect_sftp_to, wait_for_s3_ready};
 use crate::protocols::test_env::ProtocolTestEnvironment;
 use anyhow::{Result, anyhow};

crates/e2e_test/src/protocols/sftp_compliance_tests.rs

@@ -118,23 +118,29 @@ 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_IDLE_TIMEOUT,
-    ENV_SFTP_PART_SIZE, ENV_SFTP_READ_CACHE_WINDOW_BYTES, ENV_SFTP_READ_ONLY,
+    ENV_CONSOLE_ENABLE, ENV_RUSTFS_ADDRESS, ENV_SFTP_ADDRESS, ENV_SFTP_ENABLE, ENV_SFTP_HOST_KEY_DIR, ENV_SFTP_PART_SIZE,
+    ENV_SFTP_READ_CACHE_WINDOW_BYTES, ENV_SFTP_READ_ONLY,
 };
 use sha2::{Digest, Sha256};
 use std::path::PathBuf;
 use std::pin::Pin;
 use std::process::Stdio;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+#[cfg(target_os = "linux")]
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::task::{Context, Poll};
 use std::time::{Duration, Instant};
 use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncSeekExt, AsyncWrite, AsyncWriteExt, BufReader, ReadBuf};
 use tokio::net::TcpStream;
 use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 use tokio::process::{Child, Command};
-use tokio::time::{sleep, timeout};
+#[cfg(target_os = "linux")]
+use tokio::time::sleep;
+use tokio::time::timeout;
 use tracing::info;
 
 // Cross-case constants used by every spawn helper. Pinned to 5 MiB so
@@ -400,12 +406,14 @@ fn capture_server_stdout(child: &mut Child) -> Arc<tokio::sync::Mutex<Vec<String
 /// or "SFTP session task panicked" log emits a finish. The session-
 /// lifecycle cases (CMPTST-24, CMPTST-25, CMPTST-26) read both fields
 /// to assert the watchdog killed silent sessions on the expected path.
+#[cfg(target_os = "linux")]
 #[derive(Default)]
 struct SessionCounters {
     entered: AtomicUsize,
     finished: AtomicUsize,
 }
 
+#[cfg(target_os = "linux")]
 impl SessionCounters {
     fn new() -> Arc<Self> {
         Arc::new(Self {
@@ -419,6 +427,7 @@ 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;
@@ -448,6 +457,7 @@ fn watch_session_lifecycle_events(child: &mut Child, counters: Arc<SessionCounte
 /// best-effort: if ss is missing on the host the function returns
 /// Ok(None) and the caller skips the assertion. The contract is zero
 /// CLOSE_WAIT entries attributable to the test.
+#[cfg(target_os = "linux")]
 async fn count_close_wait_on_port(port: u16) -> Result<Option<usize>> {
     let output = match Command::new("ss").args(["-tn", "state", "CLOSE-WAIT"]).output().await {
         Ok(o) => o,
@@ -1166,6 +1176,7 @@ pub(crate) mod cmptst_23 {
 }
 
 // CMPTST-24: concurrent half-close burst does not leak server-side session tasks.
+#[cfg(target_os = "linux")]
 pub(crate) mod cmptst_24 {
     use super::*;
 
@@ -1544,6 +1555,7 @@ pub(crate) mod cmptst_24 {
 }
 
 // CMPTST-25: wedge-kill watchdog kills sessions parked behind a CLOSE_WAIT socket.
+#[cfg(target_os = "linux")]
 pub(crate) mod cmptst_25 {
     use super::*;
 
@@ -2076,6 +2088,7 @@ 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::*;
 

crates/protocols/src/sftp/driver.rs

@@ -1116,6 +1116,7 @@ impl<S: StorageBackend + Send + Sync + 'static> Drop for SftpDriver<S> {
 
 #[cfg(test)]
 mod tests {
+    use super::super::constants::protocol;
     use super::super::state::WritePhase;
     use super::super::test_support::{TEST_PART_SIZE, build_driver, build_readonly_driver, file_handle, write_handle};
     use super::*;
@@ -1123,8 +1124,48 @@ mod tests {
     use crate::common::gateway::{with_test_auth_override, with_test_iam_unavailable};
     use russh_sftp::server::Handler;
     use rustfs_utils::path;
+    use std::collections::HashMap;
     use std::sync::Arc;
 
+    #[tokio::test]
+    async fn init_advertises_sftp_v3_without_extensions() {
+        let backend = Arc::new(DummyBackend::new());
+        let mut driver = build_driver(backend, TEST_PART_SIZE);
+        let extensions = HashMap::from([("posix-rename@openssh.com".to_string(), "1".to_string())]);
+
+        let advertised = driver
+            .init(protocol::SFTP_VERSION, extensions)
+            .await
+            .expect("init must succeed");
+
+        assert_eq!(advertised.version, protocol::SFTP_VERSION);
+        assert!(advertised.extensions.is_empty(), "server must not advertise unsupported extensions");
+    }
+
+    #[tokio::test]
+    async fn init_from_newer_client_still_advertises_sftp_v3() {
+        let backend = Arc::new(DummyBackend::new());
+        let mut driver = build_driver(backend, TEST_PART_SIZE);
+
+        let advertised = driver
+            .init(protocol::SFTP_VERSION + 3, HashMap::new())
+            .await
+            .expect("version negotiation must still reply");
+
+        assert_eq!(advertised.version, protocol::SFTP_VERSION);
+        assert!(advertised.extensions.is_empty());
+    }
+
+    #[test]
+    fn unimplemented_packet_returns_op_unsupported() {
+        let backend = Arc::new(DummyBackend::new());
+        let driver = build_driver(backend, TEST_PART_SIZE);
+
+        let err = <SftpDriver<DummyBackend> as Handler>::unimplemented(&driver);
+
+        assert!(matches!(StatusCode::from(err), StatusCode::OpUnsupported));
+    }
+
     #[tokio::test]
     async fn fstat_on_file_handle_returns_cached_attrs() {
         let backend = Arc::new(DummyBackend::new());