gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-10 15:16:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(rpc): negotiate authenticated file writes (#5880)

Browse Source 
* fix(rpc): negotiate authenticated file writes

* fix(rpc): share capability probe failures

* test(rpc): cover dedicated capability route

* fix(rpc): satisfy capability cache lints

* fix(rpc): retry timed out capability probes

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
cxymds
2026-08-09 21:19:47 +08:00
committed by GitHub
parent 1be636b914
commit 8f9633ee83
11 changed files with 1134 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/api/mod.rs
+5 -5
View File
@@ -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,
};
}
crates/ecstore/src/cluster/rpc/http_auth.rs
+46 -1
View File
@@ -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<HmacSha256> {
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<Vec<u8>> {
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();
crates/ecstore/src/cluster/rpc/internode_data_transport.rs
+653 -16
View File
@@ -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<std::result::Result<Arc<dyn InternodeDataTransport>, 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<Option<Uuid>, PutFileCapabilityProbeFailure>;
#[derive(Debug, Clone)]
struct PutFileCapabilityFlight {
generation: u64,
v1_was_pinned: bool,
outcome: Arc<OnceCell<PutFileCapabilityProbeOutcome>>,
}
#[derive(Debug, Default)]
struct PutFileCapabilityCacheState {
cached: Option<PutFileCapabilityState>,
generation: u64,
in_flight: Option<PutFileCapabilityFlight>,
}
type PutFileCapabilityCacheEntry = Arc<tokio::sync::RwLock<PutFileCapabilityCacheState>>;
static PUT_FILE_CAPABILITY_CACHE: LazyLock<parking_lot::RwLock<HashMap<String, PutFileCapabilityCacheEntry>>> =
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<PutFileCapabilityState>, now: Instant) -> Option<Option<Uuid>> {
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<FileWriter> {
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<FileReader> {
@@ -228,6 +311,134 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport {
}
}
impl TcpHttpInternodeDataTransport {
async fn put_file_auth_capability(&self, endpoint: &str) -> Result<Option<Uuid>> {
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<Option<Uuid>> {
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<F, Fut>(endpoint: &str, probe: F) -> Result<Option<Uuid>>
where
F: FnOnce() -> Fut,
Fut: Future<Output = Result<Option<Uuid>>>,
{
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(&current.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<Uuid> {
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<Uuid>) -> 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<W> {
inner: W,
url: String,
@@ -450,6 +678,28 @@ pub fn build_internode_data_transport_from_env() -> Result<Arc<dyn InternodeData
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use tokio::sync::{Barrier, Notify};
async fn wait_for_capability_flight_waiters(entry: &PutFileCapabilityCacheEntry, waiters: usize) {
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let strong_count = entry
.read()
.await
.in_flight
.as_ref()
.map(|flight| Arc::strong_count(&flight.outcome))
.unwrap_or_default();
if strong_count > 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;
crates/ecstore/src/cluster/rpc/mod.rs
+5 -5
View File
@@ -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;
crates/ecstore/src/cluster/rpc/remote_disk.rs
+21
View File
@@ -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(
crates/ecstore/src/storage_api_contracts/mod.rs
+4 -3
View File
@@ -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,
};
}
crates/io-metrics/src/internode_metrics.rs
+2
View File
@@ -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");
crates/rio/src/http_reader.rs
+90 -3
View File
@@ -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<io::Error> {
let source = error.get_ref()?.downcast_ref::<InternodeHttpError>()?;
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::<InternodeHttpError>())
.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::<InternodeHttpError>())
.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::<InternodeHttpError>())
.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"));
crates/storage-api/src/lib.rs
+12
View File
@@ -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<u8>,
}
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";
rustfs/src/storage/rpc/http_service.rs
+281 -27
View File
@@ -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<uuid::Uuid> = LazyLock::new(uuid::Uuid::new_v4);
static PUT_FILE_CAPABILITY_SERVER_EPOCH: LazyLock<uuid::Uuid> = LazyLock::new(uuid::Uuid::new_v4);
static PUT_FILE_AUTH_STRICT: LazyLock<bool> = 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<String>,
put_file_nonce: Option<uuid::Uuid>,
put_file_server_epoch: Option<uuid::Uuid>,
}
#[derive(Debug, Default, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct PutFileCapabilityQuery {
put_file_capability: Option<u16>,
put_file_challenge: Option<uuid::Uuid>,
}
fn put_file_auth_nonce(query: &PutFileQuery) -> io::Result<Option<uuid::Uuid>> {
@@ -352,6 +363,10 @@ fn put_file_auth_nonce(query: &PutFileQuery) -> io::Result<Option<uuid::Uuid>> {
}
}
fn put_file_server_epoch_matches(query: &PutFileQuery) -> bool {
query.put_file_server_epoch == Some(*PUT_FILE_CAPABILITY_SERVER_EPOCH)
}
impl<S> Service<Request<Incoming>> for InternodeRpcService<S>
where
S: Service<Request<Incoming>, Response = Response<Body>> + Clone + Send + 'static,
@@ -403,7 +418,18 @@ async fn handle_internode_rpc(req: Request<Incoming>) -> Response<Body> {
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::<PutFileCapabilityQuery>(&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<Incoming>) -> Response<Body> {
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<Body> {
response
}
fn put_file_capability_response(challenge: uuid::Uuid) -> Response<Body> {
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<Incoming>) -> Response<Body> {
async fn handle_put_file(req: Request<Incoming>, require_auth: bool) -> Response<Body> {
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<Incoming>) -> Response<Body> {
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::<Bytes>::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::<Bytes>::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::<Bytes>::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<tokio::sync::oneshot::Sender<()>>);
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, io::Error>(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);
rustfs/src/storage/storage_api.rs
+15 -5
View File
@@ -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<Vec<u8>> {
ecstore_rpc::sign_put_file_capability(challenge, server_epoch, version)
}
pub(crate) fn verify_tonic_rpc_signature_with_bootstrap(
audience: &str,
path: &str,