fix(copy): authorize a versioned copy-source against its version

crates/core/src/api/request.rs

@@ -160,7 +160,15 @@ pub fn build_s3_operation(
                     raw_query: query.map(|q| q.to_string()),
                 })
             } else {
-                Ok(S3Operation::GetObject { bucket, key })
+                // `?versionId=` is deliberately not carried here: the read
+                // path does not address versions, so the backend returns the
+                // current object and authorizing a version would describe a
+                // read that never happens. See `S3Operation::GetObject::version`.
+                Ok(S3Operation::GetObject {
+                    bucket,
+                    key,
+                    version: None,
+                })
             }
         }
         Method::HEAD => Ok(S3Operation::HeadObject { bucket, key }),

crates/core/src/auth/authorize.rs

@@ -291,6 +291,60 @@ mod tests {
         .is_err());
     }
 
+    /// The hole this exists to close: on the bundled policy, a caller granted
+    /// `get_object` over a prefix must not be able to read — or copy out — an
+    /// older version of an object in it. A version-scoped read is its own
+    /// action, so the `get_object` grant simply does not match.
+    #[test]
+    fn get_object_grant_does_not_permit_a_versioned_read() {
+        let id = identity_with(AccessScope {
+            bucket: "b".into(),
+            prefixes: vec!["public/".into()],
+            actions: vec![Action::GetObject, Action::PutObject],
+        });
+        let versioned = S3Operation::GetObject {
+            bucket: "b".into(),
+            key: "public/src.txt".into(),
+            version: Some("v1".into()),
+        };
+        assert!(
+            authorize(&id, &versioned, &bucket("b", false)).is_err(),
+            "get_object must not imply reading previous versions"
+        );
+    }
+
+    /// ...and a caller who *is* granted the version action can, within their
+    /// prefix. Denying by default must stay expressible, not absolute.
+    #[test]
+    fn get_object_version_grant_permits_a_versioned_read_in_prefix() {
+        let id = identity_with(AccessScope {
+            bucket: "b".into(),
+            prefixes: vec!["public/".into()],
+            actions: vec![Action::GetObjectVersion],
+        });
+        let versioned = |key: &str| S3Operation::GetObject {
+            bucket: "b".into(),
+            key: key.into(),
+            version: Some("v1".into()),
+        };
+        assert!(authorize(&id, &versioned("public/src.txt"), &bucket("b", false)).is_ok());
+        // The prefix still binds.
+        assert!(authorize(&id, &versioned("private/src.txt"), &bucket("b", false)).is_err());
+    }
+
+    /// An anonymous caller never reads a previous version, even on a bucket
+    /// that allows anonymous reads: the anonymous grant covers current-object
+    /// reads only.
+    #[test]
+    fn versioned_read_denied_for_anonymous() {
+        let versioned = S3Operation::GetObject {
+            bucket: "b".into(),
+            key: "x.txt".into(),
+            version: Some("v1".into()),
+        };
+        assert!(authorize(&ResolvedIdentity::Anonymous, &versioned, &bucket("b", true)).is_err());
+    }
+
     /// The other half of a copy's authorization: the gateway resolves the copy
     /// source through a synthetic `GetObject`, so the source key is prefix-
     /// checked as a read. A caller scoped to `public/` cannot name a source
@@ -305,6 +359,7 @@ mod tests {
         let read = |key: &str| S3Operation::GetObject {
             bucket: "b".into(),
             key: key.into(),
+            version: None,
         };
         assert!(authorize(&id, &read("public/src.txt"), &bucket("b", false)).is_ok());
         assert!(authorize(&id, &read("private/src.txt"), &bucket("b", false)).is_err());

crates/core/src/proxy.rs

@@ -1084,6 +1084,7 @@ where
                 let src_op = S3Operation::GetObject {
                     bucket: src_bucket.clone(),
                     key: src_key.clone(),
+                    version: src_version.clone(),
                 };
                 let src = self
                     .bucket_registry
@@ -3404,7 +3405,12 @@ mod tests {
     #[derive(Clone)]
     struct RecordingRegistry {
         seen: Arc<std::sync::Mutex<Vec<(String, Action, String)>>>,
+        /// Versions seen on read authorizations, in order.
+        versions: Arc<std::sync::Mutex<Vec<Option<String>>>>,
         deny: Option<Action>,
+        /// Deny any read that names an object version, standing in for a
+        /// registry whose policy is not version-aware.
+        deny_versioned_reads: bool,
     }
 
     impl BucketRegistry for RecordingRegistry {
@@ -3419,9 +3425,23 @@ mod tests {
                 operation.action(),
                 operation.key().to_string(),
             ));
+            if let S3Operation::GetObject { version, .. } = operation {
+                self.versions.lock().unwrap().push(version.clone());
+            }
             if self.deny == Some(operation.action()) {
                 return Err(ProxyError::AccessDenied);
             }
+            if self.deny_versioned_reads
+                && matches!(
+                    operation,
+                    S3Operation::GetObject {
+                        version: Some(_),
+                        ..
+                    }
+                )
+            {
+                return Err(ProxyError::AccessDenied);
+            }
             Ok(ResolvedBucket {
                 config: test_bucket_config(name),
                 list_rewrite: None,
@@ -3452,7 +3472,9 @@ mod tests {
                 },
                 RecordingRegistry {
                     seen: seen.clone(),
+                    versions: Arc::new(std::sync::Mutex::new(Vec::new())),
                     deny: None,
+                    deny_versioned_reads: false,
                 },
                 MockCreds,
                 None,
@@ -3506,7 +3528,9 @@ mod tests {
                 },
                 RecordingRegistry {
                     seen: Arc::new(std::sync::Mutex::new(Vec::new())),
+                    versions: Arc::new(std::sync::Mutex::new(Vec::new())),
                     deny: Some(Action::GetObject),
+                    deny_versioned_reads: false,
                 },
                 MockCreds,
                 None,
@@ -3542,7 +3566,9 @@ mod tests {
                 },
                 RecordingRegistry {
                     seen: Arc::new(std::sync::Mutex::new(Vec::new())),
+                    versions: Arc::new(std::sync::Mutex::new(Vec::new())),
                     deny: Some(Action::PutObject),
+                    deny_versioned_reads: false,
                 },
                 MockCreds,
                 None,
@@ -3562,4 +3588,140 @@ mod tests {
             assert!(captured.lock().unwrap().is_none());
         });
     }
+
+    /// A versioned copy-source reads bytes that no other operation can reach —
+    /// the read path ignores `?versionId=` and serves the current object. The
+    /// version must therefore reach the registry on the source authorization,
+    /// or a policy that would reject it never sees it.
+    #[test]
+    fn copy_source_version_reaches_the_source_authorization() {
+        run(async {
+            let versions = Arc::new(std::sync::Mutex::new(Vec::new()));
+            let gw = ProxyGateway::new(
+                CaptureHeadersBackend {
+                    captured: Arc::new(std::sync::Mutex::new(None)),
+                },
+                RecordingRegistry {
+                    seen: Arc::new(std::sync::Mutex::new(Vec::new())),
+                    versions: versions.clone(),
+                    deny: None,
+                    deny_versioned_reads: false,
+                },
+                MockCreds,
+                None,
+            );
+            let mut headers = HeaderMap::new();
+            headers.insert(
+                "x-amz-copy-source",
+                "/src-bucket/obj.txt?versionId=v42".parse().unwrap(),
+            );
+            let action = gw
+                .resolve_request(Method::PUT, "/dst-bucket/obj.txt", None, &headers, None)
+                .await;
+            assert!(matches!(action, HandlerAction::Response(_)));
+            assert_eq!(
+                versions.lock().unwrap().clone(),
+                vec![Some("v42".to_string())],
+                "the source read must be authorized against the version it copies"
+            );
+        });
+    }
+
+    /// An unversioned copy-source authorizes an unversioned read — the version
+    /// field describes the read that actually happens, not the request shape.
+    #[test]
+    fn unversioned_copy_source_authorizes_an_unversioned_read() {
+        run(async {
+            let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
+            let gw = ProxyGateway::new(
+                CaptureHeadersBackend {
+                    captured: Arc::new(std::sync::Mutex::new(None)),
+                },
+                RecordingRegistry {
+                    seen: seen.clone(),
+                    versions: Arc::new(std::sync::Mutex::new(Vec::new())),
+                    deny: None,
+                    // Would reject a versioned read; this copy names no version.
+                    deny_versioned_reads: true,
+                },
+                MockCreds,
+                None,
+            );
+            let mut headers = HeaderMap::new();
+            headers.insert("x-amz-copy-source", "/src-bucket/obj.txt".parse().unwrap());
+            let action = gw
+                .resolve_request(Method::PUT, "/dst-bucket/obj.txt", None, &headers, None)
+                .await;
+            match action {
+                HandlerAction::Response(resp) => assert_eq!(resp.status, 200),
+                other => panic!(
+                    "expected 200 Response, got {:?}",
+                    std::mem::discriminant(&other)
+                ),
+            }
+        });
+    }
+
+    /// A registry whose policy is not version-aware can now refuse the read,
+    /// and the refusal lands before the backend is contacted — no versioned
+    /// bytes move.
+    #[test]
+    fn registry_can_deny_a_versioned_copy_source() {
+        run(async {
+            let captured = Arc::new(std::sync::Mutex::new(None));
+            let gw = ProxyGateway::new(
+                CaptureHeadersBackend {
+                    captured: captured.clone(),
+                },
+                RecordingRegistry {
+                    seen: Arc::new(std::sync::Mutex::new(Vec::new())),
+                    versions: Arc::new(std::sync::Mutex::new(Vec::new())),
+                    deny: None,
+                    deny_versioned_reads: true,
+                },
+                MockCreds,
+                None,
+            );
+            let mut headers = HeaderMap::new();
+            headers.insert(
+                "x-amz-copy-source",
+                "/src-bucket/obj.txt?versionId=v42".parse().unwrap(),
+            );
+            let action = gw
+                .resolve_request(Method::PUT, "/dst-bucket/obj.txt", None, &headers, None)
+                .await;
+            match action {
+                HandlerAction::Response(resp) => assert_eq!(resp.status, 403),
+                other => panic!(
+                    "expected 403 Response, got {:?}",
+                    std::mem::discriminant(&other)
+                ),
+            }
+            assert!(
+                captured.lock().unwrap().is_none(),
+                "a denied versioned read must never reach the backend"
+            );
+        });
+    }
+
+    /// A plain `GET` carrying `?versionId=` stays unversioned: the proxy does
+    /// not address versions on the read path, so authorizing it as versioned
+    /// would describe a read that never happens.
+    #[test]
+    fn plain_get_with_version_id_authorizes_an_unversioned_read() {
+        let headers = HeaderMap::new();
+        let op = crate::api::request::parse_s3_request(
+            &Method::GET,
+            "/b/obj.txt",
+            Some("versionId=v42"),
+            &headers,
+            crate::api::request::HostStyle::Path,
+            None,
+        )
+        .unwrap();
+        assert!(
+            matches!(op, S3Operation::GetObject { version: None, .. }),
+            "expected an unversioned GetObject, got {op:?}"
+        );
+    }
 }

crates/core/src/types.rs

@@ -189,6 +189,15 @@ pub struct AccessScope {
 #[serde(rename_all = "snake_case")]
 pub enum Action {
     GetObject,
+    /// Reading a *specific* object version, as opposed to the current one.
+    ///
+    /// A distinct action, mirroring S3's own split between `s3:GetObject` and
+    /// `s3:GetObjectVersion`: a scope that grants reads of an object does not
+    /// thereby grant reads of everything that object used to be. Prefix-scoped
+    /// policy cannot express "may read version X", so a caller must hold this
+    /// action explicitly — a grant of `GetObject` alone denies version-scoped
+    /// reads rather than silently permitting them.
+    GetObjectVersion,
     HeadObject,
     PutObject,
     ListBucket,
@@ -347,6 +356,24 @@ pub enum S3Operation {
     GetObject {
         bucket: String,
         key: String,
+        /// The object version this read will actually return, or `None` for the
+        /// current version.
+        ///
+        /// Set only where the backend read is genuinely version-scoped — today
+        /// that is the source half of a [`CopyObject`](Self::CopyObject), whose
+        /// `x-amz-copy-source` may carry `?versionId=`. A plain `GET` leaves
+        /// this `None` even when the client sends `?versionId=`, because the
+        /// proxy does not address versions on the read path and serves the
+        /// current object regardless; authorizing such a request as versioned
+        /// would describe a read that never happens.
+        ///
+        /// A `Some(_)` here changes the authorization action to
+        /// [`Action::GetObjectVersion`], so the shared policy
+        /// ([`authorize`](crate::auth::authorize)) denies it unless the caller
+        /// holds that action explicitly. Registries with their own policy
+        /// should make the same distinction: this names bytes that are
+        /// otherwise unreachable through the proxy.
+        version: Option<String>,
     },
     HeadObject {
         bucket: String,
@@ -439,7 +466,13 @@ impl S3Operation {
     /// The authorization action for this operation.
     pub fn action(&self) -> Action {
         match self {
-            S3Operation::GetObject { .. } => Action::GetObject,
+            // A version-scoped read is its own action: it reaches bytes that
+            // are otherwise unreachable, so holding `GetObject` must not imply
+            // it. See `Action::GetObjectVersion`.
+            S3Operation::GetObject { version: None, .. } => Action::GetObject,
+            S3Operation::GetObject {
+                version: Some(_), ..
+            } => Action::GetObjectVersion,
             S3Operation::HeadObject { .. } => Action::HeadObject,
             // A copy's destination is a write; the source read is authorized
             // separately at dispatch (see `execute_copy`).
@@ -503,6 +536,7 @@ mod tests {
         let op = S3Operation::GetObject {
             bucket: "b".into(),
             key: "k".into(),
+            version: None,
         };
         assert_eq!(op.action(), Action::GetObject);
 
@@ -532,6 +566,7 @@ mod tests {
         let op = S3Operation::GetObject {
             bucket: "my-bucket".into(),
             key: "k".into(),
+            version: None,
         };
         assert_eq!(op.bucket(), Some("my-bucket"));
 
@@ -543,6 +578,7 @@ mod tests {
         let op = S3Operation::GetObject {
             bucket: "b".into(),
             key: "my/key.txt".into(),
+            version: None,
         };
         assert_eq!(op.key(), "my/key.txt");
 

docs/configuration/roles.md

@@ -92,6 +92,7 @@ actions = ["get_object", "head_object", "put_object"]
 | Action | S3 Operation |
 |--------|-------------|
 | `get_object` | GET (download) |
+| `get_object_version` | GET/copy of a *specific* object version (`?versionId=`) |
 | `head_object` | HEAD (metadata) |
 | `put_object` | PUT (upload) |
 | `delete_object` | DELETE |

docs/reference/operations.md

@@ -35,6 +35,8 @@
 
 Credentials are *not* compared between the two ends. The copy is signed with the destination's credentials and the source contributes only its bucket name and prefix, so the source's own credentials are never used; whether the destination's credentials may read the source is an IAM question S3 answers with `AccessDenied`. (Comparing them would also be meaningless: a credential-injecting middleware such as `AwsBackendAuth` runs only against the destination config, so a source resolved from the same `auth_type=oidc` connection could never match.) The caller's own authorization is unaffected — the source is separately authorized as a read.
 
+**Versioned sources are authorized as versioned reads.** An `x-amz-copy-source` may carry `?versionId=`, and that version is copied — bytes the read path cannot otherwise reach, since a plain `GET` ignores `?versionId=` and serves the current object. The version is therefore carried on the synthetic `GetObject` used to authorize the source (`S3Operation::GetObject::version`), which makes the authorization action `get_object_version` rather than `get_object` — mirroring S3's own split between `s3:GetObject` and `s3:GetObjectVersion`. This is not merely advice to registry implementors: the bundled policy (`auth::authorize`, which `multistore-static-config` calls) enforces it, because a version-scoped read resolves to its own [action](../configuration/roles.md). A caller granted `get_object` over a prefix therefore cannot read — or copy out — an older version of an object in it, and an anonymous caller never can; the scope must list `get_object_version` explicitly. Prefix-scoped policy cannot express "may read version X", so the default is to deny rather than silently permit. A registry with its own policy should draw the same distinction. A plain `GET` leaves the field `None` even when the client sends `?versionId=` — authorizing it as versioned would describe a read that never happens.
+
 **Path-mapped deployments must map the copy-source.** `CopyObject` names its source in a header, so a proxy that rewrites client paths onto internal bucket names must rewrite the copy-source too — otherwise the source resolves as a client-facing name the registry has never heard of and every copy fails with `404 NoSuchBucket`. The client signed that header, so it must not be mutated; pass the mapped value alongside it via `RequestInfo::with_copy_source`, and signature verification keeps using the header as sent. `PathMapping::rewrite_copy_source` produces the mapped value:
 
 ```rust