mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-06 13:27:43 +00:00
安正超
982 lines
39 KiB
Rust
982 lines
39 KiB
Rust
// 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::config::{RustFSBufferConfig, WorkloadProfile, get_global_buffer_config, is_buffer_profile_enabled};
|
|
use crate::error::ApiError;
|
|
use crate::server::cors;
|
|
use crate::storage::ecfs::ListObjectUnorderedQuery;
|
|
use http::header::{IF_MATCH, IF_MODIFIED_SINCE, IF_NONE_MATCH, IF_UNMODIFIED_SINCE};
|
|
use http::{HeaderMap, HeaderValue, StatusCode};
|
|
use metrics::counter;
|
|
use rustfs_ecstore::bucket::metadata_sys;
|
|
use rustfs_ecstore::bucket::metadata_sys::get_replication_config;
|
|
use rustfs_ecstore::bucket::object_lock::objectlock_sys;
|
|
use rustfs_ecstore::bucket::replication::ReplicationConfigurationExt;
|
|
use rustfs_ecstore::error::StorageError;
|
|
use rustfs_ecstore::new_object_layer_fn;
|
|
use rustfs_ecstore::store_api::{BucketOperations, ObjectInfo, ObjectToDelete};
|
|
use rustfs_storage_api::BucketOptions;
|
|
use rustfs_targets::EventName;
|
|
use rustfs_targets::arn::{TargetID, TargetIDError};
|
|
use rustfs_utils::http::{
|
|
AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
|
|
SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, contains_key_str, insert_str, remove_str,
|
|
};
|
|
use s3s::dto::{
|
|
Delimiter, LambdaFunctionConfiguration, NotificationConfigurationFilter, ObjectLockConfiguration, ObjectLockEnabled,
|
|
ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockRetention, ObjectLockRetentionMode, QueueConfiguration,
|
|
TopicConfiguration,
|
|
};
|
|
use s3s::{S3Error, S3ErrorCode, S3Response, S3Result};
|
|
use serde_urlencoded::from_bytes;
|
|
use std::collections::HashMap;
|
|
use std::ops::Add;
|
|
use std::sync::Arc;
|
|
use time::OffsetDateTime;
|
|
use time::format_description::well_known::Rfc3339;
|
|
use time::{format_description::FormatItem, macros::format_description};
|
|
use tracing::{debug, warn};
|
|
|
|
pub const RFC1123: &[FormatItem<'_>] =
|
|
format_description!("[weekday repr:short], [day] [month repr:short] [year] [hour]:[minute]:[second] GMT");
|
|
|
|
fn format_object_lock_timestamp(timestamp: OffsetDateTime) -> String {
|
|
timestamp.format(&Rfc3339).unwrap_or_default()
|
|
}
|
|
|
|
fn has_object_lock_retention_metadata(metadata: &HashMap<String, String>) -> bool {
|
|
metadata.contains_key(AMZ_OBJECT_LOCK_MODE_LOWER) || metadata.contains_key(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER)
|
|
}
|
|
|
|
pub(crate) fn remove_object_lock_retention_metadata(metadata: &mut HashMap<String, String>) -> bool {
|
|
let removed_mode = metadata.remove(AMZ_OBJECT_LOCK_MODE_LOWER).is_some();
|
|
let removed_retain_until_date = metadata.remove(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER).is_some();
|
|
let removed_timestamp = contains_key_str(metadata, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP);
|
|
remove_str(metadata, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP);
|
|
|
|
removed_mode || removed_retain_until_date || removed_timestamp
|
|
}
|
|
|
|
fn remove_object_lock_legal_hold_metadata(metadata: &mut HashMap<String, String>) -> bool {
|
|
let removed_legal_hold = metadata.remove(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).is_some();
|
|
let removed_timestamp = contains_key_str(metadata, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP);
|
|
remove_str(metadata, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP);
|
|
|
|
removed_legal_hold || removed_timestamp
|
|
}
|
|
|
|
pub(crate) fn remove_object_lock_metadata_for_copy(metadata: &mut HashMap<String, String>) -> bool {
|
|
let removed_retention = remove_object_lock_retention_metadata(metadata);
|
|
let removed_legal_hold = remove_object_lock_legal_hold_metadata(metadata);
|
|
|
|
removed_retention || removed_legal_hold
|
|
}
|
|
|
|
/// Apply bucket default Object Lock retention to object metadata if no explicit retention is set.
|
|
pub(crate) fn apply_lock_retention(object_lock_config: Option<ObjectLockConfiguration>, metadata: &mut HashMap<String, String>) {
|
|
if has_object_lock_retention_metadata(metadata) {
|
|
return;
|
|
}
|
|
|
|
let Some(config) = object_lock_config else { return };
|
|
|
|
if config.object_lock_enabled.as_ref().map(|e| e.as_str()) != Some(ObjectLockEnabled::ENABLED) {
|
|
return;
|
|
}
|
|
|
|
let Some(default_retention) = config.rule.and_then(|r| r.default_retention) else { return };
|
|
let Some(mode) = default_retention.mode else { return };
|
|
|
|
let now = OffsetDateTime::now_utc();
|
|
let retain_until = match (default_retention.days, default_retention.years) {
|
|
(Some(days), _) => now.saturating_add(time::Duration::days(days as i64)),
|
|
(None, Some(years)) => objectlock_sys::add_years(now, years),
|
|
_ => return,
|
|
};
|
|
|
|
if let Ok(date_str) = retain_until.format(&Rfc3339) {
|
|
metadata.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), mode.as_str().to_string());
|
|
metadata.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), date_str);
|
|
insert_str(metadata, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, format_object_lock_timestamp(now));
|
|
}
|
|
}
|
|
|
|
pub(crate) fn apply_default_lock_retention_metadata(
|
|
object_lock_configuration: Option<ObjectLockConfiguration>,
|
|
metadata: &mut HashMap<String, String>,
|
|
) -> bool {
|
|
if has_object_lock_retention_metadata(metadata) {
|
|
return false;
|
|
}
|
|
|
|
let mut default_retention_metadata = HashMap::new();
|
|
apply_lock_retention(object_lock_configuration, &mut default_retention_metadata);
|
|
if default_retention_metadata.is_empty() {
|
|
return false;
|
|
}
|
|
|
|
metadata.extend(default_retention_metadata);
|
|
true
|
|
}
|
|
|
|
pub(crate) async fn apply_bucket_default_lock_retention(
|
|
bucket: &str,
|
|
metadata: &mut HashMap<String, String>,
|
|
has_explicit_retention: bool,
|
|
) -> S3Result<()> {
|
|
if has_explicit_retention {
|
|
return Ok(());
|
|
}
|
|
|
|
if has_object_lock_retention_metadata(metadata) {
|
|
return Ok(());
|
|
}
|
|
|
|
let object_lock_configuration = match metadata_sys::get_object_lock_config(bucket).await {
|
|
Ok((cfg, _created)) => Some(cfg),
|
|
Err(err) => {
|
|
if err == StorageError::ConfigNotFound {
|
|
None
|
|
} else {
|
|
warn!("get_object_lock_config err {:?}", err);
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::InternalError,
|
|
"Failed to load Object Lock configuration".to_string(),
|
|
));
|
|
}
|
|
}
|
|
};
|
|
|
|
apply_default_lock_retention_metadata(object_lock_configuration, metadata);
|
|
Ok(())
|
|
}
|
|
|
|
/// Calculate adaptive buffer size with workload profile support.
|
|
///
|
|
/// This enhanced version supports different workload profiles for optimal performance
|
|
/// across various use cases (AI/ML, web workloads, secure storage, etc.).
|
|
///
|
|
/// # Arguments
|
|
/// * `file_size` - The size of the file in bytes, or -1 if unknown
|
|
/// * `profile` - Optional workload profile. If None, uses auto-detection or GeneralPurpose
|
|
///
|
|
/// # Returns
|
|
/// Optimal buffer size in bytes based on the workload profile and file size
|
|
///
|
|
/// # Examples
|
|
/// ```ignore
|
|
/// // Use general purpose profile (default)
|
|
/// let buffer_size = get_adaptive_buffer_size_with_profile(1024 * 1024, None);
|
|
///
|
|
/// // Use AI training profile for large model files
|
|
/// let buffer_size = get_adaptive_buffer_size_with_profile(
|
|
/// 500 * 1024 * 1024,
|
|
/// Some(WorkloadProfile::AiTraining)
|
|
/// );
|
|
///
|
|
/// // Use secure storage profile for compliance scenarios
|
|
/// let buffer_size = get_adaptive_buffer_size_with_profile(
|
|
/// 10 * 1024 * 1024,
|
|
/// Some(WorkloadProfile::SecureStorage)
|
|
/// );
|
|
///
|
|
/// // Use custom profile for specialized requirements
|
|
/// let custom_profile = WorkloadProfile::custom(
|
|
/// 32 * 1024, // min_size: 32KB
|
|
/// 2 * 1024 * 1024, // max_size: 2MB
|
|
/// 256 * 1024, // default_unknown: 256KB
|
|
/// vec![
|
|
/// (1024 * 1024, 64 * 1024), // < 1MB: 64KB
|
|
/// (10 * 1024 * 1024, 128 * 1024), // 1MB-10MB: 128KB
|
|
/// (i64::MAX, 512 * 1024), // >= 10MB: 512KB
|
|
/// ],
|
|
/// );
|
|
/// let buffer_size = get_adaptive_buffer_size_with_profile(
|
|
/// 5 * 1024 * 1024,
|
|
/// Some(custom_profile)
|
|
/// );
|
|
/// ```
|
|
///
|
|
#[allow(dead_code)]
|
|
pub(crate) fn get_adaptive_buffer_size_with_profile(file_size: i64, profile: Option<WorkloadProfile>) -> usize {
|
|
let config = match profile {
|
|
Some(p) => RustFSBufferConfig::new(p),
|
|
None => {
|
|
// Auto-detect OS environment or use general purpose
|
|
RustFSBufferConfig::with_auto_detect()
|
|
}
|
|
};
|
|
|
|
config.get_buffer_size(file_size)
|
|
}
|
|
|
|
/// Get adaptive buffer size using global workload profile configuration.
|
|
///
|
|
/// This is the primary buffer sizing function that uses the workload profile
|
|
/// system configured at startup to provide optimal buffer sizes for different scenarios.
|
|
///
|
|
/// The function automatically selects buffer sizes based on:
|
|
/// - Configured workload profile (default: GeneralPurpose)
|
|
/// - File size characteristics
|
|
/// - Optional performance metrics collection
|
|
///
|
|
/// # Arguments
|
|
/// * `file_size` - The size of the file in bytes, or -1 if unknown
|
|
///
|
|
/// # Returns
|
|
/// Optimal buffer size in bytes based on the configured workload profile
|
|
///
|
|
/// # Performance Metrics
|
|
/// When compiled with the `metrics` feature flag, this function tracks:
|
|
/// - Buffer size distribution
|
|
/// - Selection frequency
|
|
/// - Buffer-to-file size ratios
|
|
///
|
|
/// # Examples
|
|
/// ```ignore
|
|
/// // Uses configured profile (default: GeneralPurpose)
|
|
/// let buffer_size = get_buffer_size_opt_in(file_size);
|
|
/// ```
|
|
pub(crate) fn get_buffer_size_opt_in(file_size: i64) -> usize {
|
|
let buffer_size = if is_buffer_profile_enabled() {
|
|
// Use globally configured workload profile (enabled by default in Phase 3)
|
|
let config = get_global_buffer_config();
|
|
config.get_buffer_size(file_size)
|
|
} else {
|
|
// Opt-out mode: Use GeneralPurpose profile for consistent behavior
|
|
let config = RustFSBufferConfig::new(WorkloadProfile::GeneralPurpose);
|
|
config.get_buffer_size(file_size)
|
|
};
|
|
|
|
// Optional performance metrics collection for monitoring and optimization
|
|
{
|
|
use metrics::histogram;
|
|
histogram!("rustfs_buffer_size_bytes").record(buffer_size as f64);
|
|
counter!("rustfs_buffer_size_selections_total").increment(1);
|
|
|
|
if file_size > 0 {
|
|
let ratio = buffer_size as f64 / file_size as f64;
|
|
histogram!("rustfs_buffer_to_file_ratio").record(ratio);
|
|
}
|
|
}
|
|
|
|
buffer_size
|
|
}
|
|
|
|
/// Validate object key for control characters and log special characters
|
|
///
|
|
/// This function:
|
|
/// 1. Rejects keys containing control characters (null bytes, newlines, carriage returns)
|
|
/// 2. Logs debug information for keys containing spaces, plus signs, or percent signs
|
|
///
|
|
/// The s3s library handles URL decoding, so keys are already decoded when they reach this function.
|
|
/// This validation ensures that invalid characters that could cause issues are rejected early.
|
|
pub(crate) fn validate_object_key(key: &str, operation: &str) -> S3Result<()> {
|
|
// Validate object key doesn't contain control characters
|
|
if key.contains(['\0', '\n', '\r']) {
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::InvalidArgument,
|
|
format!("Object key contains invalid control characters: {key:?}"),
|
|
));
|
|
}
|
|
|
|
// Log debug info for keys with special characters to help diagnose encoding issues
|
|
if key.contains([' ', '+', '%']) {
|
|
debug!("{} object with special characters in key: {:?}", operation, key);
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Validate that 'allow-unordered' parameter is not used with a delimiter
|
|
///
|
|
/// This function:
|
|
/// 1. Checks if a delimiter is specified in the ListObjects request
|
|
/// 2. Parses the query string to check for the 'allow-unordered' parameter
|
|
/// 3. Rejects the request if both 'delimiter' and 'allow-unordered=true' are present
|
|
///
|
|
/// According to S3 compatibility requirements, unordered listing cannot be combined with
|
|
/// hierarchical directory traversal (delimited listing). This validation ensures
|
|
/// conflicting parameters are caught before processing the request.
|
|
pub(crate) fn validate_list_object_unordered_with_delimiter(
|
|
delimiter: Option<&Delimiter>,
|
|
query_string: Option<&str>,
|
|
) -> S3Result<()> {
|
|
if delimiter.is_none() {
|
|
return Ok(());
|
|
}
|
|
|
|
let Some(query) = query_string else {
|
|
return Ok(());
|
|
};
|
|
|
|
if let Ok(params) = from_bytes::<ListObjectUnorderedQuery>(query.as_bytes())
|
|
&& params.allow_unordered.as_deref() == Some("true")
|
|
{
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::InvalidArgument,
|
|
"The allow-unordered parameter cannot be used when delimiter is specified.".to_string(),
|
|
));
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
pub(crate) fn parse_object_lock_retention(retention: Option<ObjectLockRetention>) -> S3Result<HashMap<String, String>> {
|
|
let mut eval_metadata = HashMap::new();
|
|
|
|
if let Some(v) = retention {
|
|
let mode = match v.mode {
|
|
Some(mode) => match mode.as_str() {
|
|
ObjectLockRetentionMode::COMPLIANCE | ObjectLockRetentionMode::GOVERNANCE => mode.as_str().to_string(),
|
|
_ => {
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::MalformedXML,
|
|
"The XML you provided was not well-formed or did not validate against our published schema".to_string(),
|
|
));
|
|
}
|
|
},
|
|
None => String::default(),
|
|
};
|
|
|
|
let now = OffsetDateTime::now_utc();
|
|
// Validate retain_until_date is in the future (S3 requirement)
|
|
// Only validate when both mode and date are provided (not clearing retention)
|
|
let retain_until_date = if let Some(date) = v.retain_until_date {
|
|
let retain_until = OffsetDateTime::from(date);
|
|
// Only validate future date when mode is set (not clearing retention)
|
|
if !mode.is_empty() && retain_until <= now {
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::InvalidArgument,
|
|
"The retain until date must be in the future".to_string(),
|
|
));
|
|
}
|
|
retain_until.format(&Rfc3339).unwrap()
|
|
} else {
|
|
String::default()
|
|
};
|
|
|
|
// This is intentional behavior. Empty string represents "retention cleared" which is different from "retention never set". Consistent with minio
|
|
eval_metadata.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), mode);
|
|
eval_metadata.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), retain_until_date);
|
|
insert_str(
|
|
&mut eval_metadata,
|
|
SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP,
|
|
format_object_lock_timestamp(now),
|
|
);
|
|
}
|
|
Ok(eval_metadata)
|
|
}
|
|
|
|
pub(crate) fn parse_object_lock_legal_hold(legal_hold: Option<ObjectLockLegalHold>) -> S3Result<HashMap<String, String>> {
|
|
let mut eval_metadata = HashMap::new();
|
|
if let Some(v) = legal_hold {
|
|
let status = match v.status {
|
|
Some(status) => match status.as_str() {
|
|
ObjectLockLegalHoldStatus::OFF | ObjectLockLegalHoldStatus::ON => status.as_str().to_string(),
|
|
_ => {
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::MalformedXML,
|
|
"The XML you provided was not well-formed or did not validate against our published schema".to_string(),
|
|
));
|
|
}
|
|
},
|
|
None => String::default(),
|
|
};
|
|
let now = OffsetDateTime::now_utc();
|
|
// This is intentional behavior. Empty string represents "status cleared" which is different from "status never set".
|
|
eval_metadata.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), status);
|
|
insert_str(
|
|
&mut eval_metadata,
|
|
SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP,
|
|
format_object_lock_timestamp(now),
|
|
);
|
|
}
|
|
Ok(eval_metadata)
|
|
}
|
|
|
|
pub(crate) async fn validate_bucket_object_lock_enabled(bucket: &str) -> S3Result<()> {
|
|
match metadata_sys::get_object_lock_config(bucket).await {
|
|
Ok((cfg, _created)) => {
|
|
if cfg.object_lock_enabled != Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)) {
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::InvalidRequest,
|
|
"Object Lock is not enabled for this bucket".to_string(),
|
|
));
|
|
}
|
|
}
|
|
Err(err) => {
|
|
if err == StorageError::ConfigNotFound {
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::InvalidRequest,
|
|
"Bucket is missing ObjectLockConfiguration".to_string(),
|
|
));
|
|
}
|
|
warn!("get_object_lock_config err {:?}", err);
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::InternalError,
|
|
"Failed to get bucket ObjectLockConfiguration".to_string(),
|
|
));
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Validates HTTP conditional request headers for a single object according to
|
|
/// RFC 7232 (HTTP/1.1 conditional requests) and S3 API semantics.
|
|
///
|
|
/// This function evaluates the following headers, if present, in the standard
|
|
/// conditional request order:
|
|
/// - `If-Match`
|
|
/// - `If-None-Match`
|
|
/// - `If-Modified-Since`
|
|
/// - `If-Unmodified-Since`
|
|
///
|
|
/// The `headers` parameter provides the incoming HTTP request headers, and
|
|
/// `info` supplies the object's current metadata (ETag and modification time)
|
|
/// used to evaluate those preconditions.
|
|
///
|
|
/// The function returns `Ok(())` if either no relevant conditional headers are
|
|
/// set or all specified preconditions are satisfied. If any precondition fails,
|
|
/// it returns an `Err(S3Error)` with an appropriate S3 error code (for example,
|
|
/// `PreconditionFailed` for failed `If-Match` / `If-Unmodified-Since`, or
|
|
/// `NotModified` for `If-None-Match` / `If-Modified-Since` when the resource
|
|
/// has not changed), allowing the caller to translate this into the correct
|
|
/// HTTP response.
|
|
pub(crate) fn check_preconditions(headers: &HeaderMap, info: &ObjectInfo) -> S3Result<()> {
|
|
let mod_time = info.mod_time;
|
|
let etag = info.etag.as_deref();
|
|
let if_match = non_empty_header_value(headers, IF_MATCH);
|
|
let if_none_match = non_empty_header_value(headers, IF_NONE_MATCH);
|
|
let if_modified_since = non_empty_header_value(headers, IF_MODIFIED_SINCE);
|
|
let if_unmodified_since = non_empty_header_value(headers, IF_UNMODIFIED_SINCE);
|
|
|
|
if mod_time.is_none() && etag.is_none() {
|
|
return Ok(());
|
|
}
|
|
|
|
// If-Match: requires ETag to exist
|
|
if let Some(if_match_val) = if_match {
|
|
match etag {
|
|
Some(e) if is_etag_equal(e, if_match_val) => {}
|
|
_ => return Err(S3Error::new(S3ErrorCode::PreconditionFailed)),
|
|
}
|
|
}
|
|
|
|
// If-Unmodified-Since (only when If-Match is absent)
|
|
if if_match.is_none()
|
|
&& let Some(t) = mod_time
|
|
&& let Some(if_unmodified_since) = if_unmodified_since
|
|
&& let Ok(given_time) = time::PrimitiveDateTime::parse(if_unmodified_since, &RFC1123).map(|dt| dt.assume_utc())
|
|
&& t > given_time.add(time::Duration::seconds(1))
|
|
{
|
|
return Err(S3Error::new(S3ErrorCode::PreconditionFailed));
|
|
}
|
|
|
|
// If-None-Match
|
|
if let Some(if_none_match) = if_none_match
|
|
&& let Some(e) = etag
|
|
&& is_etag_equal(e, if_none_match)
|
|
{
|
|
let mut error_headers = HeaderMap::new();
|
|
if let Ok(etag_header) = parse_etag(e) {
|
|
error_headers.insert("etag", etag_header);
|
|
}
|
|
if let Some(t) = mod_time
|
|
&& let Ok(last_modified_str) = t.format(&RFC1123)
|
|
&& let Ok(last_modified_header) = HeaderValue::from_str(&last_modified_str)
|
|
{
|
|
error_headers.insert("last-modified", last_modified_header);
|
|
}
|
|
|
|
let mut s3_error = S3Error::new(S3ErrorCode::NotModified);
|
|
s3_error.set_message("Not Modified".to_string());
|
|
s3_error.set_status_code(StatusCode::NOT_MODIFIED);
|
|
s3_error.set_headers(error_headers);
|
|
return Err(s3_error);
|
|
}
|
|
|
|
// If-Modified-Since (only when If-None-Match is absent — semantics per RFC 7232; dates use RFC 1123 format)
|
|
if if_none_match.is_none()
|
|
&& let Some(t) = mod_time
|
|
&& let Some(if_modified_since) = if_modified_since
|
|
&& let Ok(given_time) = time::PrimitiveDateTime::parse(if_modified_since, &RFC1123).map(|dt| dt.assume_utc())
|
|
&& t < given_time.add(time::Duration::seconds(1))
|
|
{
|
|
let mut error_headers = HeaderMap::new();
|
|
if let Some(e) = etag
|
|
&& let Ok(etag_header) = parse_etag(e)
|
|
{
|
|
error_headers.insert("etag", etag_header);
|
|
}
|
|
if let Ok(last_modified_str) = t.format(&RFC1123)
|
|
&& let Ok(last_modified_header) = HeaderValue::from_str(&last_modified_str)
|
|
{
|
|
error_headers.insert("last-modified", last_modified_header);
|
|
}
|
|
|
|
let mut s3_error = S3Error::new(S3ErrorCode::NotModified);
|
|
s3_error.set_message("Not Modified".to_string());
|
|
s3_error.set_status_code(StatusCode::NOT_MODIFIED);
|
|
s3_error.set_headers(error_headers);
|
|
|
|
return Err(s3_error);
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
fn non_empty_header_value(headers: &HeaderMap, name: http::header::HeaderName) -> Option<&str> {
|
|
headers
|
|
.get(name)
|
|
.and_then(|v| v.to_str().ok())
|
|
.map(str::trim)
|
|
.filter(|v| !v.is_empty())
|
|
}
|
|
|
|
/// Compares an object ETag with an ETag value from an HTTP header.
|
|
///
|
|
/// This helper implements HTTP ETag comparison semantics for headers such as
|
|
/// `If-Match` and `If-None-Match`:
|
|
/// - Supports the wildcard `*`, which matches any `object_etag`.
|
|
/// - Supports comma-separated ETag lists (e.g., `"etag1", "etag2"`), returning
|
|
/// `true` if any entry matches `object_etag`.
|
|
/// - Automatically trims surrounding whitespace and double quotes from both the
|
|
/// header entries and `object_etag` before comparison.
|
|
///
|
|
/// # Parameters
|
|
/// - `object_etag`: The ETag associated with the stored object.
|
|
/// - `header_etag`: The raw ETag header value received in the request, which may
|
|
/// be a wildcard, a single ETag, or a comma-separated list of ETags.
|
|
///
|
|
/// # Returns
|
|
/// `true` if the header value matches the object ETag according to the above
|
|
/// HTTP ETag comparison rules, otherwise `false`.
|
|
pub(crate) fn is_etag_equal(object_etag: &str, header_etag: &str) -> bool {
|
|
let header_etag = header_etag.trim();
|
|
if header_etag == "*" {
|
|
return true;
|
|
}
|
|
header_etag
|
|
.split(',')
|
|
.map(|s| s.trim().trim_matches('"'))
|
|
.any(|e| e == object_etag.trim_matches('"'))
|
|
}
|
|
|
|
/// Converts an object ETag string into an HTTP `HeaderValue` for use in response headers.
|
|
///
|
|
/// This function normalizes ETag values by ensuring they are wrapped in double quotes,
|
|
/// which is the standard HTTP ETag format according to RFC 7232. If the input ETag
|
|
/// already contains quotes, they are removed before adding new quotes to avoid duplication.
|
|
///
|
|
/// # Parameters
|
|
/// - `object_etag`: The ETag string, which may or may not already be wrapped in quotes.
|
|
/// Must not be empty after removing quotes.
|
|
///
|
|
/// # Returns
|
|
/// - `Ok(HeaderValue)`: The normalized ETag value wrapped in double quotes, ready for use
|
|
/// in HTTP response headers.
|
|
/// - `Err(S3Error)`: Returns `InvalidArgument` if the ETag value is empty (after removing quotes)
|
|
/// or contains characters that cannot be represented in an HTTP header value.
|
|
///
|
|
/// # Note
|
|
/// When this function returns an error, callers should log the error but continue with
|
|
/// the operation (e.g., return NotModified without the ETag header) rather than propagating
|
|
/// the InvalidArgument error, as the original operation semantics should be preserved.
|
|
pub fn parse_etag(object_etag: &str) -> Result<HeaderValue, S3Error> {
|
|
let etag_trimmed = object_etag.trim_matches('"');
|
|
|
|
if etag_trimmed.trim().is_empty() {
|
|
return Err(S3Error::with_message(S3ErrorCode::InvalidArgument, "ETag cannot be empty".to_string()));
|
|
}
|
|
|
|
let etag_quoted = format!("\"{}\"", etag_trimmed);
|
|
|
|
HeaderValue::from_str(&etag_quoted).map_err(|e| {
|
|
warn!(
|
|
"Failed to convert ETag to HeaderValue (ETag: {}): {}. The ETag header will be omitted from the response.",
|
|
etag_quoted, e
|
|
);
|
|
S3Error::with_message(
|
|
S3ErrorCode::InvalidArgument,
|
|
format!("Invalid ETag header value (ETag: {}): {}", etag_quoted, e),
|
|
)
|
|
})
|
|
}
|
|
|
|
/// Auxiliary functions: extract prefixes and suffixes
|
|
pub(crate) fn extract_prefix_suffix(filter: Option<&NotificationConfigurationFilter>) -> (String, String) {
|
|
if let Some(filter) = filter
|
|
&& let Some(filter_rules) = &filter.key
|
|
{
|
|
let mut prefix = String::new();
|
|
let mut suffix = String::new();
|
|
if let Some(rules) = &filter_rules.filter_rules {
|
|
for rule in rules {
|
|
if let (Some(name), Some(value)) = (rule.name.as_ref(), rule.value.as_ref()) {
|
|
if name.as_str().eq_ignore_ascii_case("prefix") {
|
|
prefix = value.clone();
|
|
} else if name.as_str().eq_ignore_ascii_case("suffix") {
|
|
suffix = value.clone();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return (prefix, suffix);
|
|
}
|
|
(String::new(), String::new())
|
|
}
|
|
|
|
/// Auxiliary functions: Handle configuration
|
|
pub(crate) fn process_queue_configurations<F>(
|
|
event_rules: &mut Vec<(Vec<EventName>, String, String, Vec<TargetID>)>,
|
|
configurations: Option<Vec<QueueConfiguration>>,
|
|
target_id_parser: F,
|
|
) -> Result<(), TargetIDError>
|
|
where
|
|
F: Fn(&str) -> Result<TargetID, TargetIDError>,
|
|
{
|
|
if let Some(configs) = configurations {
|
|
for cfg in configs {
|
|
let events = cfg.events.iter().filter_map(|e| EventName::parse(e.as_ref()).ok()).collect();
|
|
let (prefix, suffix) = extract_prefix_suffix(cfg.filter.as_ref());
|
|
let target_id = target_id_parser(&cfg.queue_arn)?;
|
|
event_rules.push((events, prefix, suffix, vec![target_id]));
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub(crate) fn process_topic_configurations<F>(
|
|
event_rules: &mut Vec<(Vec<EventName>, String, String, Vec<TargetID>)>,
|
|
configurations: Option<Vec<TopicConfiguration>>,
|
|
target_id_parser: F,
|
|
) -> Result<(), TargetIDError>
|
|
where
|
|
F: Fn(&str) -> Result<TargetID, TargetIDError>,
|
|
{
|
|
if let Some(configs) = configurations {
|
|
for cfg in configs {
|
|
let events = cfg.events.iter().filter_map(|e| EventName::parse(e.as_ref()).ok()).collect();
|
|
let (prefix, suffix) = extract_prefix_suffix(cfg.filter.as_ref());
|
|
let target_id = target_id_parser(&cfg.topic_arn)?;
|
|
event_rules.push((events, prefix, suffix, vec![target_id]));
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub(crate) fn process_lambda_configurations<F>(
|
|
event_rules: &mut Vec<(Vec<EventName>, String, String, Vec<TargetID>)>,
|
|
configurations: Option<Vec<LambdaFunctionConfiguration>>,
|
|
target_id_parser: F,
|
|
) -> Result<(), TargetIDError>
|
|
where
|
|
F: Fn(&str) -> Result<TargetID, TargetIDError>,
|
|
{
|
|
if let Some(configs) = configurations {
|
|
for cfg in configs {
|
|
let events = cfg.events.iter().filter_map(|e| EventName::parse(e.as_ref()).ok()).collect();
|
|
let (prefix, suffix) = extract_prefix_suffix(cfg.filter.as_ref());
|
|
let target_id = target_id_parser(&cfg.lambda_function_arn)?;
|
|
event_rules.push((events, prefix, suffix, vec![target_id]));
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub(crate) async fn has_replication_rules(bucket: &str, objects: &[ObjectToDelete]) -> bool {
|
|
let (cfg, _created) = match get_replication_config(bucket).await {
|
|
Ok(replication_config) => replication_config,
|
|
Err(_err) => {
|
|
return false;
|
|
}
|
|
};
|
|
|
|
for object in objects {
|
|
if cfg.has_active_rules(&object.object_name, true) {
|
|
return true;
|
|
}
|
|
}
|
|
false
|
|
}
|
|
|
|
/// Helper function to get store and validate bucket exists
|
|
pub(crate) async fn get_validated_store(bucket: &str) -> S3Result<Arc<rustfs_ecstore::store::ECStore>> {
|
|
let Some(store) = new_object_layer_fn() else {
|
|
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
|
|
};
|
|
|
|
// Validate bucket exists
|
|
store
|
|
.get_bucket_info(bucket, &BucketOptions::default())
|
|
.await
|
|
.map_err(ApiError::from)?;
|
|
|
|
Ok(store)
|
|
}
|
|
|
|
/// Quick check if CORS processing is needed (lightweight check for Origin header)
|
|
/// This avoids unnecessary function calls for non-CORS requests
|
|
#[inline]
|
|
pub(crate) fn needs_cors_processing(headers: &HeaderMap) -> bool {
|
|
headers.contains_key(cors::standard::ORIGIN)
|
|
}
|
|
|
|
/// Apply CORS headers to response based on bucket CORS configuration and request origin
|
|
///
|
|
/// This function:
|
|
/// 1. Reads the Origin header from the request
|
|
/// 2. Retrieves the bucket's CORS configuration
|
|
/// 3. Matches the origin against CORS rules
|
|
/// 4. Validates AllowedHeaders if request headers are present
|
|
/// 5. Returns one of:
|
|
/// - `None`: bucket has no CORS config (or request has no valid `Origin`)
|
|
/// - `Some(empty headers)`: bucket CORS exists but request is denied / no rule matched
|
|
/// - `Some(non-empty headers)`: bucket CORS exists and request matched
|
|
///
|
|
/// Note: This function should only be called if `needs_cors_processing()` returns true
|
|
/// to avoid unnecessary overhead for non-CORS requests.
|
|
pub(crate) async fn apply_cors_headers(bucket: &str, method: &http::Method, headers: &HeaderMap) -> Option<HeaderMap> {
|
|
use http::HeaderValue;
|
|
|
|
fn is_credentialed_request(headers: &HeaderMap) -> bool {
|
|
headers.contains_key(http::header::AUTHORIZATION)
|
|
|| headers.contains_key(http::header::COOKIE)
|
|
|| headers.contains_key("x-amz-security-token")
|
|
|| headers.contains_key("x-amz-content-sha256")
|
|
}
|
|
|
|
// Get Origin header from request
|
|
let origin = headers.get(cors::standard::ORIGIN)?.to_str().ok()?;
|
|
|
|
// Get CORS configuration for the bucket
|
|
let cors_config = match metadata_sys::get_cors_config(bucket).await {
|
|
Ok((config, _)) => config,
|
|
Err(_) => return None, // No CORS config, no headers to add
|
|
};
|
|
|
|
// Early return if no CORS rules configured
|
|
if cors_config.cors_rules.is_empty() {
|
|
return Some(HeaderMap::new());
|
|
}
|
|
|
|
// Check if method is supported and get its string representation
|
|
const SUPPORTED_METHODS: &[&str] = &["GET", "PUT", "POST", "DELETE", "HEAD", "OPTIONS"];
|
|
let method_str = method.as_str();
|
|
if !SUPPORTED_METHODS.contains(&method_str) {
|
|
return Some(HeaderMap::new());
|
|
}
|
|
|
|
// Use Access-Control-Request-Method if present (for preflight and non-preflight requests),
|
|
// otherwise fall back to the actual HTTP method.
|
|
let is_preflight = method == http::Method::OPTIONS;
|
|
let requested_method = headers
|
|
.get(cors::request::ACCESS_CONTROL_REQUEST_METHOD)
|
|
.and_then(|v| v.to_str().ok())
|
|
.unwrap_or(method_str);
|
|
|
|
// Get requested headers from preflight request
|
|
let requested_headers = if is_preflight {
|
|
headers
|
|
.get(cors::request::ACCESS_CONTROL_REQUEST_HEADERS)
|
|
.and_then(|v| v.to_str().ok())
|
|
.map(|h| h.split(',').map(|s| s.trim().to_lowercase()).collect::<Vec<_>>())
|
|
} else {
|
|
None
|
|
};
|
|
|
|
// Find matching CORS rule
|
|
for rule in cors_config.cors_rules.iter() {
|
|
// Check if origin matches
|
|
let origin_matches = rule.allowed_origins.iter().any(|allowed_origin| {
|
|
if allowed_origin == "*" {
|
|
true
|
|
} else {
|
|
// Exact match or pattern match (support wildcards like https://*.example.com)
|
|
allowed_origin == origin || matches_origin_pattern(allowed_origin, origin)
|
|
}
|
|
});
|
|
|
|
if !origin_matches {
|
|
continue;
|
|
}
|
|
|
|
// Check if method is allowed
|
|
let method_allowed = rule
|
|
.allowed_methods
|
|
.iter()
|
|
.any(|allowed_method| allowed_method.as_str() == requested_method);
|
|
|
|
if !method_allowed {
|
|
continue;
|
|
}
|
|
|
|
// Validate AllowedHeaders if present in the request
|
|
if let Some(ref req_headers) = requested_headers {
|
|
if let Some(ref allowed_headers) = rule.allowed_headers {
|
|
// Check if all requested headers are allowed
|
|
let all_headers_allowed = req_headers.iter().all(|req_header| {
|
|
allowed_headers.iter().any(|allowed_header| {
|
|
let allowed_lower = allowed_header.to_lowercase();
|
|
// "*" allows all headers, or exact match
|
|
allowed_lower == "*" || allowed_lower == *req_header
|
|
})
|
|
});
|
|
|
|
if !all_headers_allowed {
|
|
// If not all headers are allowed, skip this rule
|
|
continue;
|
|
}
|
|
} else if !req_headers.is_empty() {
|
|
// If no AllowedHeaders specified but headers were requested, skip this rule
|
|
// Unless the rule explicitly allows all headers
|
|
continue;
|
|
}
|
|
}
|
|
|
|
// Found matching rule, build response headers
|
|
let mut response_headers = HeaderMap::new();
|
|
|
|
// Access-Control-Allow-Origin
|
|
// Credentials mode + wildcard allow list requires echoing the request origin.
|
|
// Browsers reject `Access-Control-Allow-Origin: *` with `credentials: include`.
|
|
let has_wildcard_origin = rule.allowed_origins.iter().any(|o| o == "*");
|
|
let credentialed_request = is_credentialed_request(headers);
|
|
let mut origin_reflected = false;
|
|
|
|
if has_wildcard_origin {
|
|
if credentialed_request {
|
|
if let Ok(origin_value) = HeaderValue::from_str(origin) {
|
|
response_headers.insert(cors::response::ACCESS_CONTROL_ALLOW_ORIGIN, origin_value);
|
|
origin_reflected = true;
|
|
}
|
|
} else {
|
|
response_headers.insert(cors::response::ACCESS_CONTROL_ALLOW_ORIGIN, HeaderValue::from_static("*"));
|
|
}
|
|
} else if let Ok(origin_value) = HeaderValue::from_str(origin) {
|
|
response_headers.insert(cors::response::ACCESS_CONTROL_ALLOW_ORIGIN, origin_value);
|
|
origin_reflected = true;
|
|
}
|
|
|
|
// Vary: Origin whenever origin is reflected (non-"*" allow-origin).
|
|
// This prevents proxy/browser caches from reusing CORS headers across different origins.
|
|
if origin_reflected {
|
|
response_headers.insert(cors::standard::VARY, HeaderValue::from_static("Origin"));
|
|
}
|
|
|
|
// Credentials mode requires explicit allow-credentials.
|
|
if credentialed_request {
|
|
response_headers.insert(cors::response::ACCESS_CONTROL_ALLOW_CREDENTIALS, HeaderValue::from_static("true"));
|
|
}
|
|
|
|
// Access-Control-Allow-Methods (required for preflight)
|
|
if is_preflight || !rule.allowed_methods.is_empty() {
|
|
let methods_str = rule.allowed_methods.iter().map(|m| m.as_str()).collect::<Vec<_>>().join(", ");
|
|
if let Ok(methods_value) = HeaderValue::from_str(&methods_str) {
|
|
response_headers.insert(cors::response::ACCESS_CONTROL_ALLOW_METHODS, methods_value);
|
|
}
|
|
}
|
|
|
|
// Access-Control-Allow-Headers (required for preflight if headers were requested)
|
|
if is_preflight && let Some(ref allowed_headers) = rule.allowed_headers {
|
|
let headers_str = allowed_headers.iter().map(|h| h.as_str()).collect::<Vec<_>>().join(", ");
|
|
if let Ok(headers_value) = HeaderValue::from_str(&headers_str) {
|
|
response_headers.insert(cors::response::ACCESS_CONTROL_ALLOW_HEADERS, headers_value);
|
|
}
|
|
}
|
|
|
|
// Access-Control-Expose-Headers (for actual requests)
|
|
if !is_preflight && let Some(ref expose_headers) = rule.expose_headers {
|
|
let expose_headers_str = expose_headers.iter().map(|h| h.as_str()).collect::<Vec<_>>().join(", ");
|
|
if let Ok(expose_value) = HeaderValue::from_str(&expose_headers_str) {
|
|
response_headers.insert(cors::response::ACCESS_CONTROL_EXPOSE_HEADERS, expose_value);
|
|
}
|
|
}
|
|
|
|
// Access-Control-Max-Age (for preflight requests)
|
|
if is_preflight
|
|
&& let Some(max_age) = rule.max_age_seconds
|
|
&& let Ok(max_age_value) = HeaderValue::from_str(&max_age.to_string())
|
|
{
|
|
response_headers.insert(cors::response::ACCESS_CONTROL_MAX_AGE, max_age_value);
|
|
}
|
|
|
|
return Some(response_headers);
|
|
}
|
|
|
|
Some(HeaderMap::new()) // No matching rule found
|
|
}
|
|
/// Check if an origin matches a pattern (supports wildcards like https://*.example.com)
|
|
pub(crate) fn matches_origin_pattern(pattern: &str, origin: &str) -> bool {
|
|
// Simple wildcard matching: * matches any sequence
|
|
if pattern.contains('*') {
|
|
let pattern_parts: Vec<&str> = pattern.split('*').collect();
|
|
if pattern_parts.len() == 2 {
|
|
origin.starts_with(pattern_parts[0]) && origin.ends_with(pattern_parts[1])
|
|
} else {
|
|
false
|
|
}
|
|
} else {
|
|
pattern == origin
|
|
}
|
|
}
|
|
|
|
/// Wrap S3Response with CORS headers if needed
|
|
/// This function performs a lightweight check first to avoid unnecessary CORS processing
|
|
/// for non-CORS requests (requests without Origin header)
|
|
pub(crate) async fn wrap_response_with_cors<T>(
|
|
bucket: &str,
|
|
method: &http::Method,
|
|
headers: &HeaderMap,
|
|
output: T,
|
|
) -> S3Response<T> {
|
|
let mut response = S3Response::new(output);
|
|
|
|
// Quick check: only process CORS if Origin header is present
|
|
if needs_cors_processing(headers)
|
|
&& let Some(cors_headers) = apply_cors_headers(bucket, method, headers).await
|
|
{
|
|
for (key, value) in cors_headers.iter() {
|
|
response.headers.insert(key, value.clone());
|
|
}
|
|
}
|
|
|
|
response
|
|
}
|
|
|
|
/// Parse part number from Option<i32> to Option<usize> with validation
|
|
/// This function checks that the part number is greater than 0 and
|
|
/// converts it to usize, returning an error if invalid
|
|
///
|
|
/// # Arguments
|
|
/// * `part_number` - The optional part number as i32
|
|
/// * `op` - The operation name for logging purposes
|
|
///
|
|
/// # Returns
|
|
/// * `Ok(Some(usize))` if part number is valid
|
|
/// * `Ok(None)` if part number is None
|
|
/// * `Err(S3Error)` if part number is invalid (0 or overflow)
|
|
#[inline]
|
|
pub(crate) fn parse_part_number_i32_to_usize(part_number: Option<i32>, op: &'static str) -> S3Result<Option<usize>> {
|
|
match part_number {
|
|
None => Ok(None),
|
|
Some(n) if n <= 0 => Err(S3Error::with_message(
|
|
S3ErrorCode::InvalidArgument,
|
|
format!("{op}: invalid partNumber {n}, must be a positive integer"),
|
|
)),
|
|
Some(n) => Ok(Some(n as usize)),
|
|
}
|
|
}
|