diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index f4938ea95..cc31820ec 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -436,13 +436,14 @@ pub mod rpc { pub use crate::cluster::rpc::{ AuthenticatedChannel, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, PeerS3Client, S3PeerSys, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, - ScannerBucketListing, ScannerPeerActivity, TONIC_RPC_PREFIX, TonicInterceptor, 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_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, + 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, }; } diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index 881cee734..bcbfb5c5b 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -27,7 +27,10 @@ //! Advisory: 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; +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, +}; use base64::Engine as _; use base64::engine::general_purpose; use hmac::{Hmac, KeyInit, Mac}; @@ -71,6 +74,7 @@ const RPC_REPLAY_SCOPE_VERSION_V3: &str = "3"; 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 UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD"; const UNSIGNED_PAYLOAD_NONCE: &str = "unsigned"; const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes @@ -475,6 +479,13 @@ fn signature_payload(url: &str, method: &Method, timestamp: i64) -> String { format!("{url}|{method}|{timestamp}") } +fn canonical_path_and_query(url: &str) -> std::io::Result { + let uri: Uri = url.parse().map_err(|_| std::io::Error::other("Invalid RPC URL"))?; + uri.path_and_query() + .map(ToString::to_string) + .ok_or_else(|| std::io::Error::other("Invalid RPC URL")) +} + fn redacted_rpc_path(url: &str) -> String { url.parse::() .ok() @@ -502,6 +513,76 @@ fn verify_signature(secret: &str, url: &str, method: &Method, timestamp: i64, si mac.verify_slice(&signature).is_ok() } +fn update_put_file_auth_mac( + mac: &mut HmacSha256, + url: &str, + method: &Method, + nonce: Uuid, + body_sha256: &str, +) -> std::io::Result<()> { + if !valid_content_sha256(body_sha256) || body_sha256 == UNSIGNED_PAYLOAD { + return Err(std::io::Error::other("Invalid RPC content SHA-256")); + } + let path_and_query = canonical_path_and_query(url)?; + mac.update(HTTP_PUT_FILE_AUTH_DOMAIN); + for part in [ + path_and_query.as_bytes(), + b"|", + method.as_str().as_bytes(), + b"|", + nonce.as_bytes(), + b"|", + body_sha256.as_bytes(), + ] { + mac.update(part); + } + Ok(()) +} + +fn put_file_auth_mac(url: &str, method: &Method, nonce: Uuid, body_sha256: &str) -> std::io::Result<[u8; 32]> { + let mut mac = ::new_from_slice(get_shared_secret()?.as_bytes()) + .map_err(|_| std::io::Error::other("Invalid RPC HMAC key"))?; + update_put_file_auth_mac(&mut mac, url, method, nonce, body_sha256)?; + Ok(mac.finalize().into_bytes().into()) +} + +fn verify_put_file_auth_mac(url: &str, method: &Method, nonce: Uuid, body_sha256: &str, signature: &[u8]) -> std::io::Result<()> { + let mut mac = ::new_from_slice(get_shared_secret()?.as_bytes()) + .map_err(|_| std::io::Error::other("Invalid RPC HMAC key"))?; + update_put_file_auth_mac(&mut mac, url, method, nonce, body_sha256)?; + mac.verify_slice(signature) + .map_err(|_| std::io::Error::other("Invalid put_file auth trailer")) +} + +pub fn build_put_file_auth_trailer(url: &str, method: &Method, nonce: Uuid, body_sha256: &str) -> std::io::Result> { + let mac = put_file_auth_mac(url, method, nonce, body_sha256)?; + let mut trailer = Vec::with_capacity(PUT_FILE_AUTH_TRAILER_LEN); + trailer.extend_from_slice(PUT_FILE_AUTH_TRAILER_MAGIC); + trailer.extend_from_slice(body_sha256.as_bytes()); + trailer.extend_from_slice(&mac); + Ok(trailer) +} + +pub fn verify_put_file_auth_trailer(url: &str, method: &Method, nonce: Uuid, trailer: &[u8]) -> std::io::Result { + if trailer.len() != PUT_FILE_AUTH_TRAILER_LEN { + return Err(std::io::Error::other("Invalid put_file auth trailer length")); + } + if &trailer[..PUT_FILE_AUTH_TRAILER_MAGIC.len()] != PUT_FILE_AUTH_TRAILER_MAGIC { + return Err(std::io::Error::other("Invalid put_file auth trailer")); + } + let digest_start = PUT_FILE_AUTH_TRAILER_MAGIC.len(); + let digest_end = digest_start + PUT_FILE_AUTH_TRAILER_DIGEST_LEN; + let body_sha256 = std::str::from_utf8(&trailer[digest_start..digest_end]) + .map_err(|_| std::io::Error::other("Invalid RPC content SHA-256"))?; + if !valid_content_sha256(body_sha256) || body_sha256 == UNSIGNED_PAYLOAD { + return Err(std::io::Error::other("Invalid RPC content SHA-256")); + } + let mac_start = digest_end; + let mac_end = mac_start + PUT_FILE_AUTH_TRAILER_MAC_LEN; + verify_put_file_auth_mac(url, method, nonce, body_sha256, &trailer[mac_start..mac_end])?; + Ok(body_sha256.to_string()) +} + 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()); @@ -806,7 +887,13 @@ fn tonic_rpc_metric_operation(path: &str) -> &'static str { } } -fn check_and_record_nonce(nonce: Uuid, signed_at: i64, rpc_path: &str) -> std::io::Result<()> { +fn check_and_record_nonce_with_scope( + nonce: Uuid, + signed_at: i64, + rpc_path: &str, + operation: &'static str, + backend: &'static str, +) -> std::io::Result<()> { let wall_time = OffsetDateTime::now_utc().unix_timestamp(); let (result, metrics) = { let mut cache = LOCAL_RPC_NONCE_CACHE @@ -826,8 +913,8 @@ fn check_and_record_nonce(nonce: Uuid, signed_at: i64, rpc_path: &str) -> std::i expires_at, capacity: *REPLAY_CACHE_CAPACITY, metric_scope: RpcReplayCacheMetricScope { - operation: tonic_rpc_metric_operation(rpc_path), - backend: INTERNODE_TRANSPORT_BACKEND_GRPC, + operation, + backend, rpc_path, }, }) @@ -836,6 +923,37 @@ fn check_and_record_nonce(nonce: Uuid, signed_at: i64, rpc_path: &str) -> std::i result } +fn check_and_record_tonic_nonce(nonce: Uuid, signed_at: i64, rpc_path: &str) -> std::io::Result<()> { + check_and_record_nonce_with_scope( + nonce, + signed_at, + rpc_path, + tonic_rpc_metric_operation(rpc_path), + INTERNODE_TRANSPORT_BACKEND_GRPC, + ) +} + +pub fn check_and_record_signed_rpc_nonce( + headers: &HeaderMap, + nonce: Uuid, + rpc_path: &str, + operation: &'static str, + backend: &'static str, +) -> std::io::Result<()> { + if nonce.is_nil() { + return Err(std::io::Error::other("Invalid RPC nonce")); + } + let timestamp_header = headers + .get(TIMESTAMP_HEADER) + .and_then(|value| value.to_str().ok()) + .ok_or_else(|| std::io::Error::other("Missing timestamp header"))?; + let timestamp = timestamp_header + .parse::() + .map_err(|_| std::io::Error::other("Invalid timestamp format"))?; + check_timestamp(timestamp)?; + check_and_record_nonce_with_scope(nonce, timestamp, rpc_path, operation, backend) +} + /// Build headers with authentication signature pub fn build_auth_headers(url: &str, method: &Method, headers: &mut HeaderMap) -> std::io::Result<()> { let auth_headers = gen_signature_headers(url, method)?; @@ -1095,7 +1213,7 @@ fn verify_tonic_replay_scope_signature(audience: &str, path: &str, headers: &Hea if boot_epoch != tonic_rpc_boot_epoch() { return Err(std::io::Error::other("RPC boot epoch is stale")); } - check_and_record_nonce(nonce, signed_at, path) + check_and_record_tonic_nonce(nonce, signed_at, path) } /// Verify gRPC authentication, preferring v2 without downgrade on malformed v2 metadata. @@ -1160,6 +1278,8 @@ pub fn tonic_rpc_auth_failure_reason(error: &std::io::Error) -> &'static str { "Invalid unsigned RPC nonce" => "invalid_unsigned_v2_nonce", "Missing RPC content SHA-256" => "missing_content_sha256", "Invalid RPC content SHA-256" => "invalid_content_sha256", + "Invalid put_file auth trailer length" => "invalid_put_file_auth_trailer_length", + "Invalid put_file auth trailer" => "invalid_put_file_auth_trailer", "Missing signature header" => "missing_v1_signature", "Invalid signature" => "invalid_v1_signature", "Invalid RPC HMAC key" => "invalid_hmac_key", @@ -1286,7 +1406,7 @@ fn verify_tonic_rpc_signature_with_strictness( return Err(std::io::Error::other("Invalid RPC v2 signature")); } if let Some(nonce) = parsed_nonce { - check_and_record_nonce(nonce, timestamp, path)?; + check_and_record_tonic_nonce(nonce, timestamp, path)?; } Ok(()) } @@ -2031,6 +2151,8 @@ mod tests { ("Request timestamp expired", "timestamp_expired"), ("Missing RPC content SHA-256", "missing_content_sha256"), ("Invalid RPC content SHA-256", "invalid_content_sha256"), + ("Invalid put_file auth trailer length", "invalid_put_file_auth_trailer_length"), + ("Invalid put_file auth trailer", "invalid_put_file_auth_trailer"), ] { assert_eq!( tonic_rpc_auth_failure_reason(&std::io::Error::other(message)), @@ -2178,6 +2300,37 @@ mod tests { assert_eq!(tampered.to_string(), "RPC content SHA-256 mismatch"); } + #[test] + fn put_file_auth_trailer_binds_url_nonce_and_body_digest() { + ensure_test_rpc_secret(); + let url = concat!( + "/rustfs/rpc/put_file_stream?disk=disk-a&volume=bucket&path=object%2Fpart.1", + "&append=false&size=11&put_file_auth=digest-trailer-v1&put_file_nonce=11111111-2222-4333-8444-555555555555" + ); + let nonce = Uuid::parse_str("11111111-2222-4333-8444-555555555555").expect("nonce"); + let body_sha256 = hex_simd::encode_to_string(Sha256::digest(b"hello world"), hex_simd::AsciiCase::Lower); + let trailer = build_put_file_auth_trailer(url, &Method::PUT, nonce, &body_sha256).expect("trailer should build"); + + assert_eq!(trailer.len(), PUT_FILE_AUTH_TRAILER_LEN); + let verified = verify_put_file_auth_trailer(url, &Method::PUT, nonce, &trailer).expect("trailer should verify"); + assert_eq!(verified, body_sha256); + + let different_url = url.replace("size=11", "size=12"); + let err = verify_put_file_auth_trailer(&different_url, &Method::PUT, nonce, &trailer) + .expect_err("trailer must bind the signed URL"); + assert_eq!(err.to_string(), "Invalid put_file auth trailer"); + + let err = + verify_put_file_auth_trailer(url, &Method::PUT, Uuid::new_v4(), &trailer).expect_err("trailer must bind the nonce"); + assert_eq!(err.to_string(), "Invalid put_file auth trailer"); + + let mut tampered = trailer; + tampered[PUT_FILE_AUTH_TRAILER_MAGIC.len()] = b'0'; + let err = + verify_put_file_auth_trailer(url, &Method::PUT, nonce, &tampered).expect_err("trailer must bind the digest bytes"); + assert_eq!(err.to_string(), "Invalid put_file auth trailer"); + } + #[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 2d228f43e..4192a9792 100644 --- a/crates/ecstore/src/cluster/rpc/internode_data_transport.rs +++ b/crates/ecstore/src/cluster/rpc/internode_data_transport.rs @@ -12,14 +12,15 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::cluster::rpc::{build_auth_headers, verify_ns_scanner_capability}; +use crate::cluster::rpc::{build_auth_headers, build_put_file_auth_trailer, verify_ns_scanner_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, WALK_DIR_BODY_SHA256_QUERY, - WALK_DIR_STREAM_COMPLETION_QUERY, WALK_DIR_STREAM_COMPLETION_V1, + 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, }; use async_trait::async_trait; use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE}; @@ -29,9 +30,11 @@ use rustfs_config::{ }; use rustfs_rio::{HttpReader, HttpWriter}; use sha2::{Digest, Sha256}; +use std::pin::Pin; use std::sync::{Arc, OnceLock}; +use std::task::{Context, Poll}; use std::time::Duration; -use tokio::io::AsyncReadExt; +use tokio::io::{AsyncReadExt, AsyncWrite}; use uuid::Uuid; static INTERNODE_DATA_TRANSPORT: OnceLock, String>> = OnceLock::new(); @@ -166,10 +169,12 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport { } async fn open_write(&self, request: WriteStreamRequest) -> Result { - let url = build_put_file_stream_url(&request); + let nonce = Uuid::new_v4(); + let url = build_put_file_stream_url(&request, Some(nonce)); let mut headers = json_headers(); build_auth_headers(&url, &Method::PUT, &mut headers)?; - Ok(Box::new(HttpWriter::new(url, Method::PUT, headers).await?)) + let writer = HttpWriter::new(url.clone(), Method::PUT, headers).await?; + Ok(Box::new(PutFileAuthWriter::new(writer, url, nonce))) } async fn open_walk_dir(&self, request: WalkDirStreamRequest) -> Result { @@ -236,8 +241,8 @@ fn build_read_file_stream_url(request: &ReadStreamRequest) -> String { ) } -fn build_put_file_stream_url(request: &WriteStreamRequest) -> String { - format!( +fn build_put_file_stream_url(request: &WriteStreamRequest, auth_nonce: Option) -> String { + let mut url = format!( "{}{}?disk={}&volume={}&path={}&append={}&size={}", request.endpoint, PUT_FILE_STREAM_PATH, @@ -246,7 +251,104 @@ fn build_put_file_stream_url(request: &WriteStreamRequest) -> String { urlencoding::encode(&request.path), request.append, request.size - ) + ); + if let Some(nonce) = auth_nonce { + url.push_str(&format!( + "&{}={}&{}={}", + PUT_FILE_AUTH_QUERY, PUT_FILE_AUTH_V1, PUT_FILE_NONCE_QUERY, nonce + )); + } + url +} + +struct PutFileAuthWriter { + inner: W, + url: String, + nonce: Uuid, + hasher: Sha256, + trailer: Option>, + trailer_offset: usize, +} + +impl PutFileAuthWriter { + fn new(inner: W, url: String, nonce: Uuid) -> Self { + Self { + inner, + url, + nonce, + hasher: Sha256::new(), + trailer: None, + trailer_offset: 0, + } + } + + fn ensure_trailer(&mut self) -> std::io::Result<()> { + if self.trailer.is_some() { + return Ok(()); + } + let digest = hex_simd::encode_to_string(self.hasher.clone().finalize(), hex_simd::AsciiCase::Lower); + self.trailer = Some(build_put_file_auth_trailer(&self.url, &Method::PUT, self.nonce, &digest)?); + Ok(()) + } + + fn poll_write_trailer(&mut self, cx: &mut Context<'_>) -> Poll> + where + W: AsyncWrite + Unpin, + { + self.ensure_trailer()?; + let Some(trailer) = self.trailer.as_ref() else { + return Poll::Ready(Err(std::io::Error::other("put_file auth trailer missing"))); + }; + while self.trailer_offset < trailer.len() { + let written = match Pin::new(&mut self.inner).poll_write(cx, &trailer[self.trailer_offset..]) { + Poll::Ready(Ok(0)) => { + return Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::WriteZero, + "failed to write put_file auth trailer", + ))); + } + Poll::Ready(Ok(written)) => written, + Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), + Poll::Pending => return Poll::Pending, + }; + self.trailer_offset += written; + } + Poll::Ready(Ok(())) + } +} + +impl AsyncWrite for PutFileAuthWriter +where + W: AsyncWrite + Unpin, +{ + fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + if self.trailer.is_some() { + return Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "cannot write after put_file auth trailer", + ))); + } + match Pin::new(&mut self.inner).poll_write(cx, buf) { + Poll::Ready(Ok(written)) => { + self.hasher.update(&buf[..written]); + Poll::Ready(Ok(written)) + } + other => other, + } + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match self.poll_write_trailer(cx) { + Poll::Ready(Ok(())) => {} + Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), + Poll::Pending => return Poll::Pending, + } + Pin::new(&mut self.inner).poll_shutdown(cx) + } } fn build_walk_dir_url(request: &WalkDirStreamRequest) -> String { @@ -455,14 +557,17 @@ mod tests { #[test] fn put_file_stream_url_encodes_query_values() { - let url = build_put_file_stream_url(&WriteStreamRequest { - endpoint: "http://node1:9000".to_string(), - disk: "http://node1:9000/data/rustfs0".to_string(), - volume: "bucket".to_string(), - path: "object/part.1".to_string(), - append: false, - size: 4096, - }); + let url = build_put_file_stream_url( + &WriteStreamRequest { + endpoint: "http://node1:9000".to_string(), + disk: "http://node1:9000/data/rustfs0".to_string(), + volume: "bucket".to_string(), + path: "object/part.1".to_string(), + append: false, + size: 4096, + }, + None, + ); assert_eq!( url, @@ -470,6 +575,63 @@ 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 url = build_put_file_stream_url( + &WriteStreamRequest { + endpoint: "http://node1:9000".to_string(), + disk: "http://node1:9000/data/rustfs0".to_string(), + volume: "bucket".to_string(), + path: "object/part.1".to_string(), + append: false, + size: 4096, + }, + Some(nonce), + ); + + assert_eq!( + url, + concat!( + "http://node1:9000/rustfs/rpc/put_file_stream?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" + ) + ); + } + + #[tokio::test] + async fn put_file_auth_writer_appends_trailer_on_shutdown() { + use tokio::io::AsyncWriteExt; + + let _ = rustfs_credentials::set_global_rpc_secret("put-file-auth-writer-test-secret".to_string()); + let nonce = Uuid::parse_str("11111111-2222-4333-8444-555555555555").expect("nonce"); + let url = concat!( + "http://node1:9000/rustfs/rpc/put_file_stream?disk=disk-a&volume=bucket&path=object%2Fpart.1", + "&append=false&size=11&put_file_auth=digest-trailer-v1&put_file_nonce=11111111-2222-4333-8444-555555555555" + ) + .to_string(); + let mut sink = Vec::new(); + + { + let mut writer = PutFileAuthWriter::new(&mut sink, url.clone(), nonce); + writer.write_all(b"hello world").await.expect("body write should succeed"); + writer.shutdown().await.expect("shutdown should append auth trailer"); + let err = writer + .write_all(b"!") + .await + .expect_err("post-trailer writes must be rejected"); + assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe); + } + + assert_eq!(&sink[..11], b"hello world"); + let trailer = &sink[11..]; + let expected_digest = hex_simd::encode_to_string(Sha256::digest(b"hello world"), hex_simd::AsciiCase::Lower); + let verified = crate::cluster::rpc::verify_put_file_auth_trailer(&url, &Method::PUT, nonce, trailer) + .expect("emitted trailer should verify"); + assert_eq!(verified, expected_digest); + } + #[test] fn walk_dir_url_encodes_disk_ref() { let url = build_walk_dir_url(&WalkDirStreamRequest { diff --git a/crates/ecstore/src/cluster/rpc/mod.rs b/crates/ecstore/src/cluster/rpc/mod.rs index f13af6f09..118497c99 100644 --- a/crates/ecstore/src/cluster/rpc/mod.rs +++ b/crates/ecstore/src/cluster/rpc/mod.rs @@ -32,10 +32,11 @@ pub use client::{ // Re-exported through `api::rpc`; not every item is consumed inside this crate. #[allow(unused_imports)] pub use http_auth::{ - TONIC_RPC_PREFIX, build_auth_headers, 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_rpc_signature, verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest, + 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, }; diff --git a/crates/ecstore/src/storage_api_contracts/mod.rs b/crates/ecstore/src/storage_api_contracts/mod.rs index 8a288c6b1..f0d33c363 100644 --- a/crates/ecstore/src/storage_api_contracts/mod.rs +++ b/crates/ecstore/src/storage_api_contracts/mod.rs @@ -27,9 +27,11 @@ pub(crate) mod 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, 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, + 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, }; } diff --git a/crates/storage-api/src/lib.rs b/crates/storage-api/src/lib.rs index 15002fa3b..7c61251dc 100644 --- a/crates/storage-api/src/lib.rs +++ b/crates/storage-api/src/lib.rs @@ -17,6 +17,14 @@ pub const WALK_DIR_STREAM_COMPLETION_QUERY: &str = "walk_dir_stream_completion"; pub const WALK_DIR_STREAM_COMPLETION_V1: &str = "error-v1"; 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_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; 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 dad31bc4c..e998ee8a6 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -16,8 +16,9 @@ use crate::server::RPC_PREFIX; 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, NS_SCANNER_PROTOCOL_VERSION, NsScannerCapabilityResponse, StorageDiskRpcExt as _, - WALK_DIR_STREAM_COMPLETION_V1, WalkDirOptions, find_local_disk_by_ref, sign_ns_scanner_capability, verify_rpc_signature, + DEFAULT_READ_BUFFER_SIZE, 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, }; #[cfg(test)] use crate::storage::storage_api::rpc_consumer::http_service::{ @@ -72,6 +73,12 @@ 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_AUTH_STRICT: LazyLock = LazyLock::new(|| { + rustfs_utils::get_env_bool( + rustfs_config::ENV_INTERNODE_RPC_BODY_DIGEST_STRICT, + rustfs_config::DEFAULT_INTERNODE_RPC_BODY_DIGEST_STRICT, + ) +}); macro_rules! log_internode_rpc_response_failure { ($status:expr, $rpc_path:expr, $method:expr, $operation:expr, $reason:expr, $result:expr, Some(($context_key:expr, $context_value:expr)), Some($error_text:expr)) => {{ @@ -311,13 +318,34 @@ fn validate_walk_dir_completion_request(query: &WalkDirQuery, body: &[u8]) -> Op Some(propagate_completion_errors) } -#[derive(Debug, Default, serde::Deserialize)] +#[derive(Clone, Debug, Default, serde::Deserialize)] struct PutFileQuery { disk: String, volume: String, path: String, append: bool, size: i64, + put_file_auth: Option, + put_file_nonce: Option, +} + +fn put_file_auth_nonce(query: &PutFileQuery) -> io::Result> { + match query.put_file_auth.as_deref() { + None => { + if *PUT_FILE_AUTH_STRICT { + return Err(io::Error::other("put_file auth required")); + } + Ok(None) + } + Some(PUT_FILE_AUTH_V1) => { + let nonce = query + .put_file_nonce + .filter(|nonce| !nonce.is_nil()) + .ok_or_else(|| io::Error::other("Invalid RPC nonce"))?; + Ok(Some(nonce)) + } + Some(_) => Err(io::Error::other("Unsupported put_file auth version")), + } } impl Service> for InternodeRpcService @@ -1117,10 +1145,48 @@ where async fn handle_put_file(req: Request) -> Response { let method = req.method().clone(); let path = req.uri().path().to_string(); + let url = req.uri().to_string(); let query = match parse_query::(&req) { Ok(query) => query, Err(response) => return *response, }; + let auth_nonce = match put_file_auth_nonce(&query) { + Ok(nonce) => nonce, + Err(e) => { + log_internode_rpc_response_failure!( + StatusCode::FORBIDDEN, + &path, + &method, + Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + "put_file_auth_invalid", + "rejected", + Some(("disk", query.disk.as_str())), + Some(&e) + ); + return response_with_status(StatusCode::FORBIDDEN, format!("invalid put_file auth: {e}")); + } + }; + if let Some(nonce) = auth_nonce + && let Err(e) = check_and_record_signed_rpc_nonce( + req.headers(), + nonce, + PUT_FILE_STREAM_PATH, + INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + ) + { + log_internode_rpc_response_failure!( + StatusCode::FORBIDDEN, + &path, + &method, + Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + "put_file_replay_rejected", + "rejected", + Some(("disk", query.disk.as_str())), + Some(&e) + ); + return response_with_status(StatusCode::FORBIDDEN, format!("invalid put_file auth: {e}")); + } let Some(disk) = find_local_disk_by_ref(&query.disk).await else { log_internode_rpc_response_failure!( @@ -1156,14 +1222,16 @@ async fn handle_put_file(req: Request) -> Response { } }; - let copied = match write_body_chunks_to_writer(req.into_body().into_data_stream(), &mut file).await { - Ok(copied) => copied, - Err(e) => { - let message = put_file_stage_error_message("write_body", &query, &e); - log_internode_put_file_stage_failure!("write_body", query, e); - return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message); - } - }; + let copied = + match write_put_file_body_chunks_to_writer(req.into_body().into_data_stream(), &mut file, &query, auth_nonce, &url).await + { + Ok(copied) => copied, + Err(e) => { + let message = put_file_stage_error_message("write_body", &query, &e); + log_internode_put_file_stage_failure!("write_body", query, e); + return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message); + } + }; let metrics = runtime_sources::current_internode_metrics(); metrics.record_incoming_request_for_operation_and_backend( @@ -1222,6 +1290,112 @@ where Ok(copied) } +async fn write_put_file_body_chunks_to_writer( + body: S, + writer: &mut W, + query: &PutFileQuery, + auth_nonce: Option, + url: &str, +) -> io::Result +where + S: futures::TryStream + Unpin, + E: Into, + W: tokio::io::AsyncWrite + Unpin, +{ + let Some(nonce) = auth_nonce else { + return write_body_chunks_to_writer(body, writer).await; + }; + + let expected_size = (!query.append && query.size >= 0) + .then(|| { + u64::try_from(query.size) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "put_file auth size cannot be represented")) + }) + .transpose()?; + let mut body = body; + let mut remaining = expected_size; + let mut copied = 0_u64; + let mut trailer = Vec::with_capacity(PUT_FILE_AUTH_TRAILER_LEN); + let mut hasher = Sha256::new(); + + while let Some(bytes) = body.try_next().await.map_err(io::Error::other)? { + if let Some(remaining) = remaining.as_mut() { + let chunk_len = u64::try_from(bytes.len()).unwrap_or(u64::MAX); + let data_len = usize::try_from((*remaining).min(chunk_len)) + .map_err(|_| io::Error::other("put_file body length cannot be represented"))?; + if data_len > 0 { + hasher.update(&bytes[..data_len]); + copied = copied + .checked_add( + u64::try_from(data_len).map_err(|_| io::Error::other("put_file body length cannot be represented"))?, + ) + .ok_or_else(|| io::Error::other("put_file body length overflow"))?; + *remaining -= + u64::try_from(data_len).map_err(|_| io::Error::other("put_file body length cannot be represented"))?; + writer.write_all(&bytes[..data_len]).await?; + } + + if data_len < bytes.len() { + trailer.extend_from_slice(&bytes[data_len..]); + if trailer.len() > PUT_FILE_AUTH_TRAILER_LEN { + return Err(io::Error::new(io::ErrorKind::InvalidData, "put_file auth trailer has trailing data")); + } + } + } else { + let write_len = trailer + .len() + .saturating_add(bytes.len()) + .saturating_sub(PUT_FILE_AUTH_TRAILER_LEN); + if write_len > 0 { + let buffered_write_len = write_len.min(trailer.len()); + if buffered_write_len > 0 { + writer.write_all(&trailer[..buffered_write_len]).await?; + hasher.update(&trailer[..buffered_write_len]); + copied = copied + .checked_add( + u64::try_from(buffered_write_len) + .map_err(|_| io::Error::other("put_file body length cannot be represented"))?, + ) + .ok_or_else(|| io::Error::other("put_file body length overflow"))?; + trailer = trailer.split_off(buffered_write_len); + } + + let chunk_write_len = write_len - buffered_write_len; + if chunk_write_len > 0 { + writer.write_all(&bytes[..chunk_write_len]).await?; + hasher.update(&bytes[..chunk_write_len]); + } + copied = copied + .checked_add( + u64::try_from(chunk_write_len) + .map_err(|_| io::Error::other("put_file body length cannot be represented"))?, + ) + .ok_or_else(|| io::Error::other("put_file body length overflow"))?; + trailer.extend_from_slice(&bytes[chunk_write_len..]); + } else { + trailer.extend_from_slice(&bytes); + } + } + } + + if remaining.is_some_and(|remaining| remaining != 0) { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!("body size mismatch: expected {} bytes, received {copied}", query.size), + )); + } + if trailer.len() != PUT_FILE_AUTH_TRAILER_LEN { + return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "put_file auth trailer is incomplete")); + } + + let expected = verify_put_file_auth_trailer(url, &Method::PUT, nonce, &trailer)?; + let actual = hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower); + if actual != expected { + return Err(io::Error::new(io::ErrorKind::InvalidData, "put_file body digest mismatch")); + } + Ok(copied) +} + fn parse_query(req: &Request) -> Result where T: DeserializeOwned + Default, @@ -1308,11 +1482,12 @@ mod tests { 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_stage_error_message, read_file_body_stream, - remote_scanner_claim_rejection, response_with_disk_error, supports_walk_dir_stream_completion, + ns_scanner_server_epoch_matches, put_body_size_mismatch, put_file_auth_nonce, put_file_stage_error_message, + 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_body_chunks_to_writer, + verify_walk_dir_body_digest, walk_dir_response_body, 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 bytes::Bytes; use http::{HeaderMap, HeaderValue, Method, StatusCode, Uri}; use http_body_util::BodyExt; @@ -1433,6 +1608,8 @@ mod tests { path: "tmp/object/part.1".to_string(), append: false, size: 1024, + put_file_auth: None, + put_file_nonce: None, }; let msg = put_file_stage_error_message("write_body", &query, &"connection reset"); @@ -1452,6 +1629,8 @@ mod tests { path: "object/part.1".to_string(), append, size, + put_file_auth: None, + put_file_nonce: None, }; // Truncated (or over-long) body on the create path is rejected. @@ -1579,6 +1758,165 @@ mod tests { assert_eq!(out, b"hello world"); } + #[test] + fn put_file_auth_nonce_accepts_v1_requests_with_non_nil_nonce() { + let nonce = uuid::Uuid::new_v4(); + let query = PutFileQuery { + disk: "disk-a".to_string(), + volume: "bucket".to_string(), + path: "object/part.1".to_string(), + append: false, + size: 11, + put_file_auth: Some("digest-trailer-v1".to_string()), + put_file_nonce: Some(nonce), + }; + + assert_eq!(put_file_auth_nonce(&query).expect("v1 auth should parse"), Some(nonce)); + + let mut append = query.clone(); + append.append = true; + assert_eq!(put_file_auth_nonce(&append).expect("append auth should parse"), Some(nonce)); + + let mut unknown_size = query.clone(); + unknown_size.size = -1; + assert_eq!(put_file_auth_nonce(&unknown_size).expect("unknown-size auth should parse"), Some(nonce)); + + let mut nil = query.clone(); + nil.put_file_nonce = Some(uuid::Uuid::nil()); + assert!(put_file_auth_nonce(&nil).is_err()); + + let mut unknown = query; + unknown.put_file_auth = Some("digest-trailer-v2".to_string()); + assert!(put_file_auth_nonce(&unknown).is_err()); + } + + #[tokio::test] + async fn put_file_auth_body_writes_only_data_and_verifies_trailer() { + let _ = rustfs_credentials::set_global_rpc_secret("put-file-auth-body-test-secret".to_string()); + let nonce = uuid::Uuid::parse_str("11111111-2222-4333-8444-555555555555").expect("nonce"); + let url = concat!( + "/rustfs/rpc/put_file_stream?disk=disk-a&volume=bucket&path=object%2Fpart.1", + "&append=false&size=11&put_file_auth=digest-trailer-v1&put_file_nonce=11111111-2222-4333-8444-555555555555" + ); + let digest = hex_simd::encode_to_string(sha2::Sha256::digest(b"hello world"), hex_simd::AsciiCase::Lower); + let trailer = build_put_file_auth_trailer(url, &Method::PUT, nonce, &digest).expect("trailer should build"); + let query = PutFileQuery { + disk: "disk-a".to_string(), + volume: "bucket".to_string(), + path: "object/part.1".to_string(), + append: false, + size: 11, + put_file_auth: Some("digest-trailer-v1".to_string()), + put_file_nonce: Some(nonce), + }; + let mut second = b"world".to_vec(); + second.extend_from_slice(&trailer[..7]); + let body = iter(vec![ + Ok::(Bytes::from_static(b"hello ")), + Ok(Bytes::from(second)), + Ok(Bytes::copy_from_slice(&trailer[7..])), + ]); + let mut writer = Vec::new(); + + let copied = write_put_file_body_chunks_to_writer(body, &mut writer, &query, Some(nonce), url) + .await + .expect("authenticated body should verify"); + + assert_eq!(copied, 11); + assert_eq!(writer, b"hello world"); + } + + #[tokio::test] + async fn put_file_auth_body_rejects_tampered_data() { + let _ = rustfs_credentials::set_global_rpc_secret("put-file-auth-body-test-secret".to_string()); + let nonce = uuid::Uuid::parse_str("22222222-3333-4444-8555-666666666666").expect("nonce"); + let url = concat!( + "/rustfs/rpc/put_file_stream?disk=disk-a&volume=bucket&path=object%2Fpart.1", + "&append=false&size=11&put_file_auth=digest-trailer-v1&put_file_nonce=22222222-3333-4444-8555-666666666666" + ); + let signed_digest = hex_simd::encode_to_string(sha2::Sha256::digest(b"hello world"), hex_simd::AsciiCase::Lower); + let trailer = build_put_file_auth_trailer(url, &Method::PUT, nonce, &signed_digest).expect("trailer should build"); + let query = PutFileQuery { + disk: "disk-a".to_string(), + volume: "bucket".to_string(), + path: "object/part.1".to_string(), + append: false, + size: 11, + put_file_auth: Some("digest-trailer-v1".to_string()), + put_file_nonce: Some(nonce), + }; + let mut payload = b"hello worle".to_vec(); + payload.extend_from_slice(&trailer); + let body = iter(vec![Ok::(Bytes::from(payload))]); + let mut writer = Vec::new(); + + let err = write_put_file_body_chunks_to_writer(body, &mut writer, &query, Some(nonce), url) + .await + .expect_err("tampered body must fail digest verification"); + + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + assert_eq!(err.to_string(), "put_file body digest mismatch"); + } + + #[tokio::test] + async fn put_file_auth_append_body_uses_trailing_auth_record() { + let _ = rustfs_credentials::set_global_rpc_secret("put-file-auth-body-test-secret".to_string()); + let nonce = uuid::Uuid::parse_str("33333333-4444-4555-8666-777777777777").expect("nonce"); + let url = concat!( + "/rustfs/rpc/put_file_stream?disk=disk-a&volume=bucket&path=object%2Fpart.1", + "&append=true&size=0&put_file_auth=digest-trailer-v1&put_file_nonce=33333333-4444-4555-8666-777777777777" + ); + let digest = hex_simd::encode_to_string(sha2::Sha256::digest(b"append-data"), hex_simd::AsciiCase::Lower); + let trailer = build_put_file_auth_trailer(url, &Method::PUT, nonce, &digest).expect("trailer should build"); + let query = PutFileQuery { + disk: "disk-a".to_string(), + volume: "bucket".to_string(), + path: "object/part.1".to_string(), + append: true, + size: 0, + put_file_auth: Some("digest-trailer-v1".to_string()), + put_file_nonce: Some(nonce), + }; + let mut payload = b"append-data".to_vec(); + payload.extend_from_slice(&trailer); + let body = iter(vec![Ok::(Bytes::from(payload))]); + let mut writer = Vec::new(); + + let copied = write_put_file_body_chunks_to_writer(body, &mut writer, &query, Some(nonce), url) + .await + .expect("append body should verify"); + + assert_eq!(copied, 11); + assert_eq!(writer, b"append-data"); + } + + #[tokio::test] + async fn put_file_auth_append_body_rejects_missing_trailer() { + let nonce = uuid::Uuid::parse_str("44444444-5555-4666-8777-888888888888").expect("nonce"); + let url = concat!( + "/rustfs/rpc/put_file_stream?disk=disk-a&volume=bucket&path=object%2Fpart.1", + "&append=true&size=0&put_file_auth=digest-trailer-v1&put_file_nonce=44444444-5555-4666-8777-888888888888" + ); + let query = PutFileQuery { + disk: "disk-a".to_string(), + volume: "bucket".to_string(), + path: "object/part.1".to_string(), + append: true, + size: 0, + put_file_auth: Some("digest-trailer-v1".to_string()), + put_file_nonce: Some(nonce), + }; + let body = iter(vec![Ok::(Bytes::from_static(b"append-data"))]); + let mut writer = Vec::new(); + + let err = write_put_file_body_chunks_to_writer(body, &mut writer, &query, Some(nonce), url) + .await + .expect_err("missing trailer must fail"); + + assert_eq!(err.kind(), io::ErrorKind::UnexpectedEof); + assert_eq!(err.to_string(), "put_file auth trailer is incomplete"); + } + #[tokio::test] async fn walk_dir_body_surfaces_background_failure_after_data() { let body = walk_dir_response_body(true, |mut writer| async move { diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 533635fb4..d7fd898ea 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -216,10 +216,12 @@ pub(crate) mod rpc_consumer { NS_SCANNER_SESSION_ID_QUERY, NS_SCANNER_SESSION_SEQUENCE_QUERY, WALK_DIR_BODY_SHA256_QUERY, }; pub(crate) use super::super::storage_contracts::{ - NS_SCANNER_PROTOCOL_VERSION, NsScannerCapabilityResponse, WALK_DIR_STREAM_COMPLETION_V1, + NS_SCANNER_PROTOCOL_VERSION, NsScannerCapabilityResponse, PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_V1, + WALK_DIR_STREAM_COMPLETION_V1, }; pub(crate) use super::super::{ - StorageDiskRpcExt, WalkDirOptions, find_local_disk_by_ref, sign_ns_scanner_capability, verify_rpc_signature, + 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, }; } @@ -498,13 +500,15 @@ 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, - normalize_tonic_rpc_audience, sign_ns_scanner_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, - tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason, verify_rpc_signature, - verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_rpc_signature_with_bootstrap, + check_and_record_signed_rpc_nonce, normalize_tonic_rpc_audience, 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_canonical_body_digest, + verify_tonic_mutation_body_digest, verify_tonic_rpc_signature_with_bootstrap, }; #[cfg(test)] pub(crate) use rustfs_ecstore::api::rpc::{ - gen_signature_headers, gen_tonic_signature_headers, set_tonic_canonical_body_digest, verify_tonic_rpc_response_proof, + build_put_file_auth_trailer, gen_signature_headers, gen_tonic_signature_headers, set_tonic_canonical_body_digest, + verify_tonic_rpc_response_proof, }; } @@ -1655,6 +1659,25 @@ pub(crate) fn verify_rpc_signature(url: &str, method: &http::Method, headers: &h ecstore_rpc::verify_rpc_signature(url, method, headers) } +pub(crate) fn check_and_record_signed_rpc_nonce( + headers: &http::HeaderMap, + nonce: uuid::Uuid, + rpc_path: &str, + operation: &'static str, + backend: &'static str, +) -> std::io::Result<()> { + ecstore_rpc::check_and_record_signed_rpc_nonce(headers, nonce, rpc_path, operation, backend) +} + +pub(crate) fn verify_put_file_auth_trailer( + url: &str, + method: &http::Method, + nonce: uuid::Uuid, + trailer: &[u8], +) -> std::io::Result { + ecstore_rpc::verify_put_file_auth_trailer(url, method, nonce, trailer) +} + pub(crate) fn sign_ns_scanner_capability(challenge: uuid::Uuid, server_epoch: uuid::Uuid) -> std::io::Result> { ecstore_rpc::sign_ns_scanner_capability(challenge, server_epoch) }