diff --git a/bindings/dotnet/Cargo.toml b/bindings/dotnet/Cargo.toml index 9044b08bf45c..99826d9526d8 100644 --- a/bindings/dotnet/Cargo.toml +++ b/bindings/dotnet/Cargo.toml @@ -87,6 +87,7 @@ opendal = { version = ">=0", path = "../../core", features = [ "services-redis", # FIXME: rocksdb will lead to "cannot allocate memory in static TLS block" while linking. # "services-rocksdb", + "services-sftp", "services-sled", "services-tikv", "services-vercel-artifacts", @@ -110,6 +111,5 @@ tokio = { version = "1.49.0", features = ["full"] } features = [ # Depend on unix-only dependency stacks in current implementations. "services-monoiofs", - "services-sftp", ] path = "../../core" diff --git a/bindings/java/Cargo.toml b/bindings/java/Cargo.toml index f75d42e76780..34025ae1c05e 100644 --- a/bindings/java/Cargo.toml +++ b/bindings/java/Cargo.toml @@ -87,6 +87,7 @@ opendal = { version = ">=0", path = "../../core", default-features = false, feat "services-redis", # FIXME: rocksdb will lead to "cannot allocate memory in static TLS block" while linking. # "services-rocksdb", + "services-sftp", "services-sled", "services-tikv", "services-vercel-artifacts", @@ -103,12 +104,3 @@ opendal = { version = ">=0", path = "../../core", default-features = false, feat "services-yandex-disk", ] } tokio = { version = "1.49.0", features = ["full"] } - -# This is not optimal. See also the Cargo issue: -# https://github.com/rust-lang/cargo/issues/1197#issuecomment-1641086954 -[target.'cfg(unix)'.dependencies.opendal] -features = [ - # Depend on "openssh" which depends on "tokio-pipe" that is unavailable on Windows. - "services-sftp", -] -path = "../../core" diff --git a/bindings/python/Cargo.toml b/bindings/python/Cargo.toml index e3b9680fd7f1..87b4239e80d9 100644 --- a/bindings/python/Cargo.toml +++ b/bindings/python/Cargo.toml @@ -75,6 +75,7 @@ services-all = [ 'services-redb', 'services-redis', 'services-seafile', + 'services-sftp', 'services-sled', 'services-sqlite', 'services-swift', @@ -187,10 +188,6 @@ services-yandex-disk = ["opendal/services-yandex-disk"] # we build cp311-abi3 and cp310 wheels now, move this to pyo3 after we drop cp310 abi3 = ["pyo3/abi3-py311"] -# Conditionally include sftp only on non-windows targets -[target.'cfg(not(windows))'.features] -services-all = ["services-sftp"] - [lib] crate-type = ["cdylib"] doc = false diff --git a/core/Cargo.lock b/core/Cargo.lock index 892e9acdd8c9..eb422ab8aedf 100644 --- a/core/Cargo.lock +++ b/core/Cargo.lock @@ -17,6 +17,16 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" +[[package]] +name = "aead" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1973cfbc1a2daf9cf550e74e1f088c28e7f7d8c1e1418fb6c9dc5184b7e84c99" +dependencies = [ + "crypto-common 0.2.1", + "inout 0.2.2", +] + [[package]] name = "aes" version = "0.8.4" @@ -37,6 +47,22 @@ dependencies = [ "cipher 0.5.1", "cpubits", "cpufeatures 0.3.0", + "zeroize", +] + +[[package]] +name = "aes-gcm" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fdf011db2e21ce0d575593d749db5554b47fed37aff429e4dc50bc91ac93a028" +dependencies = [ + "aead", + "aes 0.9.1", + "cipher 0.5.1", + "ctr", + "ghash", + "subtle", + "zeroize", ] [[package]] @@ -211,9 +237,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3c3610892ee6e0cbce8ae2700349fcf8f98adb0dbfbee85aec3c9179d29cc072" dependencies = [ "base64ct", - "blake2", + "blake2 0.10.6", "cpufeatures 0.2.17", - "password-hash", + "password-hash 0.5.0", +] + +[[package]] +name = "argon2" +version = "0.6.0-rc.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7af50940b73bf4e16c15c448a2b121c63f2d68e3e54b6a8731673cb4aa0cdff5" +dependencies = [ + "base64ct", + "blake2 0.11.0-rc.6", + "cpufeatures 0.3.0", + "password-hash 0.6.1", ] [[package]] @@ -385,7 +423,7 @@ dependencies = [ "futures-core", "futures-io", "futures-lite", - "gloo-timers", + "gloo-timers 0.3.0", "kv-log-macro", "log", "memchr", @@ -495,22 +533,6 @@ dependencies = [ "weak-table", ] -[[package]] -name = "awaitable" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "70af449c9a763cb655c6a1e5338b42d99c67190824ff90658c1e30be844c0775" -dependencies = [ - "awaitable-error", - "cfg-if 1.0.4", -] - -[[package]] -name = "awaitable-error" -version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d5b3469636cdf8543cceab175efca534471f36eee12fb8374aba00eb5e7e7f8a" - [[package]] name = "aws-config" version = "1.8.11" @@ -1037,7 +1059,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cffb0e931875b666fc4fcb20fee52e9bbd1ef836fd9e9e04ec21555f9f85f7ef" dependencies = [ "fastrand", - "gloo-timers", + "gloo-timers 0.3.0", "tokio", ] @@ -1053,6 +1075,12 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4c7f02d4ea65f2c1853089ffd8d2787bdbc63de2f0d29dedbcf8ccdfa0ccd4cf" +[[package]] +name = "base16ct" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd307490d624467aa6f74b0eabb77633d1f758a7b25f12bceb0b22e08d9726f6" + [[package]] name = "base64" version = "0.21.7" @@ -1094,12 +1122,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9a0f5948f30df5f43ac29d310b7476793be97c50787e6ef4a63d960a0d0be827" dependencies = [ "base64 0.22.1", - "blowfish", + "blowfish 0.9.1", "getrandom 0.3.4", "subtle", "zeroize", ] +[[package]] +name = "bcrypt-pbkdf" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "144e573728da132683b9488acd528274c790e07fc06ff81ee29f9d8f8b1041e0" +dependencies = [ + "blowfish 0.10.0", + "pbkdf2 0.13.0", + "sha2 0.11.0", +] + [[package]] name = "bindgen" version = "0.72.1" @@ -1156,6 +1195,15 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "blake2" +version = "0.11.0-rc.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "061f1a09225e328e1ffbb378d2d49923c0ca5fee19fb5ac1cc9c1e9d52b93690" +dependencies = [ + "digest 0.11.3", +] + [[package]] name = "blake3" version = "1.8.5" @@ -1186,6 +1234,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cdd35008169921d80bc60d3d0ab416eecb028c4cd653352907921d95084790be" dependencies = [ "hybrid-array", + "zeroize", ] [[package]] @@ -1229,6 +1278,16 @@ dependencies = [ "cipher 0.4.4", ] +[[package]] +name = "blowfish" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "62ce3946557b35e71d1bbe07ec385073ce9eda05043f95de134eb578fcf1a298" +dependencies = [ + "byteorder", + "cipher 0.5.1", +] + [[package]] name = "bnum" version = "0.12.1" @@ -1525,8 +1584,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601" dependencies = [ "cfg-if 1.0.4", + "cipher 0.5.1", "cpufeatures 0.3.0", "rand_core 0.10.1", + "zeroize", ] [[package]] @@ -1589,6 +1650,7 @@ dependencies = [ "block-buffer 0.12.0", "crypto-common 0.2.1", "inout 0.2.2", + "zeroize", ] [[package]] @@ -1915,17 +1977,6 @@ dependencies = [ "crossbeam-utils", ] -[[package]] -name = "concurrent_arena" -version = "0.1.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a07f0a549fe58f8477a15f0f1c3aa8ced03a3cdeaa38a661530572f21ea963a0" -dependencies = [ - "arc-swap", - "parking_lot 0.12.5", - "triomphe", -] - [[package]] name = "condtype" version = "1.3.0" @@ -2219,6 +2270,23 @@ dependencies = [ "zeroize", ] +[[package]] +name = "crypto-bigint" +version = "0.7.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a52aa3fcda4e6302a9f48734f234d35d4721b96f8fe07d073f07ce9df4f0271" +dependencies = [ + "cpubits", + "ctutils", + "getrandom 0.4.2", + "hybrid-array", + "num-traits", + "rand_core 0.10.1", + "serdect", + "subtle", + "zeroize", +] + [[package]] name = "crypto-common" version = "0.1.7" @@ -2235,7 +2303,19 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77727bb15fa921304124b128af125e7e3b968275d1b108b379190264f4423710" dependencies = [ + "getrandom 0.4.2", "hybrid-array", + "rand_core 0.10.1", +] + +[[package]] +name = "crypto-primes" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3633a51a39c69ebbaa4feaa694bd83d241e4093901c84a0963b19d9bb3f0cf8f" +dependencies = [ + "crypto-bigint 0.7.5", + "rand_core 0.10.1", ] [[package]] @@ -2287,6 +2367,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7d5515a3834141de9eafb9717ad39eea8247b5674e6066c404e8c4b365d2a29e" dependencies = [ "cmov", + "subtle", ] [[package]] @@ -2299,7 +2380,24 @@ dependencies = [ "cpufeatures 0.2.17", "curve25519-dalek-derive", "digest 0.10.7", - "fiat-crypto", + "fiat-crypto 0.2.9", + "rustc_version", + "subtle", + "zeroize", +] + +[[package]] +name = "curve25519-dalek" +version = "5.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5eed333089e2e1c1ac8c6c0398e5e2497b4c9926ca6d0365ed1e099afa5bc23" +dependencies = [ + "cfg-if 1.0.4", + "cpufeatures 0.3.0", + "curve25519-dalek-derive", + "digest 0.11.3", + "fiat-crypto 0.3.0", + "rand_core 0.10.1", "rustc_version", "subtle", "zeroize", @@ -2451,6 +2549,17 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "delegate" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "780eb241654bf097afb00fc5f054a09b687dad862e485fdcf8399bb056565370" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "der" version = "0.6.1" @@ -2468,7 +2577,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" dependencies = [ "const-oid 0.9.6", - "pem-rfc7468", + "pem-rfc7468 0.7.0", + "zeroize", +] + +[[package]] +name = "der" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a69dedd701da44b0536442edf09c81a64b0ab97a7a4a5e3d1971f00027cbc63d" +dependencies = [ + "const-oid 0.10.2", + "pem-rfc7468 1.0.0", "zeroize", ] @@ -2556,17 +2676,6 @@ dependencies = [ "syn 2.0.118", ] -[[package]] -name = "derive_destructure2" -version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "64b697ac90ff296f0fc031ee5a61c7ac31fb9fff50e3fb32873b09223613fc0c" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.118", -] - [[package]] name = "derive_more" version = "2.1.1" @@ -2850,6 +2959,21 @@ dependencies = [ "spki 0.7.3", ] +[[package]] +name = "ecdsa" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0681a4fc24c767085329728d8dfba959af91228aa4610cca4f8ce317ba46ae0" +dependencies = [ + "der 0.8.1", + "digest 0.11.3", + "elliptic-curve 0.14.1", + "rfc6979 0.6.0", + "signature 3.0.0", + "spki 0.8.0", + "zeroize", +] + [[package]] name = "ed25519" version = "2.2.3" @@ -2860,20 +2984,46 @@ dependencies = [ "signature 2.2.0", ] +[[package]] +name = "ed25519" +version = "3.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29fcf32e6c73d1079f83ab4d782de2d81620346a5f38c6237a86a22f8368980a" +dependencies = [ + "pkcs8 0.11.0", + "signature 3.0.0", +] + [[package]] name = "ed25519-dalek" version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "70e796c081cee67dc755e1a36a0a172b897fab85fc3f6bc48307991f64e4eca9" dependencies = [ - "curve25519-dalek", - "ed25519", + "curve25519-dalek 4.1.3", + "ed25519 2.2.3", "serde", "sha2 0.10.9", "subtle", "zeroize", ] +[[package]] +name = "ed25519-dalek" +version = "3.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ebaa1a2bf1290ab3bfe5a7b771d050ebffab2711c19a81691c683a5144a25de" +dependencies = [ + "curve25519-dalek 5.0.0", + "ed25519 3.0.0", + "rand_core 0.10.1", + "serde", + "sha2 0.11.0", + "signature 3.0.0", + "subtle", + "zeroize", +] + [[package]] name = "edge_test_aws_s3_assume_role_with_web_identity" version = "0.58.1" @@ -2956,8 +3106,8 @@ dependencies = [ "ff 0.13.1", "generic-array 0.14.7", "group 0.13.0", - "hkdf", - "pem-rfc7468", + "hkdf 0.12.4", + "pem-rfc7468 0.7.0", "pkcs8 0.10.2", "rand_core 0.6.4", "sec1 0.7.3", @@ -2965,6 +3115,28 @@ dependencies = [ "zeroize", ] +[[package]] +name = "elliptic-curve" +version = "0.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d65aa39b3a5c1c9c1b745c9a019234bb7a21b77abcb4f4d266d706e2d577d65" +dependencies = [ + "base16ct 1.0.0", + "crypto-bigint 0.7.5", + "crypto-common 0.2.1", + "digest 0.11.3", + "ff 0.14.0", + "group 0.14.0", + "hkdf 0.13.0", + "hybrid-array", + "pem-rfc7468 1.0.0", + "pkcs8 0.11.0", + "rand_core 0.10.1", + "sec1 0.8.1", + "subtle", + "zeroize", +] + [[package]] name = "encoding_rs" version = "0.8.35" @@ -2986,6 +3158,18 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "enum_dispatch" +version = "0.3.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa18ce2bc66555b3218614519ac839ddb759a7d6720732f979ef8d13be147ecd" +dependencies = [ + "once_cell", + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -3196,12 +3380,28 @@ dependencies = [ "subtle", ] +[[package]] +name = "ff" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1f686ab92a9fb0eaf188f6c6c87b89490baa6fdb0db4544ba4dc47f7942489f" +dependencies = [ + "rand_core 0.10.1", + "subtle", +] + [[package]] name = "fiat-crypto" version = "0.2.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" +[[package]] +name = "fiat-crypto" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64cd1e32ddd350061ae6edb1b082d7c54915b5c672c389143b9a63403a109f24" + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -3778,6 +3978,17 @@ dependencies = [ "zeroize", ] +[[package]] +name = "generic-array" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb130435a959a8d525e6bca66ff6c40981a300ee96d70e3ef56f046556d614a3" +dependencies = [ + "generic-array 0.14.7", + "rustversion", + "typenum", +] + [[package]] name = "geo" version = "0.32.0" @@ -3888,6 +4099,15 @@ dependencies = [ "prost 0.14.3", ] +[[package]] +name = "ghash" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2eecf2d5dc9b66b732b97707a0210906b1d30523eb773193ab777c0c84b3e8d5" +dependencies = [ + "polyval", +] + [[package]] name = "git-version" version = "0.3.9" @@ -3939,6 +4159,18 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "gloo-timers" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "482ce8a491a501da4cd806bd190275363d674f2845005c6ddbd5d3e1dd54495d" +dependencies = [ + "futures-channel", + "futures-core", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "goosefs-sdk" version = "0.1.8" @@ -4017,6 +4249,17 @@ dependencies = [ "subtle", ] +[[package]] +name = "group" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7fd1a1c7a5206c5b7a3f5a0d7ccd3ff85d0c8f5133d62a02680255b0004af5f4" +dependencies = [ + "ff 0.14.0", + "rand_core 0.10.1", + "subtle", +] + [[package]] name = "h2" version = "0.3.27" @@ -4303,6 +4546,12 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" +[[package]] +name = "hex-literal" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e712f64ec3850b98572bffac52e2c6f282b29fe6c5fa6d42334b30be438d95c1" + [[package]] name = "hf-xet" version = "1.5.4" @@ -4383,6 +4632,15 @@ dependencies = [ "hmac 0.12.1", ] +[[package]] +name = "hkdf" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4aaa26c720c68b866f2c96ef5c1264b3e6f473fe5d4ce61cd44bbe913e553018" +dependencies = [ + "hmac 0.13.0", +] + [[package]] name = "hmac" version = "0.12.1" @@ -4522,11 +4780,14 @@ checksum = "135b12329e5e3ce057a9f972339ea52bc954fe1e9358ef27f95e89716fbc5424" [[package]] name = "hybrid-array" -version = "0.4.12" +version = "0.4.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9155a582abd142abc056962c29e3ce5ff2ad5469f4246b537ed42c5deba857da" +checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c" dependencies = [ + "ctutils", + "subtle", "typenum", + "zeroize", ] [[package]] @@ -4930,6 +5191,18 @@ dependencies = [ "cfg-if 1.0.4", ] +[[package]] +name = "internal-russh-num-bigint" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae8e22120c32fb4d19ec55fba35015f57095cd95a2e3b732e44457f5915b2ee8" +dependencies = [ + "num-integer", + "num-traits", + "rand 0.10.1", + "rand_core 0.10.1", +] + [[package]] name = "io-uring" version = "0.6.4" @@ -5161,15 +5434,15 @@ checksum = "0529410abe238729a60b108898784df8984c87f6054c9c4fcacc47e4803c1ce1" dependencies = [ "aws-lc-rs", "base64 0.22.1", - "ed25519-dalek", + "ed25519-dalek 2.2.0", "getrandom 0.2.17", "hmac 0.12.1", "js-sys", "p256 0.13.2", - "p384", + "p384 0.13.1", "pem", "rand 0.8.6", - "rsa", + "rsa 0.9.10", "serde", "serde_json", "sha2 0.10.9", @@ -5177,6 +5450,26 @@ dependencies = [ "simple_asn1", ] +[[package]] +name = "keccak" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e24a010dd405bd7ed803e5253182815b41bf2e6a80cc3bfc066658e03a198aa" +dependencies = [ + "cfg-if 1.0.4", + "cpufeatures 0.3.0", +] + +[[package]] +name = "kem" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "01737161ba802849cfd486b5bd209d38ba4943494c249a8126005170c7621edd" +dependencies = [ + "crypto-common 0.2.1", + "rand_core 0.10.1", +] + [[package]] name = "konst" version = "0.4.3" @@ -5693,6 +5986,12 @@ dependencies = [ "digest 0.11.3", ] +[[package]] +name = "md5" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ebb8d8732c6a6df3d8f032a82911cfc747e00efb95cc46e8d0acd5b5b88570c" + [[package]] name = "mea" version = "0.6.4" @@ -5867,12 +6166,37 @@ dependencies = [ "parking_lot 0.12.5", ] +[[package]] +name = "ml-kem" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e15f3e5b957493873e396a66914e83e616b6afe335cdef7efe5c6e1216aba66" +dependencies = [ + "hybrid-array", + "kem", + "module-lattice", + "pkcs8 0.11.0", + "rand_core 0.10.1", + "sha3 0.11.0", +] + [[package]] name = "mod_use" version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "21909324aa58f5c284d91cac514c6d081210901dc372d7c8ea6a9d7e0406097a" +[[package]] +name = "module-lattice" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c61b87c9683ab7cb1c6871d261ad5479b6b10ceb52c4352aaca3b5d35a8febe" +dependencies = [ + "ctutils", + "hybrid-array", + "num-traits", +] + [[package]] name = "moka" version = "0.12.15" @@ -5933,7 +6257,7 @@ dependencies = [ "md-5 0.10.6", "mongocrypt", "mongodb-internal-macros", - "pbkdf2", + "pbkdf2 0.12.2", "percent-encoding", "rand 0.9.4", "rustc_version_runtime", @@ -5984,7 +6308,7 @@ dependencies = [ "memchr", "mio 0.8.11", "monoio-macros", - "nix", + "nix 0.26.4", "once_cell", "pin-project-lite", "socket2 0.5.10", @@ -6090,6 +6414,18 @@ dependencies = [ "pin-utils", ] +[[package]] +name = "nix" +version = "0.31.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d" +dependencies = [ + "bitflags 2.11.1", + "cfg-if 1.0.4", + "cfg_aliases", + "libc", +] + [[package]] name = "noisy_float" version = "0.2.1" @@ -6174,17 +6510,6 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c6673768db2d862beb9b39a78fdcb1a69439615d5794a1be50caa9bc92c81967" -[[package]] -name = "num-derive" -version = "0.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ed3955f1a9c7c0c15e092f9c887db08b1fc683305fdf6eb6684f22555355e202" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.118", -] - [[package]] name = "num-integer" version = "0.1.46" @@ -7498,8 +7823,8 @@ dependencies = [ "futures", "log", "opendal-core", - "openssh", - "openssh-sftp-client", + "russh", + "russh-sftp", "serde", "tokio", ] @@ -7689,98 +8014,6 @@ dependencies = [ "uuid", ] -[[package]] -name = "openssh" -version = "0.11.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d534c4bfecb0ed71dea4db444a5922a294d15cf40e700548f27295e1feb0ef18" -dependencies = [ - "libc", - "once_cell", - "shell-escape", - "tempfile", - "thiserror 2.0.18", - "tokio", -] - -[[package]] -name = "openssh-sftp-client" -version = "0.15.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e7a575af2d0eaa7c8f96368924994a3ae1d484fd2b19750cc1049c1100a42e82" -dependencies = [ - "bytes", - "derive_destructure2", - "futures-core", - "once_cell", - "openssh", - "openssh-sftp-client-lowlevel", - "openssh-sftp-error", - "pin-project", - "scopeguard", - "tokio", - "tokio-io-utility", - "tokio-util", - "tracing", -] - -[[package]] -name = "openssh-sftp-client-lowlevel" -version = "0.7.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a6d1a0e0eeb46100745a2c383c842042e1f04aa57a9c18aa41a16b6d4d58aeb0" -dependencies = [ - "awaitable", - "bytes", - "concurrent_arena", - "derive_destructure2", - "openssh-sftp-error", - "openssh-sftp-protocol", - "pin-project", - "tokio", - "tokio-io-utility", -] - -[[package]] -name = "openssh-sftp-error" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "12a702f18f0595b4578b21fd120ae7aa45f4298a8b28ddcb2397ace6f5a8251a" -dependencies = [ - "awaitable-error", - "openssh", - "openssh-sftp-protocol-error", - "ssh_format_error", - "thiserror 2.0.18", - "tokio", -] - -[[package]] -name = "openssh-sftp-protocol" -version = "0.24.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9c862e0c56553146306507f55958c11ff554e02c46de287e6976e50d815b350" -dependencies = [ - "bitflags 2.11.1", - "num-derive", - "num-traits", - "openssh-sftp-protocol-error", - "serde", - "ssh_format", - "vec-strings", -] - -[[package]] -name = "openssh-sftp-protocol-error" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42b54df62ccfd9a7708a83a9d60c46293837e478f9f4c0829360dcfa60ede8d2" -dependencies = [ - "serde", - "thiserror 2.0.18", - "vec-strings", -] - [[package]] name = "openssl" version = "0.10.79" @@ -7942,10 +8175,23 @@ checksum = "c9863ad85fa8f4460f9c48cb909d38a0d689dba1f6f6988a5e3e0d31071bcd4b" dependencies = [ "ecdsa 0.16.9", "elliptic-curve 0.13.8", - "primeorder", + "primeorder 0.13.6", "sha2 0.10.9", ] +[[package]] +name = "p256" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2c9239b2dbc807adbbe147e8cf72ea7450c3a0aabe62cb8e75ff4ec22e1f72a" +dependencies = [ + "ecdsa 0.17.0", + "elliptic-curve 0.14.1", + "primefield", + "primeorder 0.14.0", + "sha2 0.11.0", +] + [[package]] name = "p384" version = "0.13.1" @@ -7954,10 +8200,38 @@ checksum = "fe42f1670a52a47d448f14b6a5c61dd78fce51856e68edaa38f7ae3a46b8d6b6" dependencies = [ "ecdsa 0.16.9", "elliptic-curve 0.13.8", - "primeorder", + "primeorder 0.13.6", "sha2 0.10.9", ] +[[package]] +name = "p384" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d17b851e6b3e378ab4ecb07fa2ed23f4d15f075735f8fec9fa1e7bdce5f8301f" +dependencies = [ + "ecdsa 0.17.0", + "elliptic-curve 0.14.1", + "fiat-crypto 0.3.0", + "primefield", + "primeorder 0.14.0", + "sha2 0.11.0", +] + +[[package]] +name = "p521" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ad64cc32c2dc466317c12ee5853e61f159f9eab1fe7efade0395dc2e7b43449" +dependencies = [ + "base16ct 1.0.0", + "ecdsa 0.17.0", + "elliptic-curve 0.14.1", + "primefield", + "primeorder 0.14.0", + "sha2 0.11.0", +] + [[package]] name = "page_size" version = "0.6.0" @@ -7968,6 +8242,26 @@ dependencies = [ "winapi", ] +[[package]] +name = "pageant" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4f3a5ae18f65a85c67a77d18d42d3606c07948e3c17c1e5f74852b26589e88a5" +dependencies = [ + "base16ct 1.0.0", + "byteorder", + "bytes", + "delegate", + "futures", + "log", + "rand 0.10.1", + "sha2 0.11.0", + "thiserror 2.0.18", + "tokio", + "windows 0.62.2", + "windows-strings 0.5.1", +] + [[package]] name = "papaya" version = "0.2.4" @@ -8043,6 +8337,15 @@ dependencies = [ "subtle", ] +[[package]] +name = "password-hash" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aab41826031698d6ffcd9cff78ef56ef998e39dc7e5067cdfebe373842d4723b" +dependencies = [ + "phc", +] + [[package]] name = "paste" version = "1.0.15" @@ -8069,10 +8372,20 @@ checksum = "f8ed6a7761f76e3b9f92dfb0a60a6a6477c61024b775147ff0973a02653abaf2" dependencies = [ "digest 0.10.7", "hmac 0.12.1", - "password-hash", + "password-hash 0.5.0", "sha2 0.10.9", ] +[[package]] +name = "pbkdf2" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112d82ceb8c5bf524d9af484d4e4970c9fd5a0cc15ba14ad93dccd28873b0629" +dependencies = [ + "digest 0.11.3", + "hmac 0.13.0", +] + [[package]] name = "pdqselect" version = "0.1.0" @@ -8098,6 +8411,15 @@ dependencies = [ "base64ct", ] +[[package]] +name = "pem-rfc7468" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6305423e0e7738146434843d1694d621cce767262b2a86910beab705e4493d9" +dependencies = [ + "base64ct", +] + [[package]] name = "percent-encoding" version = "2.3.2" @@ -8131,6 +8453,16 @@ dependencies = [ "indexmap 2.14.0", ] +[[package]] +name = "phc" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44dc769b75f93afdddd8c7fa12d685292ddeff1e66f7f0f3a234cf1818afe892" +dependencies = [ + "base64ct", + "ctutils", +] + [[package]] name = "phf" version = "0.11.3" @@ -8282,6 +8614,16 @@ dependencies = [ "spki 0.7.3", ] +[[package]] +name = "pkcs1" +version = "0.8.0-rc.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "986d2e952779af96ea048f160fd9194e1751b4faea78bcf3ceb456efe008088e" +dependencies = [ + "der 0.8.1", + "spki 0.8.0", +] + [[package]] name = "pkcs5" version = "0.7.1" @@ -8291,12 +8633,29 @@ dependencies = [ "aes 0.8.4", "cbc 0.1.2", "der 0.7.10", - "pbkdf2", - "scrypt", + "pbkdf2 0.12.2", + "scrypt 0.11.0", "sha2 0.10.9", "spki 0.7.3", ] +[[package]] +name = "pkcs5" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63d440a804ec8d6fafbb6b84471e013286658d373248927692ab3366686220ca" +dependencies = [ + "aes 0.9.1", + "aes-gcm", + "cbc 0.2.0", + "der 0.8.1", + "pbkdf2 0.13.0", + "rand_core 0.10.1", + "scrypt 0.12.0", + "sha2 0.11.0", + "spki 0.8.0", +] + [[package]] name = "pkcs8" version = "0.9.0" @@ -8314,11 +8673,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7" dependencies = [ "der 0.7.10", - "pkcs5", + "pkcs5 0.7.1", "rand_core 0.6.4", "spki 0.7.3", ] +[[package]] +name = "pkcs8" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "451913da69c775a56034ea8d9003d27ee8948e12443eae7c038ba100a4f21cb7" +dependencies = [ + "der 0.8.1", + "pkcs5 0.8.1", + "rand_core 0.10.1", + "spki 0.8.0", +] + [[package]] name = "pkg-config" version = "0.3.33" @@ -8360,17 +8731,39 @@ dependencies = [ ] [[package]] -name = "polling" -version = "3.11.0" +name = "polling" +version = "3.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218" +dependencies = [ + "cfg-if 1.0.4", + "concurrent-queue", + "hermit-abi", + "pin-project-lite", + "rustix 1.1.4", + "windows-sys 0.61.2", +] + +[[package]] +name = "poly1305" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e2d0073b297041425c7c3df6eb4792d598a15323fe63346852b092eca02904c" +dependencies = [ + "cpufeatures 0.3.0", + "universal-hash", + "zeroize", +] + +[[package]] +name = "polyval" +version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218" +checksum = "f0fa31d631f2b2cb2a544d0aa321ce847a94764d701ca2becc411138b93d49cd" dependencies = [ - "cfg-if 1.0.4", - "concurrent-queue", - "hermit-abi", - "pin-project-lite", - "rustix 1.1.4", - "windows-sys 0.61.2", + "cpubits", + "cpufeatures 0.3.0", + "universal-hash", ] [[package]] @@ -8438,6 +8831,20 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "primefield" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c555a6e4eb7d4e158fcb028c835c3b8642206ddc279b5c6b202ef9a8bdb592f4" +dependencies = [ + "crypto-bigint 0.7.5", + "crypto-common 0.2.1", + "ff 0.14.0", + "rand_core 0.10.1", + "subtle", + "zeroize", +] + [[package]] name = "primeorder" version = "0.13.6" @@ -8447,6 +8854,19 @@ dependencies = [ "elliptic-curve 0.13.8", ] +[[package]] +name = "primeorder" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c9f42978c78a00e3d68f69fc03e57a234debae69da4020a4fb588fcdcd07b06" +dependencies = [ + "elliptic-curve 0.14.1", + "once_cell", + "primefield", + "serdect", + "wnaf", +] + [[package]] name = "probe" version = "0.5.2" @@ -9221,7 +9641,7 @@ dependencies = [ "pem", "percent-encoding", "reqsign-core", - "rsa", + "rsa 0.9.10", "serde", "serde_json", "sha1 0.11.0", @@ -9243,7 +9663,7 @@ dependencies = [ "jiff", "log", "percent-encoding", - "rsa", + "rsa 0.9.10", "serde", "serde_json", "sha1 0.11.0", @@ -9274,7 +9694,7 @@ dependencies = [ "percent-encoding", "reqsign-aws-v4", "reqsign-core", - "rsa", + "rsa 0.9.10", "serde", "serde_json", "tokio", @@ -9474,6 +9894,16 @@ dependencies = [ "subtle", ] +[[package]] +name = "rfc6979" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4a459cddafb3fe76b31fd8f1108007566c40301feb64dc7b54656eb7388172b" +dependencies = [ + "crypto-bigint 0.7.5", + "hmac 0.13.0", +] + [[package]] name = "ring" version = "0.17.14" @@ -9564,7 +9994,7 @@ dependencies = [ "num-bigint-dig", "num-integer", "num-traits", - "pkcs1", + "pkcs1 0.7.5", "pkcs8 0.10.2", "rand_core 0.6.4", "sha2 0.10.9", @@ -9574,6 +10004,25 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rsa" +version = "0.10.0-rc.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30b2aa4ba0d89f73d1e332df05be0eeab8840351c36ca5654341dfdb57bb3caf" +dependencies = [ + "const-oid 0.10.2", + "crypto-bigint 0.7.5", + "crypto-primes", + "digest 0.11.3", + "pkcs1 0.8.0-rc.4", + "pkcs8 0.11.0", + "rand_core 0.10.1", + "sha2 0.11.0", + "signature 3.0.0", + "spki 0.8.0", + "zeroize", +] + [[package]] name = "rstar" version = "0.8.4" @@ -9641,6 +10090,122 @@ version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4ade083ccbb4bf536df69d1f6432cc23deb7acccff86b183f3923a6fd56a1153" +[[package]] +name = "russh" +version = "0.62.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8b67b5a0d8068c89dcbe9d95df986af7a851d1f3c604525274c37468e60464f" +dependencies = [ + "aes 0.9.1", + "aws-lc-rs", + "bitflags 2.11.1", + "block-padding 0.4.2", + "byteorder", + "bytes", + "cbc 0.2.0", + "cipher 0.5.1", + "crypto-bigint 0.7.5", + "ctr", + "curve25519-dalek 5.0.0", + "data-encoding", + "delegate", + "der 0.8.1", + "digest 0.11.3", + "ecdsa 0.17.0", + "ed25519-dalek 3.0.0", + "elliptic-curve 0.14.1", + "enum_dispatch", + "flate2", + "futures", + "generic-array 1.4.2", + "getrandom 0.4.2", + "ghash", + "hex-literal", + "hmac 0.13.0", + "inout 0.2.2", + "internal-russh-num-bigint", + "keccak", + "log", + "md5", + "ml-kem", + "module-lattice", + "num-bigint", + "p256 0.14.0", + "p384 0.14.0", + "p521", + "pageant", + "pbkdf2 0.13.0", + "pkcs1 0.8.0-rc.4", + "pkcs5 0.8.1", + "pkcs8 0.11.0", + "polyval", + "rand 0.10.1", + "rand_core 0.10.1", + "rsa 0.10.0-rc.18", + "russh-cryptovec", + "russh-util", + "salsa20 0.11.0", + "scrypt 0.12.0", + "sec1 0.8.1", + "sha1 0.11.0", + "sha2 0.11.0", + "sha3 0.12.0", + "signature 3.0.0", + "spki 0.8.0", + "ssh-encoding", + "ssh-key", + "subtle", + "thiserror 2.0.18", + "tokio", + "typenum", + "universal-hash", + "zeroize", +] + +[[package]] +name = "russh-cryptovec" +version = "0.62.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3aec6cb630dbe85d72ffd7bcd95f07e1bd69f9f270ee8adfa1afe443a6331438" +dependencies = [ + "log", + "nix 0.31.3", + "ssh-encoding", + "windows-sys 0.61.2", +] + +[[package]] +name = "russh-sftp" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed8949eca4163c18a8f59ff96d32cf61e9c13b9735e21ef32b3907f4aafa1a9" +dependencies = [ + "bitflags 2.11.1", + "bytes", + "chrono", + "dashmap 6.2.1", + "gloo-timers 0.4.0", + "log", + "serde", + "serde_bytes", + "thiserror 2.0.18", + "tokio", + "tokio-util", + "wasm-bindgen-futures", +] + +[[package]] +name = "russh-util" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "668424a5dde0bcb45b55ba7de8476b93831b4aa2fa6947e145f3b053e22c60b6" +dependencies = [ + "chrono", + "tokio", + "wasm-bindgen", + "wasm-bindgen-futures", +] + [[package]] name = "rust-ini" version = "0.21.3" @@ -9882,6 +10447,16 @@ dependencies = [ "cipher 0.4.4", ] +[[package]] +name = "salsa20" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f874456e72520ff1375a06c588eaf074b0f01f9e9e1aada45bd9b7954a6e42c" +dependencies = [ + "cfg-if 1.0.4", + "cipher 0.5.1", +] + [[package]] name = "same-file" version = "1.0.6" @@ -9918,12 +10493,24 @@ version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0516a385866c09368f0b5bcd1caff3366aace790fcd46e2bb032697bb172fd1f" dependencies = [ - "password-hash", - "pbkdf2", - "salsa20", + "password-hash 0.5.0", + "pbkdf2 0.12.2", + "salsa20 0.10.2", "sha2 0.10.9", ] +[[package]] +name = "scrypt" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d87af57419b594aa23fa95f09f0e06d80d84ba01c26148c43844cad6ff4485f0" +dependencies = [ + "cfg-if 1.0.4", + "pbkdf2 0.13.0", + "salsa20 0.11.0", + "sha2 0.11.0", +] + [[package]] name = "sct" version = "0.7.1" @@ -9968,6 +10555,20 @@ dependencies = [ "zeroize", ] +[[package]] +name = "sec1" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d56d437c2f19203ce5f7122e507831de96f3d2d4d3be5af44a0b0a09d8a80e4d" +dependencies = [ + "base16ct 1.0.0", + "ctutils", + "der 0.8.1", + "hybrid-array", + "subtle", + "zeroize", +] + [[package]] name = "security-framework" version = "2.11.1" @@ -10129,6 +10730,16 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "serdect" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66cf8fedced2fcf12406bcb34223dffb92eaf34908ede12fed414c82b7f00b3e" +dependencies = [ + "base16ct 1.0.0", + "serde", +] + [[package]] name = "sha-1" version = "0.10.1" @@ -10190,6 +10801,27 @@ dependencies = [ "digest 0.11.3", ] +[[package]] +name = "sha3" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be176f1a57ce4e3d31c1a166222d9768de5954f811601fb7ca06fc8203905ce1" +dependencies = [ + "digest 0.11.3", + "keccak", +] + +[[package]] +name = "sha3" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc9bad02c26382724b2d2692c6f179285e4b54eeecd7968f52a50059c3c11759" +dependencies = [ + "digest 0.11.3", + "keccak", + "sponge-cursor", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -10199,12 +10831,6 @@ dependencies = [ "lazy_static", ] -[[package]] -name = "shell-escape" -version = "0.1.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "45bb67a18fa91266cc7807181f62f9178a6873bfad7dc788c42e6430db40184f" - [[package]] name = "shellexpand" version = "3.1.2" @@ -10264,6 +10890,16 @@ dependencies = [ "rand_core 0.6.4", ] +[[package]] +name = "signature" +version = "3.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28d567dcbaf0049cb8ac2608a76cd95ff9e4412e1899d389ee400918ca7537f5" +dependencies = [ + "digest 0.11.3", + "rand_core 0.10.1", +] + [[package]] name = "simd-adler32" version = "0.3.9" @@ -10447,6 +11083,22 @@ dependencies = [ "der 0.7.10", ] +[[package]] +name = "spki" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d9efca8738c78ee9484207732f728b1ef517bbb1833d6fc0879ca898a522f6f" +dependencies = [ + "base64ct", + "der 0.8.1", +] + +[[package]] +name = "sponge-cursor" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a0219bd7d979d58245a4f41f695e1ac9f8befdffadd7f61f1bae9e39abc6620" + [[package]] name = "sqlx" version = "0.8.6" @@ -10555,7 +11207,7 @@ dependencies = [ "futures-util", "generic-array 0.14.7", "hex", - "hkdf", + "hkdf 0.12.4", "hmac 0.12.1", "itoa", "log", @@ -10564,7 +11216,7 @@ dependencies = [ "once_cell", "percent-encoding", "rand 0.8.6", - "rsa", + "rsa 0.9.10", "serde", "sha1 0.10.6", "sha2 0.10.9", @@ -10593,7 +11245,7 @@ dependencies = [ "futures-core", "futures-util", "hex", - "hkdf", + "hkdf 0.12.4", "hmac 0.12.1", "home", "itoa", @@ -10638,23 +11290,62 @@ dependencies = [ ] [[package]] -name = "ssh_format" -version = "0.14.1" +name = "ssh-cipher" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d801accda99469cde6d73da741422610fdf6508a72d9a69d1b55cb241c720597" +dependencies = [ + "aead", + "aes 0.9.1", + "aes-gcm", + "chacha20", + "cipher 0.5.1", + "ctutils", + "des", + "poly1305", + "ssh-encoding", + "zeroize", +] + +[[package]] +name = "ssh-encoding" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24ab31081d1c9097c327ec23550858cb5ffb4af6b866c1ef4d728455f01f3304" +checksum = "7b54d0ed0498daf3f78d82e00e28c8eec9d75a067c4cfbcc7a0f7d0f4077749e" dependencies = [ + "base64ct", "bytes", - "serde", - "ssh_format_error", + "crypto-bigint 0.7.5", + "ctutils", + "digest 0.11.3", + "pem-rfc7468 1.0.0", + "zeroize", ] [[package]] -name = "ssh_format_error" -version = "0.1.0" +name = "ssh-key" +version = "0.7.0-rc.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "be3c6519de7ca611f71ef7e8a56eb57aa1c818fecb5242d0a0f39c83776c210c" +checksum = "f9a32fae177b74a22aa9c5b01bf7e68b33545be32d9e381e248058d2adc15ce3" dependencies = [ - "serde", + "argon2 0.6.0-rc.8", + "bcrypt-pbkdf", + "ctutils", + "ed25519-dalek 3.0.0", + "hex", + "hmac 0.13.0", + "p256 0.14.0", + "p384 0.14.0", + "p521", + "rand_core 0.10.1", + "rsa 0.10.0-rc.18", + "sec1 0.8.1", + "sha1 0.11.0", + "sha2 0.11.0", + "signature 3.0.0", + "ssh-cipher", + "ssh-encoding", + "zeroize", ] [[package]] @@ -10863,7 +11554,7 @@ dependencies = [ "ahash 0.8.12", "ammonia", "anyhow", - "argon2", + "argon2 0.5.3", "async-channel 2.5.0", "async-stream", "base64 0.22.1", @@ -10903,7 +11594,7 @@ dependencies = [ "object_store", "parking_lot 0.12.5", "path-clean", - "pbkdf2", + "pbkdf2 0.12.2", "phf 0.13.1", "pin-project-lite", "quick_cache", @@ -10917,7 +11608,7 @@ dependencies = [ "roaring", "rust-stemmers", "rust_decimal", - "scrypt", + "scrypt 0.11.0", "semver", "serde", "serde_json", @@ -11219,12 +11910,6 @@ dependencies = [ "synchrony", ] -[[package]] -name = "thin-vec" -version = "0.2.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b0f7e269b48f0a7dd0146680fa24b50cc67fc0373f086a5b2f99bd084639b482" - [[package]] name = "thiserror" version = "1.0.69" @@ -11424,16 +12109,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "tokio-io-utility" -version = "0.7.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8d672654d175710e52c7c41f6aec77c62b3c0954e2a7ebce9049d1e94ed7c263" -dependencies = [ - "bytes", - "tokio", -] - [[package]] name = "tokio-macros" version = "2.7.0" @@ -11542,6 +12217,7 @@ dependencies = [ "futures-core", "futures-io", "futures-sink", + "futures-util", "pin-project-lite", "tokio", ] @@ -11919,11 +12595,6 @@ name = "triomphe" version = "0.1.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dd69c5aa8f924c7519d6372789a74eac5b94fb0f8fcf0d4a97eb0bfc3e785f39" -dependencies = [ - "arc-swap", - "serde", - "stable_deref_trait", -] [[package]] name = "try-lock" @@ -12060,6 +12731,16 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" +[[package]] +name = "universal-hash" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f4987bdc12753382e0bec4a65c50738ffaabc998b9cdd1f952fb5f39b0048a96" +dependencies = [ + "crypto-common 0.2.1", + "ctutils", +] + [[package]] name = "unsigned-varint" version = "0.8.0" @@ -12150,16 +12831,6 @@ version = "0.2.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" -[[package]] -name = "vec-strings" -version = "0.4.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c8509489e2a7ee219522238ad45fd370bec6808811ac15ac6b07453804e77659" -dependencies = [ - "serde", - "thin-vec", -] - [[package]] name = "version_check" version = "0.9.5" @@ -13087,6 +13758,17 @@ dependencies = [ "wasmparser", ] +[[package]] +name = "wnaf" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab12e7090f27e2ffd9322651492942d50c2926094af30601e1964337db39daf1" +dependencies = [ + "ff 0.14.0", + "group 0.14.0", + "hybrid-array", +] + [[package]] name = "writeable" version = "0.6.3" @@ -13352,9 +14034,9 @@ dependencies = [ [[package]] name = "zeroize" -version = "1.8.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" [[package]] name = "zerotrie" diff --git a/core/services/sftp/Cargo.toml b/core/services/sftp/Cargo.toml index 6774740990f9..f61da6348345 100644 --- a/core/services/sftp/Cargo.toml +++ b/core/services/sftp/Cargo.toml @@ -38,10 +38,10 @@ bytes = { workspace = true } fastpool = "1.0.2" futures = { workspace = true } log = { workspace = true } -openssh = "0.11.0" -openssh-sftp-client = { version = "0.15.3", features = ["openssh", "tracing"] } +russh = "0.62.4" +russh-sftp = "2.3.0" serde = { workspace = true, features = ["derive"] } -tokio = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] } [dev-dependencies] anyhow = "1.0" diff --git a/core/services/sftp/src/backend.rs b/core/services/sftp/src/backend.rs index ba977218d3a3..3f90222f7ba3 100644 --- a/core/services/sftp/src/backend.rs +++ b/core/services/sftp/src/backend.rs @@ -15,16 +15,23 @@ // specific language governing permissions and limitations // under the License. -use std::path::Path; -use std::path::PathBuf; use std::sync::Arc; use log::debug; -use openssh::KnownHosts; +use russh_sftp::client::RawSftpSession; +use russh_sftp::client::error::Error as SftpClientError; +use russh_sftp::extensions::HardlinkExtension; +use russh_sftp::protocol::{FileAttributes, OpenFlags, Packet, StatusCode}; use super::SFTP_SCHEME; use super::config::SftpConfig; +use super::core::KnownHostsStrategy; +use super::core::PIPELINE_DEPTH; +use super::core::POSIX_RENAME; use super::core::SftpCore; +use super::core::close_handle_detached; +use super::core::is_eof; +use super::core::is_sftp_failure; use super::core::is_sftp_protocol_error; use super::core::parse_sftp_error; use super::core::to_metadata; @@ -33,10 +40,7 @@ use super::reader::*; use opendal_core::raw::*; use opendal_core::*; -/// SFTP services support. (only works on unix) -/// -/// If you are interested in working on windows, please refer to [this](https://github.com/apache/opendal/issues/2963) issue. -/// Welcome to leave your comments or make contributions. +/// SFTP services support. /// /// Warning: Maximum number of file holdings is depending on the remote system configuration. /// @@ -50,7 +54,7 @@ pub struct SftpBuilder { impl SftpBuilder { /// set endpoint for sftp backend. - /// The format is same as `openssh`, using either `[user@]hostname` or `ssh://[user@]hostname[:port]`. A username or port that is specified in the endpoint overrides the one set in the builder (but does not change the builder). + /// The format is either `[user@]hostname[:port]` or `ssh://[user@]hostname[:port]`, and the port defaults to 22. A username that is specified in the endpoint overrides the one set in the builder (but does not change the builder). pub fn endpoint(mut self, endpoint: &str) -> Self { self.config.endpoint = if endpoint.is_empty() { None @@ -143,11 +147,11 @@ impl Builder for SftpBuilder { Some(v) => { let v = v.to_lowercase(); if v == "strict" { - KnownHosts::Strict + KnownHostsStrategy::Strict } else if v == "accept" { - KnownHosts::Accept + KnownHostsStrategy::Accept } else if v == "add" { - KnownHosts::Add + KnownHostsStrategy::Add } else { return Err(Error::new( ErrorKind::ConfigInvalid, @@ -155,7 +159,7 @@ impl Builder for SftpBuilder { )); } } - None => KnownHosts::Strict, + None => KnownHostsStrategy::Strict, }; let info = ServiceInfo::new(SFTP_SCHEME, root.as_str(), ""); @@ -197,6 +201,81 @@ impl Builder for SftpBuilder { } } +/// Copies a remote file through the client, keeping several reads and writes +/// in flight. +/// +/// SFTP has no server-side copy, so the payload has to make the round trip. +async fn stream_copy( + session: &Arc, + src: &str, + dst: &str, + chunk: u64, +) -> Result<()> { + let mut offset = 0u64; + + loop { + let reads = (0..PIPELINE_DEPTH).map(|i| { + let session = session.clone(); + let src = src.to_string(); + let at = offset + i as u64 * chunk; + async move { session.read(src, at, chunk as u32).await } + }); + + let mut writes = Vec::with_capacity(PIPELINE_DEPTH); + let mut consumed = 0u64; + let mut eof = false; + + for (i, result) in futures::future::join_all(reads) + .await + .into_iter() + .enumerate() + { + let data = match result { + Ok(data) => data.data, + Err(e) if is_eof(&e) => { + eof = true; + break; + } + Err(e) => return Err(parse_sftp_error(e)), + }; + + let len = data.len() as u64; + if len == 0 { + eof = true; + break; + } + + let session = session.clone(); + let dst = dst.to_string(); + let at = offset + i as u64 * chunk; + writes.push(async move { session.write(dst, at, data).await }); + consumed += len; + + // A short read leaves a gap, so the rest of this batch is dropped + // and re-requested from the corrected offset on the next pass. + if len < chunk { + break; + } + } + + futures::future::try_join_all(writes) + .await + .map_err(parse_sftp_error)?; + + if consumed == 0 { + break; + } + + offset += consumed; + + if eof { + break; + } + } + + Ok(()) +} + #[derive(Clone, Debug)] pub struct SftpBackend { pub core: Arc, @@ -223,36 +302,40 @@ impl Service for SftpBackend { path: &str, _: OpCreateDir, ) -> Result { - let client = self.core.connect().await?; - let mut fs = client.fs(); - fs.set_cwd(&self.core.root); + let conn = self.core.connect().await?; - let paths = Path::new(&path).components(); - let mut current = PathBuf::from(&self.core.root); - for p in paths { - current = current.join(p); - let res = fs.create_dir(p).await; + // Create every missing component of the requested directory chain. + let mut current = self.core.abs_path(""); + for component in path.split('/').filter(|v| !v.is_empty()) { + if !current.is_empty() && !current.ends_with('/') { + current.push('/'); + } + current.push_str(component); - if let Err(e) = res { + if let Err(e) = conn + .session + .mkdir(current.as_str(), FileAttributes::default()) + .await + && !is_sftp_protocol_error(&e) + { // ignore error if dir already exists - if !is_sftp_protocol_error(&e) { - return Err(parse_sftp_error(e)); - } + return Err(parse_sftp_error(e)); } - fs.set_cwd(¤t); } Ok(RpCreateDir::default()) } async fn stat(&self, _ctx: &OperationContext, path: &str, _: OpStat) -> Result { - let client = self.core.connect().await?; - let mut fs = client.fs(); - fs.set_cwd(&self.core.root); + let conn = self.core.connect().await?; - let meta: Metadata = to_metadata(fs.metadata(path).await.map_err(parse_sftp_error)?); + let attrs = conn + .session + .stat(self.core.abs_path(path)) + .await + .map_err(parse_sftp_error)?; - Ok(RpStat::new(meta)) + Ok(RpStat::new(to_metadata(&attrs.attrs))) } fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result { let output: oio::StreamReader = { @@ -297,26 +380,48 @@ impl Service for SftpBackend { let from = from.to_string(); let to = to.to_string(); Ok(oio::OneShotCopier::new(async move { - let client = backend.core.connect().await?; - - let mut fs = client.fs(); - fs.set_cwd(&backend.core.root); - if let Some((dir, _)) = to.rsplit_once('/') { backend .create_dir(&ctx, dir, OpCreateDir::default()) .await?; } - let src = fs.canonicalize(&from).await.map_err(parse_sftp_error)?; - let dst = fs.canonicalize(&to).await.map_err(parse_sftp_error)?; - let mut src_file = client.open(&src).await.map_err(parse_sftp_error)?; - let mut dst_file = client.create(dst).await.map_err(parse_sftp_error)?; + let core = &backend.core; + let conn = core.connect().await?; + let session = conn.session.clone(); - src_file - .copy_all_to(&mut dst_file) + let src = session + .open( + core.abs_path(&from), + OpenFlags::READ, + FileAttributes::default(), + ) + .await + .map_err(parse_sftp_error)? + .handle; + + let dst = match session + .open( + core.abs_path(&to), + OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE, + FileAttributes::default(), + ) .await - .map_err(parse_sftp_error)?; + { + Ok(handle) => handle.handle, + Err(e) => { + close_handle_detached(session, src); + return Err(parse_sftp_error(e)); + } + }; + + let chunk = conn.read_len.min(conn.write_len) as u64; + let result = stream_copy(&session, &src, &dst, chunk).await; + + // Release both handles regardless of how the transfer ended. + close_handle_detached(session.clone(), src); + close_handle_detached(session, dst); + result?; Ok(Metadata::default()) })) @@ -329,15 +434,66 @@ impl Service for SftpBackend { to: &str, _: OpRename, ) -> Result { - let client = self.core.connect().await?; - - let mut fs = client.fs(); - fs.set_cwd(&self.core.root); - if let Some((dir, _)) = to.rsplit_once('/') { self.create_dir(ctx, dir, OpCreateDir::default()).await?; } - fs.rename(from, to).await.map_err(parse_sftp_error)?; + + let conn = self.core.connect().await?; + let from = self.core.abs_path(from); + let to = self.core.abs_path(to); + + if conn.posix_rename { + // `posix-rename@openssh.com` replaces an existing destination + // atomically, which plain SFTP v3 `rename` refuses to do. + let payload: Vec = HardlinkExtension { + oldpath: from, + newpath: to, + } + .try_into() + .map_err(|err| { + Error::new( + ErrorKind::Unexpected, + "sftp failed to encode rename request", + ) + .set_source(err) + })?; + + match conn + .session + .extended(POSIX_RENAME, payload) + .await + .map_err(parse_sftp_error)? + { + Packet::Status(status) if status.status_code == StatusCode::Ok => {} + Packet::Status(status) => { + return Err(parse_sftp_error(SftpClientError::Status(status))); + } + _ => { + return Err(Error::new( + ErrorKind::Unexpected, + "sftp rename returned an unexpected packet", + )); + } + } + } else { + // Plain SFTP v3 `rename` refuses an existing destination. Try it + // first and only clear the destination once it is the thing in the + // way, so a missing or unreadable source cannot destroy the target. + match conn.session.rename(from.clone(), to.clone()).await { + Ok(_) => {} + Err(e) if is_sftp_failure(&e) => { + conn.session + .remove(to.clone()) + .await + .map_err(parse_sftp_error)?; + conn.session + .rename(from, to) + .await + .map_err(parse_sftp_error)?; + } + Err(e) => return Err(parse_sftp_error(e)), + } + } Ok(RpRename::default()) } diff --git a/core/services/sftp/src/core.rs b/core/services/sftp/src/core.rs index 84ec65004102..91e36e714379 100644 --- a/core/services/sftp/src/core.rs +++ b/core/services/sftp/src/core.rs @@ -15,18 +15,164 @@ // specific language governing permissions and limitations // under the License. +use std::fmt::Debug; +use std::sync::Arc; + use fastpool::{ManageObject, ObjectStatus, bounded}; use log::debug; use opendal_core::raw::*; use opendal_core::*; -use openssh::KnownHosts; -use openssh::SessionBuilder; -use openssh_sftp_client::Sftp; -use openssh_sftp_client::SftpOptions; -use std::fmt::Debug; -use std::path::Path; -use std::path::PathBuf; -use std::sync::Arc; +use russh::client; +use russh::keys::PrivateKeyWithHashAlg; +use russh_sftp::client::RawSftpSession; +use russh_sftp::client::rawsession::Limits; +use russh_sftp::protocol::FileAttributes; + +/// Largest payload we ever ask the remote server for in a single SFTP packet. +/// +/// The effective value is additionally clamped by the limits advertised through +/// the `limits@openssh.com` extension. +const MAX_CHUNK_SIZE: u32 = 256 * 1024; + +/// Conservative payload size used when the server does not advertise limits. +/// +/// SFTP v3 has no negotiation for this, and 32 KiB is the value the OpenSSH +/// client itself falls back to. +const DEFAULT_CHUNK_SIZE: u32 = 32 * 1024; + +/// Number of SFTP read/write packets kept in flight at once. +/// +/// SFTP is a request/response protocol, so throughput on links with non-trivial +/// latency is bound by the number of outstanding requests rather than bandwidth. +pub const PIPELINE_DEPTH: usize = 8; + +/// OpenSSH extension providing `rename` with POSIX overwrite semantics. +pub const POSIX_RENAME: &str = "posix-rename@openssh.com"; + +/// Number of SSH handshakes allowed to run at the same time. +/// +/// Servers cap how many connections may be mid-authentication at once +/// (`MaxStartups` defaults to `10:30:100` in OpenSSH, which starts dropping +/// connections past ten), so growing the pool in a burst must be throttled. +const MAX_CONCURRENT_HANDSHAKES: usize = 4; + +/// Number of agent identities tried before giving up. +/// +/// Each attempt counts against the server's `MaxAuthTries`, which defaults to +/// six in OpenSSH and drops the connection once exceeded. +#[cfg(unix)] +const MAX_AGENT_IDENTITIES: usize = 4; + +/// Specifies how `SftpBackend` validates the remote host key. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum KnownHostsStrategy { + /// The host key must be present in `known_hosts` and match. + /// + /// This mirrors `ssh -o StrictHostKeyChecking=yes`. + Strict, + /// Accept a previously unseen host key and record it in `known_hosts`. + /// + /// A changed key is still rejected. This mirrors + /// `ssh -o StrictHostKeyChecking=accept-new`. + Add, + /// Accept whatever key the server presents. + /// + /// This mirrors `ssh -o StrictHostKeyChecking=no`. + Accept, +} + +/// Validates the remote host key according to [`KnownHostsStrategy`]. +struct SshHandler { + host: String, + port: u16, + strategy: KnownHostsStrategy, +} + +impl client::Handler for SshHandler { + type Error = russh::Error; + + async fn check_server_key( + &mut self, + server_public_key: &russh::keys::ssh_key::PublicKey, + ) -> std::result::Result { + match self.strategy { + KnownHostsStrategy::Accept => Ok(true), + KnownHostsStrategy::Strict => { + Ok( + russh::keys::check_known_hosts(&self.host, self.port, server_public_key) + .unwrap_or(false), + ) + } + KnownHostsStrategy::Add => { + match russh::keys::check_known_hosts(&self.host, self.port, server_public_key) { + Ok(true) => Ok(true), + // The host is unknown: record it and continue. + Ok(false) => { + if let Err(err) = russh::keys::known_hosts::learn_known_hosts( + &self.host, + self.port, + server_public_key, + ) { + // Nothing was pinned, so every later connection also + // takes this branch and `add` silently behaves like + // `accept`. Surface it rather than hiding it. + log::warn!( + "sftp accepted host key for {}:{} but could not record it in known_hosts, \ + so it will not be verified on future connections: {err}", + self.host, + self.port + ); + } + Ok(true) + } + // The host is known but the key changed: refuse to connect. + Err(_) => Ok(false), + } + } + } + } +} + +/// A detachable reference to a live SFTP session. +/// +/// Readers, writers, and listers hold one of these instead of the pooled +/// connection itself, so a long-running stream returns its slot to the pool +/// immediately. Remote servers usually cap concurrent SSH connections +/// (`MaxStartups`), so keeping a slot for the lifetime of a stream would +/// otherwise exhaust the server well before the pool. +#[derive(Clone)] +pub struct SftpSessionRef { + pub session: Arc, + /// Largest payload accepted by the server for a single read. + pub read_len: u32, + /// Largest payload accepted by the server for a single write. + pub write_len: u32, + /// Whether the server offers `posix-rename@openssh.com`. + pub posix_rename: bool, + /// Keeps the SSH connection open for as long as any reference is alive. + _ssh: Arc>, +} + +/// A live SFTP session together with the SSH connection that carries it. +pub struct SftpConnection { + inner: SftpSessionRef, +} + +impl SftpConnection { + /// Returns a reference that keeps the session alive after the pooled + /// connection is released. + pub fn session_ref(&self) -> SftpSessionRef { + self.inner.clone() + } +} + +impl std::ops::Deref for SftpConnection { + type Target = SftpSessionRef; + + fn deref(&self) -> &Self::Target { + &self.inner + } +} pub struct SftpCore { pub info: ServiceInfo, @@ -53,7 +199,7 @@ impl SftpCore { root: String, user: Option, key: Option, - known_hosts_strategy: KnownHosts, + known_hosts_strategy: KnownHostsStrategy, ) -> Self { let client = bounded::Pool::new( bounded::PoolConfig::new(64), @@ -63,6 +209,7 @@ impl SftpCore { user, key, known_hosts_strategy, + handshakes: tokio::sync::Semaphore::new(MAX_CONCURRENT_HANDSHAKES), }, ); @@ -75,6 +222,36 @@ impl SftpCore { } } + /// Resolves an OpenDAL path against the configured root. + /// + /// An empty root leaves the path relative, so the remote server resolves it + /// against the login directory. + pub fn abs_path(&self, path: &str) -> String { + let path = path.trim_start_matches('/'); + + if self.root.is_empty() { + let path = path.trim_end_matches('/'); + if path.is_empty() { + return ".".to_string(); + } + return path.to_string(); + } + + let mut buf = String::with_capacity(self.root.len() + path.len()); + buf.push_str(&self.root); + buf.push_str(path); + + // A trailing separator carries no meaning in SFTP, and OpenSSH on + // Windows rejects such a path outright instead of reporting that it + // does not exist, which turns a missing directory into an opaque + // failure. + while buf.len() > 1 && buf.ends_with('/') { + buf.pop(); + } + + buf + } + pub async fn connect(&self) -> Result> { let fut = self.client.get(); @@ -95,56 +272,247 @@ pub struct Manager { root: String, user: Option, key: Option, - known_hosts_strategy: KnownHosts, + known_hosts_strategy: KnownHostsStrategy, + /// Throttles concurrent SSH handshakes, see [`MAX_CONCURRENT_HANDSHAKES`]. + handshakes: tokio::sync::Semaphore, } -impl ManageObject for Manager { - type Object = Sftp; - type Error = Error; +impl Manager { + /// Authenticates the session, preferring the configured key and otherwise + /// falling back to an SSH agent and the usual default key locations. + async fn authenticate( + &self, + handle: &mut client::Handle, + user: &str, + ) -> Result<()> { + if let Some(key) = &self.key { + let private_key = russh::keys::load_secret_key(key, None).map_err(|err| { + Error::new(ErrorKind::ConfigInvalid, "sftp failed to load private key") + .set_source(err) + })?; + + if self + .try_publickey(handle, user, private_key) + .await? + .is_some() + { + return Ok(()); + } - async fn create(&self) -> Result { - let mut session = SessionBuilder::default(); + return Err(Error::new( + ErrorKind::PermissionDenied, + "sftp public key authentication failed", + )); + } - if let Some(user) = &self.user { - session.user(user.clone()); + if self.try_agent(handle, user).await? { + return Ok(()); } - if let Some(key) = &self.key { - session.keyfile(key); + for candidate in default_key_paths() { + let Ok(private_key) = russh::keys::load_secret_key(&candidate, None) else { + continue; + }; + if self + .try_publickey(handle, user, private_key) + .await? + .is_some() + { + return Ok(()); + } } - session.known_hosts_check(self.known_hosts_strategy.clone()); + Err(Error::new( + ErrorKind::PermissionDenied, + "sftp authentication failed: no usable credentials, set `key` to a private key path", + )) + } + + async fn try_publickey( + &self, + handle: &mut client::Handle, + user: &str, + private_key: russh::keys::PrivateKey, + ) -> Result> { + // RSA keys must be signed with the strongest hash the server accepts; + // servers commonly reject the legacy SHA-1 `ssh-rsa` signatures. + let hash_alg = handle + .best_supported_rsa_hash() + .await + .map_err(parse_ssh_error)? + .flatten(); - let session = session - .connect(&self.endpoint) + let result = handle + .authenticate_publickey( + user, + PrivateKeyWithHashAlg::new(Arc::new(private_key), hash_alg), + ) .await .map_err(parse_ssh_error)?; - let sftp = Sftp::from_session(session, SftpOptions::default()) + Ok(result.success().then_some(())) + } + + async fn try_agent(&self, handle: &mut client::Handle, user: &str) -> Result { + #[cfg(unix)] + { + use russh::keys::agent::AgentIdentity; + + let Ok(mut agent) = russh::keys::agent::client::AgentClient::connect_env().await else { + return Ok(false); + }; + let Ok(identities) = agent.request_identities().await else { + return Ok(false); + }; + + let hash_alg = handle + .best_supported_rsa_hash() + .await + .map_err(parse_ssh_error)? + .flatten(); + + // Every attempt counts against the server's `MaxAuthTries` (6 by + // default), and exhausting it disconnects instead of reporting a + // clean authentication failure. + for identity in identities.into_iter().take(MAX_AGENT_IDENTITIES) { + let AgentIdentity::PublicKey { key, .. } = identity else { + continue; + }; + + let Ok(result) = handle + .authenticate_publickey_with(user, key, hash_alg, &mut agent) + .await + else { + continue; + }; + if result.success() { + return Ok(true); + } + } + } + + let _ = (handle, user); + Ok(false) + } +} + +impl ManageObject for Manager { + type Object = SftpConnection; + type Error = Error; + + async fn create(&self) -> Result { + let _permit = self.handshakes.acquire().await.map_err(|err| { + Error::new(ErrorKind::Unexpected, "sftp connection limiter closed").set_source(err) + })?; + + let (endpoint_user, host, port) = parse_endpoint(&self.endpoint)?; + + // A user encoded in the endpoint takes precedence over the builder value. + let user = endpoint_user + .or_else(|| self.user.clone()) + .or_else(default_user) + .ok_or_else(|| { + Error::new( + ErrorKind::ConfigInvalid, + "sftp user is not set and cannot be inferred from the environment", + ) + })?; + + let handler = SshHandler { + host: host.clone(), + port, + strategy: self.known_hosts_strategy, + }; + + let mut handle = client::connect( + Arc::new(client::Config::default()), + (host.as_str(), port), + handler, + ) + .await + .map_err(parse_ssh_error)?; + + self.authenticate(&mut handle, &user).await?; + + let channel = handle + .channel_open_session() + .await + .map_err(parse_ssh_error)?; + channel + .request_subsystem(true, "sftp") .await - .map_err(parse_sftp_error)?; + .map_err(parse_ssh_error)?; + + let config = russh_sftp::client::Config { + // Every request is a single round trip, so allow slow servers more + // room than the crate default of 10 seconds. + request_timeout_secs: 60, + ..Default::default() + }; + let mut session = RawSftpSession::new_with_config(channel.into_stream(), config); + + let version = session.init().await.map_err(parse_sftp_error)?; + + // SFTP v3 `rename` fails when the destination exists; the OpenSSH + // extension provides POSIX overwrite semantics instead. + let posix_rename = version.extensions.contains_key(POSIX_RENAME); + + let mut read_len = DEFAULT_CHUNK_SIZE; + let mut write_len = DEFAULT_CHUNK_SIZE; + if version + .extensions + .get(russh_sftp::extensions::LIMITS) + .is_some_and(|v| v == "1") + { + let limits = Limits::from(session.limits().await.map_err(parse_sftp_error)?); + session.set_limits(limits); + + if let Some(len) = limits.read_len { + read_len = clamp_chunk(len); + } + if let Some(len) = limits.write_len { + write_len = clamp_chunk(len); + } + + // The server rejects a *serialized* packet larger than this, so the + // payload must leave room for the request header. OpenSSH reserves + // the same 1 KiB between its packet and read/write limits. + if let Some(packet_len) = limits.packet_len { + let budget = clamp_chunk(packet_len.saturating_sub(1024)); + read_len = read_len.min(budget); + write_len = write_len.min(budget); + } + } + + let session = Arc::new(session); if !self.root.is_empty() { - let mut fs = sftp.fs(); - - let paths = Path::new(&self.root).components(); - let mut current = PathBuf::new(); - for p in paths { - current.push(p); - let res = fs.create_dir(p).await; - - if let Err(e) = res { - // ignore error if dir already exists - if !is_sftp_protocol_error(&e) { - return Err(parse_sftp_error(e)); - } + // Create the root directory chain, ignoring components that exist. + let mut current = String::new(); + for component in self.root.split('/').filter(|v| !v.is_empty()) { + current.push('/'); + current.push_str(component); + + if let Err(e) = session + .mkdir(current.as_str(), FileAttributes::default()) + .await + && !is_sftp_protocol_error(&e) + { + return Err(parse_sftp_error(e)); } - fs.set_cwd(¤t); } } debug!("sftp connection created at {}", self.root); - Ok(sftp) + Ok(SftpConnection { + inner: SftpSessionRef { + session, + read_len, + write_len, + posix_rename, + _ssh: Arc::new(handle), + }, + }) } // Check if connect valid by checking the root path. @@ -153,28 +521,115 @@ impl ManageObject for Manager { o: &mut Self::Object, _: &ObjectStatus, ) -> Result<(), Self::Error> { - match o.fs().metadata("./").await { + match o.session.stat(".").await { Ok(_) => Ok(()), Err(e) => Err(parse_sftp_error(e)), } } } +fn clamp_chunk(len: u64) -> u32 { + len.min(MAX_CHUNK_SIZE as u64).max(1) as u32 +} + +/// Releases a server-side handle without awaiting the reply. +/// +/// Handles are remote resources, so abandoning a reader, writer, or lister +/// early must not leak them. `Drop` cannot await, so the close is dispatched +/// onto the current runtime. Without a runtime the session is already being +/// torn down, and the server releases the handle along with the channel. +pub fn close_handle_detached(session: Arc, handle: String) { + if let Ok(runtime) = tokio::runtime::Handle::try_current() { + runtime.spawn(async move { + if let Err(err) = session.close(handle).await { + debug!("sftp failed to close handle: {err}"); + } + }); + } +} + +fn default_user() -> Option { + std::env::var("USER") + .or_else(|_| std::env::var("USERNAME")) + .ok() + .filter(|v| !v.is_empty()) +} + +fn default_key_paths() -> Vec { + let Some(home) = std::env::home_dir() else { + return Vec::new(); + }; + let ssh = home.join(".ssh"); + + ["id_ed25519", "id_ecdsa", "id_rsa"] + .iter() + .map(|name| ssh.join(name)) + .collect() +} + +/// Splits an endpoint into its user, host, and port parts. +/// +/// Accepts both `[user@]host[:port]` and `ssh://[user@]host[:port]`, and +/// defaults the port to 22. +fn parse_endpoint(endpoint: &str) -> Result<(Option, String, u16)> { + let invalid = || { + Error::new( + ErrorKind::ConfigInvalid, + "sftp endpoint is invalid, expected `[user@]host[:port]`", + ) + }; + + let raw = endpoint.trim(); + let raw = raw.strip_prefix("ssh://").unwrap_or(raw); + // Drop any trailing path component; the root is configured separately. + let raw = raw.split('/').next().unwrap_or(raw); + + let (user, host_port) = match raw.rsplit_once('@') { + Some((user, host_port)) if !user.is_empty() => (Some(user.to_string()), host_port), + _ => (None, raw), + }; + + let (host, port) = if let Some(rest) = host_port.strip_prefix('[') { + // Bracketed IPv6 literal, optionally followed by a port. + let (host, tail) = rest.split_once(']').ok_or_else(invalid)?; + let port = match tail.strip_prefix(':') { + Some(port) => port.parse::().map_err(|_| invalid())?, + None => 22, + }; + (host.to_string(), port) + } else if host_port.matches(':').count() > 1 { + // An unbracketed IPv6 literal cannot be told apart from `host:port`. + return Err(invalid()); + } else { + match host_port.rsplit_once(':') { + Some((host, port)) => ( + host.to_string(), + port.parse::().map_err(|_| invalid())?, + ), + None => (host_port.to_string(), 22), + } + }; + + if host.is_empty() { + return Err(invalid()); + } + + Ok((user, host, port)) +} + mod error { - use openssh::Error as SshError; - use openssh_sftp_client::Error as SftpClientError; - use openssh_sftp_client::error::SftpErrorKind; + use russh_sftp::client::error::Error as SftpClientError; + use russh_sftp::protocol::StatusCode; use opendal_core::Error; use opendal_core::ErrorKind; pub fn parse_sftp_error(e: SftpClientError) -> Error { let kind = match &e { - SftpClientError::UnsupportedSftpProtocol { version: _ } => ErrorKind::Unsupported, - SftpClientError::SftpError(kind, _msg) => match kind { - SftpErrorKind::NoSuchFile => ErrorKind::NotFound, - SftpErrorKind::PermDenied => ErrorKind::PermissionDenied, - SftpErrorKind::OpUnsupported => ErrorKind::Unsupported, + SftpClientError::Status(status) => match status.status_code { + StatusCode::NoSuchFile => ErrorKind::NotFound, + StatusCode::PermissionDenied => ErrorKind::PermissionDenied, + StatusCode::OpUnsupported => ErrorKind::Unsupported, _ => ErrorKind::Unexpected, }, _ => ErrorKind::Unexpected, @@ -190,56 +645,53 @@ mod error { err } - pub fn parse_ssh_error(e: SshError) -> Error { + pub fn parse_ssh_error(e: russh::Error) -> Error { Error::new(ErrorKind::Unexpected, "ssh error").set_source(e) } pub(crate) fn is_not_found(e: &SftpClientError) -> bool { - matches!(e, SftpClientError::SftpError(SftpErrorKind::NoSuchFile, _)) + matches!(e, SftpClientError::Status(status) if status.status_code == StatusCode::NoSuchFile) } pub(crate) fn is_sftp_protocol_error(e: &SftpClientError) -> bool { - matches!(e, SftpClientError::SftpError(_, _)) + matches!(e, SftpClientError::Status(_)) } pub(crate) fn is_sftp_failure(e: &SftpClientError) -> bool { - matches!(e, SftpClientError::SftpError(SftpErrorKind::Failure, _)) + matches!(e, SftpClientError::Status(status) if status.status_code == StatusCode::Failure) + } + + pub(crate) fn is_eof(e: &SftpClientError) -> bool { + matches!(e, SftpClientError::Status(status) if status.status_code == StatusCode::Eof) } } pub(super) use error::*; mod utils { - use openssh_sftp_client::metadata::MetaData as SftpMeta; + use russh_sftp::protocol::{FileAttributes, FileType}; use opendal_core::EntryMode; use opendal_core::Metadata; use opendal_core::raw::Timestamp; - pub fn to_metadata(meta: SftpMeta) -> Metadata { - let mode = meta - .file_type() - .map(|filetype| { - if filetype.is_file() { - EntryMode::FILE - } else if filetype.is_dir() { - EntryMode::DIR - } else { - EntryMode::Unknown - } - }) - .unwrap_or(EntryMode::Unknown); + pub fn to_metadata(attrs: &FileAttributes) -> Metadata { + let mode = match attrs.file_type() { + FileType::File => EntryMode::FILE, + FileType::Dir => EntryMode::DIR, + _ => EntryMode::Unknown, + }; let mut metadata = Metadata::new(mode); - if let Some(size) = meta.len() { + if let Some(size) = attrs.size { metadata.set_content_length(size); } - if let Some(modified) = meta.modified() - && let Ok(m) = Timestamp::try_from(modified.as_system_time()) + if let Some(mtime) = attrs.mtime + && let Ok(ts) = Timestamp::from_second(mtime as i64) { - metadata.set_last_modified(m); + metadata.set_last_modified(ts); } metadata @@ -247,3 +699,76 @@ mod utils { } pub(super) use utils::*; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_endpoint() { + let cases = [ + ("127.0.0.1", (None, "127.0.0.1", 22)), + ("ssh://127.0.0.1", (None, "127.0.0.1", 22)), + ("ssh://127.0.0.1:2222", (None, "127.0.0.1", 2222)), + ("ssh://foo@127.0.0.1:2222", (Some("foo"), "127.0.0.1", 2222)), + ("foo@example.com", (Some("foo"), "example.com", 22)), + ("example.com:2222", (None, "example.com", 2222)), + ("[::1]:2222", (None, "::1", 2222)), + ("[::1]", (None, "::1", 22)), + ("ssh://127.0.0.1:2222/ignored", (None, "127.0.0.1", 2222)), + ]; + + for (input, (user, host, port)) in cases { + let actual = parse_endpoint(input).expect("endpoint must parse"); + assert_eq!( + actual, + (user.map(str::to_string), host.to_string(), port), + "endpoint: {input}" + ); + } + } + + #[test] + fn test_parse_endpoint_invalid() { + // An unbracketed IPv6 literal is ambiguous with `host:port` and must + // not be parsed as host ":" on port 1. + for input in ["", "host:port", "[::1", "::1", "fe80::1:22"] { + assert!( + parse_endpoint(input).is_err(), + "endpoint must be rejected: {input}" + ); + } + } + + #[test] + fn test_abs_path() { + let core = |root: &str| { + SftpCore::new( + ServiceInfo::new(crate::SFTP_SCHEME, root, ""), + Capability::default(), + "127.0.0.1".to_string(), + root.to_string(), + None, + None, + KnownHostsStrategy::Strict, + ) + }; + + let rooted = core("/upload/"); + assert_eq!(rooted.abs_path("a/b.txt"), "/upload/a/b.txt"); + assert_eq!(rooted.abs_path("/a/b.txt"), "/upload/a/b.txt"); + + // Trailing separators are dropped; OpenSSH on Windows rejects them. + assert_eq!(rooted.abs_path("a/b/"), "/upload/a/b"); + assert_eq!(rooted.abs_path(""), "/upload"); + + // An empty root keeps paths relative to the login directory. + let rootless = core(""); + assert_eq!(rootless.abs_path("a/b.txt"), "a/b.txt"); + assert_eq!(rootless.abs_path("a/b/"), "a/b"); + assert_eq!(rootless.abs_path(""), "."); + + // The server root must survive normalisation. + assert_eq!(core("/").abs_path(""), "/"); + } +} diff --git a/core/services/sftp/src/deleter.rs b/core/services/sftp/src/deleter.rs index 6d11fb8424ba..b6a6c71d27ed 100644 --- a/core/services/sftp/src/deleter.rs +++ b/core/services/sftp/src/deleter.rs @@ -35,19 +35,17 @@ impl SftpDeleter { impl oio::OneShotDelete for SftpDeleter { async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> { - let client = self.core.connect().await?; - - let mut fs = client.fs(); - fs.set_cwd(&self.core.root); + let conn = self.core.connect().await?; + let abs_path = self.core.abs_path(&path); let res = if path.ends_with('/') { - fs.remove_dir(path).await + conn.session.rmdir(abs_path).await } else { - fs.remove_file(path).await + conn.session.remove(abs_path).await }; match res { - Ok(()) => Ok(()), + Ok(_) => Ok(()), Err(e) if is_not_found(&e) => Ok(()), Err(e) => Err(parse_sftp_error(e)), } diff --git a/core/services/sftp/src/docs.md b/core/services/sftp/src/docs.md index 3ef264241aaa..924ba7051a38 100644 --- a/core/services/sftp/src/docs.md +++ b/core/services/sftp/src/docs.md @@ -23,6 +23,29 @@ accepted values, defaults, and environment interaction. For security reasons, it doesn't support password login. Use SSH key-based authentication (e.g., configure your public key on the server via `ssh-copy-id` and provide the private key here). +### Authentication + +`SftpBackend` authenticates with a public key, in this order: + +1. The private key at `key`, when set. +2. The SSH agent advertised by `SSH_AUTH_SOCK`, on Unix. +3. `~/.ssh/id_ed25519`, `~/.ssh/id_ecdsa`, and `~/.ssh/id_rsa`. + +Setting `key` disables the fallbacks, so an unusable key fails instead of +silently authenticating as a different identity. + +### Host key verification + +`known_hosts_strategy` selects how the remote host key is checked against +`~/.ssh/known_hosts`: + +- `strict` (default): the key must already be recorded and match. This + corresponds to `ssh -o StrictHostKeyChecking=yes`. +- `add`: an unknown key is accepted and recorded, while a changed key is + rejected. This corresponds to `ssh -o StrictHostKeyChecking=accept-new`. +- `accept`: any key is accepted without being recorded. This corresponds to + `ssh -o StrictHostKeyChecking=no`. + ## Example ### Via Builder diff --git a/core/services/sftp/src/lister.rs b/core/services/sftp/src/lister.rs index 286afd0536a1..900c8a540d47 100644 --- a/core/services/sftp/src/lister.rs +++ b/core/services/sftp/src/lister.rs @@ -15,75 +15,143 @@ // specific language governing permissions and limitations // under the License. -use std::pin::Pin; +use std::collections::VecDeque; -use futures::StreamExt; -use openssh_sftp_client::fs::DirEntry; -use openssh_sftp_client::fs::ReadDir; +use russh_sftp::protocol::{FileAttributes, FileMode}; +use super::core::SftpSessionRef; +use super::core::close_handle_detached; +use super::core::is_eof; use super::core::parse_sftp_error; use super::core::to_metadata; use opendal_core::Result; use opendal_core::raw::oio; use opendal_core::raw::oio::Entry; +/// Lists a remote directory one `SSH_FXP_READDIR` batch at a time. +/// +/// Entries are yielded as they arrive rather than buffering the whole +/// directory, which keeps memory flat for large directories. pub struct SftpLister { - dir: Pin>, + /// Keeps the session alive while the directory handle is open. + conn: SftpSessionRef, + handle: String, prefix: String, + buffer: VecDeque<(String, FileAttributes)>, + /// Whether the directory's own entry still has to be emitted. + emit_self: bool, + finished: bool, +} + +impl Drop for SftpLister { + fn drop(&mut self) { + if self.finished { + return; + } + + // `list_with_limit` callers routinely stop before the directory ends, + // so the handle has to be released here too. + close_handle_detached(self.conn.session.clone(), std::mem::take(&mut self.handle)); + } } impl SftpLister { - pub fn new(dir: ReadDir, path: String) -> Self { + pub fn new(conn: SftpSessionRef, handle: String, path: String) -> Self { let prefix = if path == "/" { "".to_owned() } else { path }; SftpLister { - dir: Box::pin(dir), + conn, + handle, prefix, + buffer: VecDeque::new(), + emit_self: true, + finished: false, + } + } + + /// Builds the entry for the directory being listed. + /// + /// The metadata comes from the open handle rather than a `.` entry: + /// OpenSSH on Windows omits `.` and `..` from `readdir` altogether, so a + /// listing that relied on them would silently drop the directory itself. + async fn self_entry(&self) -> Entry { + let attrs = match self.conn.session.fstat(self.handle.as_str()).await { + Ok(attrs) => attrs.attrs, + Err(_) => { + let mut attrs = FileAttributes::default(); + attrs.set_type(FileMode::DIR); + attrs + } + }; + + let path = if self.prefix.is_empty() { + "/" + } else { + self.prefix.as_str() + }; + + Entry::new(path, to_metadata(&attrs)) + } + + /// Pulls the next batch of entries, returning `false` once the server + /// reports the end of the directory. + async fn fill(&mut self) -> Result { + if self.finished { + return Ok(false); + } + + match self.conn.session.readdir(self.handle.as_str()).await { + Ok(name) => { + self.buffer + .extend(name.files.into_iter().map(|f| (f.filename, f.attrs))); + Ok(true) + } + Err(e) if is_eof(&e) => { + self.finished = true; + // Release the directory handle as soon as the listing ends. + let _ = self + .conn + .session + .close(std::mem::take(&mut self.handle)) + .await; + Ok(false) + } + Err(e) => { + self.finished = true; + Err(parse_sftp_error(e)) + } } } } impl oio::List for SftpLister { async fn next(&mut self) -> Result> { + if self.emit_self { + self.emit_self = false; + return Ok(Some(self.self_entry().await)); + } + loop { - let item = self - .dir - .next() - .await - .transpose() - .map_err(parse_sftp_error)?; - - match item { - Some(e) => { - if e.filename().to_str() == Some("..") { - continue; - } else if e.filename().to_str() == Some(".") { - let mut path = self.prefix.as_str(); - if self.prefix.is_empty() { - path = "/"; - } - return Ok(Some(Entry::new(path, to_metadata(e.metadata())))); - } else { - return Ok(Some(map_entry(self.prefix.as_str(), e))); - } + let Some((filename, attrs)) = self.buffer.pop_front() else { + if self.fill().await? { + continue; } - None => return Ok(None), + return Ok(None); + }; + + // Emitted up front from the handle, so both are dropped here. + if filename == "." || filename == ".." { + continue; } - } - } -} -fn map_entry(prefix: &str, value: DirEntry) -> Entry { - let path = format!( - "{}{}{}", - prefix, - value.filename().to_str().unwrap(), - if value.file_type().unwrap().is_dir() { - "/" - } else { - "" - } - ); + let path = format!( + "{}{}{}", + self.prefix, + filename, + if attrs.file_type().is_dir() { "/" } else { "" } + ); - Entry::new(path.as_str(), to_metadata(value.metadata())) + return Ok(Some(Entry::new(path.as_str(), to_metadata(&attrs)))); + } + } } diff --git a/core/services/sftp/src/reader.rs b/core/services/sftp/src/reader.rs index a2cf46881c18..354800ebf7d6 100644 --- a/core/services/sftp/src/reader.rs +++ b/core/services/sftp/src/reader.rs @@ -15,71 +15,132 @@ // specific language governing permissions and limitations // under the License. +use bytes::Bytes; +use russh_sftp::protocol::{FileAttributes, OpenFlags}; + use super::backend::*; -use super::core::Manager; +use super::core::PIPELINE_DEPTH; +use super::core::SftpSessionRef; +use super::core::close_handle_detached; +use super::core::is_eof; use super::core::is_not_found; use super::core::is_sftp_failure; use super::core::parse_sftp_error; use super::lister::SftpLister; use super::writer::SftpWriter; -use bytes::BytesMut; -use fastpool::bounded; use opendal_core::raw::*; use opendal_core::*; -use openssh_sftp_client::file::File; -use std::io::SeekFrom; -use tokio::io::AsyncSeekExt; +/// Streams a remote file by issuing several SFTP reads concurrently. +/// +/// SFTP is a request/response protocol, so a single outstanding read caps +/// throughput at one packet per round trip. Each call therefore fans out up to +/// [`PIPELINE_DEPTH`] reads and returns them as one multi-chunk [`Buffer`]. pub struct SftpReadStream { - /// Keep the connection alive while data stream is alive. - _conn: bounded::Object, - - file: File, - chunk: usize, - size: Option, - read: usize, - buf: BytesMut, + /// Keeps the session alive while data stream is alive. + conn: SftpSessionRef, + + handle: String, + offset: u64, + /// Remaining bytes to read when the caller asked for a bounded range. + remaining: Option, + finished: bool, +} + +impl Drop for SftpReadStream { + fn drop(&mut self) { + // A ranged or abandoned read never reaches EOF, so the remote handle + // has to be released here rather than at the end of the stream. + close_handle_detached(self.conn.session.clone(), std::mem::take(&mut self.handle)); + } } impl SftpReadStream { - pub fn new(conn: bounded::Object, file: File, size: Option) -> Self { + pub fn new(conn: SftpSessionRef, handle: String, offset: u64, size: Option) -> Self { Self { - _conn: conn, - file, - size: size.map(|v| v as usize), - chunk: 2 * 1024 * 1024, - read: 0, - buf: BytesMut::new(), + conn, + handle, + offset, + remaining: size, + finished: false, } } } impl oio::ReadStream for SftpReadStream { async fn read(&mut self) -> Result { - if self.read >= self.size.unwrap_or(usize::MAX) { + if self.finished || self.remaining == Some(0) { return Ok(Buffer::new()); } - let size = if let Some(size) = self.size { - (size - self.read).min(self.chunk) - } else { - self.chunk - }; - self.buf.reserve(size); + let chunk = self.conn.read_len as u64; - let Some(bytes) = self - .file - .read(size as u32, self.buf.split_off(0)) - .await - .map_err(parse_sftp_error)? - else { + // Plan a batch of reads covering contiguous offsets. + let mut planned = 0u64; + let mut wants = Vec::with_capacity(PIPELINE_DEPTH); + let mut inflight = Vec::with_capacity(PIPELINE_DEPTH); + for _ in 0..PIPELINE_DEPTH { + let want = match self.remaining { + Some(remaining) => remaining.saturating_sub(planned).min(chunk), + None => chunk, + }; + if want == 0 { + break; + } + + let offset = self.offset + planned; + planned += want; + wants.push(want); + + let session = self.conn.session.clone(); + let handle = self.handle.clone(); + inflight.push(async move { session.read(handle, offset, want as u32).await }); + } + + if inflight.is_empty() { + return Ok(Buffer::new()); + } + + let results = futures::future::join_all(inflight).await; + + // Assemble the batch in order. A short read means the server gave us + // less than we asked for, so everything planned after it is discarded + // and re-requested from the corrected offset on the next call. + let mut parts = Vec::with_capacity(results.len()); + let mut consumed = 0u64; + for (result, want) in results.into_iter().zip(wants) { + match result { + Ok(data) => { + let len = data.data.len() as u64; + if len > 0 { + parts.push(Bytes::from(data.data)); + consumed += len; + } + + if len < want { + self.finished = len == 0; + break; + } + } + Err(e) if is_eof(&e) => { + self.finished = true; + break; + } + Err(e) => return Err(parse_sftp_error(e)), + } + } + + if consumed == 0 { + self.finished = true; return Ok(Buffer::new()); - }; + } + + self.offset += consumed; + if let Some(remaining) = self.remaining.as_mut() { + *remaining -= consumed; + } - self.read += bytes.len(); - self.buf = bytes; - let bs = self.buf.split(); - Ok(Buffer::from(bs.freeze())) + Ok(Buffer::from(parts)) } } @@ -100,29 +161,19 @@ impl SftpReader { impl oio::StreamRead for SftpReader { async fn open(&self, range: BytesRange) -> Result<(RpRead, Box)> { - let backend = &self.backend; - let path = self.path.as_str(); + let core = &self.backend.core; + let path = core.abs_path(&self.path); - let client = backend.core.connect().await?; - - let mut fs = client.fs(); - fs.set_cwd(&backend.core.root); - - let path = fs.canonicalize(path).await.map_err(parse_sftp_error)?; - - let mut f = client - .open(path.as_path()) + let conn = core.connect().await?; + let handle = conn + .session + .open(path, OpenFlags::READ, FileAttributes::default()) .await - .map_err(parse_sftp_error)?; - - if range.offset() != 0 { - f.seek(SeekFrom::Start(range.offset())) - .await - .map_err(new_std_io_error)?; - } + .map_err(parse_sftp_error)? + .handle; let rp = RpRead::default(); - let stream = SftpReadStream::new(client, f, range.size()); + let stream = SftpReadStream::new(conn.session_ref(), handle, range.offset(), range.size()); Ok((rp, Box::new(stream) as Box)) } @@ -160,33 +211,29 @@ impl SftpLazyWriter { .await?; } - let client = self.backend.core.connect().await?; - - let mut fs = client.fs(); - fs.set_cwd(&self.backend.core.root); - let path = fs - .canonicalize(&self.path) - .await - .map_err(parse_sftp_error)?; + let core = &self.backend.core; + let path = core.abs_path(&self.path); + let conn = core.connect().await?; - let mut option = client.options(); + let mut flags = OpenFlags::WRITE | OpenFlags::CREATE; if self.op.if_not_exists() { - option.create_new(true); - } else { - option.create(true); + flags |= OpenFlags::EXCLUDE; } - if self.op.append() { - option.append(true); + flags |= OpenFlags::APPEND; } else { - option.write(true).truncate(true); + flags |= OpenFlags::TRUNCATE; } - let res = option.open(&path).await; - let file = match res { - Ok(f) => f, + let res = conn + .session + .open(path.as_str(), flags, FileAttributes::default()) + .await; + + let handle = match res { + Ok(handle) => handle.handle, Err(e) if self.op.if_not_exists() && is_sftp_failure(&e) => { - if fs.metadata(&path).await.is_ok() { + if conn.session.stat(path.as_str()).await.is_ok() { return Err(Error::new( ErrorKind::ConditionNotMatch, "file already exists, doesn't match the condition if_not_exists", @@ -198,7 +245,20 @@ impl SftpLazyWriter { Err(e) => return Err(parse_sftp_error(e)), }; - self.inner = Some(SftpWriter::new(file)); + // Appending starts at the current end of the file; the server + // honours `APPEND`, but the local offset must match so that the + // pipelined writes address the right region. + let offset = if self.op.append() { + conn.session + .stat(path.as_str()) + .await + .map(|attrs| attrs.attrs.size.unwrap_or(0)) + .unwrap_or(0) + } else { + 0 + }; + + self.inner = Some(SftpWriter::new(conn.session_ref(), handle, offset)); } Ok(self.inner.as_mut().expect("writer must be initialized")) @@ -238,14 +298,16 @@ impl SftpLazyLister { impl oio::List for SftpLazyLister { async fn next(&mut self) -> Result> { if self.inner.is_none() { - let client = self.backend.core.connect().await?; - let mut fs = client.fs(); - fs.set_cwd(&self.backend.core.root); - - let file_path = format!("./{}", self.path); - - self.inner = Some(match fs.open_dir(&file_path).await { - Ok(dir) => Some(SftpLister::new(dir.read_dir(), self.path.clone())), + let core = &self.backend.core; + let conn = core.connect().await?; + let dir_path = core.abs_path(&self.path); + + self.inner = Some(match conn.session.opendir(dir_path).await { + Ok(handle) => Some(SftpLister::new( + conn.session_ref(), + handle.handle, + self.path.clone(), + )), Err(e) if is_not_found(&e) => None, Err(e) => return Err(parse_sftp_error(e)), }); diff --git a/core/services/sftp/src/writer.rs b/core/services/sftp/src/writer.rs index 542db18a96fe..c8ef7d56ff01 100644 --- a/core/services/sftp/src/writer.rs +++ b/core/services/sftp/src/writer.rs @@ -15,45 +15,93 @@ // specific language governing permissions and limitations // under the License. -use std::pin::Pin; - use bytes::Buf; -use openssh_sftp_client::file::File; -use openssh_sftp_client::file::TokioCompatFile; -use tokio::io::AsyncWriteExt; +use super::core::PIPELINE_DEPTH; +use super::core::SftpSessionRef; +use super::core::close_handle_detached; +use super::core::parse_sftp_error; use opendal_core::raw::*; use opendal_core::*; +/// Writes a remote file by streaming `Buffer`s into an open SFTP handle. +/// +/// Writes are split into packets no larger than the size the server advertises +/// and several packets are kept in flight, so throughput does not collapse to +/// one round trip per packet. pub struct SftpWriter { - /// TODO: maybe we can use `File` directly? - file: Pin>, + /// Keeps the session alive while the remote handle is open. + conn: SftpSessionRef, + handle: String, + offset: u64, + closed: bool, +} + +impl Drop for SftpWriter { + fn drop(&mut self) { + if self.closed { + return; + } + + // An aborted write never calls `close`, so release the handle here. + close_handle_detached(self.conn.session.clone(), std::mem::take(&mut self.handle)); + } } impl SftpWriter { - pub fn new(file: File) -> Self { + pub fn new(conn: SftpSessionRef, handle: String, offset: u64) -> Self { SftpWriter { - file: Box::pin(TokioCompatFile::new(file)), + conn, + handle, + offset, + closed: false, } } } impl oio::Write for SftpWriter { async fn write(&mut self, mut bs: Buffer) -> Result<()> { + let chunk = self.conn.write_len as usize; + let mut inflight = Vec::with_capacity(PIPELINE_DEPTH); + while bs.has_remaining() { - let n = self - .file - .write(bs.chunk()) + let piece = bs.chunk(); + let take = piece.len().min(chunk); + let data = piece[..take].to_vec(); + bs.advance(take); + + let offset = self.offset; + self.offset += take as u64; + + let session = self.conn.session.clone(); + let handle = self.handle.clone(); + inflight.push(async move { session.write(handle, offset, data).await }); + + if inflight.len() >= PIPELINE_DEPTH { + futures::future::try_join_all(std::mem::take(&mut inflight)) + .await + .map_err(parse_sftp_error)?; + } + } + + if !inflight.is_empty() { + futures::future::try_join_all(inflight) .await - .map_err(new_std_io_error)?; - bs.advance(n); + .map_err(parse_sftp_error)?; } Ok(()) } async fn close(&mut self) -> Result { - self.file.shutdown().await.map_err(new_std_io_error)?; + if !self.closed { + self.conn + .session + .close(self.handle.as_str()) + .await + .map_err(parse_sftp_error)?; + self.closed = true; + } Ok(Metadata::default()) }