diff --git a/rustfs/src/app/context/envelope.rs b/rustfs/src/app/context/envelope.rs index 573c17dde0..4d623adbdb 100644 --- a/rustfs/src/app/context/envelope.rs +++ b/rustfs/src/app/context/envelope.rs @@ -538,6 +538,12 @@ impl AuthorizationRequest for RequestEnvelope { fn requested_copy_source(&self) -> Option<&RequestedCopySource> { self.requested_copy_source() } + fn install_requested_copy_source(&mut self, source: RequestedCopySource) { + self.requested_copy_source = Some(source); + } + fn install_copy_source_bucket_generation(&mut self, guard: CopySourceBucketGenerationGuard) { + self.copy_source_bucket_generation = Some(guard); + } fn policy_headers(&self) -> &HeaderMap { self.post_object_policy_headers().map_or(self.headers(), |headers| &headers.0) } diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index 882195b676..18642439c6 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -681,16 +681,22 @@ pub(crate) fn odm_read_generation(req: &R, buc } } -async fn load_copy_source_bucket_generation(fs: &FS, bucket: &str) -> S3Result { - let store = fs - .server_ctx() - .object_store() - .ok_or_else(|| s3_error!(InternalError, "object store is not initialized"))?; - let incarnation_id = store.bucket_incarnation_id(bucket).await.map_err(ApiError::from)?; - Ok(CopySourceBucketGenerationGuard { - bucket: bucket.to_string(), - incarnation_id, - }) +/// The legacy decoder's copy source as owned parts; `None` for an access-point or Outposts ARN. +fn legacy_copy_source(source: &CopySource) -> Option<(String, String, Option)> { + match source { + CopySource::AccessPoint { .. } | CopySource::Outpost { .. } => None, + CopySource::Bucket { bucket, key, version_id } => { + Some((bucket.to_string(), key.to_string(), version_id.as_ref().map(|v| v.to_string()))) + } + } +} + +fn copy_source_object<'a>( + bucket: &'a str, + key: &'a str, + version_id: Option<&'a str>, +) -> access_owned::object_commit::CopySourceObject<'a> { + access_owned::object_commit::CopySourceObject { bucket, key, version_id } } pub(crate) fn apply_bucket_generation_guard( @@ -1046,6 +1052,8 @@ pub(crate) trait AuthorizationRequest { fn requested_legal_hold(&self) -> Option<&RequestedObjectLegalHold>; fn requested_object_tags(&self) -> Option<&RequestedObjectTags>; fn requested_copy_source(&self) -> Option<&RequestedCopySource>; + fn install_requested_copy_source(&mut self, source: RequestedCopySource); + fn install_copy_source_bucket_generation(&mut self, guard: CopySourceBucketGenerationGuard); /// The headers that describe what the request stores: a POST upload's form /// fields when present, otherwise the request headers. fn policy_headers(&self) -> &HeaderMap; @@ -1141,6 +1149,12 @@ impl AuthorizationRequest for S3Request { fn requested_copy_source(&self) -> Option<&RequestedCopySource> { self.extensions.get::() } + fn install_requested_copy_source(&mut self, source: RequestedCopySource) { + self.extensions.insert(source); + } + fn install_copy_source_bucket_generation(&mut self, guard: CopySourceBucketGenerationGuard) { + self.extensions.insert(guard); + } fn policy_headers(&self) -> &HeaderMap { self.extensions .get::() @@ -3080,31 +3094,11 @@ impl S3Access for FS { async fn copy_object(&self, req: &mut S3Request) -> S3Result<()> { let bucket = req.input.bucket.clone(); 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)), - CopySource::Bucket { bucket, key, version_id } => { - (bucket.to_string(), key.to_string(), version_id.as_ref().map(|v| v.to_string())) - } + let Some((src_bucket, src_key, version_id)) = legacy_copy_source(&req.input.copy_source) else { + return access_owned::object_commit::authorize_copy_source(req, self.server_ctx(), None).await; }; - let source = match version_id.as_deref() { - Some(version) => format!("{src_bucket}/{src_key}?versionId={}", crate::auth::canonical_policy_version_id(version)), - None => format!("{src_bucket}/{src_key}"), - }; - req.extensions.insert(RequestedCopySource(source)); - let source_bucket_generation = load_copy_source_bucket_generation(self, &src_bucket).await; - - { - let req_info = ext_req_info_mut(&mut req.extensions)?; - req_info.bucket = Some(src_bucket.clone()); - req_info.object = Some(src_key.clone()); - req_info.version_id = version_id.clone(); - - // GHSA-3ppv: a versioned copy source must authorize against - // s3:GetObjectVersion, not s3:GetObject. - authorize_request(req, versioned_read_action(version_id.as_deref())).await?; - } - req.extensions.insert(source_bucket_generation?); + let source = copy_source_object(&src_bucket, &src_key, version_id.as_deref()); + access_owned::object_commit::authorize_copy_source(req, self.server_ctx(), Some(source)).await?; let req_info = ext_req_info_mut(&mut req.extensions)?; @@ -4246,31 +4240,11 @@ impl S3Access for FS { async fn upload_part_copy(&self, req: &mut S3Request) -> S3Result<()> { let bucket = req.input.bucket.clone(); 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)), - CopySource::Bucket { bucket, key, version_id } => { - (bucket.to_string(), key.to_string(), version_id.as_ref().map(|v| v.to_string())) - } - }; - let source = match version_id.as_deref() { - Some(version) => format!("{src_bucket}/{src_key}?versionId={}", crate::auth::canonical_policy_version_id(version)), - None => format!("{src_bucket}/{src_key}"), - }; - req.extensions.insert(RequestedCopySource(source)); - let source_bucket_generation = load_copy_source_bucket_generation(self, &src_bucket).await; - - { - let req_info = ext_req_info_mut(&mut req.extensions)?; - req_info.bucket = Some(src_bucket.clone()); - req_info.object = Some(src_key.clone()); - req_info.version_id = version_id.clone(); - - // GHSA-3ppv: a versioned copy source must authorize against - // s3:GetObjectVersion, not s3:GetObject. - authorize_request(req, versioned_read_action(version_id.as_deref())).await?; - } - req.extensions.insert(source_bucket_generation?); + let source = legacy_copy_source(&req.input.copy_source); + let source = source + .as_ref() + .map(|(bucket, key, version_id)| copy_source_object(bucket, key, version_id.as_deref())); + access_owned::object_commit::authorize_copy_source(req, self.server_ctx(), source).await?; let req_info = ext_req_info_mut(&mut req.extensions)?; req_info.bucket = Some(req.input.bucket.clone()); @@ -4699,7 +4673,7 @@ mod tests { assert_eq!(unavailable.code(), &S3ErrorCode::InternalError); } - struct UnreadyIam; + pub(super) struct UnreadyIam; impl IamInterface for UnreadyIam { fn handle(&self) -> Arc> { @@ -4711,7 +4685,7 @@ mod tests { } } - struct TestKms; + pub(super) struct TestKms; impl KmsInterface for TestKms { fn handle(&self) -> Arc { @@ -4719,7 +4693,7 @@ mod tests { } } - fn build_request(input: T, method: Method) -> S3Request { + pub(super) fn build_request(input: T, method: Method) -> S3Request { S3Request { input, method, @@ -4733,7 +4707,7 @@ mod tests { } } - fn ensure_req_info(req: &mut S3Request) { + pub(super) fn ensure_req_info(req: &mut S3Request) { req.extensions.insert(ReqInfo::default()); } diff --git a/rustfs/src/storage/access_owned/object_commit.rs b/rustfs/src/storage/access_owned/object_commit.rs index 16700784e6..63b21978c5 100644 --- a/rustfs/src/storage/access_owned/object_commit.rs +++ b/rustfs/src/storage/access_owned/object_commit.rs @@ -20,3 +20,72 @@ //! policy evaluation itself (`authorize_request`), or the operation's use case. //! Upstream: the legacy hook in `storage/access.rs` and the batch's `server/gateway_handlers/object_commit.rs`. //! Downstream: the shared shapes in `storage/access.rs`. + +use super::super::{ + AuthorizationRequest, CopySourceBucketGenerationGuard, RequestedCopySource, S3Result, ServerContextSlot, authorize_request, + missing_req_info_error, s3_error, versioned_read_action, +}; +use crate::error::ApiError; + +/// The object a copy names as its source, as decoded from `x-amz-copy-source`. +pub(crate) struct CopySourceObject<'a> { + pub(crate) bucket: &'a str, + pub(crate) key: &'a str, + pub(crate) version_id: Option<&'a str>, +} + +/// The source half of a copy, which runs before anything about the destination is decided. +/// +/// `None` is an access-point or Outposts ARN source and is `NotImplemented` before any policy is +/// read. Otherwise the source is recorded for the `x-amz-copy-source` condition, its bucket +/// incarnation is captured, reading it is authorized (`s3:GetObjectVersion` when a version is +/// named) and only then is a failed capture surfaced or the incarnation handed on. `ReqInfo` is +/// left naming the source; the caller points it at the destination. +pub(crate) async fn authorize_copy_source( + req: &mut R, + server: &ServerContextSlot, + source: Option>, +) -> S3Result<()> { + let Some(source) = source else { + return Err(s3_error!(NotImplemented)); + }; + let requested = match source.version_id { + Some(version) => format!( + "{}/{}?versionId={}", + source.bucket, + source.key, + crate::auth::canonical_policy_version_id(version) + ), + None => format!("{}/{}", source.bucket, source.key), + }; + req.install_requested_copy_source(RequestedCopySource(requested)); + let generation = load_copy_source_bucket_generation(server, source.bucket).await; + + let req_info = req.req_info_mut().ok_or_else(missing_req_info_error)?; + req_info.bucket = Some(source.bucket.to_owned()); + req_info.object = Some(source.key.to_owned()); + req_info.version_id = source.version_id.map(str::to_owned); + // GHSA-3ppv: a versioned copy source must authorize against s3:GetObjectVersion. + authorize_request(req, versioned_read_action(source.version_id)).await?; + + req.install_copy_source_bucket_generation(generation?); + Ok(()) +} + +async fn load_copy_source_bucket_generation( + server: &ServerContextSlot, + bucket: &str, +) -> S3Result { + let store = server + .object_store() + .ok_or_else(|| s3_error!(InternalError, "object store is not initialized"))?; + let incarnation_id = store.bucket_incarnation_id(bucket).await.map_err(ApiError::from)?; + Ok(CopySourceBucketGenerationGuard { + bucket: bucket.to_string(), + incarnation_id, + }) +} + +#[cfg(test)] +#[path = "object_commit_test.rs"] +mod tests; diff --git a/rustfs/src/storage/access_owned/object_commit_test.rs b/rustfs/src/storage/access_owned/object_commit_test.rs new file mode 100644 index 0000000000..70e7beb3da --- /dev/null +++ b/rustfs/src/storage/access_owned/object_commit_test.rs @@ -0,0 +1,310 @@ +// 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. + +//! Parity of the copy-source authorization across both request edges: the same decision and the +//! same state handed on, for an `S3Request` and for a `RequestEnvelope`, in both directions. + +use super::{CopySourceObject, authorize_copy_source}; +use crate::app::{RequestEnvelope, RequestHead}; +use crate::storage::access::tests::{TestKms, UnreadyIam, build_request, ensure_req_info}; +use crate::storage::access::{ + AuthorizationRequest, CopyObjectInput, CopySource, CopySourceBucketGenerationGuard, ECStore, FS, ReqInfo, S3Access as _, + S3ErrorCode, S3Result, UploadPartCopyInput, +}; +use crate::storage::storage_api::contract::bucket::{BucketOperations as _, MakeBucketOptions}; +use crate::storage::storage_api::runtime_sources_consumer::{AppContext, ServerContextSlot}; +use http::{Extensions, HeaderMap, Method, Uri}; +use serial_test::serial; +use std::sync::Arc; + +const VERSION: &str = "0194e0f1-0000-7000-8000-000000000000"; + +/// What one call decided and handed on. +#[derive(Debug, PartialEq)] +struct Observed { + refusal: Option, + requested_source: Option, + generation: Option<(String, uuid::Uuid)>, + req_info: Option<(Option, Option, Option)>, +} + +fn observe( + req: &R, + result: S3Result<()>, + generation: Option<&CopySourceBucketGenerationGuard>, +) -> Observed { + Observed { + refusal: result.err().map(|error| error.code().clone()), + requested_source: req.requested_copy_source().map(|source| source.0.clone()), + generation: generation.map(|guard| (guard.bucket.clone(), guard.incarnation_id)), + req_info: req + .req_info() + .map(|info| (info.bucket.clone(), info.object.clone(), info.version_id.clone())), + } +} + +/// Runs the copy-source authorization once on each edge with the same facts: `served` is the +/// request's server context, `capture` the one the source incarnation is read from. +async fn on_both_edges( + served: &Arc, + capture: &ServerContextSlot, + with_req_info: bool, + source: Option<(&str, &str, Option<&str>)>, +) -> (Observed, Observed) { + let extensions = || { + let mut extensions = Extensions::new(); + if with_req_info { + extensions.insert(ReqInfo::default()); + } + extensions.insert(Arc::clone(served)); + extensions + }; + let object = || source.map(|(bucket, key, version_id)| CopySourceObject { bucket, key, version_id }); + + let mut legacy = build_request((), Method::PUT); + legacy.extensions = extensions(); + let result = authorize_copy_source(&mut legacy, capture, object()).await; + let generation = legacy.extensions.get::(); + let legacy = observe(&legacy, result, generation); + + let head = RequestHead { + method: Method::PUT, + uri: Uri::from_static("/"), + headers: HeaderMap::new(), + credentials: None, + region: None, + service: None, + }; + let mut envelope = RequestEnvelope::from_parts(head, extensions()); + let result = authorize_copy_source(&mut envelope, capture, object()).await; + let owned = observe(&envelope, result, envelope.copy_source_bucket_generation()); + (legacy, owned) +} + +async fn source_bucket(store: &ECStore, actions: &[&str]) -> String { + let bucket = format!("copy-source-parity-{}", uuid::Uuid::new_v4()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create the copy source bucket"); + if actions.is_empty() { + return bucket; + } + let policy = serde_json::json!({"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":"*"}, + "Action":actions,"Resource":[format!("arn:aws:s3:::{bucket}/*")]}]}) + .to_string(); + let mut metadata = (*crate::storage::get_bucket_metadata(&bucket) + .await + .expect("source bucket metadata is cached")) + .clone(); + metadata.policy_config = Some(serde_json::from_str(&policy).expect("source policy parses")); + metadata.policy_config_json = policy.into_bytes(); + crate::storage::storage_api::set_bucket_metadata(bucket.clone(), metadata) + .await + .expect("publish the source policy"); + bucket +} + +#[test] +#[serial] +fn copy_source_authorization_decides_and_hands_on_the_same_on_both_edges() { + crate::app::gating_test_env::run_large_stack_test("copy-source-parity", copy_source_parity_inner); +} + +async fn copy_source_parity_inner() { + let store = crate::app::gating_test_env::shared_gating_ecstore().await; + let server = Arc::new(ServerContextSlot::new()); + assert!(server.install(Arc::new(AppContext::new(Arc::clone(&store), Arc::new(UnreadyIam), Arc::new(TestKms))))); + let unready = Arc::new(ServerContextSlot::new()); + + let readable = source_bucket(&store, &["s3:GetObject"]).await; + let version_readable = source_bucket(&store, &["s3:GetObjectVersion"]).await; + let closed = source_bucket(&store, &[]).await; + let incarnation = |bucket: String| { + let store = Arc::clone(&store); + async move { + let id = store + .bucket_incarnation_id(&bucket) + .await + .expect("read the source incarnation"); + Some((bucket, id)) + } + }; + let named = + |bucket: &str, version: Option<&str>| Some((Some(bucket.to_owned()), Some("key".to_owned()), version.map(str::to_owned))); + let nothing_named = Some((None, None, None)); + + let denied = |requested: String, req_info| Observed { + refusal: Some(S3ErrorCode::AccessDenied), + requested_source: Some(requested), + generation: None, + req_info, + }; + let cases = [ + ( + "an ARN source is NotImplemented before any policy or state", + Arc::clone(&server), + true, + None, + Observed { + refusal: Some(S3ErrorCode::NotImplemented), + requested_source: None, + generation: None, + req_info: nothing_named.clone(), + }, + ), + ( + "a source without a read policy is refused and hands on no incarnation", + Arc::clone(&server), + true, + Some((closed.as_str(), "key", None)), + denied(format!("{closed}/key"), named(&closed, None)), + ), + ( + "a named version needs s3:GetObjectVersion, not s3:GetObject", + Arc::clone(&server), + true, + Some((readable.as_str(), "key", Some(VERSION))), + denied(format!("{readable}/key?versionId={VERSION}"), named(&readable, Some(VERSION))), + ), + ( + "a deny wins over a failed incarnation capture", + Arc::clone(&unready), + true, + Some((closed.as_str(), "key", None)), + denied(format!("{closed}/key"), named(&closed, None)), + ), + ( + "a failed incarnation capture is surfaced after the allow", + Arc::clone(&unready), + true, + Some((readable.as_str(), "key", None)), + Observed { + refusal: Some(S3ErrorCode::InternalError), + requested_source: Some(format!("{readable}/key")), + generation: None, + req_info: named(&readable, None), + }, + ), + ( + "a request without ReqInfo is refused as internal", + Arc::clone(&server), + false, + Some((readable.as_str(), "key", None)), + Observed { + refusal: Some(S3ErrorCode::InternalError), + requested_source: Some(format!("{readable}/key")), + generation: None, + req_info: None, + }, + ), + ( + "a readable source is allowed and hands on its incarnation", + Arc::clone(&server), + true, + Some((readable.as_str(), "key", None)), + Observed { + refusal: None, + requested_source: Some(format!("{readable}/key")), + generation: incarnation(readable.clone()).await, + req_info: named(&readable, None), + }, + ), + ( + "a readable version is allowed under s3:GetObjectVersion", + Arc::clone(&server), + true, + Some((version_readable.as_str(), "key", Some(VERSION))), + Observed { + refusal: None, + requested_source: Some(format!("{version_readable}/key?versionId={VERSION}")), + generation: incarnation(version_readable.clone()).await, + req_info: named(&version_readable, Some(VERSION)), + }, + ), + ]; + // Every case is checked before failing, so one run names every case a defect breaks. + let mut mismatches = Vec::new(); + for (case, slot, with_req_info, source, expected) in cases { + let (legacy, owned) = on_both_edges(&server, &slot, with_req_info, source).await; + if legacy != expected || owned != expected { + mismatches.push(format!("{case}: expected {expected:?}, legacy {legacy:?}, owned {owned:?}")); + } + } + assert!(mismatches.is_empty(), "{}", mismatches.join("\n")); +} + +#[test] +#[serial] +fn copy_hooks_decide_the_source_before_the_destination() { + crate::app::gating_test_env::run_large_stack_test("copy-source-first", copy_hooks_source_first_inner); +} + +/// The destination half refuses a `Tagging` header without the REPLACE directive as +/// `InvalidRequest`; that refusal must never be reached for a source the caller may not read. +async fn copy_hooks_source_first_inner() { + let store = crate::app::gating_test_env::shared_gating_ecstore().await; + let server = ServerContextSlot::new(); + assert!(server.install(Arc::new(AppContext::new(Arc::clone(&store), Arc::new(UnreadyIam), Arc::new(TestKms))))); + let fs = FS::with_server_ctx(server); + let closed = source_bucket(&store, &[]).await; + let readable = source_bucket(&store, &["s3:GetObject"]).await; + let destination = source_bucket(&store, &["s3:PutObject"]).await; + let arn = "arn:aws:s3:us-east-1:123456789012:accesspoint/my-access-point/object/key"; + + for (case, spelling, copy_expected, part_expected) in [ + ( + "a denied source", + format!("{closed}/key"), + S3ErrorCode::AccessDenied, + S3ErrorCode::AccessDenied, + ), + ("an ARN source", arn.to_owned(), S3ErrorCode::NotImplemented, S3ErrorCode::NotImplemented), + // Control: with the source readable, the destination half is reached and refuses. + ( + "a readable source", + format!("{readable}/key"), + S3ErrorCode::InvalidRequest, + S3ErrorCode::AccessDenied, + ), + ] { + let source = || CopySource::parse(&spelling).expect("the request parser accepts the source"); + let copy = CopyObjectInput::builder() + .bucket(destination.clone()) + .key("copy".to_string()) + .copy_source(source()) + .tagging(Some("a=b".to_string())) + .build() + .expect("copy input"); + let mut copy = build_request(copy, Method::PUT); + ensure_req_info(&mut copy); + copy.extensions.insert(fs.server_ctx().clone()); + let copy_error = fs.copy_object(&mut copy).await.expect_err(case); + assert_eq!(copy_error.code(), ©_expected, "CopyObject: {case}"); + + let part = UploadPartCopyInput::builder() + .bucket(closed.clone()) + .key("copy".to_string()) + .copy_source(source()) + .upload_id("unused".to_string()) + .part_number(1) + .build() + .expect("part copy input"); + let mut part = build_request(part, Method::PUT); + ensure_req_info(&mut part); + part.extensions.insert(fs.server_ctx().clone()); + let part_error = fs.upload_part_copy(&mut part).await.expect_err(case); + assert_eq!(part_error.code(), &part_expected, "UploadPartCopy: {case}"); + } +}