diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 98dfbcdbd..4571f145b 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -442,11 +442,11 @@ pub mod rpc { ScannerBucketListing, ScannerPeerActivity, TONIC_RPC_PREFIX, TonicInterceptor, build_put_file_auth_trailer, check_and_record_signed_rpc_nonce, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers, gen_tonic_signature_interceptor, node_service_time_out_client, node_service_time_out_client_no_auth, - normalize_tonic_rpc_audience, set_tonic_canonical_body_digest, sign_ns_scanner_capability, sign_tonic_rpc_response_proof, - tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason, - verify_put_file_auth_trailer, verify_rpc_signature, verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest, - verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof, verify_tonic_rpc_signature, - verify_tonic_rpc_signature_with_bootstrap, + normalize_tonic_rpc_audience, set_tonic_canonical_body_digest, sign_ns_scanner_capability, sign_put_file_capability, + sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, + tonic_rpc_auth_failure_reason, verify_put_file_auth_trailer, verify_put_file_capability, verify_rpc_signature, + verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, + verify_tonic_rpc_response_proof, verify_tonic_rpc_signature, verify_tonic_rpc_signature_with_bootstrap, }; } diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index bcbfb5c5b..eed40d141 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -29,7 +29,7 @@ use crate::cluster::rpc::context_propagation::{inject_request_id_into_http_headers, inject_trace_context_into_http_headers}; use crate::storage_api_contracts::internode::{ NS_SCANNER_PROTOCOL_VERSION, PUT_FILE_AUTH_TRAILER_DIGEST_LEN, PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_TRAILER_MAC_LEN, - PUT_FILE_AUTH_TRAILER_MAGIC, + PUT_FILE_AUTH_TRAILER_MAGIC, PUT_FILE_CAPABILITY_VERSION, }; use base64::Engine as _; use base64::engine::general_purpose; @@ -75,6 +75,7 @@ const RPC_RESPONSE_PROOF_DOMAIN: &[u8] = b"rustfs-rpc-response-proof-v1\0"; const RPC_REPLAY_SCOPE_DOMAIN: &[u8] = b"rustfs-rpc-replay-scope-v3\0"; const RPC_BOOT_EPOCH_PROOF_DOMAIN: &[u8] = b"rustfs-rpc-boot-epoch-proof-v1\0"; const HTTP_PUT_FILE_AUTH_DOMAIN: &[u8] = b"rustfs-http-put-file-auth-v1\0"; +const HTTP_PUT_FILE_CAPABILITY_AUTH_DOMAIN: &[u8] = b"rustfs-http-put-file-capability-v1\0"; const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD"; const UNSIGNED_PAYLOAD_NONCE: &str = "unsigned"; const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes @@ -583,6 +584,36 @@ pub fn verify_put_file_auth_trailer(url: &str, method: &Method, nonce: Uuid, tra Ok(body_sha256.to_string()) } +fn update_put_file_capability_mac(mac: &mut HmacSha256, challenge: Uuid, server_epoch: Uuid, version: u16) { + mac.update(HTTP_PUT_FILE_CAPABILITY_AUTH_DOMAIN); + mac.update(challenge.as_bytes()); + mac.update(server_epoch.as_bytes()); + mac.update(&version.to_be_bytes()); +} + +fn put_file_capability_mac(challenge: Uuid, server_epoch: Uuid, version: u16) -> std::io::Result { + if challenge.is_nil() || server_epoch.is_nil() || version != PUT_FILE_CAPABILITY_VERSION { + return Err(std::io::Error::other("Invalid put_file capability scope")); + } + let mut mac = HmacSha256::new_from_slice(get_shared_secret()?.as_bytes()) + .map_err(|_| std::io::Error::other("Invalid RPC HMAC secret"))?; + update_put_file_capability_mac(&mut mac, challenge, server_epoch, version); + Ok(mac) +} + +pub fn sign_put_file_capability(challenge: Uuid, server_epoch: Uuid, version: u16) -> std::io::Result> { + Ok(put_file_capability_mac(challenge, server_epoch, version)? + .finalize() + .into_bytes() + .to_vec()) +} + +pub fn verify_put_file_capability(challenge: Uuid, server_epoch: Uuid, version: u16, proof: &[u8]) -> std::io::Result<()> { + put_file_capability_mac(challenge, server_epoch, version)? + .verify_slice(proof) + .map_err(|_| std::io::Error::new(std::io::ErrorKind::PermissionDenied, "Invalid put_file capability proof")) +} + fn update_ns_scanner_capability_mac(mac: &mut HmacSha256, challenge: Uuid, server_epoch: Uuid) { mac.update(NS_SCANNER_CAPABILITY_AUTH_DOMAIN); mac.update(&NS_SCANNER_PROTOCOL_VERSION.to_be_bytes()); @@ -2331,6 +2362,20 @@ mod tests { assert_eq!(err.to_string(), "Invalid put_file auth trailer"); } + #[test] + fn put_file_capability_proof_binds_challenge_epoch_and_version() { + ensure_test_rpc_secret(); + let challenge = Uuid::parse_str("11111111-2222-4333-8444-555555555555").expect("challenge"); + let server_epoch = Uuid::parse_str("aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee").expect("server epoch"); + let proof = sign_put_file_capability(challenge, server_epoch, PUT_FILE_CAPABILITY_VERSION) + .expect("capability proof should build"); + + assert!(verify_put_file_capability(challenge, server_epoch, PUT_FILE_CAPABILITY_VERSION, &proof).is_ok()); + assert!(verify_put_file_capability(Uuid::new_v4(), server_epoch, PUT_FILE_CAPABILITY_VERSION, &proof).is_err()); + assert!(verify_put_file_capability(challenge, Uuid::new_v4(), PUT_FILE_CAPABILITY_VERSION, &proof).is_err()); + assert!(verify_put_file_capability(challenge, server_epoch, PUT_FILE_CAPABILITY_VERSION + 1, &proof).is_err()); + } + #[test] fn tier_mutation_rpc_contract_requires_method_bound_v2_body_digest() { ensure_test_rpc_secret(); diff --git a/crates/ecstore/src/cluster/rpc/internode_data_transport.rs b/crates/ecstore/src/cluster/rpc/internode_data_transport.rs index 4192a9792..b57e6e4ae 100644 --- a/crates/ecstore/src/cluster/rpc/internode_data_transport.rs +++ b/crates/ecstore/src/cluster/rpc/internode_data_transport.rs @@ -12,15 +12,18 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::cluster::rpc::{build_auth_headers, build_put_file_auth_trailer, verify_ns_scanner_capability}; +use crate::cluster::rpc::{ + build_auth_headers, build_put_file_auth_trailer, verify_ns_scanner_capability, verify_put_file_capability, +}; use crate::disk::error::{Error, Result}; use crate::disk::{FileReader, FileWriter}; use crate::storage_api_contracts::internode::{ NS_SCANNER_BODY_SHA256_QUERY, NS_SCANNER_CAPABILITY_CHALLENGE_QUERY, NS_SCANNER_CYCLE_QUERY, NS_SCANNER_LEADER_EPOCH_QUERY, NS_SCANNER_PROTOCOL_VERSION, NS_SCANNER_PROTOCOL_VERSION_QUERY, NS_SCANNER_REQUEST_ID_QUERY, NS_SCANNER_SERVER_EPOCH_QUERY, NS_SCANNER_SESSION_ID_QUERY, NS_SCANNER_SESSION_SEQUENCE_QUERY, NsScannerCapabilityResponse, PUT_FILE_AUTH_QUERY, - PUT_FILE_AUTH_V1, PUT_FILE_NONCE_QUERY, WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_STREAM_COMPLETION_QUERY, - WALK_DIR_STREAM_COMPLETION_V1, + PUT_FILE_AUTH_V1, PUT_FILE_CAPABILITY_CHALLENGE_QUERY, PUT_FILE_CAPABILITY_QUERY, PUT_FILE_CAPABILITY_VERSION, + PUT_FILE_NONCE_QUERY, PUT_FILE_SERVER_EPOCH_QUERY, PutFileCapabilityResponse, WALK_DIR_BODY_SHA256_QUERY, + WALK_DIR_STREAM_COMPLETION_QUERY, WALK_DIR_STREAM_COMPLETION_V1, }; use async_trait::async_trait; use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE}; @@ -30,20 +33,29 @@ use rustfs_config::{ }; use rustfs_rio::{HttpReader, HttpWriter}; use sha2::{Digest, Sha256}; +use std::collections::HashMap; +use std::future::Future; use std::pin::Pin; -use std::sync::{Arc, OnceLock}; +use std::sync::{Arc, LazyLock, OnceLock}; use std::task::{Context, Poll}; -use std::time::Duration; +use std::time::{Duration, Instant}; use tokio::io::{AsyncReadExt, AsyncWrite}; +use tokio::sync::OnceCell; use uuid::Uuid; static INTERNODE_DATA_TRANSPORT: OnceLock, String>> = OnceLock::new(); const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream"; const PUT_FILE_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream"; +const PUT_FILE_AUTH_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream_v1"; +const PUT_FILE_CAPABILITY_PATH: &str = "/rustfs/rpc/put_file_capability"; const WALK_DIR_PATH: &str = "/rustfs/rpc/walk_dir"; const NS_SCANNER_PATH: &str = "/rustfs/rpc/ns_scanner"; const NS_SCANNER_MAX_CAPABILITY_RESPONSE_SIZE: usize = 1024; +const PUT_FILE_MAX_CAPABILITY_RESPONSE_SIZE: usize = 1024; +const PUT_FILE_LEGACY_CAPABILITY_TTL: Duration = Duration::from_secs(30); +const PUT_FILE_V1_CAPABILITY_TTL: Duration = Duration::from_secs(30); +const PUT_FILE_CAPABILITY_PROBE_TIMEOUT: Duration = Duration::from_secs(5); const CONTENT_TYPE_JSON: &str = "application/json"; const CONTENT_TYPE_MSGPACK: &str = "application/msgpack"; @@ -54,6 +66,73 @@ fn unsupported_transport_message(transport: &str) -> String { ) } +#[derive(Debug, Clone, Copy)] +enum PutFileCapabilityState { + LegacyUntil(Instant), + V1 { server_epoch: Uuid, revalidate_after: Instant }, +} + +#[derive(Debug)] +struct PutFileCapabilityProbeFailure(Error); + +impl PutFileCapabilityProbeFailure { + fn to_error(&self) -> Error { + match &self.0 { + Error::Io(error) => rustfs_rio::clone_internode_http_io_error(error) + .map(Error::Io) + .unwrap_or_else(|| self.0.clone()), + _ => self.0.clone(), + } + } +} + +type PutFileCapabilityProbeOutcome = std::result::Result, PutFileCapabilityProbeFailure>; + +#[derive(Debug, Clone)] +struct PutFileCapabilityFlight { + generation: u64, + v1_was_pinned: bool, + outcome: Arc>, +} + +#[derive(Debug, Default)] +struct PutFileCapabilityCacheState { + cached: Option, + generation: u64, + in_flight: Option, +} + +type PutFileCapabilityCacheEntry = Arc>; + +static PUT_FILE_CAPABILITY_CACHE: LazyLock>> = + LazyLock::new(|| parking_lot::RwLock::new(HashMap::new())); + +fn put_file_capability_cache_entry(endpoint: &str) -> PutFileCapabilityCacheEntry { + if let Some(entry) = PUT_FILE_CAPABILITY_CACHE.read().get(endpoint).cloned() { + return entry; + } + PUT_FILE_CAPABILITY_CACHE + .write() + .entry(endpoint.to_owned()) + .or_insert_with(|| Arc::new(tokio::sync::RwLock::new(PutFileCapabilityCacheState::default()))) + .clone() +} + +fn fresh_put_file_capability(state: Option, now: Instant) -> Option> { + match state { + Some(PutFileCapabilityState::V1 { + server_epoch, + revalidate_after, + }) if now < revalidate_after => Some(Some(server_epoch)), + Some(PutFileCapabilityState::LegacyUntil(expires_at)) if now < expires_at => Some(None), + Some(PutFileCapabilityState::V1 { .. }) | Some(PutFileCapabilityState::LegacyUntil(_)) | None => None, + } +} + +fn put_file_capability_status_is_legacy(status: u16) -> bool { + status == 404 +} + #[derive(Debug, Clone, Copy, Eq, PartialEq)] pub struct InternodeDataTransportCapabilities { /// Backend can open a streaming remote disk reader. @@ -169,12 +248,16 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport { } async fn open_write(&self, request: WriteStreamRequest) -> Result { - let nonce = Uuid::new_v4(); - let url = build_put_file_stream_url(&request, Some(nonce)); + let server_epoch = self.put_file_auth_capability(&request.endpoint).await?; + let nonce = server_epoch.map(|_| Uuid::new_v4()); + let url = build_put_file_stream_url(&request, nonce.zip(server_epoch)); let mut headers = json_headers(); build_auth_headers(&url, &Method::PUT, &mut headers)?; let writer = HttpWriter::new(url.clone(), Method::PUT, headers).await?; - Ok(Box::new(PutFileAuthWriter::new(writer, url, nonce))) + match nonce { + Some(nonce) => Ok(Box::new(PutFileAuthWriter::new(writer, url, nonce))), + None => Ok(Box::new(writer)), + } } async fn open_walk_dir(&self, request: WalkDirStreamRequest) -> Result { @@ -228,6 +311,134 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport { } } +impl TcpHttpInternodeDataTransport { + async fn put_file_auth_capability(&self, endpoint: &str) -> Result> { + resolve_put_file_auth_capability(endpoint, || async { + tokio::time::timeout(PUT_FILE_CAPABILITY_PROBE_TIMEOUT, self.probe_put_file_auth(endpoint)) + .await + .map_err(|_| { + Error::from(rustfs_rio::internode_http_timeout_error( + &Method::GET, + &format!("{endpoint}{PUT_FILE_CAPABILITY_PATH}"), + )) + })? + }) + .await + } + + async fn probe_put_file_auth(&self, endpoint: &str) -> Result> { + let challenge = Uuid::new_v4(); + let url = build_put_file_capability_url(endpoint, challenge); + let mut headers = msgpack_headers(); + build_auth_headers(&url, &Method::GET, &mut headers)?; + let reader = match HttpReader::new(url, Method::GET, headers, None).await { + Ok(reader) => reader, + Err(err) => { + let err = Error::from(err); + if matches!( + err.internode_http_error_kind(), + Some(rustfs_rio::InternodeHttpErrorKind::HttpStatus(status)) + if put_file_capability_status_is_legacy(status.as_u16()) + ) { + return Ok(None); + } + return Err(err); + } + }; + let mut body = Vec::new(); + reader + .take(u64::try_from(PUT_FILE_MAX_CAPABILITY_RESPONSE_SIZE + 1).unwrap_or(u64::MAX)) + .read_to_end(&mut body) + .await?; + Ok(Some(verify_put_file_capability_response(challenge, &body)?)) + } +} + +async fn resolve_put_file_auth_capability(endpoint: &str, probe: F) -> Result> +where + F: FnOnce() -> Fut, + Fut: Future>>, +{ + let entry = put_file_capability_cache_entry(endpoint); + { + let state = entry.read().await; + if let Some(cached) = fresh_put_file_capability(state.cached, Instant::now()) { + return Ok(cached); + } + } + + let flight = { + let mut state = entry.write().await; + if let Some(cached) = fresh_put_file_capability(state.cached, Instant::now()) { + return Ok(cached); + } + if let Some(flight) = state.in_flight.clone() { + flight + } else { + state.generation = state + .generation + .checked_add(1) + .ok_or_else(|| Error::other("put_file capability probe generation exhausted"))?; + let flight = PutFileCapabilityFlight { + generation: state.generation, + v1_was_pinned: matches!(state.cached, Some(PutFileCapabilityState::V1 { .. })), + outcome: Arc::new(OnceCell::new()), + }; + state.in_flight = Some(flight.clone()); + flight + } + }; + + let outcome = flight + .outcome + .get_or_init(|| async { probe().await.map_err(PutFileCapabilityProbeFailure) }) + .await; + + { + let mut state = entry.write().await; + let is_current_flight = state + .in_flight + .as_ref() + .is_some_and(|current| current.generation == flight.generation && Arc::ptr_eq(¤t.outcome, &flight.outcome)); + if is_current_flight { + match outcome { + Ok(Some(server_epoch)) => { + state.cached = Some(PutFileCapabilityState::V1 { + server_epoch: *server_epoch, + revalidate_after: Instant::now() + PUT_FILE_V1_CAPABILITY_TTL, + }); + } + Ok(None) if !flight.v1_was_pinned => { + state.cached = Some(PutFileCapabilityState::LegacyUntil(Instant::now() + PUT_FILE_LEGACY_CAPABILITY_TTL)); + } + Ok(None) | Err(_) => {} + } + state.in_flight = None; + } + } + + match outcome { + Ok(Some(server_epoch)) => Ok(Some(*server_epoch)), + Ok(None) if flight.v1_was_pinned => Err(Error::other("remote put_file capability downgrade rejected")), + Ok(None) => Ok(None), + Err(failure) => Err(failure.to_error()), + } +} + +fn verify_put_file_capability_response(challenge: Uuid, body: &[u8]) -> Result { + if body.is_empty() || body.len() > PUT_FILE_MAX_CAPABILITY_RESPONSE_SIZE { + return Err(Error::other("invalid remote put_file capability response size")); + } + let response: PutFileCapabilityResponse = + rmp_serde::from_slice(body).map_err(|_| Error::other("invalid remote put_file capability response"))?; + if response.version != PUT_FILE_CAPABILITY_VERSION || response.server_epoch.is_nil() { + return Err(Error::other("incompatible remote put_file capability response")); + } + verify_put_file_capability(challenge, response.server_epoch, response.version, &response.proof) + .map_err(|err| Error::other(format!("remote put_file capability authentication failed: {err}")))?; + Ok(response.server_epoch) +} + fn build_read_file_stream_url(request: &ReadStreamRequest) -> String { format!( "{}{}?disk={}&volume={}&path={}&offset={}&length={}", @@ -241,26 +452,43 @@ fn build_read_file_stream_url(request: &ReadStreamRequest) -> String { ) } -fn build_put_file_stream_url(request: &WriteStreamRequest, auth_nonce: Option) -> String { +fn build_put_file_stream_url(request: &WriteStreamRequest, auth_scope: Option<(Uuid, Uuid)>) -> String { + let stream_path = if auth_scope.is_some() { + PUT_FILE_AUTH_STREAM_PATH + } else { + PUT_FILE_STREAM_PATH + }; let mut url = format!( "{}{}?disk={}&volume={}&path={}&append={}&size={}", request.endpoint, - PUT_FILE_STREAM_PATH, + stream_path, urlencoding::encode(&request.disk), urlencoding::encode(&request.volume), urlencoding::encode(&request.path), request.append, request.size ); - if let Some(nonce) = auth_nonce { + if let Some((nonce, server_epoch)) = auth_scope { url.push_str(&format!( - "&{}={}&{}={}", - PUT_FILE_AUTH_QUERY, PUT_FILE_AUTH_V1, PUT_FILE_NONCE_QUERY, nonce + "&{}={}&{}={}&{}={}", + PUT_FILE_AUTH_QUERY, PUT_FILE_AUTH_V1, PUT_FILE_NONCE_QUERY, nonce, PUT_FILE_SERVER_EPOCH_QUERY, server_epoch )); } url } +fn build_put_file_capability_url(endpoint: &str, challenge: Uuid) -> String { + format!( + "{}{}?{}={}&{}={}", + endpoint, + PUT_FILE_CAPABILITY_PATH, + PUT_FILE_CAPABILITY_QUERY, + PUT_FILE_CAPABILITY_VERSION, + PUT_FILE_CAPABILITY_CHALLENGE_QUERY, + challenge + ) +} + struct PutFileAuthWriter { inner: W, url: String, @@ -450,6 +678,28 @@ pub fn build_internode_data_transport_from_env() -> Result waiters { + return; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("capability callers should join the in-flight probe"); + } #[derive(Debug)] struct LegacyTestTransport; @@ -578,6 +828,7 @@ mod tests { #[test] fn put_file_stream_url_advertises_auth_nonce_when_enabled() { let nonce = Uuid::parse_str("11111111-2222-4333-8444-555555555555").expect("nonce"); + let server_epoch = Uuid::parse_str("aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee").expect("server epoch"); let url = build_put_file_stream_url( &WriteStreamRequest { endpoint: "http://node1:9000".to_string(), @@ -587,19 +838,405 @@ mod tests { append: false, size: 4096, }, - Some(nonce), + Some((nonce, server_epoch)), ); assert_eq!( url, concat!( - "http://node1:9000/rustfs/rpc/put_file_stream?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0", + "http://node1:9000/rustfs/rpc/put_file_stream_v1?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0", "&volume=bucket&path=object%2Fpart.1&append=false&size=4096", - "&put_file_auth=digest-trailer-v1&put_file_nonce=11111111-2222-4333-8444-555555555555" + "&put_file_auth=digest-trailer-v1&put_file_nonce=11111111-2222-4333-8444-555555555555", + "&put_file_server_epoch=aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee" ) ); } + #[test] + fn put_file_capability_url_binds_version_and_challenge() { + let challenge = Uuid::parse_str("11111111-2222-4333-8444-555555555555").expect("challenge"); + + assert_eq!( + build_put_file_capability_url("http://node1:9000", challenge), + concat!( + "http://node1:9000/rustfs/rpc/put_file_capability?put_file_capability=1", + "&put_file_challenge=11111111-2222-4333-8444-555555555555" + ) + ); + } + + #[test] + fn put_file_capability_legacy_statuses_are_exact() { + assert!(put_file_capability_status_is_legacy(404)); + for status in [200, 400, 401, 403, 405, 408, 426, 429, 500, 503] { + assert!(!put_file_capability_status_is_legacy(status)); + } + } + + #[test] + fn put_file_capability_timeout_is_retryable() { + let error = Error::from(rustfs_rio::internode_http_timeout_error( + &Method::GET, + "http://node:9000/rustfs/rpc/put_file_capability", + )); + + assert_eq!( + error.internode_http_error_kind(), + Some(rustfs_rio::InternodeHttpErrorKind::ConnectTimeout) + ); + assert!(error.is_retryable_internode_write_failure()); + } + + #[tokio::test] + async fn put_file_capability_cache_pins_v1_and_honors_live_legacy_ttl() { + let transport = TcpHttpInternodeDataTransport; + let v1_endpoint = format!("http://v1-{}.invalid", Uuid::new_v4()); + let v1_entry = put_file_capability_cache_entry(&v1_endpoint); + let server_epoch = Uuid::new_v4(); + v1_entry.write().await.cached = Some(PutFileCapabilityState::V1 { + server_epoch, + revalidate_after: Instant::now() + PUT_FILE_V1_CAPABILITY_TTL, + }); + assert_eq!( + transport.put_file_auth_capability(&v1_endpoint).await.expect("v1 cache"), + Some(server_epoch) + ); + let cache_probe_called = AtomicBool::new(false); + assert_eq!( + resolve_put_file_auth_capability(&v1_endpoint, || async { + cache_probe_called.store(true, Ordering::SeqCst); + Ok(None) + }) + .await + .expect("live v1 cache"), + Some(server_epoch) + ); + assert!(!cache_probe_called.load(Ordering::SeqCst)); + v1_entry.write().await.cached = Some(PutFileCapabilityState::V1 { + server_epoch, + revalidate_after: Instant::now(), + }); + assert!( + resolve_put_file_auth_capability(&v1_endpoint, || async { Ok(None) }) + .await + .is_err() + ); + let replacement_epoch = Uuid::new_v4(); + assert_eq!( + resolve_put_file_auth_capability(&v1_endpoint, || async { Ok(Some(replacement_epoch)) }) + .await + .expect("authenticated replacement should refresh the epoch"), + Some(replacement_epoch) + ); + + let legacy_endpoint = format!("http://legacy-{}.invalid", Uuid::new_v4()); + let legacy_entry = put_file_capability_cache_entry(&legacy_endpoint); + legacy_entry.write().await.cached = + Some(PutFileCapabilityState::LegacyUntil(Instant::now() + PUT_FILE_LEGACY_CAPABILITY_TTL)); + assert!( + transport + .put_file_auth_capability(&legacy_endpoint) + .await + .expect("legacy cache") + .is_none() + ); + + let expired_endpoint = format!("http://expired-legacy-{}.invalid", Uuid::new_v4()); + let expired_entry = put_file_capability_cache_entry(&expired_endpoint); + expired_entry.write().await.cached = Some(PutFileCapabilityState::LegacyUntil(Instant::now())); + let reprobed = std::sync::atomic::AtomicBool::new(false); + assert_eq!( + resolve_put_file_auth_capability(&expired_endpoint, || async { + reprobed.store(true, std::sync::atomic::Ordering::SeqCst); + Ok(Some(server_epoch)) + }) + .await + .expect("expired legacy cache should reprobe"), + Some(server_epoch) + ); + assert!(reprobed.load(std::sync::atomic::Ordering::SeqCst)); + } + + #[tokio::test] + async fn legacy_put_file_capability_omits_the_auth_trailer_protocol() { + let endpoint = format!("http://legacy-selection-{}.invalid", Uuid::new_v4()); + let server_epoch = resolve_put_file_auth_capability(&endpoint, || async { Ok(None) }) + .await + .expect("legacy capability result"); + let auth_scope = server_epoch.map(|epoch| (Uuid::new_v4(), epoch)); + let url = build_put_file_stream_url( + &WriteStreamRequest { + endpoint, + disk: "http://node1:9000/data/rustfs0".to_string(), + volume: "bucket".to_string(), + path: "object/part.1".to_string(), + append: false, + size: 4096, + }, + auth_scope, + ); + + assert!(auth_scope.is_none()); + assert!(!url.contains(PUT_FILE_AUTH_QUERY)); + assert!(!url.contains(PUT_FILE_NONCE_QUERY)); + } + + #[tokio::test] + async fn put_file_capability_probe_is_singleflight_per_endpoint() { + let endpoint = format!("http://singleflight-{}.invalid", Uuid::new_v4()); + let entry = put_file_capability_cache_entry(&endpoint); + let calls = Arc::new(AtomicUsize::new(0)); + let release = Arc::new(Notify::new()); + let start = Arc::new(Barrier::new(65)); + let mut tasks = Vec::with_capacity(64); + + for _ in 0..64 { + let endpoint = endpoint.clone(); + let calls = Arc::clone(&calls); + let release = Arc::clone(&release); + let start = Arc::clone(&start); + tasks.push(tokio::spawn(async move { + start.wait().await; + resolve_put_file_auth_capability(&endpoint, || async move { + calls.fetch_add(1, Ordering::SeqCst); + release.notified().await; + Err(Error::from(rustfs_rio::new_test_internode_http_io_error( + rustfs_rio::InternodeHttpErrorKind::ConnectionRefused, + ))) + }) + .await + })); + } + + start.wait().await; + wait_for_capability_flight_waiters(&entry, 64).await; + assert_eq!(calls.load(Ordering::SeqCst), 1); + release.notify_waiters(); + + let results = tokio::time::timeout(Duration::from_secs(1), futures::future::join_all(tasks)) + .await + .expect("all callers should finish within one probe window"); + for result in results { + let error = result.expect("capability task should finish").expect_err("probe should fail"); + assert_eq!( + error.internode_http_error_kind(), + Some(rustfs_rio::InternodeHttpErrorKind::ConnectionRefused) + ); + assert!(error.is_retryable_internode_write_failure()); + } + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn put_file_capability_probe_recovers_when_initializer_is_cancelled() { + let endpoint = format!("http://cancelled-singleflight-{}.invalid", Uuid::new_v4()); + let entry = put_file_capability_cache_entry(&endpoint); + let calls = Arc::new(AtomicUsize::new(0)); + let initializer_started = Arc::new(Notify::new()); + let never_release = Arc::new(Notify::new()); + + let first = { + let endpoint = endpoint.clone(); + let calls = Arc::clone(&calls); + let initializer_started = Arc::clone(&initializer_started); + let never_release = Arc::clone(&never_release); + tokio::spawn(async move { + resolve_put_file_auth_capability(&endpoint, || async move { + calls.fetch_add(1, Ordering::SeqCst); + initializer_started.notify_one(); + never_release.notified().await; + Ok(Some(Uuid::new_v4())) + }) + .await + }) + }; + initializer_started.notified().await; + + let replacement_epoch = Uuid::new_v4(); + let second = { + let endpoint = endpoint.clone(); + let calls = Arc::clone(&calls); + tokio::spawn(async move { + resolve_put_file_auth_capability(&endpoint, || async move { + calls.fetch_add(1, Ordering::SeqCst); + Ok(Some(replacement_epoch)) + }) + .await + }) + }; + wait_for_capability_flight_waiters(&entry, 2).await; + first.abort(); + assert!(first.await.expect_err("initializer should be cancelled").is_cancelled()); + + assert_eq!( + second.await.expect("waiter should finish").expect("waiter should take over"), + Some(replacement_epoch) + ); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn put_file_capability_probe_recovers_after_all_callers_cancel() { + let endpoint = format!("http://all-cancelled-{}.invalid", Uuid::new_v4()); + let entry = put_file_capability_cache_entry(&endpoint); + let calls = Arc::new(AtomicUsize::new(0)); + let initializer_started = Arc::new(Notify::new()); + let never_release = Arc::new(Notify::new()); + + let first = { + let endpoint = endpoint.clone(); + let calls = Arc::clone(&calls); + let initializer_started = Arc::clone(&initializer_started); + let never_release = Arc::clone(&never_release); + tokio::spawn(async move { + resolve_put_file_auth_capability(&endpoint, || async move { + calls.fetch_add(1, Ordering::SeqCst); + initializer_started.notify_one(); + never_release.notified().await; + Ok(None) + }) + .await + }) + }; + initializer_started.notified().await; + let second = { + let endpoint = endpoint.clone(); + let calls = Arc::clone(&calls); + tokio::spawn(async move { + resolve_put_file_auth_capability(&endpoint, || async move { + calls.fetch_add(1, Ordering::SeqCst); + Ok(None) + }) + .await + }) + }; + wait_for_capability_flight_waiters(&entry, 2).await; + first.abort(); + second.abort(); + assert!(first.await.expect_err("initializer should be cancelled").is_cancelled()); + assert!(second.await.expect_err("waiter should be cancelled").is_cancelled()); + + let server_epoch = Uuid::new_v4(); + assert_eq!( + resolve_put_file_auth_capability(&endpoint, || async { + calls.fetch_add(1, Ordering::SeqCst); + Ok(Some(server_epoch)) + }) + .await + .expect("later caller should initialize the abandoned flight"), + Some(server_epoch) + ); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn put_file_capability_failed_wave_can_retry_immediately() { + let endpoint = format!("http://retry-after-failure-{}.invalid", Uuid::new_v4()); + let first = resolve_put_file_auth_capability(&endpoint, || async { Err(Error::Timeout) }).await; + assert!(matches!(first, Err(Error::Timeout))); + + let server_epoch = Uuid::new_v4(); + assert_eq!( + resolve_put_file_auth_capability(&endpoint, || async { Ok(Some(server_epoch)) }) + .await + .expect("new request should reprobe"), + Some(server_epoch) + ); + } + + #[tokio::test] + async fn put_file_capability_probes_different_endpoints_in_parallel() { + let first_endpoint = format!("http://parallel-a-{}.invalid", Uuid::new_v4()); + let second_endpoint = format!("http://parallel-b-{}.invalid", Uuid::new_v4()); + let probes_started = Arc::new(Barrier::new(2)); + let first_barrier = Arc::clone(&probes_started); + let second_barrier = Arc::clone(&probes_started); + + let results = tokio::time::timeout(Duration::from_secs(5), async { + tokio::join!( + resolve_put_file_auth_capability(&first_endpoint, || async move { + first_barrier.wait().await; + Ok(None) + }), + resolve_put_file_auth_capability(&second_endpoint, || async move { + second_barrier.wait().await; + Ok(None) + }) + ) + }) + .await + .expect("different endpoints should not serialize"); + assert!(results.0.expect("first result").is_none()); + assert!(results.1.expect("second result").is_none()); + } + + #[tokio::test] + async fn stale_put_file_capability_flight_cannot_overwrite_newer_state() { + let endpoint = format!("http://stale-flight-{}.invalid", Uuid::new_v4()); + let entry = put_file_capability_cache_entry(&endpoint); + let probe_started = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let stale_epoch = Uuid::new_v4(); + let newer_epoch = Uuid::new_v4(); + + let task = { + let endpoint = endpoint.clone(); + let probe_started = Arc::clone(&probe_started); + let release = Arc::clone(&release); + tokio::spawn(async move { + resolve_put_file_auth_capability(&endpoint, || async move { + probe_started.notify_one(); + release.notified().await; + Ok(Some(stale_epoch)) + }) + .await + }) + }; + probe_started.notified().await; + { + let mut state = entry.write().await; + state.generation = state.generation.checked_add(1).expect("test generation should advance"); + state.cached = Some(PutFileCapabilityState::V1 { + server_epoch: newer_epoch, + revalidate_after: Instant::now() + PUT_FILE_V1_CAPABILITY_TTL, + }); + state.in_flight = None; + } + release.notify_one(); + assert_eq!( + task.await.expect("stale task should finish").expect("stale probe result"), + Some(stale_epoch) + ); + assert_eq!( + fresh_put_file_capability(entry.read().await.cached, Instant::now()), + Some(Some(newer_epoch)) + ); + } + + #[test] + fn put_file_capability_response_fails_closed_on_malformed_or_unbound_data() { + let _ = rustfs_credentials::set_global_rpc_secret("put-file-capability-response-test-secret".to_string()); + let challenge = Uuid::parse_str("11111111-2222-4333-8444-555555555555").expect("challenge"); + let server_epoch = Uuid::parse_str("aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee").expect("server epoch"); + let proof = crate::cluster::rpc::sign_put_file_capability(challenge, server_epoch, PUT_FILE_CAPABILITY_VERSION) + .expect("proof should build"); + let response = PutFileCapabilityResponse { + version: PUT_FILE_CAPABILITY_VERSION, + server_epoch, + proof, + }; + let body = rmp_serde::to_vec_named(&response).expect("response should encode"); + + assert_eq!( + verify_put_file_capability_response(challenge, &body).expect("response should verify"), + server_epoch + ); + assert!(verify_put_file_capability_response(Uuid::new_v4(), &body).is_err()); + assert!(verify_put_file_capability_response(challenge, &body[..body.len() - 1]).is_err()); + assert!(verify_put_file_capability_response(challenge, &[]).is_err()); + assert!(verify_put_file_capability_response(challenge, &vec![0_u8; PUT_FILE_MAX_CAPABILITY_RESPONSE_SIZE + 1]).is_err()); + } + #[tokio::test] async fn put_file_auth_writer_appends_trailer_on_shutdown() { use tokio::io::AsyncWriteExt; diff --git a/crates/ecstore/src/cluster/rpc/mod.rs b/crates/ecstore/src/cluster/rpc/mod.rs index 118497c99..6eac3b066 100644 --- a/crates/ecstore/src/cluster/rpc/mod.rs +++ b/crates/ecstore/src/cluster/rpc/mod.rs @@ -34,11 +34,11 @@ pub use client::{ pub use http_auth::{ TONIC_RPC_PREFIX, build_auth_headers, build_put_file_auth_trailer, check_and_record_signed_rpc_nonce, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience, set_tonic_canonical_body_digest, - set_tonic_mutation_body_digest, sign_ns_scanner_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, - tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason, verify_ns_scanner_capability, verify_put_file_auth_trailer, - verify_rpc_signature, verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest, - verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof, verify_tonic_rpc_signature, - verify_tonic_rpc_signature_with_bootstrap, + set_tonic_mutation_body_digest, sign_ns_scanner_capability, sign_put_file_capability, sign_tonic_rpc_response_proof, + tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason, verify_ns_scanner_capability, + verify_put_file_auth_trailer, verify_put_file_capability, verify_rpc_signature, verify_tonic_boot_epoch_response, + verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof, + verify_tonic_rpc_signature, verify_tonic_rpc_signature_with_bootstrap, }; #[cfg(test)] pub(crate) use internode_data_transport::TcpHttpInternodeDataTransport; diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 36b660f65..485858d15 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -4486,6 +4486,27 @@ mod tests { assert_eq!(snapshot.outgoing_requests_total, 0); } + #[tokio::test] + #[serial(internode_metrics)] + async fn test_remote_disk_create_file_retries_once_on_capability_probe_timeout() { + let transport = RetryingOpenWriteInternodeDataTransport::with_steps(vec![ + OpenWriteTestStep::Error(DiskError::from(rustfs_rio::internode_http_timeout_error( + &http::Method::GET, + "http://remote-node:9000/rustfs/rpc/put_file_capability", + ))), + OpenWriteTestStep::Success, + ]); + let remote_disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await; + crate::cluster::rpc::runtime_sources::reset_internode_metrics_for_test(); + + let _created = remote_disk + .create_file("orig-bucket", "bucket", "object/part.1", 4096) + .await + .expect("capability probe timeout should recover on retry"); + + assert_eq!(transport.calls().len(), 2, "create_file should retry capability probe timeouts once"); + } + #[tokio::test] async fn test_remote_disk_append_file_does_not_retry_non_retryable_open_write_error() { let transport = RetryingOpenWriteInternodeDataTransport::with_steps(vec![OpenWriteTestStep::Error(DiskError::from( diff --git a/crates/ecstore/src/storage_api_contracts/mod.rs b/crates/ecstore/src/storage_api_contracts/mod.rs index f0d33c363..11e7b800f 100644 --- a/crates/ecstore/src/storage_api_contracts/mod.rs +++ b/crates/ecstore/src/storage_api_contracts/mod.rs @@ -29,9 +29,10 @@ pub(crate) mod internode { NS_SCANNER_REQUEST_ID_QUERY, NS_SCANNER_SERVER_EPOCH_QUERY, NS_SCANNER_SESSION_ID_QUERY, NS_SCANNER_SESSION_SEQUENCE_QUERY, NsScannerCapabilityResponse, PUT_FILE_AUTH_QUERY, PUT_FILE_AUTH_TRAILER_DIGEST_LEN, PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_TRAILER_MAC_LEN, PUT_FILE_AUTH_TRAILER_MAGIC, PUT_FILE_AUTH_V1, - PUT_FILE_NONCE_QUERY, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, - SCANNER_ACTIVITY_PROTOCOL_VERSION, WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_STREAM_COMPLETION_QUERY, - WALK_DIR_STREAM_COMPLETION_V1, + PUT_FILE_CAPABILITY_CHALLENGE_QUERY, PUT_FILE_CAPABILITY_QUERY, PUT_FILE_CAPABILITY_VERSION, PUT_FILE_NONCE_QUERY, + PUT_FILE_SERVER_EPOCH_QUERY, PutFileCapabilityResponse, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, + SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION, WALK_DIR_BODY_SHA256_QUERY, + WALK_DIR_STREAM_COMPLETION_QUERY, WALK_DIR_STREAM_COMPLETION_V1, }; } diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index 020201779..62ffe091c 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -22,6 +22,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; pub const INTERNODE_OPERATION_READ_FILE_STREAM: &str = "read_file_stream"; pub const INTERNODE_OPERATION_PUT_FILE_STREAM: &str = "put_file_stream"; +pub const INTERNODE_OPERATION_PUT_FILE_CAPABILITY: &str = "put_file_capability"; pub const INTERNODE_OPERATION_WALK_DIR: &str = "walk_dir"; pub const INTERNODE_OPERATION_NS_SCANNER: &str = "ns_scanner"; pub const INTERNODE_OPERATION_GRPC_READ_ALL: &str = "grpc_read_all"; @@ -999,6 +1000,7 @@ mod tests { fn operation_metric_names_and_low_cardinality_values_are_stable() { assert_eq!(INTERNODE_OPERATION_READ_FILE_STREAM, "read_file_stream"); assert_eq!(INTERNODE_OPERATION_PUT_FILE_STREAM, "put_file_stream"); + assert_eq!(INTERNODE_OPERATION_PUT_FILE_CAPABILITY, "put_file_capability"); assert_eq!(INTERNODE_OPERATION_WALK_DIR, "walk_dir"); assert_eq!(INTERNODE_OPERATION_GRPC_READ_ALL, "grpc_read_all"); assert_eq!(INTERNODE_OPERATION_GRPC_WRITE_ALL, "grpc_write_all"); diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index 14289bda6..531c4edfb 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -19,8 +19,8 @@ use http::{HeaderMap, Version}; use pin_project_lite::pin_project; use reqwest::{Certificate, Client, Identity, Method, RequestBuilder}; use rustfs_io_metrics::internode_metrics::{ - INTERNODE_OPERATION_NS_SCANNER, INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, - INTERNODE_OPERATION_WALK_DIR, + INTERNODE_OPERATION_NS_SCANNER, INTERNODE_OPERATION_PUT_FILE_CAPABILITY, INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR, }; use rustfs_tls_runtime::load_cert_bundle_der_bytes; use rustfs_utils::{get_env_bool, get_env_opt_str, get_env_opt_u64, get_env_opt_usize}; @@ -43,6 +43,8 @@ use tracing::{error, warn}; const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream"; const PUT_FILE_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream"; +const PUT_FILE_AUTH_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream_v1"; +const PUT_FILE_CAPABILITY_PATH: &str = "/rustfs/rpc/put_file_capability"; const WALK_DIR_PATH: &str = "/rustfs/rpc/walk_dir"; const NS_SCANNER_PATH: &str = "/rustfs/rpc/ns_scanner"; const HTTP_VERSION_09_LABEL: &str = "http/0.9"; @@ -261,6 +263,31 @@ pub fn new_test_internode_http_io_error(kind: InternodeHttpErrorKind) -> io::Err InternodeHttpError::new_for_test(kind).into_io_error() } +/// Build a retryable internode timeout error with the request's operation context. +#[doc(hidden)] +pub fn internode_http_timeout_error(method: &Method, url: &str) -> io::Error { + internode_kind_error(method, url, internode_rpc_operation(url), InternodeHttpErrorKind::ConnectTimeout) +} + +/// Clone an internode HTTP I/O error while retaining its structured classification. +/// +/// The underlying transport source is intentionally omitted because it is not +/// cloneable. The request context and remote disk marker remain available to +/// retry and error-mapping code. +#[doc(hidden)] +pub fn clone_internode_http_io_error(error: &io::Error) -> Option { + let source = error.get_ref()?.downcast_ref::()?; + Some( + InternodeHttpError { + kind: source.kind, + context: source.context.clone(), + remote_disk_error: source.remote_disk_error, + source: None, + } + .into_io_error(), + ) +} + #[doc(hidden)] pub fn new_test_remote_file_not_found_http_io_error() -> io::Error { InternodeHttpError::with_remote_disk_error( @@ -1223,7 +1250,8 @@ fn internode_rpc_operation(url: &str) -> Option<&'static str> { let url = reqwest::Url::parse(url).ok()?; match url.path() { READ_FILE_STREAM_PATH => Some(INTERNODE_OPERATION_READ_FILE_STREAM), - PUT_FILE_STREAM_PATH => Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + PUT_FILE_STREAM_PATH | PUT_FILE_AUTH_STREAM_PATH => Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + PUT_FILE_CAPABILITY_PATH => Some(INTERNODE_OPERATION_PUT_FILE_CAPABILITY), WALK_DIR_PATH => Some(INTERNODE_OPERATION_WALK_DIR), NS_SCANNER_PATH => Some(INTERNODE_OPERATION_NS_SCANNER), _ => None, @@ -1920,6 +1948,14 @@ mod tests { internode_rpc_operation(&format!("http://node:9000{PUT_FILE_STREAM_PATH}?disk=d")), Some(INTERNODE_OPERATION_PUT_FILE_STREAM) ); + assert_eq!( + internode_rpc_operation(&format!("http://node:9000{PUT_FILE_AUTH_STREAM_PATH}?disk=d")), + Some(INTERNODE_OPERATION_PUT_FILE_STREAM) + ); + assert_eq!( + internode_rpc_operation(&format!("http://node:9000{PUT_FILE_CAPABILITY_PATH}?put_file_capability=1")), + Some(INTERNODE_OPERATION_PUT_FILE_CAPABILITY) + ); assert_eq!( internode_rpc_operation(&format!("http://node:9000{WALK_DIR_PATH}?disk=d")), Some(INTERNODE_OPERATION_WALK_DIR) @@ -1935,6 +1971,21 @@ mod tests { ); } + #[test] + fn internode_http_timeout_error_retains_operation_context() { + let error = + internode_http_timeout_error(&Method::GET, "http://node:9000/rustfs/rpc/put_file_capability?put_file_capability=1"); + let source = error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("timeout should retain internode classification"); + + assert_eq!(source.kind(), InternodeHttpErrorKind::ConnectTimeout); + assert_eq!(source.context().method(), "GET"); + assert_eq!(source.context().target(), PUT_FILE_CAPABILITY_PATH); + assert_eq!(source.context().operation(), Some(INTERNODE_OPERATION_PUT_FILE_CAPABILITY)); + } + #[test] fn http_version_metrics_labels_are_low_cardinality() { assert_eq!(http_version_metric_label(Version::HTTP_09), HTTP_VERSION_09_LABEL); @@ -2385,6 +2436,42 @@ mod tests { assert!(source.context().target().contains(PUT_FILE_STREAM_PATH)); } + #[test] + fn cloned_internode_http_error_retains_classification_and_context() { + let original = internode_status_error( + &Method::GET, + "http://node:9000/rustfs/rpc/put_file_capability?put_file_capability=1", + Some(INTERNODE_OPERATION_PUT_FILE_CAPABILITY), + reqwest::StatusCode::SERVICE_UNAVAILABLE, + ); + let cloned = clone_internode_http_io_error(&original).expect("internode error should clone"); + let source = cloned + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("clone should retain internode source"); + + assert_eq!( + source.kind(), + InternodeHttpErrorKind::HttpStatus(reqwest::StatusCode::SERVICE_UNAVAILABLE) + ); + assert!(source.kind().is_retryable()); + assert_eq!(source.context().method(), "GET"); + assert_eq!(source.context().target(), PUT_FILE_CAPABILITY_PATH); + assert_eq!(source.context().operation(), Some(INTERNODE_OPERATION_PUT_FILE_CAPABILITY)); + } + + #[test] + fn cloned_internode_http_error_retains_remote_disk_marker() { + let original = new_test_remote_file_not_found_http_io_error(); + let cloned = clone_internode_http_io_error(&original).expect("internode error should clone"); + let source = cloned + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("clone should retain internode source"); + + assert!(source.is_remote_file_not_found()); + } + #[test] fn loopback_urls_bypass_proxy_selection() { assert!(should_bypass_proxy_for_url("http://127.0.0.1:9000/stream")); diff --git a/crates/storage-api/src/lib.rs b/crates/storage-api/src/lib.rs index 7c61251dc..e011327f9 100644 --- a/crates/storage-api/src/lib.rs +++ b/crates/storage-api/src/lib.rs @@ -20,11 +20,23 @@ pub const WALK_DIR_BODY_SHA256_QUERY: &str = "walk_dir_body_sha256"; pub const PUT_FILE_AUTH_QUERY: &str = "put_file_auth"; pub const PUT_FILE_AUTH_V1: &str = "digest-trailer-v1"; pub const PUT_FILE_NONCE_QUERY: &str = "put_file_nonce"; +pub const PUT_FILE_CAPABILITY_QUERY: &str = "put_file_capability"; +pub const PUT_FILE_CAPABILITY_CHALLENGE_QUERY: &str = "put_file_challenge"; +pub const PUT_FILE_CAPABILITY_VERSION: u16 = 1; +pub const PUT_FILE_SERVER_EPOCH_QUERY: &str = "put_file_server_epoch"; pub const PUT_FILE_AUTH_TRAILER_MAGIC: &[u8; 16] = b"RFS-PUT-AUTH-V1\0"; pub const PUT_FILE_AUTH_TRAILER_DIGEST_LEN: usize = 64; pub const PUT_FILE_AUTH_TRAILER_MAC_LEN: usize = 32; pub const PUT_FILE_AUTH_TRAILER_LEN: usize = PUT_FILE_AUTH_TRAILER_MAGIC.len() + PUT_FILE_AUTH_TRAILER_DIGEST_LEN + PUT_FILE_AUTH_TRAILER_MAC_LEN; + +#[derive(Debug, serde::Deserialize, serde::Serialize)] +#[serde(deny_unknown_fields)] +pub struct PutFileCapabilityResponse { + pub version: u16, + pub server_epoch: uuid::Uuid, + pub proof: Vec, +} pub const NS_SCANNER_BODY_SHA256_QUERY: &str = "ns_scanner_body_sha256"; pub const NS_SCANNER_CAPABILITY_CHALLENGE_QUERY: &str = "ns_scanner_challenge"; pub const NS_SCANNER_CYCLE_QUERY: &str = "ns_scanner_cycle"; diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index 3842cf3d2..0e2cfe61e 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -17,15 +17,15 @@ use crate::storage::request_context::spawn_traced; use crate::storage::storage_api::DiskError; use crate::storage::storage_api::rpc_consumer::http_service::{ DEFAULT_READ_BUFFER_SIZE, DeleteOptions, DiskStore, NS_SCANNER_PROTOCOL_VERSION, NsScannerCapabilityResponse, - PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_V1, StorageDiskRpcExt as _, WALK_DIR_STREAM_COMPLETION_V1, WalkDirOptions, - check_and_record_signed_rpc_nonce, find_local_disk_by_ref, sign_ns_scanner_capability, verify_put_file_auth_trailer, - verify_rpc_signature, + PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_V1, PUT_FILE_CAPABILITY_VERSION, PutFileCapabilityResponse, StorageDiskRpcExt as _, + WALK_DIR_STREAM_COMPLETION_V1, WalkDirOptions, check_and_record_signed_rpc_nonce, find_local_disk_by_ref, + sign_ns_scanner_capability, sign_put_file_capability, verify_put_file_auth_trailer, verify_rpc_signature, }; #[cfg(test)] use crate::storage::storage_api::rpc_consumer::http_service::{ NS_SCANNER_BODY_SHA256_QUERY, NS_SCANNER_CAPABILITY_CHALLENGE_QUERY, NS_SCANNER_CYCLE_QUERY, NS_SCANNER_LEADER_EPOCH_QUERY, NS_SCANNER_REQUEST_ID_QUERY, NS_SCANNER_SERVER_EPOCH_QUERY, NS_SCANNER_SESSION_ID_QUERY, NS_SCANNER_SESSION_SEQUENCE_QUERY, - WALK_DIR_BODY_SHA256_QUERY, + PUT_FILE_CAPABILITY_CHALLENGE_QUERY, PUT_FILE_CAPABILITY_QUERY, WALK_DIR_BODY_SHA256_QUERY, }; use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; use crate::storage::storage_api::tonic_rpc_auth_failure_reason; @@ -36,8 +36,8 @@ use http_body_util::{BodyExt, Limited}; use hyper::body::Incoming; use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; use rustfs_io_metrics::internode_metrics::{ - INTERNODE_OPERATION_NS_SCANNER, INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, - INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + INTERNODE_OPERATION_NS_SCANNER, INTERNODE_OPERATION_PUT_FILE_CAPABILITY, INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, }; use rustfs_utils::net::bytes_stream; use s3s::Body; @@ -70,11 +70,14 @@ const EVENT_RPC_BACKGROUND_TASK_FAILED: &str = "rpc_background_task_failed"; const RPC_OPERATION_UNKNOWN: &str = "unknown"; const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream"; const PUT_FILE_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream"; +const PUT_FILE_AUTH_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream_v1"; +const PUT_FILE_CAPABILITY_PATH: &str = "/rustfs/rpc/put_file_capability"; const WALK_DIR_PATH: &str = "/rustfs/rpc/walk_dir"; const NS_SCANNER_PATH: &str = "/rustfs/rpc/ns_scanner"; const NS_SCANNER_REQUEST_BODY_TIMEOUT: Duration = Duration::from_secs(15); const NS_SCANNER_STREAM_BUFFER_SIZE: usize = 64 * 1024; static NS_SCANNER_SERVER_EPOCH: LazyLock = LazyLock::new(uuid::Uuid::new_v4); +static PUT_FILE_CAPABILITY_SERVER_EPOCH: LazyLock = LazyLock::new(uuid::Uuid::new_v4); static PUT_FILE_AUTH_STRICT: LazyLock = LazyLock::new(|| { rustfs_utils::get_env_bool( rustfs_config::ENV_INTERNODE_RPC_BODY_DIGEST_STRICT, @@ -331,6 +334,14 @@ struct PutFileQuery { size: i64, put_file_auth: Option, put_file_nonce: Option, + put_file_server_epoch: Option, +} + +#[derive(Debug, Default, serde::Deserialize)] +#[serde(deny_unknown_fields)] +struct PutFileCapabilityQuery { + put_file_capability: Option, + put_file_challenge: Option, } fn put_file_auth_nonce(query: &PutFileQuery) -> io::Result> { @@ -352,6 +363,10 @@ fn put_file_auth_nonce(query: &PutFileQuery) -> io::Result> { } } +fn put_file_server_epoch_matches(query: &PutFileQuery) -> bool { + query.put_file_server_epoch == Some(*PUT_FILE_CAPABILITY_SERVER_EPOCH) +} + impl Service> for InternodeRpcService where S: Service, Response = Response> + Clone + Send + 'static, @@ -403,7 +418,18 @@ async fn handle_internode_rpc(req: Request) -> Response { Err(response) => *response, }, (Method::POST, NS_SCANNER_PATH) => handle_ns_scanner(req).await, - (Method::PUT, PUT_FILE_STREAM_PATH) => handle_put_file(req).await, + (Method::GET, PUT_FILE_CAPABILITY_PATH) => match parse_query::(&req) { + Ok(query) if query.put_file_capability == Some(PUT_FILE_CAPABILITY_VERSION) => { + match query.put_file_challenge.filter(|challenge| !challenge.is_nil()) { + Some(challenge) => put_file_capability_response(challenge), + None => response_with_status(StatusCode::BAD_REQUEST, "put_file capability challenge is invalid"), + } + } + Ok(_) => response_with_status(StatusCode::UPGRADE_REQUIRED, "put_file capability is unsupported"), + Err(response) => *response, + }, + (Method::PUT, PUT_FILE_STREAM_PATH) => handle_put_file(req, false).await, + (Method::PUT, PUT_FILE_AUTH_STREAM_PATH) => handle_put_file(req, true).await, _ => response_with_status(StatusCode::NOT_FOUND, "internode rpc route not found"), }; @@ -425,7 +451,8 @@ async fn handle_internode_rpc(req: Request) -> Response { fn internode_http_operation(path: &str) -> Option<&'static str> { match path { READ_FILE_STREAM_PATH => Some(INTERNODE_OPERATION_READ_FILE_STREAM), - PUT_FILE_STREAM_PATH => Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + PUT_FILE_STREAM_PATH | PUT_FILE_AUTH_STREAM_PATH => Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + PUT_FILE_CAPABILITY_PATH => Some(INTERNODE_OPERATION_PUT_FILE_CAPABILITY), WALK_DIR_PATH => Some(INTERNODE_OPERATION_WALK_DIR), NS_SCANNER_PATH => Some(INTERNODE_OPERATION_NS_SCANNER), _ => None, @@ -494,6 +521,36 @@ fn ns_scanner_capability_response(challenge: uuid::Uuid) -> Response { response } +fn put_file_capability_response(challenge: uuid::Uuid) -> Response { + let server_epoch = *PUT_FILE_CAPABILITY_SERVER_EPOCH; + let proof = match sign_put_file_capability(challenge, server_epoch, PUT_FILE_CAPABILITY_VERSION) { + Ok(proof) => proof, + Err(err) => { + return response_with_status( + StatusCode::INTERNAL_SERVER_ERROR, + format!("put_file capability authentication is unavailable: {err}"), + ); + } + }; + match rmp_serde::to_vec_named(&PutFileCapabilityResponse { + version: PUT_FILE_CAPABILITY_VERSION, + server_epoch, + proof, + }) { + Ok(body) => { + let mut response = Response::new(Body::from(Bytes::from(body))); + response + .headers_mut() + .insert(http::header::CONTENT_TYPE, HeaderValue::from_static("application/msgpack")); + response + } + Err(err) => response_with_status( + StatusCode::INTERNAL_SERVER_ERROR, + format!("put_file capability response encoding failed: {err}"), + ), + } +} + fn ns_scanner_server_epoch_matches(server_epoch: uuid::Uuid) -> bool { server_epoch == *NS_SCANNER_SERVER_EPOCH } @@ -1245,7 +1302,7 @@ where } } -async fn handle_put_file(req: Request) -> Response { +async fn handle_put_file(req: Request, require_auth: bool) -> Response { let method = req.method().clone(); let path = req.uri().path().to_string(); let url = req.uri().to_string(); @@ -1269,11 +1326,17 @@ async fn handle_put_file(req: Request) -> Response { return response_with_status(StatusCode::FORBIDDEN, format!("invalid put_file auth: {e}")); } }; + if require_auth && auth_nonce.is_none() { + return response_with_status(StatusCode::FORBIDDEN, "invalid put_file auth: put_file auth required"); + } + if require_auth && !put_file_server_epoch_matches(&query) { + return response_with_status(StatusCode::CONFLICT, "put_file capability server epoch changed"); + } if let Some(nonce) = auth_nonce && let Err(e) = check_and_record_signed_rpc_nonce( req.headers(), nonce, - PUT_FILE_STREAM_PATH, + &path, INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, ) @@ -1576,7 +1639,9 @@ fn internode_rpc_subsystem(operation: Option<&'static str>) -> &'static str { match operation { Some(INTERNODE_OPERATION_WALK_DIR) => LOG_SUBSYSTEM_DIRECTORY_WALK, Some(INTERNODE_OPERATION_NS_SCANNER) => LOG_SUBSYSTEM_NAMESPACE_SCANNER, - Some(INTERNODE_OPERATION_READ_FILE_STREAM | INTERNODE_OPERATION_PUT_FILE_STREAM) => LOG_SUBSYSTEM_FILE_TRANSFER, + Some( + INTERNODE_OPERATION_READ_FILE_STREAM | INTERNODE_OPERATION_PUT_FILE_STREAM | INTERNODE_OPERATION_PUT_FILE_CAPABILITY, + ) => LOG_SUBSYSTEM_FILE_TRANSFER, _ => LOG_SUBSYSTEM_ROUTING, } } @@ -1599,36 +1664,43 @@ fn put_file_stage_error_message(stage: &str, query: &PutFileQuery, err: &dyn std #[cfg(test)] mod tests { use super::{ - DiskError, LOG_SUBSYSTEM_DIRECTORY_WALK, LOG_SUBSYSTEM_FILE_TRANSFER, LOG_SUBSYSTEM_NAMESPACE_SCANNER, - LOG_SUBSYSTEM_ROUTING, NS_SCANNER_BODY_SHA256_QUERY, NS_SCANNER_CAPABILITY_CHALLENGE_QUERY, NS_SCANNER_CYCLE_QUERY, - NS_SCANNER_LEADER_EPOCH_QUERY, NS_SCANNER_PATH, NS_SCANNER_REQUEST_ID_QUERY, NS_SCANNER_SERVER_EPOCH_QUERY, - NS_SCANNER_SESSION_ID_QUERY, NS_SCANNER_SESSION_SEQUENCE_QUERY, NsScannerQuery, PUT_FILE_STREAM_PATH, PutFileQuery, - READ_FILE_STREAM_PATH, WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_PATH, WalkDirQuery, append_walk_dir_completion, - internode_http_operation, internode_rpc_subsystem, is_internode_rpc_path, ns_scanner_response_body, - ns_scanner_server_epoch_matches, put_body_size_mismatch, put_file_auth_nonce, put_file_stage_error_message, - put_file_target_lock, read_file_body_stream, remote_scanner_claim_rejection, response_with_disk_error, - supports_walk_dir_stream_completion, validate_walk_dir_completion_request, verify_internode_rpc_signature, - verify_ns_scanner_body_digest, verify_walk_dir_body_digest, walk_dir_response_body, write_authenticated_put_file, - write_body_chunks_to_writer, write_put_file_body_chunks_to_writer, + DiskError, InternodeRpcService, LOG_SUBSYSTEM_DIRECTORY_WALK, LOG_SUBSYSTEM_FILE_TRANSFER, + LOG_SUBSYSTEM_NAMESPACE_SCANNER, LOG_SUBSYSTEM_ROUTING, NS_SCANNER_BODY_SHA256_QUERY, + NS_SCANNER_CAPABILITY_CHALLENGE_QUERY, NS_SCANNER_CYCLE_QUERY, NS_SCANNER_LEADER_EPOCH_QUERY, NS_SCANNER_PATH, + NS_SCANNER_REQUEST_ID_QUERY, NS_SCANNER_SERVER_EPOCH_QUERY, NS_SCANNER_SESSION_ID_QUERY, + NS_SCANNER_SESSION_SEQUENCE_QUERY, NsScannerQuery, PUT_FILE_AUTH_STREAM_PATH, PUT_FILE_CAPABILITY_PATH, + PUT_FILE_STREAM_PATH, PutFileQuery, READ_FILE_STREAM_PATH, WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_PATH, WalkDirQuery, + append_walk_dir_completion, internode_http_operation, internode_rpc_subsystem, is_internode_rpc_path, + ns_scanner_response_body, ns_scanner_server_epoch_matches, put_body_size_mismatch, put_file_auth_nonce, + put_file_capability_response, put_file_server_epoch_matches, put_file_stage_error_message, put_file_target_lock, + read_file_body_stream, remote_scanner_claim_rejection, response_with_disk_error, supports_walk_dir_stream_completion, + validate_walk_dir_completion_request, verify_internode_rpc_signature, verify_ns_scanner_body_digest, + verify_walk_dir_body_digest, walk_dir_response_body, write_authenticated_put_file, write_body_chunks_to_writer, + write_put_file_body_chunks_to_writer, }; - use crate::storage::storage_api::ecstore_rpc::build_put_file_auth_trailer; + use crate::storage::storage_api::ecstore_rpc::{build_put_file_auth_trailer, gen_signature_headers}; use crate::storage::storage_api::rpc_consumer::http_service::{DiskAPI as _, DiskOption, DiskStore, Endpoint, new_disk}; use bytes::Bytes; - use http::{HeaderMap, HeaderValue, Method, StatusCode, Uri}; - use http_body_util::BodyExt; + use http::{HeaderMap, HeaderValue, Method, Request, Response, StatusCode, Uri}; + use http_body_util::{BodyExt, Empty}; + use hyper::{client::conn::http1 as client_http1, server::conn::http1 as server_http1}; + use hyper_util::{rt::TokioIo, service::TowerToHyperService}; use metrics::with_local_recorder; use metrics_util::debugging::DebuggingRecorder; use rustfs_io_metrics::internode_metrics::{ - INTERNODE_OPERATION_NS_SCANNER, INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, - INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics, + INTERNODE_OPERATION_NS_SCANNER, INTERNODE_OPERATION_PUT_FILE_CAPABILITY, INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + global_internode_metrics, }; use sha2::Digest as _; use std::collections::HashMap; + use std::convert::Infallible; use std::future::Future as _; use std::pin::Pin; use std::task::{Context, Poll}; use tokio::io; use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::{TcpListener, TcpStream}; use tokio_stream::StreamExt; use tokio_stream::iter; @@ -1643,6 +1715,117 @@ mod tests { (disk, dir) } + #[tokio::test] + #[serial_test::serial] + async fn authenticated_put_route_checks_server_epoch_before_disk_lookup() { + let _ = rustfs_credentials::set_global_rpc_secret("put-file-auth-body-test-secret".to_string()); + let listener = match TcpListener::bind("127.0.0.1:0").await { + Ok(listener) => listener, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("test listener should bind: {err}"), + }; + let addr = listener.local_addr().expect("listener address should be available"); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.expect("test server should accept a connection"); + let fallback = tower::service_fn(|_| async { Ok::<_, Infallible>(Response::new(s3s::Body::empty())) }); + server_http1::Builder::new() + .serve_connection(TokioIo::new(socket), TowerToHyperService::new(InternodeRpcService::new(fallback))) + .await + .expect("test connection should complete"); + }); + + let stream = TcpStream::connect(addr).await.expect("test client should connect"); + let (mut sender, connection) = client_http1::handshake(TokioIo::new(stream)) + .await + .expect("HTTP/1 handshake should succeed"); + let client = tokio::spawn(async move { + connection.await.expect("test client connection should complete"); + }); + + let challenge = uuid::Uuid::new_v4(); + let capability_query = format!( + "{}={}&{}={challenge}", + super::PUT_FILE_CAPABILITY_QUERY, + super::PUT_FILE_CAPABILITY_VERSION, + super::PUT_FILE_CAPABILITY_CHALLENGE_QUERY + ); + let capability_uri = format!("{PUT_FILE_CAPABILITY_PATH}?{capability_query}"); + let mut capability_request = Request::builder() + .method(Method::GET) + .uri(&capability_uri) + .header(http::header::HOST, addr.to_string()) + .body(Empty::::new()) + .expect("capability request should build"); + capability_request + .headers_mut() + .extend(gen_signature_headers(&capability_uri, &Method::GET).expect("capability signature should build")); + let response = sender + .send_request(capability_request) + .await + .expect("capability request should complete"); + assert_eq!(response.status(), StatusCode::OK); + assert!( + !response + .into_body() + .collect() + .await + .expect("capability response should drain") + .to_bytes() + .is_empty() + ); + + let legacy_probe_uri = format!("{PUT_FILE_STREAM_PATH}?{capability_query}"); + let mut legacy_probe_request = Request::builder() + .method(Method::GET) + .uri(&legacy_probe_uri) + .header(http::header::HOST, addr.to_string()) + .body(Empty::::new()) + .expect("legacy-path probe request should build"); + legacy_probe_request + .headers_mut() + .extend(gen_signature_headers(&legacy_probe_uri, &Method::GET).expect("legacy-path signature should build")); + let response = sender + .send_request(legacy_probe_request) + .await + .expect("legacy-path probe should complete"); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + response + .into_body() + .collect() + .await + .expect("legacy-path response should drain"); + + for (server_epoch, expected_status) in [ + (None, StatusCode::CONFLICT), + (Some(uuid::Uuid::new_v4()), StatusCode::CONFLICT), + (Some(*super::PUT_FILE_CAPABILITY_SERVER_EPOCH), StatusCode::BAD_REQUEST), + ] { + let nonce = uuid::Uuid::new_v4(); + let mut uri = format!( + "{PUT_FILE_AUTH_STREAM_PATH}?disk=definitely-missing&volume=bucket&path=object&append=false&size=0&put_file_auth=digest-trailer-v1&put_file_nonce={nonce}" + ); + if let Some(server_epoch) = server_epoch { + uri.push_str(&format!("&put_file_server_epoch={server_epoch}")); + } + let headers = gen_signature_headers(&uri, &Method::PUT).expect("request signature should build"); + let mut request = Request::builder() + .method(Method::PUT) + .uri(&uri) + .header(http::header::HOST, addr.to_string()) + .body(Empty::::new()) + .expect("test request should build"); + request.headers_mut().extend(headers); + + let response = sender.send_request(request).await.expect("test request should complete"); + assert_eq!(response.status(), expected_status, "unexpected status for server epoch {server_epoch:?}"); + response.into_body().collect().await.expect("response body should drain"); + } + + drop(sender); + client.await.expect("test client task should join"); + server.await.expect("test server task should join"); + } + struct DropNotifier(Option>); impl Drop for DropNotifier { @@ -1692,6 +1875,14 @@ mod tests { Some(INTERNODE_OPERATION_READ_FILE_STREAM) ); assert_eq!(internode_http_operation(PUT_FILE_STREAM_PATH), Some(INTERNODE_OPERATION_PUT_FILE_STREAM)); + assert_eq!( + internode_http_operation(PUT_FILE_AUTH_STREAM_PATH), + Some(INTERNODE_OPERATION_PUT_FILE_STREAM) + ); + assert_eq!( + internode_http_operation(PUT_FILE_CAPABILITY_PATH), + Some(INTERNODE_OPERATION_PUT_FILE_CAPABILITY) + ); assert_eq!(internode_http_operation(WALK_DIR_PATH), Some(INTERNODE_OPERATION_WALK_DIR)); assert_eq!(internode_http_operation(NS_SCANNER_PATH), Some(INTERNODE_OPERATION_NS_SCANNER)); assert_eq!(internode_http_operation("/rustfs/rpc/unknown"), None); @@ -1718,6 +1909,43 @@ mod tests { assert_eq!(response.status(), StatusCode::FORBIDDEN); } + #[test] + fn put_file_capability_get_requires_signature() { + let challenge = uuid::Uuid::new_v4(); + let uri: Uri = format!( + "{PUT_FILE_CAPABILITY_PATH}?{}={}&{}={challenge}", + super::PUT_FILE_CAPABILITY_QUERY, + super::PUT_FILE_CAPABILITY_VERSION, + super::PUT_FILE_CAPABILITY_CHALLENGE_QUERY + ) + .parse() + .expect("uri"); + let response = verify_internode_rpc_signature(&uri, &Method::GET, &HeaderMap::new()).expect_err("response"); + assert_eq!(response.status(), StatusCode::FORBIDDEN); + } + + #[tokio::test] + async fn put_file_capability_response_is_authenticated() { + let _ = rustfs_credentials::set_global_rpc_secret("put-file-capability-server-test-secret".to_string()); + let challenge = uuid::Uuid::new_v4(); + let response = put_file_capability_response(challenge); + assert_eq!(response.status(), StatusCode::OK); + let body = http_body_util::BodyExt::collect(response.into_body()) + .await + .expect("response body should collect") + .to_bytes(); + let capability: super::PutFileCapabilityResponse = rmp_serde::from_slice(&body).expect("capability should decode"); + assert_eq!(capability.version, super::PUT_FILE_CAPABILITY_VERSION); + assert!(!capability.server_epoch.is_nil()); + crate::storage::storage_api::ecstore_rpc::verify_put_file_capability( + challenge, + capability.server_epoch, + capability.version, + &capability.proof, + ) + .expect("capability proof should verify"); + } + #[test] fn namespace_scanner_rejects_requests_from_a_prior_server_epoch() { assert!(ns_scanner_server_epoch_matches(*super::NS_SCANNER_SERVER_EPOCH)); @@ -1773,6 +2001,7 @@ mod tests { size: 1024, put_file_auth: None, put_file_nonce: None, + put_file_server_epoch: None, }; let msg = put_file_stage_error_message("write_body", &query, &"connection reset"); @@ -1794,6 +2023,7 @@ mod tests { size, put_file_auth: None, put_file_nonce: None, + put_file_server_epoch: None, }; // Truncated (or over-long) body on the create path is rejected. @@ -1816,6 +2046,10 @@ mod tests { internode_rpc_subsystem(Some(INTERNODE_OPERATION_PUT_FILE_STREAM)), LOG_SUBSYSTEM_FILE_TRANSFER ); + assert_eq!( + internode_rpc_subsystem(Some(INTERNODE_OPERATION_PUT_FILE_CAPABILITY)), + LOG_SUBSYSTEM_FILE_TRANSFER + ); assert_eq!(internode_rpc_subsystem(Some(INTERNODE_OPERATION_WALK_DIR)), LOG_SUBSYSTEM_DIRECTORY_WALK); assert_eq!( internode_rpc_subsystem(Some(INTERNODE_OPERATION_NS_SCANNER)), @@ -1932,9 +2166,19 @@ mod tests { size: 11, put_file_auth: Some("digest-trailer-v1".to_string()), put_file_nonce: Some(nonce), + put_file_server_epoch: Some(*super::PUT_FILE_CAPABILITY_SERVER_EPOCH), }; assert_eq!(put_file_auth_nonce(&query).expect("v1 auth should parse"), Some(nonce)); + assert!(put_file_server_epoch_matches(&query)); + + let mut stale_epoch = query.clone(); + stale_epoch.put_file_server_epoch = Some(uuid::Uuid::new_v4()); + assert!(!put_file_server_epoch_matches(&stale_epoch)); + + let mut missing_epoch = query.clone(); + missing_epoch.put_file_server_epoch = None; + assert!(!put_file_server_epoch_matches(&missing_epoch)); let mut append = query.clone(); append.append = true; @@ -1971,6 +2215,7 @@ mod tests { size: 11, put_file_auth: Some("digest-trailer-v1".to_string()), put_file_nonce: Some(nonce), + put_file_server_epoch: Some(*super::PUT_FILE_CAPABILITY_SERVER_EPOCH), }; let mut second = b"world".to_vec(); second.extend_from_slice(&trailer[..7]); @@ -2007,6 +2252,7 @@ mod tests { size: 11, put_file_auth: Some("digest-trailer-v1".to_string()), put_file_nonce: Some(nonce), + put_file_server_epoch: Some(*super::PUT_FILE_CAPABILITY_SERVER_EPOCH), }; let mut payload = b"hello worle".to_vec(); payload.extend_from_slice(&trailer); @@ -2039,6 +2285,7 @@ mod tests { size: 0, put_file_auth: Some("digest-trailer-v1".to_string()), put_file_nonce: Some(nonce), + put_file_server_epoch: Some(*super::PUT_FILE_CAPABILITY_SERVER_EPOCH), }; let mut payload = b"append-data".to_vec(); payload.extend_from_slice(&trailer); @@ -2068,6 +2315,7 @@ mod tests { size: 0, put_file_auth: Some("digest-trailer-v1".to_string()), put_file_nonce: Some(nonce), + put_file_server_epoch: Some(*super::PUT_FILE_CAPABILITY_SERVER_EPOCH), }; let body = iter(vec![Ok::(Bytes::from_static(b"append-data"))]); let mut writer = Vec::new(); @@ -2102,6 +2350,7 @@ mod tests { size: 11, put_file_auth: Some("digest-trailer-v1".to_string()), put_file_nonce: Some(nonce), + put_file_server_epoch: None, }; let mut payload = b"hello worle".to_vec(); payload.extend_from_slice(&trailer); @@ -2140,6 +2389,7 @@ mod tests { size: 11, put_file_auth: Some("digest-trailer-v1".to_string()), put_file_nonce: Some(nonce), + put_file_server_epoch: None, }; let err = write_authenticated_put_file( @@ -2180,6 +2430,7 @@ mod tests { size: 0, put_file_auth: Some("digest-trailer-v1".to_string()), put_file_nonce: Some(nonce), + put_file_server_epoch: None, }; let err = write_authenticated_put_file( @@ -2223,6 +2474,7 @@ mod tests { size: 0, put_file_auth: Some("digest-trailer-v1".to_string()), put_file_nonce: Some(nonce), + put_file_server_epoch: None, }; let mut payload = b"append-data".to_vec(); payload.extend_from_slice(&trailer); @@ -2279,6 +2531,7 @@ mod tests { size: 0, put_file_auth: Some("digest-trailer-v1".to_string()), put_file_nonce: Some(first_nonce), + put_file_server_epoch: None, }; let (started_tx, started_rx) = tokio::sync::oneshot::channel(); let (release_tx, release_rx) = tokio::sync::oneshot::channel(); @@ -2374,6 +2627,7 @@ mod tests { size: 0, put_file_auth: Some("digest-trailer-v1".to_string()), put_file_nonce: Some(nonce), + put_file_server_epoch: None, }; let legacy_lock = put_file_target_lock(&disk, &query); diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 3b82b0966..1db42a887 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -221,15 +221,17 @@ pub(crate) mod rpc_consumer { pub(crate) use super::super::storage_contracts::{ NS_SCANNER_BODY_SHA256_QUERY, NS_SCANNER_CAPABILITY_CHALLENGE_QUERY, NS_SCANNER_CYCLE_QUERY, NS_SCANNER_LEADER_EPOCH_QUERY, NS_SCANNER_REQUEST_ID_QUERY, NS_SCANNER_SERVER_EPOCH_QUERY, - NS_SCANNER_SESSION_ID_QUERY, NS_SCANNER_SESSION_SEQUENCE_QUERY, WALK_DIR_BODY_SHA256_QUERY, + NS_SCANNER_SESSION_ID_QUERY, NS_SCANNER_SESSION_SEQUENCE_QUERY, PUT_FILE_CAPABILITY_CHALLENGE_QUERY, + PUT_FILE_CAPABILITY_QUERY, WALK_DIR_BODY_SHA256_QUERY, }; pub(crate) use super::super::storage_contracts::{ NS_SCANNER_PROTOCOL_VERSION, NsScannerCapabilityResponse, PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_V1, - WALK_DIR_STREAM_COMPLETION_V1, + PUT_FILE_CAPABILITY_VERSION, PutFileCapabilityResponse, WALK_DIR_STREAM_COMPLETION_V1, }; pub(crate) use super::super::{ DeleteOptions, DiskStore, StorageDiskRpcExt, WalkDirOptions, check_and_record_signed_rpc_nonce, - find_local_disk_by_ref, sign_ns_scanner_capability, verify_put_file_auth_trailer, verify_rpc_signature, + find_local_disk_by_ref, sign_ns_scanner_capability, sign_put_file_capability, verify_put_file_auth_trailer, + verify_rpc_signature, }; } @@ -508,7 +510,7 @@ pub(crate) mod ecstore_rpc { pub(crate) use rustfs_ecstore::api::rpc::{ KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, PeerS3Client, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, TONIC_RPC_PREFIX, - check_and_record_signed_rpc_nonce, normalize_tonic_rpc_audience, sign_ns_scanner_capability, + check_and_record_signed_rpc_nonce, normalize_tonic_rpc_audience, sign_ns_scanner_capability, sign_put_file_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason, verify_put_file_auth_trailer, verify_rpc_signature, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_rpc_signature_with_bootstrap, @@ -516,7 +518,7 @@ pub(crate) mod ecstore_rpc { #[cfg(test)] pub(crate) use rustfs_ecstore::api::rpc::{ build_put_file_auth_trailer, gen_signature_headers, gen_tonic_signature_headers, set_tonic_canonical_body_digest, - verify_tonic_rpc_response_proof, + verify_put_file_capability, verify_tonic_rpc_response_proof, }; } @@ -1696,6 +1698,14 @@ pub(crate) fn sign_ns_scanner_capability(challenge: uuid::Uuid, server_epoch: uu ecstore_rpc::sign_ns_scanner_capability(challenge, server_epoch) } +pub(crate) fn sign_put_file_capability( + challenge: uuid::Uuid, + server_epoch: uuid::Uuid, + version: u16, +) -> std::io::Result> { + ecstore_rpc::sign_put_file_capability(challenge, server_epoch, version) +} + pub(crate) fn verify_tonic_rpc_signature_with_bootstrap( audience: &str, path: &str,