mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-29 08:27:06 +00:00
feat(s3): limit presigned PutObject content length (#6724)
feat(s3): limit presigned put content length
This commit is contained in:
@@ -0,0 +1,35 @@
|
|||||||
|
# Presigned PutObject size limit
|
||||||
|
|
||||||
|
RustFS V1 supports an optional, RustFS-specific capability on a SigV4
|
||||||
|
presigned `PutObject` URL:
|
||||||
|
|
||||||
|
```text
|
||||||
|
x-rustfs-max-content-length=<unsigned 64-bit integer>
|
||||||
|
```
|
||||||
|
|
||||||
|
The backend that creates the URL must add this query parameter to the request
|
||||||
|
URI before calculating the SigV4 presign. It is part of the canonical query;
|
||||||
|
adding, removing, or changing it after signing invalidates the signature. A
|
||||||
|
browser can then upload with a plain `PUT` and does not need a custom size
|
||||||
|
header.
|
||||||
|
|
||||||
|
RustFS validates the capability after SigV4 authentication and enforces it on
|
||||||
|
the decoded request body. A declared `Content-Length` above the limit is
|
||||||
|
rejected before storage. If the body produces more bytes than the limit while
|
||||||
|
streaming, RustFS returns `EntityTooLarge` and does not publish the object.
|
||||||
|
|
||||||
|
The V1 contract is deliberately narrow:
|
||||||
|
|
||||||
|
- The parameter is accepted only on a SigV4 presigned `PutObject` request.
|
||||||
|
- Duplicate, case-variant, malformed, negative, or overflowing values return
|
||||||
|
`InvalidRequest`.
|
||||||
|
- Requests without the parameter, including ordinary authenticated or
|
||||||
|
anonymous `PUT`, keep the existing behavior.
|
||||||
|
- The parameter on `CopyObject`, multipart, `GET`, `HEAD`, `DELETE`, bucket, or
|
||||||
|
other operations returns `InvalidRequest`.
|
||||||
|
- Unknown-length and SigV4 streaming-chunked uploads remain unsupported by the
|
||||||
|
existing PutObject admission contract and are not enabled by this feature.
|
||||||
|
|
||||||
|
This capability is per request; it is not a cumulative multipart-upload cap.
|
||||||
|
Multipart session limits are planned for V2 under a separate query/API
|
||||||
|
contract.
|
||||||
@@ -78,6 +78,7 @@ use crate::app::object_usecase::{
|
|||||||
use crate::app::runtime_sources::{
|
use crate::app::runtime_sources::{
|
||||||
AppContext, current_app_context, current_object_data_cache_for_context, current_object_store_handle_for_context,
|
AppContext, current_app_context, current_object_data_cache_for_context, current_object_store_handle_for_context,
|
||||||
};
|
};
|
||||||
|
use crate::auth::{VerifiedPresignedRequest, reject_presigned_put_max_content_length_for_other_operation};
|
||||||
use crate::capacity::record_capacity_write;
|
use crate::capacity::record_capacity_write;
|
||||||
use crate::error::ApiError;
|
use crate::error::ApiError;
|
||||||
use crate::table_catalog;
|
use crate::table_catalog;
|
||||||
@@ -397,6 +398,11 @@ impl DefaultMultipartUsecase {
|
|||||||
&self,
|
&self,
|
||||||
req: S3Request<AbortMultipartUploadInput>,
|
req: S3Request<AbortMultipartUploadInput>,
|
||||||
) -> S3Result<S3Response<AbortMultipartUploadOutput>> {
|
) -> S3Result<S3Response<AbortMultipartUploadOutput>> {
|
||||||
|
reject_presigned_put_max_content_length_for_other_operation(
|
||||||
|
&req.headers,
|
||||||
|
req.uri.query(),
|
||||||
|
req.extensions.get::<VerifiedPresignedRequest>().is_some(),
|
||||||
|
)?;
|
||||||
record_s3_op(S3Operation::AbortMultipartUpload);
|
record_s3_op(S3Operation::AbortMultipartUpload);
|
||||||
let mut opts = ObjectOptions::default();
|
let mut opts = ObjectOptions::default();
|
||||||
apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
|
apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
|
||||||
@@ -438,6 +444,11 @@ impl DefaultMultipartUsecase {
|
|||||||
&self,
|
&self,
|
||||||
req: S3Request<CompleteMultipartUploadInput>,
|
req: S3Request<CompleteMultipartUploadInput>,
|
||||||
) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
|
) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
|
||||||
|
reject_presigned_put_max_content_length_for_other_operation(
|
||||||
|
&req.headers,
|
||||||
|
req.uri.query(),
|
||||||
|
req.extensions.get::<VerifiedPresignedRequest>().is_some(),
|
||||||
|
)?;
|
||||||
let mut helper = OperationHelper::new(
|
let mut helper = OperationHelper::new(
|
||||||
&req,
|
&req,
|
||||||
EventName::ObjectCreatedCompleteMultipartUpload,
|
EventName::ObjectCreatedCompleteMultipartUpload,
|
||||||
@@ -741,6 +752,11 @@ impl DefaultMultipartUsecase {
|
|||||||
&self,
|
&self,
|
||||||
req: S3Request<CreateMultipartUploadInput>,
|
req: S3Request<CreateMultipartUploadInput>,
|
||||||
) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
|
) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
|
||||||
|
reject_presigned_put_max_content_length_for_other_operation(
|
||||||
|
&req.headers,
|
||||||
|
req.uri.query(),
|
||||||
|
req.extensions.get::<VerifiedPresignedRequest>().is_some(),
|
||||||
|
)?;
|
||||||
let helper =
|
let helper =
|
||||||
OperationHelper::new(&req, EventName::ObjectCreatedCreateMultipartUpload, S3Operation::CreateMultipartUpload)
|
OperationHelper::new(&req, EventName::ObjectCreatedCreateMultipartUpload, S3Operation::CreateMultipartUpload)
|
||||||
.suppress_event();
|
.suppress_event();
|
||||||
@@ -962,6 +978,11 @@ impl DefaultMultipartUsecase {
|
|||||||
#[instrument(level = "debug", skip(self, req))]
|
#[instrument(level = "debug", skip(self, req))]
|
||||||
#[hotpath::measure(impl_type = "MultipartUsecase")]
|
#[hotpath::measure(impl_type = "MultipartUsecase")]
|
||||||
pub async fn execute_upload_part(&self, req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
|
pub async fn execute_upload_part(&self, req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
|
||||||
|
reject_presigned_put_max_content_length_for_other_operation(
|
||||||
|
&req.headers,
|
||||||
|
req.uri.query(),
|
||||||
|
req.extensions.get::<VerifiedPresignedRequest>().is_some(),
|
||||||
|
)?;
|
||||||
let mut opts = ObjectOptions::default();
|
let mut opts = ObjectOptions::default();
|
||||||
apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
|
apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
|
||||||
let input = req.input;
|
let input = req.input;
|
||||||
@@ -1229,6 +1250,11 @@ impl DefaultMultipartUsecase {
|
|||||||
&self,
|
&self,
|
||||||
req: S3Request<ListMultipartUploadsInput>,
|
req: S3Request<ListMultipartUploadsInput>,
|
||||||
) -> S3Result<S3Response<ListMultipartUploadsOutput>> {
|
) -> S3Result<S3Response<ListMultipartUploadsOutput>> {
|
||||||
|
reject_presigned_put_max_content_length_for_other_operation(
|
||||||
|
&req.headers,
|
||||||
|
req.uri.query(),
|
||||||
|
req.extensions.get::<VerifiedPresignedRequest>().is_some(),
|
||||||
|
)?;
|
||||||
let mut opts = ObjectOptions::default();
|
let mut opts = ObjectOptions::default();
|
||||||
apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
|
apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
|
||||||
let ListMultipartUploadsInput {
|
let ListMultipartUploadsInput {
|
||||||
@@ -1276,6 +1302,11 @@ impl DefaultMultipartUsecase {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub async fn execute_list_parts(&self, req: S3Request<ListPartsInput>) -> S3Result<S3Response<ListPartsOutput>> {
|
pub async fn execute_list_parts(&self, req: S3Request<ListPartsInput>) -> S3Result<S3Response<ListPartsOutput>> {
|
||||||
|
reject_presigned_put_max_content_length_for_other_operation(
|
||||||
|
&req.headers,
|
||||||
|
req.uri.query(),
|
||||||
|
req.extensions.get::<VerifiedPresignedRequest>().is_some(),
|
||||||
|
)?;
|
||||||
let mut opts = ObjectOptions::default();
|
let mut opts = ObjectOptions::default();
|
||||||
apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
|
apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
|
||||||
let ListPartsInput {
|
let ListPartsInput {
|
||||||
@@ -1307,6 +1338,11 @@ impl DefaultMultipartUsecase {
|
|||||||
&self,
|
&self,
|
||||||
req: S3Request<UploadPartCopyInput>,
|
req: S3Request<UploadPartCopyInput>,
|
||||||
) -> S3Result<S3Response<UploadPartCopyOutput>> {
|
) -> S3Result<S3Response<UploadPartCopyOutput>> {
|
||||||
|
reject_presigned_put_max_content_length_for_other_operation(
|
||||||
|
&req.headers,
|
||||||
|
req.uri.query(),
|
||||||
|
req.extensions.get::<VerifiedPresignedRequest>().is_some(),
|
||||||
|
)?;
|
||||||
// Captured before `req.input` is destructured below.
|
// Captured before `req.input` is destructured below.
|
||||||
let copy_principal = SseKmsPrincipal::from_request(&req);
|
let copy_principal = SseKmsPrincipal::from_request(&req);
|
||||||
let source_bucket = match &req.input.copy_source {
|
let source_bucket = match &req.input.copy_source {
|
||||||
|
|||||||
@@ -16,6 +16,8 @@
|
|||||||
|
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
|
use crate::auth::{VerifiedPresignedRequest, reject_presigned_put_max_content_length_for_other_operation};
|
||||||
|
|
||||||
fn copy_namespace_lock_error(bucket: &str, object: &str, mode: &'static str, err: rustfs_lock::LockError) -> StorageError {
|
fn copy_namespace_lock_error(bucket: &str, object: &str, mode: &'static str, err: rustfs_lock::LockError) -> StorageError {
|
||||||
match err {
|
match err {
|
||||||
rustfs_lock::LockError::QuorumNotReached { required, achieved } => StorageError::NamespaceLockQuorumUnavailable {
|
rustfs_lock::LockError::QuorumNotReached { required, achieved } => StorageError::NamespaceLockQuorumUnavailable {
|
||||||
@@ -92,6 +94,11 @@ impl DefaultObjectUsecase {
|
|||||||
|
|
||||||
#[instrument(name = "execute_copy_object", level = "debug", skip(self, req))]
|
#[instrument(name = "execute_copy_object", level = "debug", skip(self, req))]
|
||||||
async fn execute_copy_object_inner(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> {
|
async fn execute_copy_object_inner(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> {
|
||||||
|
reject_presigned_put_max_content_length_for_other_operation(
|
||||||
|
&req.headers,
|
||||||
|
req.uri.query(),
|
||||||
|
req.extensions.get::<VerifiedPresignedRequest>().is_some(),
|
||||||
|
)?;
|
||||||
if let Some(context) = &self.context {
|
if let Some(context) = &self.context {
|
||||||
let _ = context.object_store();
|
let _ = context.object_store();
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -16,6 +16,9 @@
|
|||||||
|
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
|
use crate::auth::{RUSTFS_MAX_CONTENT_LENGTH_QUERY, VerifiedPresignedRequest, parse_presigned_put_max_content_length};
|
||||||
|
use crate::error::UploadLimitExceeded;
|
||||||
|
|
||||||
const DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES: i64 = 32 * 1024 * 1024;
|
const DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES: i64 = 32 * 1024 * 1024;
|
||||||
|
|
||||||
const ENV_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES";
|
const ENV_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES";
|
||||||
@@ -124,6 +127,58 @@ struct RequestBodyReadTimeout {
|
|||||||
timed_out: bool,
|
timed_out: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Enforces a maximum size on the decoded request entity while preserving the
|
||||||
|
/// streaming behavior of the underlying S3 body.
|
||||||
|
struct MaxContentLengthStream {
|
||||||
|
inner: StreamingBlob,
|
||||||
|
limit: u64,
|
||||||
|
received: u64,
|
||||||
|
exceeded: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Stream for MaxContentLengthStream {
|
||||||
|
type Item = Result<Bytes, StdError>;
|
||||||
|
|
||||||
|
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||||
|
let this = self.as_mut().get_mut();
|
||||||
|
if this.exceeded {
|
||||||
|
return Poll::Ready(None);
|
||||||
|
}
|
||||||
|
|
||||||
|
match Pin::new(&mut this.inner).poll_next(cx) {
|
||||||
|
Poll::Ready(Some(Ok(chunk))) => {
|
||||||
|
let chunk_len = u64::try_from(chunk.len()).unwrap_or(u64::MAX);
|
||||||
|
let exceeds = this.received > this.limit || chunk_len > this.limit.saturating_sub(this.received);
|
||||||
|
if exceeds {
|
||||||
|
this.exceeded = true;
|
||||||
|
return Poll::Ready(Some(Err(Box::new(UploadLimitExceeded { limit: this.limit }))));
|
||||||
|
}
|
||||||
|
|
||||||
|
this.received = this.received.saturating_add(chunk_len);
|
||||||
|
Poll::Ready(Some(Ok(chunk)))
|
||||||
|
}
|
||||||
|
other => other,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn size_hint(&self) -> (usize, Option<usize>) {
|
||||||
|
let remaining = usize::try_from(self.limit.saturating_sub(self.received)).unwrap_or(usize::MAX);
|
||||||
|
let (lower, upper) = self.inner.size_hint();
|
||||||
|
(lower.min(remaining), upper.map(|upper| upper.min(remaining)))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ByteStream for MaxContentLengthStream {
|
||||||
|
fn remaining_length(&self) -> RemainingLength {
|
||||||
|
let remaining = usize::try_from(self.limit.saturating_sub(self.received)).unwrap_or(usize::MAX);
|
||||||
|
let inner = self.inner.remaining_length();
|
||||||
|
inner
|
||||||
|
.exact()
|
||||||
|
.map(|exact| RemainingLength::new_exact(exact.min(remaining)))
|
||||||
|
.unwrap_or_else(RemainingLength::unknown)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
impl Stream for RequestBodyReadTimeout {
|
impl Stream for RequestBodyReadTimeout {
|
||||||
type Item = Result<Bytes, StdError>;
|
type Item = Result<Bytes, StdError>;
|
||||||
|
|
||||||
@@ -752,6 +807,11 @@ impl DefaultObjectUsecase {
|
|||||||
}
|
}
|
||||||
|
|
||||||
let (event_name, quota_operation, request_method_name) = Self::put_object_execution_context(&req);
|
let (event_name, quota_operation, request_method_name) = Self::put_object_execution_context(&req);
|
||||||
|
let max_content_length = parse_presigned_put_max_content_length(
|
||||||
|
&req.headers,
|
||||||
|
req.uri.query(),
|
||||||
|
req.extensions.get::<VerifiedPresignedRequest>().is_some(),
|
||||||
|
)?;
|
||||||
if req.extensions.get::<PostObjectRequestMarker>().is_some() && is_post_object_sse_kms_requested(&req.input, &req.headers)
|
if req.extensions.get::<PostObjectRequestMarker>().is_some() && is_post_object_sse_kms_requested(&req.input, &req.headers)
|
||||||
{
|
{
|
||||||
return Err(s3_error!(NotImplemented, "SSE-KMS is not supported for POST object uploads"));
|
return Err(s3_error!(NotImplemented, "SSE-KMS is not supported for POST object uploads"));
|
||||||
@@ -769,6 +829,12 @@ impl DefaultObjectUsecase {
|
|||||||
// member) instead of writing the replica.
|
// member) instead of writing the replica.
|
||||||
let inbound_replication_put = replication_request_authorized(&req)
|
let inbound_replication_put = replication_request_authorized(&req)
|
||||||
&& get_header(&req.headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true");
|
&& get_header(&req.headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true");
|
||||||
|
if max_content_length.is_some() && is_put_object_extract_requested(&req.headers) {
|
||||||
|
return Err(S3Error::with_message(
|
||||||
|
S3ErrorCode::InvalidRequest,
|
||||||
|
format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is not supported for archive extraction"),
|
||||||
|
));
|
||||||
|
}
|
||||||
if is_put_object_extract_requested(&req.headers) && !inbound_replication_put {
|
if is_put_object_extract_requested(&req.headers) && !inbound_replication_put {
|
||||||
return Box::pin(self.execute_put_object_extract(req)).await;
|
return Box::pin(self.execute_put_object_extract(req)).await;
|
||||||
}
|
}
|
||||||
@@ -842,9 +908,25 @@ impl DefaultObjectUsecase {
|
|||||||
guard_put_object_body_read_timeout(body, &bucket, &key, &request_id, content_length, put_object_body_read_timeout())
|
guard_put_object_body_read_timeout(body, &bucket, &key, &request_id, content_length, put_object_body_read_timeout())
|
||||||
};
|
};
|
||||||
|
|
||||||
|
let body = match max_content_length {
|
||||||
|
Some(limit) => StreamingBlob::new(MaxContentLengthStream {
|
||||||
|
inner: body,
|
||||||
|
limit,
|
||||||
|
received: 0,
|
||||||
|
exceeded: false,
|
||||||
|
}),
|
||||||
|
None => body,
|
||||||
|
};
|
||||||
|
|
||||||
// Resolve the authoritative decoded/plain object length (rejecting negative/unknown) before anything else consumes it.
|
// Resolve the authoritative decoded/plain object length (rejecting negative/unknown) before anything else consumes it.
|
||||||
let mut size = resolve_put_object_authoritative_size(&req.headers, content_length)?;
|
let mut size = resolve_put_object_authoritative_size(&req.headers, content_length)?;
|
||||||
|
|
||||||
|
if let Some(limit) = max_content_length
|
||||||
|
&& u64::try_from(size).is_ok_and(|size| size > limit)
|
||||||
|
{
|
||||||
|
return Err(S3Error::new(S3ErrorCode::EntityTooLarge));
|
||||||
|
}
|
||||||
|
|
||||||
// The app check preserves the existing S3 error contract; the storage
|
// The app check preserves the existing S3 error contract; the storage
|
||||||
// commit path reserves the exact net logical growth under its locks.
|
// commit path reserves the exact net logical growth under its locks.
|
||||||
let quota_check = self
|
let quota_check = self
|
||||||
@@ -1555,6 +1637,7 @@ pub(super) fn previous_current_size_from_backfill(backfill: Option<OldCurrentSiz
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use futures::StreamExt;
|
||||||
use http::{HeaderMap, HeaderName, HeaderValue, Method};
|
use http::{HeaderMap, HeaderName, HeaderValue, Method};
|
||||||
use s3s::dto::{DefaultRetention, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRule};
|
use s3s::dto::{DefaultRetention, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRule};
|
||||||
use std::pin::Pin;
|
use std::pin::Pin;
|
||||||
@@ -1594,6 +1677,39 @@ mod tests {
|
|||||||
.expect("cancelled owner must abort and reap the stalled storage task");
|
.expect("cancelled owner must abort and reap the stalled storage task");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn max_content_length_stream_rejects_the_first_chunk_over_limit() {
|
||||||
|
let inner = StreamingBlob::wrap(futures::stream::iter([
|
||||||
|
Ok::<Bytes, std::io::Error>(Bytes::from_static(b"1234")),
|
||||||
|
Ok::<Bytes, std::io::Error>(Bytes::from_static(b"56")),
|
||||||
|
]));
|
||||||
|
let mut limited = MaxContentLengthStream {
|
||||||
|
inner,
|
||||||
|
limit: 5,
|
||||||
|
received: 0,
|
||||||
|
exceeded: false,
|
||||||
|
};
|
||||||
|
|
||||||
|
assert_eq!(limited.next().await.unwrap().unwrap(), Bytes::from_static(b"1234"));
|
||||||
|
let error = limited.next().await.unwrap().unwrap_err();
|
||||||
|
assert!(error.downcast_ref::<UploadLimitExceeded>().is_some());
|
||||||
|
assert!(limited.next().await.is_none());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn max_content_length_stream_allows_exact_limit() {
|
||||||
|
let inner = StreamingBlob::from_bytes(Bytes::from_static(b"12345"));
|
||||||
|
let mut limited = MaxContentLengthStream {
|
||||||
|
inner,
|
||||||
|
limit: 5,
|
||||||
|
received: 0,
|
||||||
|
exceeded: false,
|
||||||
|
};
|
||||||
|
|
||||||
|
assert_eq!(limited.next().await.unwrap().unwrap(), Bytes::from_static(b"12345"));
|
||||||
|
assert!(limited.next().await.is_none());
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn put_request_user_metadata_cannot_suppress_bucket_default_retention() {
|
fn put_request_user_metadata_cannot_suppress_bucket_default_retention() {
|
||||||
let mut metadata =
|
let mut metadata =
|
||||||
|
|||||||
+168
-3
@@ -38,6 +38,7 @@ use subtle::ConstantTimeEq;
|
|||||||
use time::OffsetDateTime;
|
use time::OffsetDateTime;
|
||||||
use time::format_description::well_known::Rfc3339;
|
use time::format_description::well_known::Rfc3339;
|
||||||
use tracing::{debug, trace, warn};
|
use tracing::{debug, trace, warn};
|
||||||
|
use url::form_urlencoded;
|
||||||
|
|
||||||
const LOG_COMPONENT_AUTH: &str = "auth";
|
const LOG_COMPONENT_AUTH: &str = "auth";
|
||||||
const LOG_SUBSYSTEM_CREDENTIALS: &str = "credentials";
|
const LOG_SUBSYSTEM_CREDENTIALS: &str = "credentials";
|
||||||
@@ -50,6 +51,15 @@ const EVENT_KEYSTONE_CREDENTIALS_VALIDATED: &str = "keystone_credentials_validat
|
|||||||
const EVENT_KEYSTONE_CONTEXT_MISSING: &str = "keystone_context_missing";
|
const EVENT_KEYSTONE_CONTEXT_MISSING: &str = "keystone_context_missing";
|
||||||
const EVENT_SESSION_TOKEN_EXTRACTION: &str = "session_token_extraction";
|
const EVENT_SESSION_TOKEN_EXTRACTION: &str = "session_token_extraction";
|
||||||
|
|
||||||
|
/// RustFS-specific query capability for a single presigned PutObject request.
|
||||||
|
pub(crate) const RUSTFS_MAX_CONTENT_LENGTH_QUERY: &str = "x-rustfs-max-content-length";
|
||||||
|
|
||||||
|
/// Inserted by the S3 access boundary after the upstream verifier accepts a
|
||||||
|
/// request as SigV4 presigned. Downstream capability parsing must require this
|
||||||
|
/// marker instead of treating query syntax as proof of authentication.
|
||||||
|
#[derive(Debug, Clone, Copy)]
|
||||||
|
pub(crate) struct VerifiedPresignedRequest;
|
||||||
|
|
||||||
/// Performs constant-time string comparison to prevent timing attacks.
|
/// Performs constant-time string comparison to prevent timing attacks.
|
||||||
///
|
///
|
||||||
/// This function should be used when comparing sensitive values like passwords,
|
/// This function should be used when comparing sensitive values like passwords,
|
||||||
@@ -913,9 +923,11 @@ pub(crate) fn is_request_presigned_signature_v4_with_query(header: &HeaderMap, q
|
|||||||
if let Some(credential) = header.get(AMZ_CREDENTIAL) {
|
if let Some(credential) = header.get(AMZ_CREDENTIAL) {
|
||||||
return !credential.to_str().unwrap_or("").is_empty();
|
return !credential.to_str().unwrap_or("").is_empty();
|
||||||
}
|
}
|
||||||
query
|
query.is_some_and(|query| {
|
||||||
.and_then(|query| get_query_param(query, "x-amz-credential"))
|
form_urlencoded::parse(query.as_bytes())
|
||||||
.is_some_and(|credential| !credential.is_empty())
|
.find(|(name, _)| name.eq_ignore_ascii_case("x-amz-credential"))
|
||||||
|
.is_some_and(|(_, credential)| !credential.is_empty())
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Verify request has AWS PreSign Version '2'
|
/// Verify request has AWS PreSign Version '2'
|
||||||
@@ -1007,6 +1019,98 @@ pub fn get_query_param<'a>(query: &'a str, param_name: &str) -> Option<&'a str>
|
|||||||
None
|
None
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Parse the RustFS presigned PutObject size capability after authentication.
|
||||||
|
///
|
||||||
|
/// The query value is covered by SigV4 when it is present before presigning, but
|
||||||
|
/// the signature does not assign any semantics to the extension. Keep parsing
|
||||||
|
/// strict and only enable the capability for a verified SigV4 presigned request.
|
||||||
|
pub(crate) fn parse_presigned_put_max_content_length(
|
||||||
|
header: &HeaderMap,
|
||||||
|
query: Option<&str>,
|
||||||
|
verified_presigned: bool,
|
||||||
|
) -> S3Result<Option<u64>> {
|
||||||
|
let Some(query) = query else {
|
||||||
|
return Ok(None);
|
||||||
|
};
|
||||||
|
|
||||||
|
let mut value = None;
|
||||||
|
let mut decoded_query = Vec::new();
|
||||||
|
for (name, candidate) in form_urlencoded::parse(query.as_bytes()) {
|
||||||
|
decoded_query.push((name.to_string(), candidate.to_string()));
|
||||||
|
if name == RUSTFS_MAX_CONTENT_LENGTH_QUERY {
|
||||||
|
if value.is_some() {
|
||||||
|
return Err(S3Error::with_message(
|
||||||
|
S3ErrorCode::InvalidRequest,
|
||||||
|
format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} must appear exactly once"),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
value = Some(candidate.into_owned());
|
||||||
|
} else if name.eq_ignore_ascii_case(RUSTFS_MAX_CONTENT_LENGTH_QUERY) {
|
||||||
|
return Err(S3Error::with_message(
|
||||||
|
S3ErrorCode::InvalidRequest,
|
||||||
|
format!("query parameter name must be exactly {RUSTFS_MAX_CONTENT_LENGTH_QUERY}"),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let Some(value) = value else {
|
||||||
|
return Ok(None);
|
||||||
|
};
|
||||||
|
|
||||||
|
let query_value = |wanted: &str| {
|
||||||
|
decoded_query
|
||||||
|
.iter()
|
||||||
|
.find(|(name, _)| name.eq_ignore_ascii_case(wanted))
|
||||||
|
.map(|(_, value)| value.as_str())
|
||||||
|
};
|
||||||
|
let is_complete_sigv4_query = [
|
||||||
|
("x-amz-algorithm", "AWS4-HMAC-SHA256"),
|
||||||
|
("x-amz-date", ""),
|
||||||
|
("x-amz-expires", ""),
|
||||||
|
("x-amz-signedheaders", ""),
|
||||||
|
("x-amz-credential", ""),
|
||||||
|
("x-amz-signature", ""),
|
||||||
|
]
|
||||||
|
.into_iter()
|
||||||
|
.all(|(name, expected)| {
|
||||||
|
query_value(name).is_some_and(|value| !value.is_empty() && (expected.is_empty() || value == expected))
|
||||||
|
});
|
||||||
|
if !verified_presigned
|
||||||
|
|| !is_complete_sigv4_query
|
||||||
|
|| !matches!(get_request_auth_type_with_query(header, Some(query)), AuthType::Presigned)
|
||||||
|
{
|
||||||
|
return Err(S3Error::with_message(
|
||||||
|
S3ErrorCode::InvalidRequest,
|
||||||
|
format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} requires a SigV4 presigned request"),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
let limit = value.parse::<u64>().map_err(|_| {
|
||||||
|
S3Error::with_message(
|
||||||
|
S3ErrorCode::InvalidRequest,
|
||||||
|
format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} must be a non-negative 64-bit integer"),
|
||||||
|
)
|
||||||
|
})?;
|
||||||
|
|
||||||
|
Ok(Some(limit))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Reject the PutObject-only size capability when it appears on another
|
||||||
|
/// operation. Callers must invoke this after request authentication has run.
|
||||||
|
pub(crate) fn reject_presigned_put_max_content_length_for_other_operation(
|
||||||
|
header: &HeaderMap,
|
||||||
|
query: Option<&str>,
|
||||||
|
verified_presigned: bool,
|
||||||
|
) -> S3Result<()> {
|
||||||
|
if parse_presigned_put_max_content_length(header, query, verified_presigned)?.is_some() {
|
||||||
|
return Err(S3Error::with_message(
|
||||||
|
S3ErrorCode::InvalidRequest,
|
||||||
|
format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is only supported for presigned PutObject"),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
@@ -1651,6 +1755,67 @@ mod tests {
|
|||||||
assert_eq!(result, Some("value=with=equals"));
|
assert_eq!(result, Some("value=with=equals"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn presigned_put_max_content_length_requires_exactly_one_signed_query_value() {
|
||||||
|
let headers = HeaderMap::new();
|
||||||
|
let signed_prefix = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/20260827/us-east-1/s3/aws4_request&X-Amz-Signature=signature";
|
||||||
|
|
||||||
|
let query = format!("{signed_prefix}&x-rustfs-max-content-length=104857600");
|
||||||
|
assert_eq!(
|
||||||
|
parse_presigned_put_max_content_length(&headers, Some(&query), true).unwrap(),
|
||||||
|
Some(104_857_600)
|
||||||
|
);
|
||||||
|
|
||||||
|
let encoded_credential = query.replacen("X-Amz-Credential", "X%2DAmz-Credential", 1);
|
||||||
|
assert_eq!(
|
||||||
|
parse_presigned_put_max_content_length(&headers, Some(&encoded_credential), true).unwrap(),
|
||||||
|
Some(104_857_600)
|
||||||
|
);
|
||||||
|
|
||||||
|
let duplicate = format!("{query}&x-rustfs-max-content-length=1");
|
||||||
|
assert_eq!(
|
||||||
|
parse_presigned_put_max_content_length(&headers, Some(&duplicate), true)
|
||||||
|
.unwrap_err()
|
||||||
|
.code(),
|
||||||
|
&S3ErrorCode::InvalidRequest
|
||||||
|
);
|
||||||
|
|
||||||
|
let wrong_case = format!("{signed_prefix}&X-RustFS-Max-Content-Length=1");
|
||||||
|
assert_eq!(
|
||||||
|
parse_presigned_put_max_content_length(&headers, Some(&wrong_case), true)
|
||||||
|
.unwrap_err()
|
||||||
|
.code(),
|
||||||
|
&S3ErrorCode::InvalidRequest
|
||||||
|
);
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
reject_presigned_put_max_content_length_for_other_operation(&headers, Some(&query), true)
|
||||||
|
.unwrap_err()
|
||||||
|
.code(),
|
||||||
|
&S3ErrorCode::InvalidRequest
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn presigned_put_max_content_length_rejects_unsigned_or_invalid_values() {
|
||||||
|
let headers = HeaderMap::new();
|
||||||
|
let forged = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/credential&X-Amz-Signature=fake&x-rustfs-max-content-length=1";
|
||||||
|
assert_eq!(
|
||||||
|
parse_presigned_put_max_content_length(&headers, Some(forged), false)
|
||||||
|
.unwrap_err()
|
||||||
|
.code(),
|
||||||
|
&S3ErrorCode::InvalidRequest
|
||||||
|
);
|
||||||
|
for query in [
|
||||||
|
"x-rustfs-max-content-length=1",
|
||||||
|
"X-Amz-Credential=test/credential&x-rustfs-max-content-length=-1",
|
||||||
|
"X-Amz-Credential=test/credential&x-rustfs-max-content-length=18446744073709551616",
|
||||||
|
] {
|
||||||
|
let error = parse_presigned_put_max_content_length(&headers, Some(query), true).unwrap_err();
|
||||||
|
assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_credentials_is_expired() {
|
fn test_credentials_is_expired() {
|
||||||
let mut cred = create_test_credentials();
|
let mut cred = create_test_credentials();
|
||||||
|
|||||||
@@ -17,6 +17,23 @@ use crate::storage_api::error::{QuotaError, StorageError};
|
|||||||
use rustfs_kms::KmsUnavailableError;
|
use rustfs_kms::KmsUnavailableError;
|
||||||
use s3s::{S3Error, S3ErrorCode};
|
use s3s::{S3Error, S3ErrorCode};
|
||||||
|
|
||||||
|
/// Marks a request body that exceeded a presigned upload size capability.
|
||||||
|
///
|
||||||
|
/// This marker must survive the body-reader and storage layers so the client
|
||||||
|
/// receives `EntityTooLarge` instead of a generic internal error.
|
||||||
|
#[derive(Debug, Clone, Copy)]
|
||||||
|
pub(crate) struct UploadLimitExceeded {
|
||||||
|
pub limit: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl std::fmt::Display for UploadLimitExceeded {
|
||||||
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
|
write!(f, "upload exceeds the maximum content length of {} bytes", self.limit)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl std::error::Error for UploadLimitExceeded {}
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub struct ApiError {
|
pub struct ApiError {
|
||||||
pub code: S3ErrorCode,
|
pub code: S3ErrorCode,
|
||||||
@@ -274,6 +291,17 @@ impl From<StorageError> for ApiError {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if let StorageError::Io(ref io_err) = err
|
||||||
|
&& let Some(inner) = io_err.get_ref()
|
||||||
|
&& error_chain_has_type::<UploadLimitExceeded>(inner)
|
||||||
|
{
|
||||||
|
return ApiError {
|
||||||
|
code: S3ErrorCode::EntityTooLarge,
|
||||||
|
message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
|
||||||
|
source: Some(Box::new(err)),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
if let StorageError::Io(ref io_err) = err
|
if let StorageError::Io(ref io_err) = err
|
||||||
&& io_err
|
&& io_err
|
||||||
.get_ref()
|
.get_ref()
|
||||||
@@ -399,6 +427,13 @@ impl From<std::io::Error> for ApiError {
|
|||||||
source: Some(Box::new(err)),
|
source: Some(Box::new(err)),
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
if error_chain_has_type::<UploadLimitExceeded>(inner) {
|
||||||
|
return ApiError {
|
||||||
|
code: S3ErrorCode::EntityTooLarge,
|
||||||
|
message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
|
||||||
|
source: Some(Box::new(err)),
|
||||||
|
};
|
||||||
|
}
|
||||||
if error_chain_has_type::<rustfs_rio::IncompleteBody>(inner) {
|
if error_chain_has_type::<rustfs_rio::IncompleteBody>(inner) {
|
||||||
return ApiError {
|
return ApiError {
|
||||||
code: S3ErrorCode::IncompleteBody,
|
code: S3ErrorCode::IncompleteBody,
|
||||||
@@ -815,6 +850,16 @@ mod tests {
|
|||||||
assert!(api_error.source.is_some());
|
assert!(api_error.source.is_some());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn upload_limit_marker_maps_to_entity_too_large_across_io_boundaries() {
|
||||||
|
let direct: ApiError = IoError::other(UploadLimitExceeded { limit: 5 }).into();
|
||||||
|
assert_eq!(direct.code, S3ErrorCode::EntityTooLarge);
|
||||||
|
|
||||||
|
let storage: ApiError = StorageError::Io(IoError::other(IoError::other(UploadLimitExceeded { limit: 5 }))).into();
|
||||||
|
assert_eq!(storage.code, S3ErrorCode::EntityTooLarge);
|
||||||
|
assert_eq!(storage.message, ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_api_error_from_storage_io_copy_object_terminal_error_stays_internal() {
|
fn test_api_error_from_storage_io_copy_object_terminal_error_stays_internal() {
|
||||||
let io_error = IoError::other(StorageError::FileCorrupt);
|
let io_error = IoError::other(StorageError::FileCorrupt);
|
||||||
|
|||||||
@@ -16,8 +16,9 @@ use super::ObjectOptions;
|
|||||||
use super::ecfs::FS;
|
use super::ecfs::FS;
|
||||||
use super::{ECStore, PolicySys, ReplicationStatusType, StorageError, get_lock_acquire_timeout, is_err_bucket_not_found};
|
use super::{ECStore, PolicySys, ReplicationStatusType, StorageError, get_lock_acquire_timeout, is_err_bucket_not_found};
|
||||||
use crate::auth::{
|
use crate::auth::{
|
||||||
check_key_valid_with_context, get_condition_values_with_client_info, get_condition_values_with_query_and_client_info,
|
AuthType, RUSTFS_MAX_CONTENT_LENGTH_QUERY, VerifiedPresignedRequest, check_key_valid_with_context,
|
||||||
get_session_token,
|
get_condition_values_with_client_info, get_condition_values_with_query_and_client_info, get_request_auth_type_with_query,
|
||||||
|
get_session_token, parse_presigned_put_max_content_length,
|
||||||
};
|
};
|
||||||
use crate::error::ApiError;
|
use crate::error::ApiError;
|
||||||
use crate::license::license_check;
|
use crate::license::license_check;
|
||||||
@@ -1770,9 +1771,28 @@ impl S3Access for FS {
|
|||||||
|
|
||||||
// Publish this server's context slot so downstream data-plane handlers
|
// Publish this server's context slot so downstream data-plane handlers
|
||||||
// resolve the same store (backlog#1052 S6).
|
// resolve the same store (backlog#1052 S6).
|
||||||
let ext = cx.extensions_mut();
|
let verified_presigned = matches!(get_request_auth_type_with_query(cx.headers(), cx.uri().query()), AuthType::Presigned);
|
||||||
ext.insert(self.server_ctx().clone());
|
{
|
||||||
ext.insert(req_info);
|
let ext = cx.extensions_mut();
|
||||||
|
ext.insert(self.server_ctx().clone());
|
||||||
|
ext.insert(req_info);
|
||||||
|
if verified_presigned {
|
||||||
|
ext.insert(VerifiedPresignedRequest);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The size capability is intentionally scoped to the single-object
|
||||||
|
// PutObject operation. Validate this at the operation-aware access
|
||||||
|
// boundary so unsupported GET/HEAD/DELETE/bucket routes cannot silently
|
||||||
|
// ignore a signed capability query.
|
||||||
|
if parse_presigned_put_max_content_length(cx.headers(), cx.uri().query(), verified_presigned)?.is_some()
|
||||||
|
&& cx.s3_op().name() != "PutObject"
|
||||||
|
{
|
||||||
|
return Err(S3Error::with_message(
|
||||||
|
S3ErrorCode::InvalidRequest,
|
||||||
|
format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is only supported for presigned PutObject"),
|
||||||
|
));
|
||||||
|
}
|
||||||
license_check().map_err(|er| match er.kind() {
|
license_check().map_err(|er| match er.kind() {
|
||||||
std::io::ErrorKind::PermissionDenied => s3_error!(AccessDenied, "{er}"),
|
std::io::ErrorKind::PermissionDenied => s3_error!(AccessDenied, "{er}"),
|
||||||
_ => {
|
_ => {
|
||||||
|
|||||||
Reference in New Issue
Block a user