From 046d5386ba9ffe6a0df39ffd78322c62e9f358d2 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 14 Jun 2026 14:26:44 +0800 Subject: [PATCH] feat: stream object zip downloads (#3380) * feat: stream object zip downloads * fix: stream zip downloads page by page * fix: prepare zip downloads before streaming --------- Signed-off-by: houseme Co-authored-by: houseme --- Cargo.lock | 18 + Cargo.toml | 2 + rustfs/Cargo.toml | 2 + rustfs/src/admin/handlers/mod.rs | 2 + .../src/admin/handlers/object_zip_download.rs | 1428 +++++++++++++++++ rustfs/src/admin/mod.rs | 7 +- rustfs/src/admin/route_policy.rs | 20 + rustfs/src/admin/route_registration_test.rs | 7 + rustfs/src/admin/router.rs | 89 + rustfs/src/server/http.rs | 4 +- rustfs/src/server/layer.rs | 89 +- 11 files changed, 1662 insertions(+), 6 deletions(-) create mode 100644 rustfs/src/admin/handlers/object_zip_download.rs diff --git a/Cargo.lock b/Cargo.lock index a68d54e70..c34e2e718 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -635,6 +635,7 @@ checksum = "e79b3f8a79cccc2898f31920fc69f304859b3bd567490f75ebf51ae1c792a9ac" dependencies = [ "compression-codecs", "compression-core", + "futures-io", "pin-project-lite", "tokio", ] @@ -778,6 +779,21 @@ dependencies = [ "tungstenite", ] +[[package]] +name = "async_zip" +version = "0.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d8c50d65ce1b0e0cb65a785ff615f78860d7754290647d3b983208daa4f85e6" +dependencies = [ + "async-compression", + "crc32fast", + "futures-lite", + "pin-project", + "thiserror 2.0.18", + "tokio", + "tokio-util", +] + [[package]] name = "atoi" version = "2.0.0" @@ -9031,6 +9047,7 @@ dependencies = [ "anyhow", "astral-tokio-tar", "async-trait", + "async_zip", "atoi 3.0.0", "atomic_enum", "aws-config", @@ -9044,6 +9061,7 @@ dependencies = [ "datafusion", "flatbuffers", "futures", + "futures-lite", "futures-util", "hashbrown 0.17.1", "hex-simd", diff --git a/Cargo.toml b/Cargo.toml index aff67c87e..55d15e575 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -125,6 +125,7 @@ rustfs-zip = { path = "./crates/zip", version = "1.0.0-beta.8" } # Async Runtime and Networking async-channel = "2.5.0" +async_zip = { version = "0.0.18", default-features = false, features = ["tokio", "deflate"] } mysql_async = { version = "0.37", default-features = false, features = ["default-rustls", "tracing"] } async-compression = { version = "0.4.42" } async-recursion = "1.1.1" @@ -133,6 +134,7 @@ async-nats = "0.49.1" axum = "0.8.9" futures = "0.3.32" futures-core = "0.3.32" +futures-lite = "2.6.1" futures-util = "0.3.32" pollster = "0.4.0" pulsar = { version = "6.8.0", default-features = false, features = ["tokio-rustls-runtime"] } diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index dfaf2b1ab..5d44790a6 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -104,6 +104,7 @@ tempfile = { workspace = true } async-trait = { workspace = true } axum.workspace = true futures.workspace = true +futures-lite.workspace = true futures-util.workspace = true hyper.workspace = true hyper-util.workspace = true @@ -141,6 +142,7 @@ time = { workspace = true, features = ["parsing", "formatting", "serde"] } astral-tokio-tar = { workspace = true } atoi = { workspace = true } atomic_enum = { workspace = true } +async_zip = { workspace = true } base64 = { workspace = true } hmac = { workspace = true } sha2 = { workspace = true } diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index 277a58804..7e0c35caa 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -31,6 +31,7 @@ pub mod kms_management; pub mod metrics; pub mod module_switch; mod notify_runtime_access; +pub mod object_zip_download; pub mod oidc; pub mod plugins_catalog; pub mod plugins_instances; @@ -75,6 +76,7 @@ mod tests { let _get_extension_catalog = extensions::GetExtensionCatalogHandler {}; let _list_extension_instances = extensions::ListExtensionInstancesHandler {}; let _get_plugin_catalog = plugins_catalog::GetPluginCatalogHandler {}; + let _create_object_zip_download = object_zip_download::CreateObjectZipDownloadHandler {}; let _list_plugin_instances = plugins_instances::ListPluginInstancesHandler {}; let _get_plugin_instance = plugins_instances::GetPluginInstanceHandler {}; let _put_plugin_instance = plugins_instances::PutPluginInstanceHandler {}; diff --git a/rustfs/src/admin/handlers/object_zip_download.rs b/rustfs/src/admin/handlers/object_zip_download.rs new file mode 100644 index 000000000..51140b2b5 --- /dev/null +++ b/rustfs/src/admin/handlers/object_zip_download.rs @@ -0,0 +1,1428 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::admin::router::{ADMIN_OBJECT_ZIP_DOWNLOADS_PATH, AdminOperation, Operation, S3Router}; +use crate::auth::{check_key_valid, get_session_token}; +use crate::error::ApiError; +use crate::license::license_check; +use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use crate::storage::access::{ReqInfo, authorize_request}; +use aes_gcm::{ + Aes256Gcm, Key, Nonce, + aead::{Aead, KeyInit}, +}; +use async_zip::{Compression, ZipEntryBuilder, base::write::ZipFileWriter}; +use futures_lite::io::AsyncWriteExt; +use http::{HeaderMap, HeaderValue, StatusCode, header}; +use hyper::{Method, Uri}; +use matchit::Params; +use rand::RngExt; +use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; +use rustfs_credentials::get_global_action_cred; +use rustfs_ecstore::{ + global::get_global_region, + new_object_layer_fn, + store_api::{BucketOperations, ListOperations, ObjectIO, ObjectOperations, ObjectOptions}, +}; +use rustfs_policy::policy::action::{Action, S3Action}; +use rustfs_storage_api::bucket::BucketOptions; +use rustfs_trusted_proxies::{ClientInfo, ValidationMode}; +use rustfs_utils::{base64_decode_url_safe_no_pad, base64_encode_url_safe_no_pad}; +use s3s::{Body, S3Request, S3Response, S3Result, dto::StreamingBlob, header::CONTENT_TYPE, s3_error}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::collections::HashSet; +use std::io; +use std::net::{IpAddr, SocketAddr}; +use time::{Duration, OffsetDateTime, format_description::well_known::Rfc3339}; +use tokio::fs::File; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; +use tokio_util::io::ReaderStream; +use url::form_urlencoded; +use uuid::Uuid; + +const OBJECT_ZIP_DOWNLOAD_TOKEN_TTL: Duration = Duration::minutes(5); +const ZIP_STREAM_BUFFER_SIZE: usize = 1024 * 1024; +const ZIP_OBJECT_BUFFER_SIZE: usize = 128 * 1024; +const ZIP_LIST_MAX_KEYS: i32 = 1000; + +pub fn register_object_zip_download_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::POST, + format!("{ADMIN_PREFIX}{ADMIN_OBJECT_ZIP_DOWNLOADS_PATH}").as_str(), + AdminOperation(&CreateObjectZipDownloadHandler {}), + )?; + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}{ADMIN_OBJECT_ZIP_DOWNLOADS_PATH}/{{id}}.zip").as_str(), + AdminOperation(&DownloadObjectZipHandler {}), + )?; + + Ok(()) +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub struct CreateObjectZipDownloadRequest { + pub bucket: String, + #[serde(default)] + pub prefix: Option, + #[serde(default)] + pub objects: Vec, + #[serde(default)] + pub prefixes: Vec, + #[serde(default)] + pub filename: Option, +} + +#[derive(Debug, Serialize)] +pub struct CreateObjectZipDownloadResponse { + pub download_url: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub expires_at: Option, +} + +impl CreateObjectZipDownloadRequest { + fn validate(&self) -> S3Result<()> { + if self.bucket.trim().is_empty() { + return Err(s3_error!(InvalidRequest, "bucket must not be empty")); + } + + if self.objects.iter().any(|object| object.is_empty()) { + return Err(s3_error!(InvalidRequest, "objects must not contain empty strings")); + } + + if self.prefixes.iter().any(|prefix| prefix.is_empty()) { + return Err(s3_error!(InvalidRequest, "prefixes must not contain empty strings")); + } + + let has_prefix = self.prefix.as_deref().is_some_and(|prefix| !prefix.is_empty()); + if !has_prefix && self.objects.is_empty() && self.prefixes.is_empty() { + return Err(s3_error!(InvalidRequest, "at least one of prefix, objects, or prefixes is required")); + } + + Ok(()) + } +} + +#[derive(Clone, Debug)] +struct ObjectZipDownloadToken { + id: String, + principal: String, + req_info: ReqInfo, + auth_context: ObjectZipDownloadAuthContext, + request: CreateObjectZipDownloadRequest, +} + +#[derive(Clone, Debug, Default)] +struct ObjectZipDownloadAuthContext { + headers: HeaderMap, + remote_addr: Option, + client_info: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +struct ObjectZipDownloadTokenPayload { + id: String, + principal: String, + req_info: ObjectZipDownloadReqInfoSnapshot, + auth_context: ObjectZipDownloadAuthContextSnapshot, + request: CreateObjectZipDownloadRequest, + expires_at_unix: i64, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +struct ObjectZipDownloadReqInfoSnapshot { + cred: Option, + is_owner: bool, + bucket: Option, + object: Option, + version_id: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +struct ObjectZipDownloadAuthContextSnapshot { + headers: Vec<(String, String)>, + remote_addr: Option, + client_info: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +struct ClientInfoSnapshot { + real_ip: IpAddr, + forwarded_host: Option, + forwarded_proto: Option, + is_from_trusted_proxy: bool, + proxy_ip: Option, + proxy_hops: usize, + validation_mode: ValidationMode, + warnings: Vec, +} + +#[derive(Debug)] +struct CreatedObjectZipDownloadToken { + id: String, + token: String, + expires_at: OffsetDateTime, +} + +struct PreparedZipArchive { + file: File, +} + +impl From<&ReqInfo> for ObjectZipDownloadReqInfoSnapshot { + fn from(req_info: &ReqInfo) -> Self { + let mut cred = req_info.cred.clone(); + if let Some(cred) = cred.as_mut() { + cred.secret_key.clear(); + cred.session_token.clear(); + } + + Self { + cred, + is_owner: req_info.is_owner, + bucket: req_info.bucket.clone(), + object: req_info.object.clone(), + version_id: req_info.version_id.clone(), + } + } +} + +impl From for ReqInfo { + fn from(snapshot: ObjectZipDownloadReqInfoSnapshot) -> Self { + Self { + cred: snapshot.cred, + is_owner: snapshot.is_owner, + bucket: snapshot.bucket, + object: snapshot.object, + version_id: snapshot.version_id, + region: get_global_region(), + request_context: None, + } + } +} + +impl From<&ObjectZipDownloadAuthContext> for ObjectZipDownloadAuthContextSnapshot { + fn from(context: &ObjectZipDownloadAuthContext) -> Self { + Self { + headers: context + .headers + .iter() + .filter_map(|(name, value)| { + value + .to_str() + .ok() + .map(|value| (name.as_str().to_string(), value.to_string())) + }) + .collect(), + remote_addr: context.remote_addr.map(|addr| addr.0), + client_info: context.client_info.as_ref().map(ClientInfoSnapshot::from), + } + } +} + +impl TryFrom for ObjectZipDownloadAuthContext { + type Error = s3s::S3Error; + + fn try_from(snapshot: ObjectZipDownloadAuthContextSnapshot) -> S3Result { + let mut headers = HeaderMap::new(); + for (name, value) in snapshot.headers { + let name = http::header::HeaderName::from_bytes(name.as_bytes()) + .map_err(|_| s3_error!(AccessDenied, "invalid or expired download token"))?; + let value = + HeaderValue::from_str(&value).map_err(|_| s3_error!(AccessDenied, "invalid or expired download token"))?; + headers.append(name, value); + } + + Ok(Self { + headers, + remote_addr: snapshot.remote_addr.map(RemoteAddr), + client_info: snapshot.client_info.map(ClientInfo::from), + }) + } +} + +impl From<&ClientInfo> for ClientInfoSnapshot { + fn from(client_info: &ClientInfo) -> Self { + Self { + real_ip: client_info.real_ip, + forwarded_host: client_info.forwarded_host.clone(), + forwarded_proto: client_info.forwarded_proto.clone(), + is_from_trusted_proxy: client_info.is_from_trusted_proxy, + proxy_ip: client_info.proxy_ip, + proxy_hops: client_info.proxy_hops, + validation_mode: client_info.validation_mode, + warnings: client_info.warnings.clone(), + } + } +} + +impl From for ClientInfo { + fn from(snapshot: ClientInfoSnapshot) -> Self { + Self { + real_ip: snapshot.real_ip, + forwarded_host: snapshot.forwarded_host, + forwarded_proto: snapshot.forwarded_proto, + is_from_trusted_proxy: snapshot.is_from_trusted_proxy, + proxy_ip: snapshot.proxy_ip, + proxy_hops: snapshot.proxy_hops, + validation_mode: snapshot.validation_mode, + warnings: snapshot.warnings, + } + } +} + +fn download_token_encryption_key() -> S3Result<[u8; 32]> { + let credentials = + get_global_action_cred().ok_or_else(|| s3_error!(InternalError, "global action credentials are not initialized"))?; + if credentials.secret_key.is_empty() { + return Err(s3_error!(InternalError, "global action credentials are not initialized")); + } + let mut hasher = Sha256::new(); + hasher.update(b"rustfs-object-zip-download-token-v1"); + hasher.update(credentials.secret_key.as_bytes()); + Ok(hasher.finalize().into()) +} + +fn encode_download_token(payload: &ObjectZipDownloadTokenPayload) -> S3Result { + let payload_json = + serde_json::to_vec(payload).map_err(|e| s3_error!(InternalError, "failed to serialize download token: {}", e))?; + let key = download_token_encryption_key()?; + let cipher = Aes256Gcm::new(&Key::::from(key)); + let mut nonce_bytes = [0_u8; 12]; + rand::rng().fill(&mut nonce_bytes); + let nonce = Nonce::from(nonce_bytes); + let ciphertext = cipher + .encrypt(&nonce, payload_json.as_slice()) + .map_err(|_| s3_error!(InternalError, "failed to encrypt download token"))?; + Ok(format!( + "{}.{}", + base64_encode_url_safe_no_pad(&nonce_bytes), + base64_encode_url_safe_no_pad(&ciphertext) + )) +} + +fn decode_download_token(token: &str) -> S3Result { + let Some((nonce_part, ciphertext_part)) = token.split_once('.') else { + return Err(s3_error!(AccessDenied, "invalid or expired download token")); + }; + if ciphertext_part.contains('.') { + return Err(s3_error!(AccessDenied, "invalid or expired download token")); + } + + let nonce = base64_decode_url_safe_no_pad(nonce_part.as_bytes()) + .map_err(|_| s3_error!(AccessDenied, "invalid or expired download token"))?; + let ciphertext = base64_decode_url_safe_no_pad(ciphertext_part.as_bytes()) + .map_err(|_| s3_error!(AccessDenied, "invalid or expired download token"))?; + let nonce: [u8; 12] = nonce + .as_slice() + .try_into() + .map_err(|_| s3_error!(AccessDenied, "invalid or expired download token"))?; + let key = download_token_encryption_key()?; + let cipher = Aes256Gcm::new(&Key::::from(key)); + let payload = cipher + .decrypt(&Nonce::from(nonce), ciphertext.as_slice()) + .map_err(|_| s3_error!(AccessDenied, "invalid or expired download token"))?; + + serde_json::from_slice(&payload).map_err(|_| s3_error!(AccessDenied, "invalid or expired download token")) +} + +fn create_download_token( + principal: String, + req_info: ReqInfo, + auth_context: ObjectZipDownloadAuthContext, + request: CreateObjectZipDownloadRequest, + now: OffsetDateTime, +) -> S3Result { + let id = Uuid::new_v4().to_string(); + let expires_at = now + OBJECT_ZIP_DOWNLOAD_TOKEN_TTL; + let payload = ObjectZipDownloadTokenPayload { + id: id.clone(), + principal, + req_info: ObjectZipDownloadReqInfoSnapshot::from(&req_info), + auth_context: ObjectZipDownloadAuthContextSnapshot::from(&auth_context), + request, + expires_at_unix: expires_at.unix_timestamp(), + }; + let token = encode_download_token(&payload)?; + + Ok(CreatedObjectZipDownloadToken { id, token, expires_at }) +} + +fn validate_download_token(id: &str, token: &str, now: OffsetDateTime) -> S3Result { + let payload = decode_download_token(token)?; + if payload.id != id { + return Err(s3_error!(AccessDenied, "invalid or expired download token")); + } + let expires_at = OffsetDateTime::from_unix_timestamp(payload.expires_at_unix) + .map_err(|_| s3_error!(AccessDenied, "invalid or expired download token"))?; + if expires_at <= now { + return Err(s3_error!(AccessDenied, "invalid or expired download token")); + } + + Ok(ObjectZipDownloadToken { + id: payload.id, + principal: payload.principal, + req_info: payload.req_info.into(), + auth_context: payload.auth_context.try_into()?, + request: payload.request, + }) +} + +fn format_expires_at(value: OffsetDateTime) -> S3Result { + value + .format(&Rfc3339) + .map_err(|e| s3_error!(InternalError, "failed to format token expiration: {}", e)) +} + +fn build_download_url(id: &str, token: &str) -> String { + format!("{ADMIN_PREFIX}{ADMIN_OBJECT_ZIP_DOWNLOADS_PATH}/{id}.zip?token={token}") +} + +async fn authenticate_object_zip_download_request(req: &mut S3Request) -> S3Result<()> { + let Some(input_cred) = &req.credentials else { + return Err(s3_error!(InvalidRequest, "authentication required")); + }; + + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + + let remote_addr = req.extensions.get::>().cloned().unwrap_or_default(); + req.extensions.insert(ReqInfo { + cred: Some(cred), + is_owner: owner, + region: get_global_region(), + request_context: req.extensions.get().cloned(), + ..Default::default() + }); + req.extensions.insert(remote_addr); + + license_check().map_err(|er| match er.kind() { + std::io::ErrorKind::PermissionDenied => s3_error!(AccessDenied, "{er}"), + _ => { + tracing::error!("license check failed due to unexpected error: {er}"); + s3_error!(InternalError, "License validation failed") + } + })?; + + Ok(()) +} + +fn authenticated_zip_download_principal(req: &S3Request) -> S3Result { + let req_info = req + .extensions + .get::() + .ok_or_else(|| s3_error!(AccessDenied, "authentication required"))?; + let credentials = req_info + .cred + .clone() + .ok_or_else(|| s3_error!(AccessDenied, "authentication required"))?; + + Ok(credentials.access_key) +} + +fn authenticated_zip_download_req_info(req: &S3Request) -> S3Result { + let mut req_info = req + .extensions + .get::() + .cloned() + .ok_or_else(|| s3_error!(AccessDenied, "authentication required"))?; + if let Some(cred) = req_info.cred.as_mut() { + cred.secret_key.clear(); + cred.session_token.clear(); + } + Ok(req_info) +} + +fn authenticated_zip_download_auth_context(req: &S3Request) -> ObjectZipDownloadAuthContext { + ObjectZipDownloadAuthContext { + headers: req.headers.clone(), + remote_addr: req.extensions.get::>().copied().flatten(), + client_info: req.extensions.get::().cloned(), + } +} + +async fn authorize_object_zip_download_scope( + req: &mut S3Request, + request: &CreateObjectZipDownloadRequest, +) -> S3Result<()> { + for prefix in listing_prefixes(request) { + authorize_object_zip_download_list_prefix(req, &request.bucket, prefix).await?; + } + + for object in &request.objects { + authorize_object_zip_download_s3_action(req, &request.bucket, Some(object), S3Action::GetObjectAction).await?; + } + + Ok(()) +} + +async fn authorize_object_zip_download_list_prefix(req: &mut S3Request, bucket: &str, prefix: &str) -> S3Result<()> { + let original_uri = req.uri.clone(); + req.uri = uri_with_prefix_query(&original_uri, prefix)?; + let result = authorize_object_zip_download_s3_action(req, bucket, None, S3Action::ListBucketAction).await; + req.uri = original_uri; + result +} + +async fn authorize_object_zip_download_s3_action( + req: &mut S3Request, + bucket: &str, + object: Option<&str>, + action: S3Action, +) -> S3Result<()> { + if let Some(req_info) = req.extensions.get_mut::() { + req_info.bucket = Some(bucket.to_string()); + req_info.object = object.map(str::to_string); + req_info.version_id = None; + } + + authorize_request(req, Action::S3Action(action)).await +} + +fn uri_with_prefix_query(uri: &Uri, prefix: &str) -> S3Result { + let existing_pairs: Vec<(String, String)> = uri + .query() + .map(|query| { + form_urlencoded::parse(query.as_bytes()) + .map(|(k, v)| (k.into_owned(), v.into_owned())) + .collect() + }) + .unwrap_or_default(); + let mut query = form_urlencoded::Serializer::new(String::new()); + for (key, value) in existing_pairs { + if key != "prefix" { + query.append_pair(&key, &value); + } + } + query.append_pair("prefix", prefix); + let path_and_query = format!("{}?{}", uri.path(), query.finish()); + let mut parts = uri.clone().into_parts(); + parts.path_and_query = Some( + path_and_query + .parse() + .map_err(|e| s3_error!(InvalidRequest, "invalid prefix authorization URI: {}", e))?, + ); + Uri::from_parts(parts).map_err(|e| s3_error!(InvalidRequest, "invalid prefix authorization URI: {}", e)) +} + +fn build_json_response( + status: StatusCode, + body: &impl Serialize, + request_id: Option<&HeaderValue>, +) -> S3Result> { + let data = serde_json::to_vec(body).map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?; + let mut header = HeaderMap::new(); + header.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + if let Some(value) = request_id { + header.insert("x-request-id", value.clone()); + } + Ok(S3Response::with_headers((status, Body::from(data)), header)) +} + +fn query_value_exact(uri: &Uri, key: &str) -> Option { + uri.query().and_then(|query| { + form_urlencoded::parse(query.as_bytes()) + .find_map(|(name, value)| if name == key { Some(value.into_owned()) } else { None }) + }) +} + +pub struct CreateObjectZipDownloadHandler {} + +#[async_trait::async_trait] +impl Operation for CreateObjectZipDownloadHandler { + async fn call(&self, mut req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authenticate_object_zip_download_request(&mut req).await?; + + let body = req + .input + .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE) + .await + .map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))?; + let request: CreateObjectZipDownloadRequest = + serde_json::from_slice(&body).map_err(|e| s3_error!(InvalidRequest, "invalid JSON: {}", e))?; + request.validate()?; + authorize_object_zip_download_scope(&mut req, &request).await?; + let principal = authenticated_zip_download_principal(&req)?; + let req_info = authenticated_zip_download_req_info(&req)?; + let auth_context = authenticated_zip_download_auth_context(&req); + let created = create_download_token(principal, req_info, auth_context, request, OffsetDateTime::now_utc())?; + + build_json_response( + StatusCode::OK, + &CreateObjectZipDownloadResponse { + download_url: build_download_url(&created.id, &created.token), + expires_at: Some(format_expires_at(created.expires_at)?), + }, + req.headers.get("x-request-id"), + ) + } +} + +pub struct DownloadObjectZipHandler {} + +#[async_trait::async_trait] +impl Operation for DownloadObjectZipHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let id = params.get("id").unwrap_or(""); + let token = query_value_exact(&req.uri, "token").unwrap_or_default(); + if id.is_empty() || token.is_empty() { + return Err(s3_error!(AccessDenied, "invalid or expired download token")); + } + + let record = validate_download_token(id, &token, OffsetDateTime::now_utc())?; + let _token_id = &record.id; + let _principal = &record.principal; + let prepared = prepare_zip_download_archive(&record).await?; + let body = build_zip_stream_body(prepared); + + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/zip")); + headers.insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store")); + headers.insert( + header::CONTENT_DISPOSITION, + HeaderValue::from_str(&format!( + "attachment; filename=\"{}\"", + sanitize_content_disposition_filename(record.request.filename.as_deref().unwrap_or("download.zip")) + )) + .map_err(|e| s3_error!(InternalError, "invalid download filename: {}", e))?, + ); + + Ok(S3Response::with_headers((StatusCode::OK, body), headers)) + } +} + +fn sanitize_content_disposition_filename(filename: &str) -> String { + let cleaned = filename + .chars() + .map(|ch| match ch { + '"' | '\\' | '\r' | '\n' => '_', + ch if ch.is_control() => '_', + ch => ch, + }) + .collect::(); + if cleaned.trim().is_empty() { + "download.zip".to_string() + } else { + cleaned + } +} + +fn zip_entry_name(base_prefix: Option<&str>, key: &str) -> S3Result { + let relative = base_prefix + .filter(|base| !base.is_empty() && key.starts_with(*base)) + .map(|base| &key[base.len()..]) + .unwrap_or(key); + let entry_name = relative.trim_start_matches('/').to_string(); + validate_zip_entry_name(&entry_name)?; + Ok(entry_name) +} + +fn validate_zip_entry_name(entry_name: &str) -> S3Result<()> { + if entry_name.is_empty() { + return Err(s3_error!(InvalidRequest, "zip entry name must not be empty")); + } + if entry_name.starts_with('/') || entry_name.contains('\\') || entry_name.split('/').any(|part| part == "..") { + return Err(s3_error!(InvalidRequest, "zip entry name must be relative")); + } + Ok(()) +} + +async fn preflight_zip_items(request: &CreateObjectZipDownloadRequest, items: &[ZipDownloadItem]) -> S3Result<()> { + let store = new_object_layer_fn().ok_or_else(|| s3_error!(InternalError, "object store not initialized"))?; + for item in items { + store + .get_object_info(&request.bucket, &item.key, &ObjectOptions::default()) + .await + .map_err(storage_error_to_s3)?; + } + Ok(()) +} + +fn storage_error_to_s3(err: rustfs_ecstore::error::Error) -> s3s::S3Error { + ApiError::from(err).into() +} + +async fn validate_zip_download_request(record: &ObjectZipDownloadToken) -> S3Result<()> { + let store = new_object_layer_fn().ok_or_else(|| s3_error!(InternalError, "object store not initialized"))?; + store + .get_bucket_info(&record.request.bucket, &BucketOptions::default()) + .await + .map_err(storage_error_to_s3)?; + + validate_zip_download_items(record, &collect_explicit_zip_items(&record.request)?).await?; + Ok(()) +} + +async fn validate_zip_download_items(record: &ObjectZipDownloadToken, items: &[ZipDownloadItem]) -> S3Result<()> { + authorize_zip_items_for_download(record, items).await?; + preflight_zip_items(&record.request, items).await +} + +fn listing_prefixes(request: &CreateObjectZipDownloadRequest) -> Vec<&str> { + if request.objects.is_empty() + && request.prefixes.is_empty() + && let Some(prefix) = request.prefix.as_deref().filter(|prefix| !prefix.is_empty()) + { + return vec![prefix]; + } + + request.prefixes.iter().map(String::as_str).collect() +} + +async fn authorize_zip_items_for_download(record: &ObjectZipDownloadToken, items: &[ZipDownloadItem]) -> S3Result<()> { + for item in items { + let mut req = S3Request { + input: Body::empty(), + method: Method::GET, + uri: format!("{ADMIN_PREFIX}{ADMIN_OBJECT_ZIP_DOWNLOADS_PATH}/{}.zip", record.id) + .parse() + .map_err(|e| s3_error!(InternalError, "failed to build download authorization URI: {}", e))?, + headers: record.auth_context.headers.clone(), + extensions: { + let mut extensions = http::Extensions::new(); + extensions.insert(record.auth_context.remote_addr); + if let Some(client_info) = record.auth_context.client_info.clone() { + extensions.insert(client_info); + } + extensions + }, + credentials: None, + region: get_global_region(), + service: None, + trailing_headers: None, + }; + let mut req_info = record.req_info.clone(); + req_info.bucket = Some(record.request.bucket.clone()); + req_info.object = Some(item.key.clone()); + req_info.version_id = None; + req.extensions.insert(req_info); + authorize_request(&mut req, Action::S3Action(S3Action::GetObjectAction)).await?; + } + Ok(()) +} + +fn collect_explicit_zip_items(request: &CreateObjectZipDownloadRequest) -> S3Result> { + let mut items = Vec::new(); + let mut seen = HashSet::new(); + for key in &request.objects { + push_zip_item(request, &mut seen, &mut items, key.clone())?; + } + Ok(items) +} + +fn push_zip_item( + request: &CreateObjectZipDownloadRequest, + seen: &mut HashSet, + items: &mut Vec, + key: String, +) -> S3Result<()> { + if seen.insert(key.clone()) { + items.push(ZipDownloadItem { + entry_name: zip_entry_name(request.prefix.as_deref(), &key)?, + key, + }); + } + Ok(()) +} + +fn explicit_zip_item_keys(items: &[ZipDownloadItem]) -> HashSet { + items.iter().map(|item| item.key.clone()).collect() +} + +fn normalized_listing_prefixes(request: &CreateObjectZipDownloadRequest) -> Vec { + let mut prefixes: Vec = listing_prefixes(request).into_iter().map(str::to_string).collect(); + prefixes.sort(); + prefixes.dedup(); + + let mut normalized = Vec::new(); + for prefix in prefixes { + if normalized.iter().any(|existing: &String| prefix.starts_with(existing)) { + continue; + } + normalized.push(prefix); + } + normalized +} + +fn build_zip_stream_body(prepared: PreparedZipArchive) -> Body { + let stream = ReaderStream::with_capacity(prepared.file, ZIP_STREAM_BUFFER_SIZE); + Body::from(StreamingBlob::wrap(stream)) +} + +async fn prepare_zip_download_archive(record: &ObjectZipDownloadToken) -> S3Result { + validate_zip_download_request(record).await?; + + let temp_file = tempfile::tempfile().map_err(|err| s3_error!(InternalError, "failed to create ZIP temp file: {}", err))?; + let std_file = temp_file; + let mut file = File::from_std(std_file); + + generate_zip_archive(record, &mut file) + .await + .map_err(|err| s3_error!(InternalError, "failed to prepare ZIP download: {}", err))?; + file.seek(std::io::SeekFrom::Start(0)) + .await + .map_err(|err| s3_error!(InternalError, "failed to rewind ZIP temp file: {}", err))?; + + Ok(PreparedZipArchive { file }) +} + +async fn generate_zip_archive(record: &ObjectZipDownloadToken, file: &mut File) -> io::Result<()> { + let mut zip = ZipFileWriter::with_tokio(file).force_zip64(); + let store = new_object_layer_fn().ok_or_else(|| io::Error::other("object store not initialized"))?; + + let explicit_items = collect_explicit_zip_items(&record.request) + .map_err(|err| io::Error::other(format!("failed to prepare explicit ZIP items: {err}")))?; + let explicit_keys = explicit_zip_item_keys(&explicit_items); + write_zip_items(&mut zip, &record.request.bucket, &explicit_items).await?; + + for prefix in normalized_listing_prefixes(&record.request) { + let mut continuation_token = None; + loop { + let listed = store + .clone() + .list_objects_v2( + &record.request.bucket, + prefix.as_str(), + continuation_token.clone(), + None, + ZIP_LIST_MAX_KEYS, + false, + None, + false, + ) + .await + .map_err(|err| io::Error::other(format!("failed to list prefix `{prefix}` for ZIP stream: {err}")))?; + + let mut page_items = Vec::new(); + for object in listed.objects { + if object.name == prefix && object.size == 0 { + continue; + } + if explicit_keys.contains(&object.name) { + continue; + } + page_items.push(ZipDownloadItem { + entry_name: zip_entry_name(record.request.prefix.as_deref(), &object.name) + .map_err(|err| io::Error::other(format!("failed to prepare ZIP entry name: {err}")))?, + key: object.name, + }); + } + validate_zip_download_items(record, &page_items) + .await + .map_err(|err| io::Error::other(format!("failed to validate ZIP page for prefix `{prefix}`: {err}")))?; + write_zip_items(&mut zip, &record.request.bucket, &page_items).await?; + + if !listed.is_truncated { + break; + } + continuation_token = listed.next_continuation_token; + if continuation_token.is_none() { + break; + } + } + } + + zip.close() + .await + .map(|_| ()) + .map_err(|err| io::Error::other(format!("failed to finish ZIP archive: {err}"))) +} + +async fn write_zip_items( + zip: &mut async_zip::tokio::write::ZipFileWriter, + bucket: &str, + items: &[ZipDownloadItem], +) -> io::Result<()> +where + W: tokio::io::AsyncWrite + Unpin, +{ + for item in items { + write_zip_item(zip, bucket, item).await.map_err(|err| { + io::Error::other(format!( + "failed to write bucket `{}` object `{}` into ZIP entry `{}`: {}", + bucket, item.key, item.entry_name, err + )) + })?; + } + Ok(()) +} + +async fn write_zip_item( + zip: &mut async_zip::tokio::write::ZipFileWriter, + bucket: &str, + item: &ZipDownloadItem, +) -> io::Result<()> +where + W: tokio::io::AsyncWrite + Unpin, +{ + let store = new_object_layer_fn().ok_or_else(|| io::Error::other("object store not initialized"))?; + let mut reader = store + .get_object_reader(bucket, &item.key, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .map_err(|err| io::Error::other(format!("failed to open object reader: {err}")))?; + let entry = ZipEntryBuilder::new(item.entry_name.clone().into(), Compression::Stored); + let mut entry_writer = zip + .write_entry_stream(entry) + .await + .map_err(|err| io::Error::other(format!("failed to start ZIP entry: {err}")))?; + + let mut buffer = vec![0_u8; ZIP_OBJECT_BUFFER_SIZE]; + loop { + let read = reader.read(&mut buffer).await?; + if read == 0 { + break; + } + entry_writer.write_all(&buffer[..read]).await?; + } + entry_writer + .close() + .await + .map_err(|err| io::Error::other(format!("failed to close ZIP entry: {err}"))) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct ZipDownloadItem { + key: String, + entry_name: String, +} + +#[cfg(test)] +mod tests { + use super::*; + use rustfs_credentials::init_global_action_credentials; + use s3s::S3ErrorCode; + + fn valid_request() -> CreateObjectZipDownloadRequest { + CreateObjectZipDownloadRequest { + bucket: "photos".to_string(), + prefix: Some("2026/".to_string()), + objects: Vec::new(), + prefixes: Vec::new(), + filename: None, + } + } + + fn assert_invalid_request(request: CreateObjectZipDownloadRequest, expected_message: &str) { + let err = request.validate().expect_err("request should be invalid"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + assert_eq!(err.message(), Some(expected_message)); + } + + #[test] + fn validation_accepts_prefix_objects_or_prefixes() { + valid_request().validate().expect("prefix selection should be valid"); + + let mut object_request = valid_request(); + object_request.prefix = None; + object_request.objects = vec!["a.txt".to_string()]; + object_request.validate().expect("object selection should be valid"); + + let mut prefix_request = valid_request(); + prefix_request.prefix = None; + prefix_request.prefixes = vec!["folder/".to_string()]; + prefix_request.validate().expect("prefixes selection should be valid"); + } + + #[test] + fn validation_rejects_empty_bucket() { + let mut request = valid_request(); + request.bucket = " ".to_string(); + + assert_invalid_request(request, "bucket must not be empty"); + } + + #[test] + fn validation_rejects_empty_selection() { + let mut request = valid_request(); + request.prefix = None; + + assert_invalid_request(request, "at least one of prefix, objects, or prefixes is required"); + } + + #[test] + fn validation_rejects_empty_objects_and_prefixes() { + let mut object_request = valid_request(); + object_request.objects = vec!["".to_string()]; + assert_invalid_request(object_request, "objects must not contain empty strings"); + + let mut prefix_request = valid_request(); + prefix_request.prefixes = vec!["".to_string()]; + assert_invalid_request(prefix_request, "prefixes must not contain empty strings"); + } + + #[test] + fn request_deserialization_defaults_optional_collections() { + let request: CreateObjectZipDownloadRequest = + serde_json::from_str(r#"{"bucket":"photos","prefix":"2026/"}"#).expect("request should deserialize"); + + assert_eq!(request.bucket, "photos"); + assert_eq!(request.prefix.as_deref(), Some("2026/")); + assert!(request.objects.is_empty()); + assert!(request.prefixes.is_empty()); + assert_eq!(request.filename, None); + } + + #[test] + fn request_deserialization_denies_unknown_fields() { + let err = + serde_json::from_str::(r#"{"bucket":"photos","prefix":"2026/","unexpected":true}"#) + .expect_err("unknown fields should be rejected"); + + assert!(err.to_string().contains("unknown field")); + } + + #[test] + fn response_serialization_skips_absent_expires_at() { + let response = CreateObjectZipDownloadResponse { + download_url: "/rustfs/admin/v3/object-zip-downloads/id.zip?token=token".to_string(), + expires_at: None, + }; + + let json = serde_json::to_value(response).expect("response should serialize"); + assert_eq!(json["download_url"], "/rustfs/admin/v3/object-zip-downloads/id.zip?token=token"); + assert!(json.get("expires_at").is_none()); + } + + #[test] + fn create_download_token_returns_browser_download_url_parts() { + let now = OffsetDateTime::UNIX_EPOCH; + let created = create_test_download_token(now); + + assert!(!created.id.is_empty()); + assert!(!created.token.is_empty()); + assert_eq!(created.expires_at, now + OBJECT_ZIP_DOWNLOAD_TOKEN_TTL); + let download_url = build_download_url(&created.id, &created.token); + assert!(download_url.starts_with("/rustfs/admin/v3/object-zip-downloads/")); + assert!(download_url.ends_with(&format!(".zip?token={}", created.token))); + } + + #[test] + fn validate_download_token_rejects_unknown_token() { + let err = validate_download_token("missing", "missing", OffsetDateTime::UNIX_EPOCH) + .expect_err("unknown token should be rejected"); + + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[test] + fn validate_download_token_rejects_mismatched_token_value() { + let now = OffsetDateTime::UNIX_EPOCH; + let created = create_test_download_token(now); + + let err = validate_download_token("different-id", &created.token, now).expect_err("wrong token id should be rejected"); + + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[test] + fn validate_download_token_rejects_expired_token() { + let now = OffsetDateTime::UNIX_EPOCH; + let created = create_test_download_token(now); + + let err = validate_download_token(&created.id, &created.token, now + OBJECT_ZIP_DOWNLOAD_TOKEN_TTL) + .expect_err("expired token should be rejected"); + + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[test] + fn validate_download_token_accepts_unexpired_matching_token() { + let now = OffsetDateTime::UNIX_EPOCH; + let created = create_test_download_token(now); + + let record = validate_download_token(&created.id, &created.token, now).expect("matching token should be accepted"); + + assert_eq!(record.id, created.id); + assert_eq!(record.principal, "alice"); + assert_eq!(record.request.bucket, "photos"); + assert_eq!(record.req_info.cred.as_ref().map(|cred| cred.access_key.as_str()), Some("alice")); + assert_eq!(record.req_info.cred.as_ref().map(|cred| cred.secret_key.as_str()), Some("")); + assert_eq!( + record + .auth_context + .headers + .get("user-agent") + .and_then(|value| value.to_str().ok()), + Some("object-zip-test") + ); + assert_eq!( + record.auth_context.remote_addr.map(|addr| addr.0), + Some("127.0.0.1:9000".parse().expect("test socket addr should parse")) + ); + assert_eq!( + record.auth_context.client_info.as_ref().map(|info| info.real_ip), + Some("203.0.113.10".parse().expect("test IP should parse")) + ); + } + + #[test] + fn validate_download_token_rejects_tampered_payload() { + let now = OffsetDateTime::UNIX_EPOCH; + let created = create_test_download_token(now); + let (nonce, ciphertext) = created + .token + .split_once('.') + .expect("token should contain nonce and ciphertext"); + let mut ciphertext = base64_decode_url_safe_no_pad(ciphertext.as_bytes()).expect("ciphertext should decode"); + let first = ciphertext.first_mut().expect("ciphertext should not be empty"); + *first ^= 0x01; + let tampered = format!("{}.{}", nonce, base64_encode_url_safe_no_pad(&ciphertext)); + + let err = validate_download_token(&created.id, &tampered, now).expect_err("tampered token should be rejected"); + + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[test] + fn object_zip_download_tokens_are_stateless_encrypted_payloads() { + let src = include_str!("object_zip_download.rs"); + + assert!( + src.contains("ObjectZipDownloadTokenPayload"), + "object ZIP download token should carry an authenticated payload" + ); + assert!( + src.contains("get_global_action_cred"), + "object ZIP download token should be verifiable by any node sharing global credentials" + ); + assert!( + src.contains("Aes256Gcm"), + "object ZIP download token payload should not expose captured authorization headers in clear text" + ); + assert!( + !src.contains(&format!("{}{}{}", "OnceLock<", "RwLock<", "HashMap")), + "object ZIP download tokens must not use process-local storage" + ); + } + + #[test] + fn create_download_token_does_not_expose_payload_plaintext() { + let created = create_test_download_token(OffsetDateTime::UNIX_EPOCH); + + assert!(!created.token.contains("alice")); + assert!(!created.token.contains("photos")); + assert!(!created.token.contains("object-zip-test")); + } + + #[test] + fn uri_with_prefix_query_preserves_existing_query_and_overrides_prefix() { + let uri: Uri = "/rustfs/admin/v3/object-zip-downloads?X-Amz-Algorithm=AWS4-HMAC-SHA256&prefix=old&marker=1" + .parse() + .expect("uri should parse"); + + let updated = uri_with_prefix_query(&uri, "new/prefix/").expect("uri should be rewritten"); + let query = updated.query().expect("query should exist"); + + assert!(query.contains("X-Amz-Algorithm=AWS4-HMAC-SHA256")); + assert!(query.contains("marker=1")); + assert!(query.contains("prefix=new%2Fprefix%2F")); + assert!(!query.contains("prefix=old")); + } + + #[test] + fn sanitize_content_disposition_filename_removes_unsafe_characters() { + assert_eq!(sanitize_content_disposition_filename("bad\"\\\r\nname.zip"), "bad____name.zip"); + assert_eq!(sanitize_content_disposition_filename(" "), "download.zip"); + } + + #[test] + fn zip_entry_name_strips_base_prefix() { + assert_eq!( + zip_entry_name(Some("2026/events/"), "2026/events/a.txt").expect("entry name should build"), + "a.txt" + ); + assert_eq!( + zip_entry_name(Some("2026/events/"), "2026/events/nested/b.txt").expect("entry name should build"), + "nested/b.txt" + ); + } + + #[test] + fn zip_entry_name_falls_back_to_full_key_when_outside_base_prefix() { + assert_eq!( + zip_entry_name(Some("2026/events/"), "other/a.txt").expect("entry name should build"), + "other/a.txt" + ); + assert_eq!( + zip_entry_name(None, "/absolute-ish.txt").expect("entry name should build"), + "absolute-ish.txt" + ); + } + + #[test] + fn zip_entry_name_rejects_empty_or_parent_directory_entries() { + assert_eq!( + zip_entry_name(Some("2026/events/"), "2026/events/") + .expect_err("empty relative entry should be rejected") + .code(), + &S3ErrorCode::InvalidRequest + ); + assert_eq!( + zip_entry_name(None, "../secret.txt") + .expect_err("parent directory entry should be rejected") + .code(), + &S3ErrorCode::InvalidRequest + ); + assert_eq!( + zip_entry_name(None, "..\\secret.txt") + .expect_err("windows parent directory entry should be rejected") + .code(), + &S3ErrorCode::InvalidRequest + ); + } + + #[test] + fn collect_explicit_zip_items_deduplicates_keys_preserving_order() { + let mut request = valid_request(); + request.prefix = Some("2026/events/".to_string()); + request.objects = vec![ + "2026/events/a.txt".to_string(), + "2026/events/a.txt".to_string(), + "2026/events/b.txt".to_string(), + ]; + + let items = collect_explicit_zip_items(&request).expect("items should collect"); + + assert_eq!( + items, + vec![ + ZipDownloadItem { + key: "2026/events/a.txt".to_string(), + entry_name: "a.txt".to_string() + }, + ZipDownloadItem { + key: "2026/events/b.txt".to_string(), + entry_name: "b.txt".to_string() + }, + ] + ); + } + + #[test] + fn push_zip_item_deduplicates_across_explicit_and_listed_keys() { + let mut request = valid_request(); + request.prefix = Some("2026/events/".to_string()); + request.objects = vec!["2026/events/a.txt".to_string()]; + + let mut seen = HashSet::new(); + let mut items = collect_explicit_zip_items(&request).expect("explicit items should collect"); + seen.extend(items.iter().map(|item| item.key.clone())); + + push_zip_item(&request, &mut seen, &mut items, "2026/events/a.txt".to_string()) + .expect("duplicate listed item should be ignored"); + push_zip_item(&request, &mut seen, &mut items, "2026/events/nested/b.txt".to_string()) + .expect("new listed item should be added"); + + assert_eq!( + items, + vec![ + ZipDownloadItem { + key: "2026/events/a.txt".to_string(), + entry_name: "a.txt".to_string() + }, + ZipDownloadItem { + key: "2026/events/nested/b.txt".to_string(), + entry_name: "nested/b.txt".to_string() + }, + ] + ); + } + + #[test] + fn listing_prefixes_uses_prefix_as_scope_only_when_no_explicit_selection() { + let request = valid_request(); + + assert_eq!(listing_prefixes(&request), vec!["2026/"]); + + let mut mixed = valid_request(); + mixed.objects = vec!["2026/a.txt".to_string()]; + mixed.prefixes = vec!["2026/nested/".to_string()]; + + assert_eq!(listing_prefixes(&mixed), vec!["2026/nested/"]); + } + + #[test] + fn normalized_listing_prefixes_removes_duplicates_and_nested_prefixes() { + let mut request = valid_request(); + request.prefix = None; + request.prefixes = vec![ + "2026/".to_string(), + "2026/reports/".to_string(), + "2026/".to_string(), + "2027/".to_string(), + ]; + + assert_eq!(normalized_listing_prefixes(&request), vec!["2026/".to_string(), "2027/".to_string()]); + } + + #[test] + fn download_handler_prepares_archive_before_streaming_response() { + let src = include_str!("object_zip_download.rs"); + let handler_block = extract_block_between_markers( + src, + "impl Operation for DownloadObjectZipHandler", + "fn sanitize_content_disposition_filename", + ); + + assert!( + handler_block.contains("prepare_zip_download_archive(&record).await?;"), + "download handler should fully prepare the ZIP archive before returning the ZIP response" + ); + assert!( + handler_block.contains("build_zip_stream_body(prepared)"), + "download handler should stream the prepared archive instead of rebuilding request scope during response" + ); + assert!( + !handler_block.contains("tokio::spawn"), + "download handler should not defer archive generation to a background task after returning 200" + ); + } + + #[test] + fn zip_download_preflight_preserves_storage_error_semantics() { + let src = include_str!("object_zip_download.rs"); + let preflight_block = extract_block_between_markers(src, "async fn preflight_zip_items", "fn storage_error_to_s3"); + let validation_block = + extract_block_between_markers(src, "async fn validate_zip_download_request", "async fn validate_zip_download_items"); + + assert!( + preflight_block.contains(".map_err(storage_error_to_s3)?;"), + "preflight object checks should preserve storage-layer S3 semantics" + ); + assert!( + validation_block.contains(".map_err(storage_error_to_s3)?;"), + "bucket and listing checks should preserve storage-layer S3 semantics" + ); + assert!( + preflight_block.contains(".get_object_info(&request.bucket, &item.key, &ObjectOptions::default())") + && !preflight_block.contains("failed to prepare object"), + "preflight object checks must not wrap expected missing object errors with a custom InternalError message" + ); + assert!( + validation_block.contains(".get_bucket_info(&record.request.bucket, &BucketOptions::default())") + && !validation_block.contains("failed to validate bucket") + && !validation_block.contains("failed to list prefix"), + "bucket checks must not wrap expected user-facing errors with custom InternalError messages" + ); + } + + #[test] + fn object_zip_download_handler_requires_admin_authorization_contract() { + let src = include_str!("object_zip_download.rs"); + let handler_block = + extract_block_between_markers(src, "impl Operation for CreateObjectZipDownloadHandler", "#[cfg(test)]"); + + assert!( + handler_block.contains("authenticate_object_zip_download_request(&mut req).await?;"), + "object ZIP download POST should authenticate the signed request" + ); + assert!( + handler_block.contains("authorize_object_zip_download_scope(&mut req, &request).await?;"), + "object ZIP download POST should authorize the requested object/list scope" + ); + assert!( + !handler_block.contains("ServerInfoAdminAction"), + "object ZIP download POST must not use broad server-info admin authorization for object reads" + ); + } + + #[test] + fn prefix_authorization_binds_body_prefix_to_list_bucket_query_conditions() { + let src = include_str!("object_zip_download.rs"); + let scope_block = extract_block_between_markers( + src, + "async fn authorize_object_zip_download_scope", + "async fn authorize_object_zip_download_s3_action", + ); + + assert!( + scope_block.contains("listing_prefixes(request)"), + "object ZIP download should authorize the actual prefixes used for listing" + ); + assert!( + !scope_block.contains("request.prefix.as_deref()"), + "base prefix alone should not force ListBucket authorization for explicit object selection" + ); + assert!( + scope_block.contains("authorize_object_zip_download_list_prefix"), + "prefix list authorization should bind body prefixes into s3:prefix conditions" + ); + } + + fn extract_block_between_markers<'a>(src: &'a str, start_marker: &str, end_marker: &str) -> &'a str { + let start = src + .find(start_marker) + .unwrap_or_else(|| panic!("Expected marker `{start_marker}` in source")); + let after_start = &src[start..]; + let end = after_start + .find(end_marker) + .unwrap_or_else(|| panic!("Expected end marker `{end_marker}` in source")); + &after_start[..end] + } + + fn create_test_download_token(now: OffsetDateTime) -> CreatedObjectZipDownloadToken { + ensure_test_signing_credentials(); + create_download_token("alice".to_string(), test_req_info(), test_auth_context(), valid_request(), now) + .expect("token should be created") + } + + fn ensure_test_signing_credentials() { + if get_global_action_cred().is_none() { + let _ = init_global_action_credentials( + Some("TESTROOTACCESSKEY".to_string()), + Some("TESTROOTSECRET1234567890".to_string()), + ); + } + } + + fn test_req_info() -> ReqInfo { + let mut credentials = rustfs_credentials::Credentials { + access_key: "alice".to_string(), + secret_key: "should-not-be-serialized".to_string(), + session_token: "should-not-be-serialized".to_string(), + ..Default::default() + }; + credentials.secret_key.clear(); + credentials.session_token.clear(); + ReqInfo { + cred: Some(credentials), + is_owner: false, + bucket: Some("photos".to_string()), + object: None, + version_id: None, + region: get_global_region(), + request_context: None, + } + } + + fn test_auth_context() -> ObjectZipDownloadAuthContext { + let mut headers = HeaderMap::new(); + headers.insert("user-agent", HeaderValue::from_static("object-zip-test")); + ObjectZipDownloadAuthContext { + headers, + remote_addr: Some(RemoteAddr("127.0.0.1:9000".parse().expect("test socket addr should parse"))), + client_info: Some(ClientInfo { + real_ip: "203.0.113.10".parse().expect("test IP should parse"), + forwarded_host: Some("console.example.test".to_string()), + forwarded_proto: Some("https".to_string()), + is_from_trusted_proxy: true, + proxy_ip: Some("192.0.2.10".parse().expect("test IP should parse")), + proxy_hops: 1, + validation_mode: ValidationMode::HopByHop, + warnings: Vec::new(), + }), + } + } +} diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index a454bc626..08a8baa5b 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -31,9 +31,9 @@ mod console_test; mod route_registration_test; use handlers::{ - audit, bucket_meta, config_admin, extensions, heal, health, kms, module_switch, oidc, plugins_catalog, plugins_instances, - pools, profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, - user, + audit, bucket_meta, config_admin, extensions, heal, health, kms, module_switch, object_zip_download, oidc, plugins_catalog, + plugins_instances, pools, profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, + table_catalog, tier, tls_debug, user, }; use router::{AdminOperation, S3Router}; use s3s::route::S3Route; @@ -71,6 +71,7 @@ fn register_admin_routes(r: &mut S3Router) -> std::io::Result<() audit::register_audit_target_route(r)?; module_switch::register_module_switch_route(r)?; extensions::register_extension_route(r)?; + object_zip_download::register_object_zip_download_route(r)?; plugins_catalog::register_plugin_catalog_route(r)?; plugins_instances::register_plugin_instance_route(r)?; diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index ec1429909..9a2fd3827 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -981,6 +981,16 @@ pub const DEFERRED_ADMIN_ROUTE_POLICIES: &[DeferredAdminRoutePolicy] = &[ "/rustfs/admin/v3/datausageinfo", DeferredRoutePolicyReason::MultipleActions, ), + deferred( + HttpMethod::Post, + "/rustfs/admin/v3/object-zip-downloads", + DeferredRoutePolicyReason::S3Action, + ), + deferred( + HttpMethod::Get, + "/rustfs/admin/v3/object-zip-downloads/{id}.zip", + DeferredRoutePolicyReason::CredentialOnly, + ), deferred(HttpMethod::Get, "/rustfs/admin/v3/metrics", DeferredRoutePolicyReason::CredentialOnly), deferred(HttpMethod::Get, "/rustfs/admin/v3/pools/list", DeferredRoutePolicyReason::MultipleActions), deferred( @@ -1220,6 +1230,16 @@ mod tests { DeferredRoutePolicyReason::MultipleActions, ); assert_deferred(HttpMethod::Get, "/rustfs/admin/v3/accountinfo", DeferredRoutePolicyReason::S3Action); + assert_deferred( + HttpMethod::Post, + "/rustfs/admin/v3/object-zip-downloads", + DeferredRoutePolicyReason::S3Action, + ); + assert_deferred( + HttpMethod::Get, + "/rustfs/admin/v3/object-zip-downloads/{id}.zip", + DeferredRoutePolicyReason::CredentialOnly, + ); } fn route_policy_inventory_keys() -> BTreeSet { diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 18fe08a3b..c477b185b 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -225,6 +225,12 @@ fn expected_admin_route_matrix() -> Vec { admin_route(Method::PUT, "/v3/module-switches"), admin_route(Method::GET, "/v4/extensions/catalog"), admin_route(Method::GET, "/v4/extensions/instances"), + admin_route(Method::POST, "/v3/object-zip-downloads"), + admin_route_sample( + Method::GET, + "/v3/object-zip-downloads/{id}.zip", + "/v3/object-zip-downloads/example-id.zip", + ), admin_route(Method::GET, "/v4/plugins/catalog"), admin_route(Method::GET, "/v4/plugins/instances"), admin_route_sample(Method::GET, "/v4/plugins/instances/{id}", "/v4/plugins/instances/example-id"), @@ -567,6 +573,7 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::PUT, &admin_path("/v3/module-switches")); assert_route(&router, Method::GET, &admin_path("/v4/extensions/catalog")); assert_route(&router, Method::GET, &admin_path("/v4/extensions/instances")); + assert_route(&router, Method::POST, &admin_path("/v3/object-zip-downloads")); assert_route(&router, Method::GET, &admin_path("/v4/plugins/catalog")); assert_route(&router, Method::GET, &admin_path("/v4/plugins/instances")); assert_route(&router, Method::GET, &admin_path("/v4/plugins/instances/example-id")); diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index dc5bc2ae6..b2211c6c0 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -99,6 +99,7 @@ use tracing::{error, warn}; use url::form_urlencoded; use uuid::Uuid; +pub const ADMIN_OBJECT_ZIP_DOWNLOADS_PATH: &str = "/v3/object-zip-downloads"; const LOG_COMPONENT_ADMIN: &str = "admin"; const LOG_SUBSYSTEM_OBJECT_LAMBDA: &str = "object_lambda"; const LOG_SUBSYSTEM_LIVE_EVENTS: &str = "live_events"; @@ -2298,6 +2299,17 @@ fn is_public_health_path(path: &str) -> bool { path == HEALTH_PREFIX || path == HEALTH_READY_PATH } +fn is_object_zip_download_token_path(method: &Method, uri: &Uri) -> bool { + if method != Method::GET { + return false; + } + + let path = canonicalize_admin_path(uri.path()); + path.starts_with(&format!("{ADMIN_PREFIX}{ADMIN_OBJECT_ZIP_DOWNLOADS_PATH}/")) + && path.ends_with(".zip") + && query_value_exact(uri, "token").is_some_and(|token| !token.is_empty()) +} + fn canonicalize_admin_path(path: &str) -> std::borrow::Cow<'_, str> { if is_admin_path(path) && let Some(suffix) = path.strip_prefix(MINIO_ADMIN_PREFIX) @@ -2441,6 +2453,12 @@ where return Ok(()); } + // Object ZIP downloads are browser-navigated with a short-lived token; + // the handler validates the token before returning any bytes. + if is_object_zip_download_token_path(&req.method, &req.uri) { + return Ok(()); + } + // Allow unauthenticated STS requests to POST / (AssumeRoleWithWebIdentity // doesn't use SigV4 — the JWT token in the request body is the authentication). // The handler dispatches on the Action parameter: AssumeRole will reject if @@ -3871,6 +3889,77 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::AccessDenied); } + #[tokio::test] + async fn check_access_allows_object_zip_download_token_navigation() { + let router: S3Router = S3Router::new(false); + let mut req = S3Request { + input: Body::from(String::new()), + method: Method::GET, + uri: "/rustfs/admin/v3/object-zip-downloads/example.zip?token=abc" + .parse() + .expect("uri should parse"), + headers: HeaderMap::new(), + extensions: http::Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + }; + + router + .check_access(&mut req) + .await + .expect("token download navigation should reach the handler"); + } + + #[tokio::test] + async fn check_access_rejects_object_zip_download_without_token() { + let router: S3Router = S3Router::new(false); + let mut req = S3Request { + input: Body::from(String::new()), + method: Method::GET, + uri: "/rustfs/admin/v3/object-zip-downloads/example.zip" + .parse() + .expect("uri should parse"), + headers: HeaderMap::new(), + extensions: http::Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + }; + + let err = router + .check_access(&mut req) + .await + .expect_err("token download without token must be denied before handler"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + + #[tokio::test] + async fn check_access_rejects_anonymous_object_zip_download_post() { + let router: S3Router = S3Router::new(false); + let mut req = S3Request { + input: Body::from(String::new()), + method: Method::POST, + uri: "/rustfs/admin/v3/object-zip-downloads?token=abc" + .parse() + .expect("uri should parse"), + headers: HeaderMap::new(), + extensions: http::Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + }; + + let err = router + .check_access(&mut req) + .await + .expect_err("token exception must not apply to POST"); + assert_eq!(err.code(), &S3ErrorCode::AccessDenied); + } + #[test] fn listen_notification_keepalive_plan_defaults_to_space_keepalive() { let uri: Uri = "/demo-bucket?events=s3:ObjectCreated:Put".parse().expect("uri should parse"); diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index f458e8595..d07ed7531 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -24,7 +24,7 @@ use crate::server::{ layer::{ BodylessStatusFixLayer, ConditionalCorsLayer, EmptyBodyContentLengthCompatLayer, HeadRequestBodyFixLayer, ObjectAttributesEtagFixLayer, PublicHealthEndpointLayer, RedirectLayer, RequestContextLayer, RequestLoggingLayer, - S3ErrorMessageCompatLayer, + S3ErrorMessageCompatLayer, redact_sensitive_uri_query, }, tls_material::{ TlsAcceptorHolder, TlsHandshakeFailureKind, build_acceptor_from_loaded, load_tls_material, spawn_reload_loop, @@ -970,7 +970,7 @@ fn process_connection( status_code = tracing::field::Empty, method = %request.method(), peer_addr = %peer_addr, - uri = %request.uri(), + uri = %redact_sensitive_uri_query(request.uri()), version = ?request.version(), user_agent = tracing::field::Empty, content_type = tracing::field::Empty, diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index c084cd39f..7380b407a 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -43,11 +43,14 @@ use std::task::{Context, Poll}; use std::time::Instant; use tower::{Layer, Service}; use tracing::{debug, error, info}; +use url::form_urlencoded; const HTTP_REQUEST_COMPLETED_EVENT: &str = "http_request_completed"; const HTTP_REQUEST_FAILED_EVENT: &str = "http_request_failed"; const LOG_COMPONENT_SERVER: &str = "server"; const LOG_SUBSYSTEM_HTTP: &str = "http"; +const REDACTED_QUERY_VALUE: &str = "redacted"; +const OBJECT_ZIP_DOWNLOADS_PATH: &str = "/v3/object-zip-downloads/"; /// A carrier that adapts [`HeaderMap`] for OpenTelemetry trace context propagation. struct HeaderMapCarrier<'a>(&'a HeaderMap); @@ -73,6 +76,55 @@ impl<'a> opentelemetry::propagation::Extractor for HeaderMapCarrier<'a> { } } +pub(crate) fn redact_sensitive_uri_query(uri: &http::Uri) -> String { + let path = uri.path(); + if !is_object_zip_download_path(path) { + return uri.to_string(); + } + + let Some(query) = uri.query() else { + return uri.to_string(); + }; + + let mut redacted_token = false; + let mut serializer = form_urlencoded::Serializer::new(String::new()); + for (key, value) in form_urlencoded::parse(query.as_bytes()) { + if key == "token" { + redacted_token = true; + serializer.append_pair(&key, REDACTED_QUERY_VALUE); + } else { + serializer.append_pair(&key, &value); + } + } + + if !redacted_token { + return uri.to_string(); + } + + let redacted_query = serializer.finish(); + let path_and_query = if redacted_query.is_empty() { + path.to_string() + } else { + format!("{path}?{redacted_query}") + }; + let mut parts = uri.clone().into_parts(); + match path_and_query.parse() { + Ok(path_and_query) => { + parts.path_and_query = Some(path_and_query); + http::Uri::from_parts(parts) + .map(|uri| uri.to_string()) + .unwrap_or_else(|_| uri.to_string()) + } + Err(_) => uri.to_string(), + } +} + +fn is_object_zip_download_path(path: &str) -> bool { + (path.starts_with(ADMIN_PREFIX) || path.starts_with(MINIO_ADMIN_PREFIX)) + && path.contains(OBJECT_ZIP_DOWNLOADS_PATH) + && path.ends_with(".zip") +} + /// Tower middleware layer that creates a canonical [`RequestContext`] from HTTP headers /// and injects it into `request.extensions()`. /// @@ -199,7 +251,7 @@ impl RequestLogContext { span_id: request_context.as_ref().and_then(|ctx| ctx.span_id.clone()), peer_addr, method: req.method().to_string(), - uri: req.uri().to_string(), + uri: redact_sensitive_uri_query(req.uri()), request_started_at: request_context, fallback_start: Instant::now(), } @@ -2667,6 +2719,41 @@ mod tests { assert_eq!(context.uri, "/bucket/object.txt"); } + #[test] + fn request_log_context_redacts_object_zip_download_tokens() { + let request = Request::builder() + .method(Method::GET) + .uri("/rustfs/admin/v3/object-zip-downloads/download-id.zip?token=secret-token&part=1") + .body(()) + .expect("request"); + + let context = RequestLogContext::from_request(&request); + + assert_eq!(context.uri, "/rustfs/admin/v3/object-zip-downloads/download-id.zip?token=redacted&part=1"); + assert!(!context.uri.contains("secret-token")); + } + + #[test] + fn request_log_context_redacts_object_zip_download_tokens_for_minio_admin_prefix() { + let request = Request::builder() + .method(Method::GET) + .uri("/minio/admin/v3/object-zip-downloads/download-id.zip?token=secret-token&part=1") + .body(()) + .expect("request"); + + let context = RequestLogContext::from_request(&request); + + assert_eq!(context.uri, "/minio/admin/v3/object-zip-downloads/download-id.zip?token=redacted&part=1"); + assert!(!context.uri.contains("secret-token")); + } + + #[test] + fn redact_sensitive_uri_query_preserves_non_zip_download_uris() { + let uri: http::Uri = "/rustfs/admin/v3/users?token=not-a-download-token".parse().expect("uri"); + + assert_eq!(redact_sensitive_uri_query(&uri), "/rustfs/admin/v3/users?token=not-a-download-token"); + } + #[test] fn request_logging_layer_emits_single_completion_event_with_standard_fields() { let writer = SharedWriter::default();