refactor(access): one owned copy-source authorization for both copies

CopyObject and UploadPartCopy authorized their copy source with two
identical blocks in their S3Access hooks. Both hooks now call
`access_owned::object_commit::authorize_copy_source`, generic over
`AuthorizationRequest`, which the gateway stack's `Served::authorize`
for both operations will call too (batch E1, rustfs/backlog#2769). The
order is the hooks' order, unchanged: an access-point or Outposts ARN
source is NotImplemented before any policy is read; 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); a failed capture is surfaced only after the allow,
and the incarnation is handed on. The destination is decided afterwards
by each hook, as before. `AuthorizationRequest` gains the two installers
the shape needs, implemented for `S3Request` and `RequestEnvelope`.

Measurement for the batch (recorded here as the coordinator asked):
none of the three E1 operations commits a 200 before its work ends, so
the gateway edges will answer with `Resp::new`, not `Resp::commit`.
The pinned 1.0.5 reference binary could not be run on this host (the
asset is Linux x86-64 only and no container runtime was usable), so the
current legacy stack was measured instead: the debug binary at e8847f6f1,
default stack, on 127.0.0.1, driven by `curl --aws-sigv4`:
- CompleteMultipartUpload with a wrong part ETag: 400 InvalidPart with
  Content-Length; success: 200 with Content-Length, not chunked.
- CopyObject, missing source: 404 NoSuchKey; success: 200 with
  Content-Length.
- UploadPartCopy, missing source: 404 NoSuchKey; range beyond the
  source: 416 InvalidRange; success: 200 with Content-Length.
The 1.0.5 source agrees (tag 1.0.5, s3s 5761ddfe): s3s writes a
keep-alive body only for CompleteMultipartUpload and only when the
output's `future` member is set (crates/s3s/src/ops/multipart.rs:27);
RustFS never sets it, and CopyObject and UploadPartCopy have no such
path.

Not built: the destination half of either hook, the gateway edges
(they arrive with each operation's own increment), and the copy-source
read guard (task H's generic form is used as is).

Lines outside tests: +105 / -59 (access.rs, envelope.rs, the new
access_owned/object_commit.rs).

Tests (6 negative cases, 2 positive in the parity table; the hook test
has 2 negative cases and 1 control per operation): copy-source parity
on both edges, comparing decision,
recorded source, handed-on incarnation and ReqInfo for an ARN source,
a source without a read policy, a named version under s3:GetObject,
a deny with a failed capture, a failed capture after an allow, a
missing ReqInfo, and the two allows; and both legacy hooks refuse a
denied source and an ARN source without ReqInfo ever naming the
destination.

Mutations (two grouped runs of the two new tests, each reverted):
- ARN source answered Ok: parity "an ARN source is NotImplemented..."
  and hook "CopyObject: a denied source"/ARN case red.
- source decision skipped: parity "a source without a read policy..."
  and "a deny wins over a failed incarnation capture" red.
- version swap dropped (always s3:GetObject): parity "a named version
  needs s3:GetObjectVersion" red.
- capture failure surfaced before the decision: parity "a deny wins
  over a failed incarnation capture" and "a failed incarnation capture
  is surfaced after the allow" red.
- destination check before the source (CopyObject hook): hook
  "CopyObject: a denied source" red.

Not verified: UploadPartCopy's source-before-destination order is not
distinguishable by an answer on the legacy hook (both halves refuse
AccessDenied); it is pinned on the gateway edge with that operation.
The gateway stack is unchanged by this commit.

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)
     }

rustfs/src/storage/access.rs

@@ -681,16 +681,22 @@ pub(crate) fn odm_read_generation<R: AuthorizationRequest + ?Sized>(req: &R, buc
     }
 }
 
-async fn load_copy_source_bucket_generation(fs: &FS, bucket: &str) -> S3Result<CopySourceBucketGenerationGuard> {
-    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<String>)> {
+    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<R: AuthorizationRequest + ?Sized>(
@@ -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<T> AuthorizationRequest for S3Request<T> {
     fn requested_copy_source(&self) -> Option<&RequestedCopySource> {
         self.extensions.get::<RequestedCopySource>()
     }
+    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::<PostObjectPolicyHeaders>()
@@ -3080,31 +3094,11 @@ impl S3Access for FS {
     async fn copy_object(&self, req: &mut S3Request<CopyObjectInput>) -> 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<UploadPartCopyInput>) -> 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<IamSys<ObjectStore>> {
@@ -4711,7 +4685,7 @@ mod tests {
         }
     }
 
-    struct TestKms;
+    pub(super) struct TestKms;
 
     impl KmsInterface for TestKms {
         fn handle(&self) -> Arc<KmsServiceManager> {
@@ -4719,7 +4693,7 @@ mod tests {
         }
     }
 
-    fn build_request<T>(input: T, method: Method) -> S3Request<T> {
+    pub(super) fn build_request<T>(input: T, method: Method) -> S3Request<T> {
         S3Request {
             input,
             method,
@@ -4733,7 +4707,7 @@ mod tests {
         }
     }
 
-    fn ensure_req_info<T>(req: &mut S3Request<T>) {
+    pub(super) fn ensure_req_info<T>(req: &mut S3Request<T>) {
         req.extensions.insert(ReqInfo::default());
     }
 

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<R: AuthorizationRequest + ?Sized>(
+    req: &mut R,
+    server: &ServerContextSlot,
+    source: Option<CopySourceObject<'_>>,
+) -> 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<CopySourceBucketGenerationGuard> {
+    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;

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<S3ErrorCode>,
+    requested_source: Option<String>,
+    generation: Option<(String, uuid::Uuid)>,
+    req_info: Option<(Option<String>, Option<String>, Option<String>)>,
+}
+
+fn observe<R: AuthorizationRequest>(
+    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<ServerContextSlot>,
+    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::<CopySourceBucketGenerationGuard>();
+    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(), &copy_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}");
+    }
+}