gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-11-14 13:48:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(storage): read and install authorization state on both edges

Browse Source 
The state an access hook hands to a use case is now read and installed
through the `AuthorizationRequest` trait. Its two implementors are the
s3s request and `RequestEnvelope`, so the legacy hooks and the gateway
edge call the same functions:

- These readers are now generic over `AuthorizationRequest`:
  `apply_bucket_generation_guard`,
  `apply_copy_source_bucket_generation_guard` (both on one shared
  fence), `odm_read_generation`, `prepare_odm_read_generation`,
  `bucket_config_mutation_incarnation`,
  `replication_request_authorized`,
  `retained_table_data_plane_publication_guards`,
  `request_context_from_req`, `resource_free_condition_values`,
  `req_info_ref`, `req_info_mut`, `load_bucket_generation_from_store`
  and `SseKmsPrincipal::from_request`.
- `load_bucket_generation` and `capture_odm_read_generation` now take
  the server context explicitly. The legacy hooks pass
  `FS::server_ctx`.
- New trait accessors: request context; the bucket, pending-delete,
  copy-source and ODM generation guards; the bucket-config mutation
  snapshot.
- New trait installers: the bucket, pending-delete and ODM generation
  guards. The legacy hooks now install those guards through them.
- New: `authorize_object_request(req, bucket, key, version_id,
  action)`, the object twin of `authorize_bucket_request`, with no
  version swap. `versioned_read_action` and `versioned_tagging_action`
  become crate-visible so callers can pick the action first. The
  GetObjectAcl, PutObjectAcl, GetObjectLegalHold and GetObjectRetention
  hooks now call it.
- Removed: `request_context_from_extensions`, which had no caller left.

Each reader keeps the same answers and the same error codes and
messages.

Not built here: the bucket-configuration mutation shape, the
body-input newtypes, and the replication-only, write-ETag and
staged-part checks.

Tests: `storage/access_readers_test.rs` covers each reader through
both forms with the state present and absent (12 refusal
cases, 11 accepted or absent ones), plus the installers on the envelope. The
existing owned-versus-legacy decision table now also runs every case
through `authorize_object_request` on the envelope.
This commit is contained in:
overtrue
2026-11-13 11:31:33 +08:00
parent be2d3ca2cf
commit 9c394af489
8 changed files with 723 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/app/context/envelope.rs
+31
View File
@@ -480,6 +480,37 @@ impl AuthorizationRequest for RequestEnvelope {
fn req_info_mut(&mut self) -> Option<&mut ReqInfo> {
self.req_info_mut()
}
fn request_context(&self) -> Option<&RequestContext> {
self.request_context.as_ref()
}
fn bucket_generation(&self) -> Option<&BucketGenerationGuard> {
self.bucket_generation.as_ref()
}
fn install_bucket_generation(&mut self, guard: BucketGenerationGuard) {
self.bucket_generation = Some(guard);
}
fn pending_delete_bucket_generation(&self) -> Option<&PendingDeleteBucketGenerationGuard> {
self.pending_delete_bucket_generation.as_ref()
}
fn install_pending_delete_bucket_generation(&mut self, guard: PendingDeleteBucketGenerationGuard) {
self.pending_delete_bucket_generation = Some(guard);
}
fn copy_source_bucket_generation(&self) -> Option<&CopySourceBucketGenerationGuard> {
self.copy_source_bucket_generation.as_ref()
}
fn odm_read_generation(&self) -> Option<&OdmReadGenerationGuard> {
self.odm_read_generation.as_ref()
}
fn install_odm_read_generation(&mut self, guard: OdmReadGenerationGuard) {
self.odm_read_generation = Some(guard);
}
fn bucket_config_mutation(&self) -> Option<&BucketConfigMutationSnapshot> {
self.bucket_config_mutation.as_ref()
}
#[cfg(test)]
fn test_extensions(&self) -> &Extensions {
&self.residual_extensions
}
fn internal_object_authorization(&self) -> bool {
false
}
rustfs/src/app/object/put.rs
+4 -4
View File
@@ -970,14 +970,14 @@ pub(super) enum PutObjectOrigin<'a> {
impl PutObjectOrigin<'_> {
fn replication_request_authorized(&self) -> bool {
match self {
Self::S3 { req, .. } => replication_request_authorized(req),
Self::S3 { req, .. } => replication_request_authorized(*req),
Self::Internal { .. } => false,
}
}
fn apply_bucket_generation_guard(&self, bucket: &str, opts: &mut ObjectOptions) -> S3Result<()> {
match self {
Self::S3 { req, .. } => apply_bucket_generation_guard(req, bucket, opts),
Self::S3 { req, .. } => apply_bucket_generation_guard(*req, bucket, opts),
Self::Internal {
expected_bucket_incarnation_id,
..
@@ -1010,14 +1010,14 @@ impl PutObjectOrigin<'_> {
fn table_data_plane_publication_guards(&self) -> Option<TableDataPlanePublicationGuards> {
match self {
Self::S3 { req, .. } => retained_table_data_plane_publication_guards(req),
Self::S3 { req, .. } => retained_table_data_plane_publication_guards(*req),
Self::Internal { .. } => None,
}
}
fn sse_principal(&self) -> Option<SseKmsPrincipal> {
match self {
Self::S3 { req, .. } => SseKmsPrincipal::from_request(req),
Self::S3 { req, .. } => SseKmsPrincipal::from_request(*req),
Self::Internal { .. } => None,
}
}
rustfs/src/app/owned_authorization_test.rs
+16 -2
View File
@@ -1536,9 +1536,19 @@ async fn owned_and_legacy_entry_points_reach_the_same_decisions() {
});
(case.facts)(&mut legacy);
let mut owned = RequestEnvelope::from_s3s(&legacy);
// The object shape both edges' access hooks call: it must reach the same decision.
let mut owned_shape = RequestEnvelope::from_s3s(&legacy);
let action = Action::S3Action(case.action);
let legacy_result = crate::app::storage_api::owned_parity::authorize_request(&mut legacy, action).await;
let owned_result = crate::app::authorize_owned_request(&mut owned, action, None).await;
let shape_result = crate::app::storage_api::owned_parity::authorize_object_request(
&mut owned_shape,
case.bucket,
case.object,
None,
action,
)
.await;
let legacy_ok = match &legacy_result {
Ok(()) => case.allowed,
Err(err) => !case.allowed && err.code() == &S3ErrorCode::AccessDenied,
@@ -1547,9 +1557,13 @@ async fn owned_and_legacy_entry_points_reach_the_same_decisions() {
Ok(()) => case.allowed,
Err(err) => !case.allowed && err.code() == &rustfs_s3_types::S3ErrorCode::AccessDenied,
};
if !legacy_ok || !owned_ok {
let shape_ok = match &shape_result {
Ok(()) => case.allowed,
Err(err) => !case.allowed && err.code() == &S3ErrorCode::AccessDenied,
};
if !legacy_ok || !owned_ok || !shape_ok {
mismatches.push(format!(
"{} (expected allowed={}): legacy {legacy_result:?}, owned {owned_result:?}",
"{} (expected allowed={}): legacy {legacy_result:?}, owned {owned_result:?}, object shape {shape_result:?}",
case.name, case.allowed
));
}
rustfs/src/app/storage_api.rs
+2 -1
View File
@@ -1848,7 +1848,8 @@ pub(crate) mod owned_parity {
DenialAuditor, DenialRateLimiter, DenialScope, FS, ObjectTagConditions, PostObjectPolicyHeaders, PostObjectRequestMarker,
ReqInfo, RequestContext, RequestedCopySource, RequestedObjectLegalHold, RequestedObjectLockRetainUntil,
RequestedObjectLockRetention, RequestedObjectTags, SUPPRESSED_DENIALS_TAG, TableDataPlanePublicationGuards,
authorization_denial_entry, authorization_denial_entry_for_view, authorize_request, parity_test_support,
authorization_denial_entry, authorization_denial_entry_for_view, authorize_object_request, authorize_request,
parity_test_support,
};
}
rustfs/src/storage/access.rs
+221 -138
View File
@@ -74,6 +74,9 @@ use url::Url;
/// state this module keeps private.
#[path = "access_owned/mod.rs"]
pub(crate) mod access_owned;
#[cfg(test)]
#[path = "access_readers_test.rs"]
mod readers_test;
fn iam_authorization_error(err: IamError) -> ApiError {
match err {
@@ -107,10 +110,8 @@ pub(crate) struct ReqInfo {
pub suppress_denial_audit: bool,
}
pub(crate) fn replication_request_authorized<T>(req: &S3Request<T>) -> bool {
req.extensions
.get::<ReqInfo>()
.is_some_and(|req_info| req_info.replication_request_authorized)
pub(crate) fn replication_request_authorized<R: AuthorizationRequest + ?Sized>(req: &R) -> bool {
req.req_info().is_some_and(|req_info| req_info.replication_request_authorized)
}
fn has_replication_only_put_headers(headers: &HeaderMap) -> bool {
@@ -437,9 +438,12 @@ pub(crate) struct BucketConfigMutationSnapshot {
incarnation_id: uuid::Uuid,
}
pub(crate) fn bucket_config_mutation_incarnation<T>(req: &S3Request<T>, bucket: &str) -> S3Result<Option<uuid::Uuid>> {
let Some(snapshot) = req.extensions.get::<BucketConfigMutationSnapshot>() else {
if req.extensions.get::<std::sync::Arc<ServerContextSlot>>().is_some() {
pub(crate) fn bucket_config_mutation_incarnation<R: AuthorizationRequest + ?Sized>(
req: &R,
bucket: &str,
) -> S3Result<Option<uuid::Uuid>> {
let Some(snapshot) = req.bucket_config_mutation() else {
if req.server_context().is_some() {
return Err(s3_error!(InternalError, "bucket config mutation snapshot is missing"));
}
return Ok(None);
@@ -561,12 +565,12 @@ impl PendingDeleteBucketGenerationGuard {
}
}
pub(crate) async fn load_bucket_generation_from_store<T>(
pub(crate) async fn load_bucket_generation_from_store<R: AuthorizationRequest + ?Sized>(
store: &crate::storage::storage_api::ECStore,
req: &S3Request<T>,
req: &R,
bucket: &str,
) -> S3Result<BucketGenerationGuard> {
if let Some(existing) = req.extensions.get::<BucketGenerationGuard>() {
if let Some(existing) = req.bucket_generation() {
return if existing.bucket == bucket {
Ok(existing.clone())
} else {
@@ -581,9 +585,15 @@ pub(crate) async fn load_bucket_generation_from_store<T>(
})
}
async fn load_bucket_generation<T>(fs: &FS, req: &S3Request<T>, bucket: &str) -> S3Result<BucketGenerationGuard> {
let store = fs
.server_ctx()
/// Loads the bucket incarnation through `server_ctx`'s object store, or returns the guard the
/// request already carries. The legacy hooks pass `FS::server_ctx`; the gateway edge passes the
/// envelope's server context.
pub(crate) async fn load_bucket_generation<R: AuthorizationRequest + ?Sized>(
server_ctx: &ServerContextSlot,
req: &R,
bucket: &str,
) -> S3Result<BucketGenerationGuard> {
let store = server_ctx
.object_store()
.ok_or_else(|| s3_error!(InternalError, "object store is not initialized"))?;
load_bucket_generation_from_store(store.as_ref(), req, bucket).await
@@ -615,33 +625,39 @@ fn odm_read_source_configured(bucket: &str) -> bool {
sys.is_module_enabled() && sys.state(bucket).is_some()
}
async fn capture_odm_read_generation<T>(fs: &FS, req: &S3Request<T>, bucket: &str) -> OdmReadGenerationGuard {
/// The on-demand-migration source generation a read may consult, captured before its
/// authorization. A capture failure is kept in the guard and surfaces only on a source miss.
pub(crate) async fn capture_odm_read_generation<R: AuthorizationRequest + ?Sized>(
server_ctx: &ServerContextSlot,
req: &R,
bucket: &str,
) -> OdmReadGenerationGuard {
if !odm_read_source_configured(bucket) {
return OdmReadGenerationGuard::Unavailable;
}
OdmReadGenerationGuard::from_result(load_bucket_generation(fs, req, bucket).await)
OdmReadGenerationGuard::from_result(load_bucket_generation(server_ctx, req, bucket).await)
}
/// Direct usecase callers have no access middleware; capture at their entry.
/// A server request without the access marker must never bind to a later source.
pub(crate) async fn prepare_odm_read_generation<T>(
pub(crate) async fn prepare_odm_read_generation<R: AuthorizationRequest + ?Sized>(
store: &crate::storage::storage_api::ECStore,
req: &mut S3Request<T>,
req: &mut R,
bucket: &str,
) {
if req.extensions.get::<OdmReadGenerationGuard>().is_some() {
if req.odm_read_generation().is_some() {
return;
}
let guard = if req.extensions.get::<std::sync::Arc<ServerContextSlot>>().is_some() || !odm_read_source_configured(bucket) {
let guard = if req.server_context().is_some() || !odm_read_source_configured(bucket) {
OdmReadGenerationGuard::Unavailable
} else {
OdmReadGenerationGuard::from_result(load_bucket_generation_from_store(store, req, bucket).await)
};
req.extensions.insert(guard);
req.install_odm_read_generation(guard);
}
pub(crate) fn odm_read_generation<T>(req: &S3Request<T>, bucket: &str) -> S3Result<Option<uuid::Uuid>> {
match req.extensions.get::<OdmReadGenerationGuard>() {
pub(crate) fn odm_read_generation<R: AuthorizationRequest + ?Sized>(req: &R, bucket: &str) -> S3Result<Option<uuid::Uuid>> {
match req.odm_read_generation() {
Some(OdmReadGenerationGuard::Ready(guard)) if guard.bucket == bucket => Ok(Some(guard.incarnation_id)),
Some(OdmReadGenerationGuard::Ready(_)) => {
Err(s3_error!(InternalError, "source generation guard does not match request bucket"))
@@ -663,60 +679,72 @@ async fn load_copy_source_bucket_generation(fs: &FS, bucket: &str) -> S3Result<C
})
}
pub(crate) fn apply_bucket_generation_guard<T>(req: &S3Request<T>, bucket: &str, opts: &mut ObjectOptions) -> S3Result<()> {
let pending_guard = req.extensions.get::<PendingDeleteBucketGenerationGuard>();
let guard = match pending_guard {
pub(crate) fn apply_bucket_generation_guard<R: AuthorizationRequest + ?Sized>(
req: &R,
bucket: &str,
opts: &mut ObjectOptions,
) -> S3Result<()> {
let guard = match req.pending_delete_bucket_generation() {
Some(PendingDeleteBucketGenerationGuard::Ready(guard)) => Some(guard),
Some(PendingDeleteBucketGenerationGuard::Failed { code, message }) => {
return Err(S3Error::with_message(code.clone(), message.clone()));
}
None => req.extensions.get::<BucketGenerationGuard>(),
None => req.bucket_generation(),
};
let Some(guard) = guard else {
if req.extensions.get::<std::sync::Arc<ServerContextSlot>>().is_some() {
return Err(s3_error!(InternalError, "bucket generation guard is missing"));
}
if let Some(publication_guards) = req.extensions.get::<TableDataPlanePublicationGuards>() {
publication_guards.add_lock_loss_fences(opts);
}
return Ok(());
};
if guard.bucket != bucket {
return Err(s3_error!(InternalError, "bucket generation guard does not match request bucket"));
}
opts.expected_bucket_incarnation_id = Some(guard.incarnation_id);
if let Some(publication_guards) = req.extensions.get::<TableDataPlanePublicationGuards>() {
publication_guards.add_lock_loss_fences(opts);
}
Ok(())
apply_generation_fence(
req,
guard.map(|guard| (guard.bucket.as_str(), guard.incarnation_id)),
bucket,
opts,
"bucket generation guard",
)
}
pub(crate) fn retained_table_data_plane_publication_guards<T>(req: &S3Request<T>) -> Option<TableDataPlanePublicationGuards> {
req.extensions.get::<TableDataPlanePublicationGuards>().cloned()
pub(crate) fn retained_table_data_plane_publication_guards<R: AuthorizationRequest + ?Sized>(
req: &R,
) -> Option<TableDataPlanePublicationGuards> {
req.publication_guards().cloned()
}
pub(crate) fn apply_copy_source_bucket_generation_guard<T>(
req: &S3Request<T>,
pub(crate) fn apply_copy_source_bucket_generation_guard<R: AuthorizationRequest + ?Sized>(
req: &R,
bucket: &str,
opts: &mut ObjectOptions,
) -> S3Result<()> {
let Some(guard) = req.extensions.get::<CopySourceBucketGenerationGuard>() else {
if req.extensions.get::<std::sync::Arc<ServerContextSlot>>().is_some() {
return Err(s3_error!(InternalError, "copy source bucket generation guard is missing"));
let guard = req.copy_source_bucket_generation();
apply_generation_fence(
req,
guard.map(|guard| (guard.bucket.as_str(), guard.incarnation_id)),
bucket,
opts,
"copy source bucket generation guard",
)
}
/// Binds `opts` to the captured incarnation of `bucket` and to the retained table publication
/// locks. A server request without a captured guard is refused: only direct use-case callers,
/// which carry no server context, run unfenced.
fn apply_generation_fence<R: AuthorizationRequest + ?Sized>(
req: &R,
guard: Option<(&str, uuid::Uuid)>,
bucket: &str,
opts: &mut ObjectOptions,
guard_name: &str,
) -> S3Result<()> {
match guard {
Some((guarded, _)) if guarded != bucket => {
return Err(S3Error::with_message(
S3ErrorCode::InternalError,
format!("{guard_name} does not match request bucket"),
));
}
if let Some(publication_guards) = req.extensions.get::<TableDataPlanePublicationGuards>() {
publication_guards.add_lock_loss_fences(opts);
Some((_, incarnation_id)) => opts.expected_bucket_incarnation_id = Some(incarnation_id),
None if req.server_context().is_some() => {
return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("{guard_name} is missing")));
}
return Ok(());
};
if guard.bucket != bucket {
return Err(s3_error!(
InternalError,
"copy source bucket generation guard does not match request bucket"
));
None => {}
}
opts.expected_bucket_incarnation_id = Some(guard.incarnation_id);
if let Some(publication_guards) = req.extensions.get::<TableDataPlanePublicationGuards>() {
if let Some(publication_guards) = req.publication_guards() {
publication_guards.add_lock_loss_fences(opts);
}
Ok(())
@@ -726,12 +754,12 @@ fn missing_req_info_error() -> S3Error {
s3_error!(InternalError, "ReqInfo not found in request extensions")
}
pub(crate) fn req_info_ref<T>(req: &S3Request<T>) -> S3Result<&ReqInfo> {
req.extensions.get::<ReqInfo>().ok_or_else(missing_req_info_error)
pub(crate) fn req_info_ref<R: AuthorizationRequest + ?Sized>(req: &R) -> S3Result<&ReqInfo> {
req.req_info().ok_or_else(missing_req_info_error)
}
pub(crate) fn req_info_mut<T>(req: &mut S3Request<T>) -> S3Result<&mut ReqInfo> {
req.extensions.get_mut::<ReqInfo>().ok_or_else(missing_req_info_error)
pub(crate) fn req_info_mut<R: AuthorizationRequest + ?Sized>(req: &mut R) -> S3Result<&mut ReqInfo> {
req.req_info_mut().ok_or_else(missing_req_info_error)
}
fn ext_req_info_mut(ext: &mut http::Extensions) -> S3Result<&mut ReqInfo> {
@@ -740,17 +768,10 @@ fn ext_req_info_mut(ext: &mut http::Extensions) -> S3Result<&mut ReqInfo> {
/// Extract the canonical `RequestContext` from a request, checking both
/// the request extensions directly and the `ReqInfo.request_context` field.
pub(crate) fn request_context_from_req<T>(req: &S3Request<T>) -> Option<RequestContext> {
request_context_from_extensions(&req.extensions)
}
/// Same lookup against the extensions alone, for callers that have already
/// moved a field out of the request and can no longer borrow it whole.
pub(crate) fn request_context_from_extensions(extensions: &http::Extensions) -> Option<RequestContext> {
extensions
.get::<RequestContext>()
pub(crate) fn request_context_from_req<R: AuthorizationRequest + ?Sized>(req: &R) -> Option<RequestContext> {
req.request_context()
.cloned()
.or_else(|| extensions.get::<ReqInfo>().and_then(|ri| ri.request_context.clone()))
.or_else(|| req.req_info().and_then(|req_info| req_info.request_context.clone()))
}
#[derive(Clone, Debug)]
@@ -990,6 +1011,18 @@ pub(crate) trait AuthorizationRequest {
fn server_context(&self) -> Option<&Arc<ServerContextSlot>>;
fn req_info(&self) -> Option<&ReqInfo>;
fn req_info_mut(&mut self) -> Option<&mut ReqInfo>;
fn request_context(&self) -> Option<&RequestContext>;
fn bucket_generation(&self) -> Option<&BucketGenerationGuard>;
fn install_bucket_generation(&mut self, guard: BucketGenerationGuard);
fn pending_delete_bucket_generation(&self) -> Option<&PendingDeleteBucketGenerationGuard>;
fn install_pending_delete_bucket_generation(&mut self, guard: PendingDeleteBucketGenerationGuard);
fn copy_source_bucket_generation(&self) -> Option<&CopySourceBucketGenerationGuard>;
fn odm_read_generation(&self) -> Option<&OdmReadGenerationGuard>;
fn install_odm_read_generation(&mut self, guard: OdmReadGenerationGuard);
fn bucket_config_mutation(&self) -> Option<&BucketConfigMutationSnapshot>;
/// The extensions no typed accessor covers, for test-only payloads.
#[cfg(test)]
fn test_extensions(&self) -> &http::Extensions;
fn internal_object_authorization(&self) -> bool;
fn multipart_staging(&self) -> bool;
fn staged_multipart_part(&self) -> bool;
@@ -1035,6 +1068,37 @@ impl<T> AuthorizationRequest for S3Request<T> {
fn req_info_mut(&mut self) -> Option<&mut ReqInfo> {
self.extensions.get_mut::<ReqInfo>()
}
fn request_context(&self) -> Option<&RequestContext> {
self.extensions.get::<RequestContext>()
}
fn bucket_generation(&self) -> Option<&BucketGenerationGuard> {
self.extensions.get::<BucketGenerationGuard>()
}
fn install_bucket_generation(&mut self, guard: BucketGenerationGuard) {
self.extensions.insert(guard);
}
fn pending_delete_bucket_generation(&self) -> Option<&PendingDeleteBucketGenerationGuard> {
self.extensions.get::<PendingDeleteBucketGenerationGuard>()
}
fn install_pending_delete_bucket_generation(&mut self, guard: PendingDeleteBucketGenerationGuard) {
self.extensions.insert(guard);
}
fn copy_source_bucket_generation(&self) -> Option<&CopySourceBucketGenerationGuard> {
self.extensions.get::<CopySourceBucketGenerationGuard>()
}
fn odm_read_generation(&self) -> Option<&OdmReadGenerationGuard> {
self.extensions.get::<OdmReadGenerationGuard>()
}
fn install_odm_read_generation(&mut self, guard: OdmReadGenerationGuard) {
self.extensions.insert(guard);
}
fn bucket_config_mutation(&self) -> Option<&BucketConfigMutationSnapshot> {
self.extensions.get::<BucketConfigMutationSnapshot>()
}
#[cfg(test)]
fn test_extensions(&self) -> &http::Extensions {
&self.extensions
}
fn internal_object_authorization(&self) -> bool {
self.extensions.get::<InternalObjectAuthorization>().is_some()
}
@@ -1349,12 +1413,11 @@ fn retain_internal_object_authorization_conditions(
/// Lives here rather than at the call site because this module already owns the request
/// plumbing every such decision needs: the verified client address and `ReqInfo`. The KMS
/// grammar has no bucket/object/tag conditions, so none of the S3 merges apply.
pub(crate) fn resource_free_condition_values<T>(
req: &S3Request<T>,
pub(crate) fn resource_free_condition_values<R: AuthorizationRequest + ?Sized>(
req: &R,
cred: &rustfs_credentials::Credentials,
) -> HashMap<String, Vec<String>> {
let remote_addr = req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0));
get_condition_values_with_client_info(&req.headers, cred, None, None, remote_addr, req.extensions.get::<ClientInfo>())
get_condition_values_with_client_info(req.headers(), cred, None, None, req.remote_addr(), req.client_info())
}
fn request_object_store<R: AuthorizationRequest + ?Sized>(req: &R) -> S3Result<Arc<ECStore>> {
@@ -1692,7 +1755,7 @@ pub(crate) fn delete_object_authorize_action(version_id: Option<&str>) -> Action
/// Note: `ActionSet::is_match` still maps a `s3:GetObjectVersion` grant onto a
/// `s3:GetObject` request. It only broadens a Version grant toward current
/// reads, never the reverse.
fn versioned_read_action(version_id: Option<&str>) -> Action {
pub(crate) fn versioned_read_action(version_id: Option<&str>) -> Action {
if version_id.is_some() {
Action::S3Action(S3Action::GetObjectVersionAction)
} else {
@@ -1700,7 +1763,7 @@ fn versioned_read_action(version_id: Option<&str>) -> Action {
}
}
fn versioned_tagging_action(action: S3Action, version_id: Option<&str>) -> Action {
pub(crate) fn versioned_tagging_action(action: S3Action, version_id: Option<&str>) -> Action {
let action = if version_id.is_some() {
match action {
S3Action::GetObjectTaggingAction => S3Action::GetObjectVersionTaggingAction,
@@ -1913,6 +1976,29 @@ pub(crate) async fn authorize_bucket_request<R: AuthorizationRequest + ?Sized>(
authorize_request(req, action).await
}
/// Authorizes `action` on one object version for either edge: the whole access hook of an
/// operation that names a bucket, a key and an optional version and asks one action, such as
/// GetObjectAcl. A caller whose action depends on the version picks it first
/// ([`versioned_read_action`], [`versioned_tagging_action`]).
///
/// # Errors
///
/// `InternalError` when the access phase installed no `ReqInfo`; otherwise what
/// [`authorize_request`] refuses.
pub(crate) async fn authorize_object_request<R: AuthorizationRequest + ?Sized>(
req: &mut R,
bucket: &str,
key: &str,
version_id: Option<&str>,
action: Action,
) -> S3Result<()> {
let req_info = req_info_mut(req)?;
req_info.bucket = Some(bucket.to_owned());
req_info.object = Some(key.to_owned());
req_info.version_id = version_id.map(str::to_owned);
authorize_request(req, action).await
}
/// Authorize the exact tag set selected by CopyObject's source reader before publication.
/// The access hook also calls this with a preliminary snapshot, so tag-dependent
/// destination Allow statements can be evaluated before opening either object.
@@ -2942,13 +3028,13 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn abort_multipart_upload(&self, req: &mut S3Request<AbortMultipartUploadInput>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
authorize_request(req, Action::S3Action(S3Action::AbortMultipartUploadAction)).await?;
req.extensions.insert(bucket_generation?);
req.install_bucket_generation(bucket_generation?);
Ok(())
}
@@ -2957,7 +3043,7 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn complete_multipart_upload(&self, req: &mut S3Request<CompleteMultipartUploadInput>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
@@ -2966,7 +3052,7 @@ impl S3Access for FS {
let specific_etag = req.input.if_match.as_ref().is_some_and(|etag| !etag.is_any());
authorize_write_etag_condition(req, specific_etag).await?;
authorize_replication_only_put_headers(req).await?;
req.extensions.insert(bucket_generation?);
req.install_bucket_generation(bucket_generation?);
Ok(())
}
@@ -2975,7 +3061,7 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn copy_object(&self, req: &mut S3Request<CopyObjectInput>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
let (src_bucket, src_key, version_id) = match &req.input.copy_source {
CopySource::AccessPoint { .. } => return Err(s3_error!(NotImplemented)),
CopySource::Outpost { .. } => return Err(s3_error!(NotImplemented)),
@@ -3071,14 +3157,14 @@ impl S3Access for FS {
authorize_request(req, Action::S3Action(S3Action::PutObjectRetentionAction)).await?;
}
req.extensions.insert(bucket_generation?);
req.install_bucket_generation(bucket_generation?);
Ok(())
}
/// Checks whether the CreateMultipartUpload request has accesses to the resources.
async fn create_multipart_upload(&self, req: &mut S3Request<CreateMultipartUploadInput>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
@@ -3098,7 +3184,7 @@ impl S3Access for FS {
authorize_request(req, Action::S3Action(S3Action::PutObjectRetentionAction)).await?;
}
req.extensions.insert(bucket_generation?);
req.install_bucket_generation(bucket_generation?);
Ok(())
}
@@ -3250,7 +3336,7 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn delete_object(&self, req: &mut S3Request<DeleteObjectInput>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
#[cfg(test)]
wait_for_object_generation_test_hook(&bucket).await;
// Preserve DeleteObject's established NoSuchBucket response instead of
@@ -3312,8 +3398,7 @@ impl S3Access for FS {
authorize_request(req, Action::S3Action(S3Action::BypassGovernanceRetentionAction)).await?;
}
req.extensions
.insert(PendingDeleteBucketGenerationGuard::from_result(bucket_generation));
req.install_pending_delete_bucket_generation(PendingDeleteBucketGenerationGuard::from_result(bucket_generation));
Ok(())
}
@@ -3564,7 +3649,7 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn get_object(&self, req: &mut S3Request<GetObjectInput>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let source_generation = capture_odm_read_generation(self, req, &bucket).await;
let source_generation = capture_odm_read_generation(self.server_ctx(), req, &bucket).await;
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
@@ -3573,7 +3658,7 @@ impl S3Access for FS {
// GHSA-3ppv: a versioned read (?versionId=...) must authorize against
// s3:GetObjectVersion, not s3:GetObject.
authorize_request(req, versioned_read_action(req.input.version_id.as_deref())).await?;
req.extensions.insert(source_generation);
req.install_odm_read_generation(source_generation);
Ok(())
}
@@ -3581,12 +3666,8 @@ impl S3Access for FS {
///
/// This method returns `Ok(())` by default.
async fn get_object_acl(&self, req: &mut S3Request<GetObjectAclInput>) -> S3Result<()> {
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, Action::S3Action(S3Action::GetObjectAclAction)).await
let (bucket, key, version_id) = (req.input.bucket.clone(), req.input.key.clone(), req.input.version_id.clone());
authorize_object_request(req, &bucket, &key, version_id.as_deref(), Action::S3Action(S3Action::GetObjectAclAction)).await
}
/// Checks whether the GetObjectAttributes request has accesses to the resources.
@@ -3613,12 +3694,15 @@ impl S3Access for FS {
///
/// This method returns `Ok(())` by default.
async fn get_object_legal_hold(&self, req: &mut S3Request<GetObjectLegalHoldInput>) -> S3Result<()> {
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, Action::S3Action(S3Action::GetObjectLegalHoldAction)).await
let (bucket, key, version_id) = (req.input.bucket.clone(), req.input.key.clone(), req.input.version_id.clone());
authorize_object_request(
req,
&bucket,
&key,
version_id.as_deref(),
Action::S3Action(S3Action::GetObjectLegalHoldAction),
)
.await
}
/// Checks whether the GetObjectLockConfiguration request has accesses to the resources.
@@ -3635,12 +3719,15 @@ impl S3Access for FS {
///
/// This method returns `Ok(())` by default.
async fn get_object_retention(&self, req: &mut S3Request<GetObjectRetentionInput>) -> S3Result<()> {
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, Action::S3Action(S3Action::GetObjectRetentionAction)).await
let (bucket, key, version_id) = (req.input.bucket.clone(), req.input.key.clone(), req.input.version_id.clone());
authorize_object_request(
req,
&bucket,
&key,
version_id.as_deref(),
Action::S3Action(S3Action::GetObjectRetentionAction),
)
.await
}
/// Checks whether the GetObjectTagging request has accesses to the resources.
@@ -3689,7 +3776,7 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn head_object(&self, req: &mut S3Request<HeadObjectInput>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let source_generation = capture_odm_read_generation(self, req, &bucket).await;
let source_generation = capture_odm_read_generation(self.server_ctx(), req, &bucket).await;
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
@@ -3703,7 +3790,7 @@ impl S3Access for FS {
req_info_mut(req)?.replication_request_authorized = true;
}
req.extensions.insert(source_generation);
req.install_odm_read_generation(source_generation);
Ok(())
}
@@ -3760,13 +3847,13 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn list_multipart_uploads(&self, req: &mut S3Request<ListMultipartUploadsInput>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
authorize_request(req, Action::S3Action(S3Action::ListBucketMultipartUploadsAction)).await?;
prepare_table_data_plane_list_access(req, &bucket, S3Action::ListBucketMultipartUploadsAction).await?;
req.extensions.insert(bucket_generation?);
req.install_bucket_generation(bucket_generation?);
Ok(())
}
@@ -3797,12 +3884,12 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn list_objects_v2(&self, req: &mut S3Request<ListObjectsV2Input>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let source_generation = capture_odm_read_generation(self, req, &bucket).await;
let source_generation = capture_odm_read_generation(self.server_ctx(), req, &bucket).await;
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
authorize_bucket_listing(req, &bucket, S3Action::ListBucketAction).await?;
req.extensions.insert(source_generation);
req.install_odm_read_generation(source_generation);
Ok(())
}
@@ -3811,13 +3898,13 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn list_parts(&self, req: &mut S3Request<ListPartsInput>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
authorize_request(req, list_parts_authorize_action()).await?;
req.extensions.insert(bucket_generation?);
req.install_bucket_generation(bucket_generation?);
Ok(())
}
@@ -4035,7 +4122,7 @@ impl S3Access for FS {
let bucket = req.input.bucket.clone();
// Snapshot before authorization, but preserve AccessDenied precedence
// by exposing any bucket-state error only after authorization succeeds.
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
#[cfg(test)]
let bucket_generation = if PUT_GENERATION_READ_QUORUM_FAILURE.try_with(|_| ()).is_ok() {
Err(ApiError::from(StorageError::ErasureReadQuorum).into())
@@ -4048,7 +4135,7 @@ impl S3Access for FS {
authorize_request(req, Action::S3Action(S3Action::PutObjectAction)).await?;
let specific_etag = req.input.if_match.as_ref().is_some_and(|etag| !etag.is_any());
authorize_write_etag_condition(req, specific_etag).await?;
req.extensions.insert(bucket_generation?);
req.install_bucket_generation(bucket_generation?);
if req.method != http::Method::POST {
authorize_replication_only_put_headers(req).await?;
@@ -4071,12 +4158,8 @@ impl S3Access for FS {
///
/// This method returns `Ok(())` by default.
async fn put_object_acl(&self, req: &mut S3Request<PutObjectAclInput>) -> S3Result<()> {
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
req_info.version_id = req.input.version_id.clone();
authorize_request(req, Action::S3Action(S3Action::PutObjectAclAction)).await
let (bucket, key, version_id) = (req.input.bucket.clone(), req.input.key.clone(), req.input.version_id.clone());
authorize_object_request(req, &bucket, &key, version_id.as_deref(), Action::S3Action(S3Action::PutObjectAclAction)).await
}
/// Checks whether the PutObjectLegalHold request has accesses to the resources.
@@ -4091,9 +4174,9 @@ impl S3Access for FS {
req_info.version_id = req.input.version_id.clone();
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
authorize_request(req, Action::S3Action(S3Action::PutObjectLegalHoldAction)).await?;
req.extensions.insert(bucket_generation?);
req.install_bucket_generation(bucket_generation?);
Ok(())
}
@@ -4115,7 +4198,7 @@ impl S3Access for FS {
req_info.version_id = req.input.version_id.clone();
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
set_requested_object_lock_retention(req);
authorize_request(req, Action::S3Action(S3Action::PutObjectRetentionAction)).await?;
@@ -4124,7 +4207,7 @@ impl S3Access for FS {
authorize_request(req, Action::S3Action(S3Action::BypassGovernanceRetentionAction)).await?;
}
req.extensions.insert(bucket_generation?);
req.install_bucket_generation(bucket_generation?);
Ok(())
}
@@ -4160,7 +4243,7 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn restore_object(&self, req: &mut S3Request<RestoreObjectInput>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
#[cfg(test)]
wait_for_object_generation_test_hook(&bucket).await;
let req_info = ext_req_info_mut(&mut req.extensions)?;
@@ -4169,7 +4252,7 @@ impl S3Access for FS {
req_info.version_id = req.input.version_id.clone();
authorize_request(req, Action::S3Action(S3Action::RestoreObjectAction)).await?;
req.extensions.insert(bucket_generation?);
req.install_bucket_generation(bucket_generation?);
Ok(())
}
@@ -4189,13 +4272,13 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn upload_part(&self, req: &mut S3Request<UploadPartInput>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
let req_info = ext_req_info_mut(&mut req.extensions)?;
req_info.bucket = Some(req.input.bucket.clone());
req_info.object = Some(req.input.key.clone());
authorize_staged_multipart_part(req).await?;
req.extensions.insert(bucket_generation?);
req.install_bucket_generation(bucket_generation?);
Ok(())
}
@@ -4204,7 +4287,7 @@ impl S3Access for FS {
/// This method returns `Ok(())` by default.
async fn upload_part_copy(&self, req: &mut S3Request<UploadPartCopyInput>) -> S3Result<()> {
let bucket = req.input.bucket.clone();
let bucket_generation = load_bucket_generation(self, req, &bucket).await;
let bucket_generation = load_bucket_generation(self.server_ctx(), req, &bucket).await;
let (src_bucket, src_key, version_id) = match &req.input.copy_source {
CopySource::AccessPoint { .. } => return Err(s3_error!(NotImplemented)),
CopySource::Outpost { .. } => return Err(s3_error!(NotImplemented)),
@@ -4237,7 +4320,7 @@ impl S3Access for FS {
req_info.version_id = None;
authorize_staged_multipart_part(req).await?;
req.extensions.insert(bucket_generation?);
req.install_bucket_generation(bucket_generation?);
Ok(())
}
rustfs/src/storage/access_readers_test.rs
+444
View File
@@ -0,0 +1,444 @@
// 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.
//! Owned-versus-legacy parity of the state an access hook hands to a use case: every reader and
//! installer must answer the same through an s3s request and through the `RequestEnvelope` built
//! from it, with the state present and absent, and each answer is pinned to its expected value so
//! the parity cannot hold vacuously.
//!
//! Not covered here: policy decisions (`app/owned_authorization_test.rs`).
//! Upstream: the readers in `storage/access.rs`; downstream: none.
use super::{
Action, AuthorizationRequest, BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, HashMap,
HeaderMap, ObjectOptions, OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, RemoteAddr, ReqInfo, RequestContext,
S3Action, S3ErrorCode, S3Request, S3Result, TableDataPlanePublicationGuards, apply_bucket_generation_guard,
apply_copy_source_bucket_generation_guard, authorize_object_request, bucket_config_mutation_incarnation,
capture_odm_read_generation, odm_read_generation, replication_request_authorized, request_context_from_req,
resource_free_condition_values, retained_table_data_plane_publication_guards,
};
use crate::app::RequestEnvelope;
use crate::runtime_sources::ServerContextSlot;
use rustfs_trusted_proxies::ClientInfo;
use std::net::SocketAddr;
const BUCKET: &str = "guarded-bucket";
fn incarnation() -> uuid::Uuid {
uuid::Uuid::from_u128(0x7107)
}
fn guard(bucket: &str) -> BucketGenerationGuard {
BucketGenerationGuard {
bucket: bucket.to_owned(),
incarnation_id: incarnation(),
}
}
fn legacy(fill: impl FnOnce(&mut http::Extensions)) -> S3Request<()> {
let mut extensions = http::Extensions::new();
fill(&mut extensions);
S3Request {
input: (),
method: http::Method::GET,
uri: "/guarded-bucket/key".parse().expect("static uri"),
headers: HeaderMap::new(),
extensions,
credentials: None,
region: None,
service: None,
trailing_headers: None,
}
}
fn server_request(extensions: &mut http::Extensions) {
extensions.insert(ServerContextSlot::new());
}
/// The incarnation a reader bound, or the refusal's code and message.
type Answer = Result<Option<uuid::Uuid>, (S3ErrorCode, String)>;
fn answer<T>(result: S3Result<T>, bound: impl FnOnce(T) -> Option<uuid::Uuid>) -> Answer {
result
.map(bound)
.map_err(|err| (err.code().clone(), err.message().unwrap_or_default().to_owned()))
}
fn refused(code: S3ErrorCode, message: &str) -> Answer {
Err((code, message.to_owned()))
}
fn bucket_fence<R: AuthorizationRequest + ?Sized>(req: &R) -> Answer {
let mut opts = ObjectOptions::default();
let result = apply_bucket_generation_guard(req, BUCKET, &mut opts);
answer(result, |()| opts.expected_bucket_incarnation_id)
}
fn copy_source_fence<R: AuthorizationRequest + ?Sized>(req: &R) -> Answer {
let mut opts = ObjectOptions::default();
let result = apply_copy_source_bucket_generation_guard(req, BUCKET, &mut opts);
answer(result, |()| opts.expected_bucket_incarnation_id)
}
fn odm<R: AuthorizationRequest + ?Sized>(req: &R) -> Answer {
answer(odm_read_generation(req, BUCKET), |id| id)
}
fn config_mutation<R: AuthorizationRequest + ?Sized>(req: &R) -> Answer {
answer(bucket_config_mutation_incarnation(req, BUCKET), |id| id)
}
/// Asserts that `read` gives `expected` through both request forms of every case.
fn assert_cases(
read: fn(&S3Request<()>) -> Answer,
owned: fn(&RequestEnvelope) -> Answer,
cases: Vec<(&str, S3Request<()>, Answer)>,
) {
for (name, req, expected) in cases {
let envelope = RequestEnvelope::from_s3s(&req);
assert_eq!(read(&req), expected, "legacy form: {name}");
assert_eq!(owned(&envelope), expected, "owned form: {name}");
}
}
#[test]
fn bucket_generation_guard_reads_the_same_through_both_forms() {
let missing = "bucket generation guard is missing";
let mismatch = "bucket generation guard does not match request bucket";
assert_cases(
bucket_fence,
bucket_fence,
vec![
(
"captured guard binds",
legacy(|ext| {
ext.insert(guard(BUCKET));
}),
Ok(Some(incarnation())),
),
(
"pending-delete guard wins over the captured guard",
legacy(|ext| {
ext.insert(guard("other-bucket"));
ext.insert(PendingDeleteBucketGenerationGuard::Ready(guard(BUCKET)));
}),
Ok(Some(incarnation())),
),
(
"guard of another bucket",
legacy(|ext| {
ext.insert(guard("other-bucket"));
}),
refused(S3ErrorCode::InternalError, mismatch),
),
(
"failed pending-delete capture",
legacy(|ext| {
ext.insert(guard(BUCKET));
ext.insert(PendingDeleteBucketGenerationGuard::Failed {
code: S3ErrorCode::NoSuchBucket,
message: "bucket is gone".to_owned(),
});
}),
refused(S3ErrorCode::NoSuchBucket, "bucket is gone"),
),
(
"server request without a guard",
legacy(server_request),
refused(S3ErrorCode::InternalError, missing),
),
("direct use-case call without a guard", legacy(|_| {}), Ok(None)),
],
);
}
#[test]
fn copy_source_generation_guard_reads_the_same_through_both_forms() {
let copy_guard = |bucket: &str| CopySourceBucketGenerationGuard {
bucket: bucket.to_owned(),
incarnation_id: incarnation(),
};
assert_cases(
copy_source_fence,
copy_source_fence,
vec![
(
"captured guard binds",
legacy(|ext| {
ext.insert(copy_guard(BUCKET));
}),
Ok(Some(incarnation())),
),
(
"guard of another bucket",
legacy(|ext| {
ext.insert(copy_guard("other-bucket"));
}),
refused(
S3ErrorCode::InternalError,
"copy source bucket generation guard does not match request bucket",
),
),
(
"the destination guard does not stand in for the source",
legacy(|ext| {
server_request(ext);
ext.insert(guard(BUCKET));
}),
refused(S3ErrorCode::InternalError, "copy source bucket generation guard is missing"),
),
("direct use-case call without a guard", legacy(|_| {}), Ok(None)),
],
);
}
#[test]
fn odm_read_generation_reads_the_same_through_both_forms() {
assert_cases(
odm,
odm,
vec![
(
"captured source generation",
legacy(|ext| {
ext.insert(OdmReadGenerationGuard::Ready(guard(BUCKET)));
}),
Ok(Some(incarnation())),
),
(
"source generation of another bucket",
legacy(|ext| {
ext.insert(OdmReadGenerationGuard::Ready(guard("other-bucket")));
}),
refused(S3ErrorCode::InternalError, "source generation guard does not match request bucket"),
),
(
"failed capture surfaces its error",
legacy(|ext| {
ext.insert(OdmReadGenerationGuard::Failed {
code: S3ErrorCode::SlowDown,
message: "capture failed".to_owned(),
});
}),
refused(S3ErrorCode::SlowDown, "capture failed"),
),
(
"no source configured",
legacy(|ext| {
ext.insert(OdmReadGenerationGuard::Unavailable);
}),
Ok(None),
),
("nothing captured", legacy(|_| {}), Ok(None)),
],
);
}
#[test]
fn bucket_config_mutation_incarnation_reads_the_same_through_both_forms() {
let snapshot = |bucket: &str| BucketConfigMutationSnapshot {
bucket: bucket.to_owned(),
incarnation_id: incarnation(),
};
assert_cases(
config_mutation,
config_mutation,
vec![
(
"captured snapshot",
legacy(|ext| {
ext.insert(snapshot(BUCKET));
}),
Ok(Some(incarnation())),
),
(
"snapshot of another bucket",
legacy(|ext| {
ext.insert(snapshot("other-bucket"));
}),
refused(
S3ErrorCode::InternalError,
"bucket config mutation snapshot does not match request bucket",
),
),
(
"server request without a snapshot",
legacy(server_request),
refused(S3ErrorCode::InternalError, "bucket config mutation snapshot is missing"),
),
("direct use-case call without a snapshot", legacy(|_| {}), Ok(None)),
],
);
}
#[test]
fn installed_state_reads_back_through_the_envelope() {
let mut envelope = RequestEnvelope::from_s3s(&legacy(server_request));
assert_eq!(
bucket_fence(&envelope),
refused(S3ErrorCode::InternalError, "bucket generation guard is missing")
);
envelope.install_bucket_generation(guard(BUCKET));
assert_eq!(bucket_fence(&envelope), Ok(Some(incarnation())));
envelope.install_pending_delete_bucket_generation(PendingDeleteBucketGenerationGuard::Failed {
code: S3ErrorCode::NoSuchBucket,
message: "bucket is gone".to_owned(),
});
assert_eq!(bucket_fence(&envelope), refused(S3ErrorCode::NoSuchBucket, "bucket is gone"));
assert_eq!(odm(&envelope), Ok(None));
envelope.install_odm_read_generation(OdmReadGenerationGuard::Ready(guard("other-bucket")));
assert_eq!(
odm(&envelope),
refused(S3ErrorCode::InternalError, "source generation guard does not match request bucket")
);
}
#[tokio::test]
async fn odm_capture_without_a_configured_source_is_unavailable_in_both_forms() {
let slot = ServerContextSlot::new();
let req = legacy(|ext| {
ext.insert(guard(BUCKET));
});
let envelope = RequestEnvelope::from_s3s(&req);
for captured in [
capture_odm_read_generation(&slot, &req, BUCKET).await,
capture_odm_read_generation(&slot, &envelope, BUCKET).await,
] {
assert!(matches!(captured, OdmReadGenerationGuard::Unavailable), "{captured:?}");
}
}
#[test]
fn replication_authorization_flag_reads_the_same_through_both_forms() {
let with_flag = |flag: bool| {
legacy(move |ext| {
ext.insert(ReqInfo {
replication_request_authorized: flag,
..Default::default()
});
})
};
for (name, req, expected) in [
("authorized replication request", with_flag(true), true),
("ordinary request", with_flag(false), false),
("request without ReqInfo", legacy(|_| {}), false),
] {
let envelope = RequestEnvelope::from_s3s(&req);
assert_eq!(replication_request_authorized(&req), expected, "legacy form: {name}");
assert_eq!(replication_request_authorized(&envelope), expected, "owned form: {name}");
}
}
#[test]
fn publication_guards_and_request_context_read_the_same_through_both_forms() {
let context = |id: &str| RequestContext {
request_id: id.to_owned(),
x_amz_request_id: id.to_owned(),
trace_id: None,
span_id: None,
start_time: std::time::Instant::now(),
};
let nested = |id: &str| ReqInfo {
request_context: Some(context(id)),
..Default::default()
};
for (name, req, expected) in [
(
"direct context",
legacy(|ext| {
ext.insert(context("direct"));
}),
Some("direct"),
),
(
"context inside ReqInfo",
legacy(|ext| {
ext.insert(nested("nested"));
}),
Some("nested"),
),
(
"the direct context wins",
legacy(|ext| {
ext.insert(context("direct"));
ext.insert(nested("nested"));
}),
Some("direct"),
),
("no context", legacy(|_| {}), None),
] {
let envelope = RequestEnvelope::from_s3s(&req);
let id = |found: Option<RequestContext>| found.map(|context| context.request_id);
assert_eq!(id(request_context_from_req(&req)).as_deref(), expected, "legacy form: {name}");
assert_eq!(id(request_context_from_req(&envelope)).as_deref(), expected, "owned form: {name}");
}
let with_guards = legacy(|ext| {
ext.insert(TableDataPlanePublicationGuards::default());
});
assert!(retained_table_data_plane_publication_guards(&with_guards).is_some());
assert!(retained_table_data_plane_publication_guards(&RequestEnvelope::from_s3s(&with_guards)).is_some());
let without = legacy(|_| {});
assert!(retained_table_data_plane_publication_guards(&without).is_none());
assert!(retained_table_data_plane_publication_guards(&RequestEnvelope::from_s3s(&without)).is_none());
}
#[test]
fn resource_free_conditions_read_the_same_client_through_both_forms() {
let peer: SocketAddr = "192.0.2.2:9000".parse().expect("peer address");
let client: SocketAddr = "198.51.100.7:443".parse().expect("client address");
let cred = rustfs_credentials::Credentials {
access_key: "condition-user".to_owned(),
..Default::default()
};
let source_ip = |conditions: &HashMap<String, Vec<String>>| conditions.get("SourceIp").cloned();
for (name, req, expected) in [
(
"forwarded client wins over the peer",
legacy(|ext| {
ext.insert(Some(RemoteAddr(peer)));
ext.insert(ClientInfo::direct(client));
}),
Some(vec!["198.51.100.7".to_owned()]),
),
(
"peer only",
legacy(|ext| {
ext.insert(Some(RemoteAddr(peer)));
}),
Some(vec!["192.0.2.2".to_owned()]),
),
] {
let envelope = RequestEnvelope::from_s3s(&req);
let legacy_conditions = resource_free_condition_values(&req, &cred);
assert_eq!(legacy_conditions, resource_free_condition_values(&envelope, &cred), "{name}");
assert_eq!(source_ip(&legacy_conditions), expected, "{name}");
}
}
#[tokio::test]
async fn object_request_without_req_info_is_refused_in_both_forms() {
let mut req = legacy(|_| {});
let mut envelope = RequestEnvelope::from_s3s(&req);
let action = Action::S3Action(S3Action::GetObjectAclAction);
for result in [
authorize_object_request(&mut req, BUCKET, "key", Some("v1"), action).await,
authorize_object_request(&mut envelope, BUCKET, "key", Some("v1"), action).await,
] {
let err = result.expect_err("a request without ReqInfo is refused");
assert_eq!(err.code(), &S3ErrorCode::InternalError);
assert_eq!(err.message(), Some("ReqInfo not found in request extensions"));
}
}
rustfs/src/storage/sse.rs
+4 -5
View File
@@ -75,7 +75,7 @@ use super::storage_api::ecstore_object::{
ReadEncryptionMode, ReadEncryptionRequest,
};
use crate::runtime_sources::current_kms_runtime_service_manager;
use crate::storage::access::{ReqInfo, request_context_from_req, resource_free_condition_values};
use crate::storage::access::{AuthorizationRequest, request_context_from_req, resource_free_condition_values};
use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
#[cfg(feature = "rio-v2")]
use aes_gcm::aead::Payload;
@@ -100,7 +100,6 @@ use rustfs_policy::policy::Args;
use rustfs_policy::policy::action::{Action, KmsAction};
use rustfs_utils::{get_env_bool, get_env_opt_str};
use s3s::S3ErrorCode;
use s3s::S3Request;
use s3s::dto::ServerSideEncryption;
use serde::{Deserialize, Serialize};
use serde_json::Value;
@@ -898,8 +897,8 @@ impl SseKmsPrincipal {
/// Anonymous requests use an empty account, matching the bucket-policy authorization
/// path. This keeps every request that crossed the S3 boundary distinct from trusted
/// internal callers, which are represented by `None`.
pub(crate) fn from_request<T>(req: &S3Request<T>) -> Option<Self> {
let req_info = req.extensions.get::<ReqInfo>()?;
pub(crate) fn from_request<R: AuthorizationRequest + ?Sized>(req: &R) -> Option<Self> {
let req_info = req.req_info()?;
let cred = req_info.cred.as_ref();
Some(Self {
@@ -910,7 +909,7 @@ impl SseKmsPrincipal {
conditions: cred.map(|cred| resource_free_condition_values(req, cred)).unwrap_or_default(),
request_audit: request_context_from_req(req).and_then(|context| kms_request_audit(&context.request_id)),
#[cfg(test)]
test_hooks: req.extensions.get::<TestAuthorizationHooks>().cloned(),
test_hooks: req.test_extensions().get::<TestAuthorizationHooks>().cloned(),
})
}
rustfs/src/storage/storage_api.rs
+1 -1
View File
@@ -175,7 +175,7 @@ pub(crate) mod owned_parity_consumer {
pub(crate) use super::super::access::{
ObjectTagConditions, PostObjectPolicyHeaders, PostObjectRequestMarker, ReqInfo, RequestedCopySource,
RequestedObjectLegalHold, RequestedObjectLockRetainUntil, RequestedObjectLockRetention, RequestedObjectTags,
TableDataPlanePublicationGuards, authorize_request, parity_test_support,
TableDataPlanePublicationGuards, authorize_object_request, authorize_request, parity_test_support,
};
pub(crate) use super::super::denial_audit::{DenialAuditor, DenialRateLimiter, SUPPRESSED_DENIALS_TAG};
pub(crate) use super::super::ecfs::FS;