diff --git a/Cargo.lock b/Cargo.lock index 6bff51998..8b3dcbde2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4595,6 +4595,12 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" +[[package]] +name = "futures-timer" +version = "3.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af43fadb8a98512d547e37b4e92e0ced13e205c061b87b4623eff01d918d6968" + [[package]] name = "futures-util" version = "0.3.34" @@ -7850,10 +7856,34 @@ version = "0.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "010378780309880b08997fae13be7834dba947d36393bd372f2b1556deb2a2f6" dependencies = [ + "phf_macros", "phf_shared 0.14.0", "serde", ] +[[package]] +name = "phf_generator" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aeb62e0959d5a1bebc965f4d15d9e2b7cea002b6b0f5ba8cde6cc26738467100" +dependencies = [ + "fastrand", + "phf_shared 0.14.0", +] + +[[package]] +name = "phf_macros" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fa8d0ca26d424d27630da600c6624696e7dec8bf7b3b492b383c5dc49e5e085" +dependencies = [ + "phf_generator", + "phf_shared 0.14.0", + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "phf_shared" version = "0.12.1" @@ -8646,6 +8676,15 @@ version = "1.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0" +[[package]] +name = "quick-xml" +version = "0.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1" +dependencies = [ + "memchr", +] + [[package]] name = "quick-xml" version = "0.42.0" @@ -9534,7 +9573,7 @@ dependencies = [ "pin-project-lite", "proptest", "prost 0.14.4", - "quick-xml", + "quick-xml 0.42.0", "rand 0.10.2", "rcgen", "regex", @@ -9552,6 +9591,7 @@ dependencies = [ "rustfs-ecstore", "rustfs-extension-schema", "rustfs-filemeta", + "rustfs-gateway", "rustfs-heal", "rustfs-heal-contracts", "rustfs-iam", @@ -9917,6 +9957,131 @@ dependencies = [ "xxhash-rust", ] +[[package]] +name = "rustfs-gateway" +version = "0.38.1" +source = "git+https://github.com/rustfs/gateway.git?rev=90b83a20c44b636dd14681ce4e697fa46eda0b9b#90b83a20c44b636dd14681ce4e697fa46eda0b9b" +dependencies = [ + "arc-swap", + "bytes", + "futures-timer", + "futures-util", + "http 1.5.0", + "http-body 1.1.0", + "http-body-util", + "httparse", + "hyper", + "rustfs-gateway-core", + "rustfs-gateway-http", + "rustfs-gateway-macros", + "rustfs-gateway-sig", + "rustfs-gateway-stream", + "rustfs-gateway-types", + "rustfs-gateway-xml", + "sha2 0.11.0", + "tokio", + "tower", +] + +[[package]] +name = "rustfs-gateway-core" +version = "0.32.0" +source = "git+https://github.com/rustfs/gateway.git?rev=90b83a20c44b636dd14681ce4e697fa46eda0b9b#90b83a20c44b636dd14681ce4e697fa46eda0b9b" +dependencies = [ + "bytes", + "http 1.5.0", + "md-5 0.11.0", + "percent-encoding", + "rustfs-gateway-http", + "rustfs-gateway-sig", + "rustfs-gateway-stream", + "rustfs-gateway-types", + "rustfs-gateway-xml", + "serde", + "serde_json", + "subtle", + "zeroize", +] + +[[package]] +name = "rustfs-gateway-http" +version = "0.1.4" +source = "git+https://github.com/rustfs/gateway.git?rev=90b83a20c44b636dd14681ce4e697fa46eda0b9b#90b83a20c44b636dd14681ce4e697fa46eda0b9b" +dependencies = [ + "hmac 0.13.0", + "http 1.5.0", + "phf 0.14.0", + "rustfs-gateway-stream", + "rustfs-gateway-types", + "sha2 0.11.0", + "smallvec", + "subtle", + "zeroize", +] + +[[package]] +name = "rustfs-gateway-macros" +version = "0.1.4" +source = "git+https://github.com/rustfs/gateway.git?rev=90b83a20c44b636dd14681ce4e697fa46eda0b9b#90b83a20c44b636dd14681ce4e697fa46eda0b9b" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "rustfs-gateway-sig" +version = "0.11.1" +source = "git+https://github.com/rustfs/gateway.git?rev=90b83a20c44b636dd14681ce4e697fa46eda0b9b#90b83a20c44b636dd14681ce4e697fa46eda0b9b" +dependencies = [ + "hmac 0.13.0", + "http 1.5.0", + "rustfs-gateway-http", + "sha1 0.11.0", + "sha2 0.11.0", + "smallvec", + "subtle", + "zeroize", +] + +[[package]] +name = "rustfs-gateway-stream" +version = "0.2.0" +source = "git+https://github.com/rustfs/gateway.git?rev=90b83a20c44b636dd14681ce4e697fa46eda0b9b#90b83a20c44b636dd14681ce4e697fa46eda0b9b" +dependencies = [ + "bitflags 2.13.2", + "bytes", + "http 1.5.0", + "http-body 1.1.0", +] + +[[package]] +name = "rustfs-gateway-types" +version = "0.19.0+aws.2026-08-04" +source = "git+https://github.com/rustfs/gateway.git?rev=90b83a20c44b636dd14681ce4e697fa46eda0b9b#90b83a20c44b636dd14681ce4e697fa46eda0b9b" +dependencies = [ + "bytes", + "crc-fast", + "hex", + "http 1.5.0", + "md-5 0.11.0", + "percent-encoding", + "rustfs-gateway-stream", + "rustfs-gateway-xml", + "sha1 0.11.0", + "sha2 0.11.0", + "unicode-normalization", + "zeroize", +] + +[[package]] +name = "rustfs-gateway-xml" +version = "0.3.0" +source = "git+https://github.com/rustfs/gateway.git?rev=90b83a20c44b636dd14681ce4e697fa46eda0b9b#90b83a20c44b636dd14681ce4e697fa46eda0b9b" +dependencies = [ + "quick-xml 0.41.0", +] + [[package]] name = "rustfs-heal" version = "1.0.0-rc.6" @@ -10290,7 +10455,7 @@ dependencies = [ "jiff", "metrics", "percent-encoding", - "quick-xml", + "quick-xml 0.42.0", "rayon", "rustc-hash", "rustfs-config", @@ -10625,7 +10790,7 @@ dependencies = [ "hyper-util", "lazy_static", "md-5 0.11.0", - "quick-xml", + "quick-xml 0.42.0", "rand 0.10.2", "rustfs-checksums", "rustfs-config", @@ -11236,7 +11401,7 @@ dependencies = [ "numeric_cast", "phf 0.14.0", "pin-project-lite", - "quick-xml", + "quick-xml 0.42.0", "regex", "s3s-rfc2047", "s3s-sigv2", diff --git a/Cargo.toml b/Cargo.toml index 74001d08f..677112a98 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -314,6 +314,9 @@ rustix = { version = "1.1.4" } rust-embed = { version = "8.12.0" } rustc-hash = { version = "2.1.3" } s3s = { git = "https://github.com/s3s-project/s3s.git", rev = "f3e17541f366696bf0cbaf380fcbd8b44c17eba4", version = "0.15.0", features = ["minio"] } +# RUSTFS_S3_STACK=gateway HTTP boundary (rustfs/backlog#1752). One pinned git rev for every gateway crate; +# `server` stays off because RustFS owns the listener, TLS and connection lifecycle. +rustfs-gateway = { git = "https://github.com/rustfs/gateway.git", rev = "90b83a20c44b636dd14681ce4e697fa46eda0b9b", default-features = false } serial_test = "4.0.1" shadow-rs = { default-features = false, version = "2.0.0" } siphasher = "1.0.3" diff --git a/deny.toml b/deny.toml index cfecab82f..c1e2bc22a 100644 --- a/deny.toml +++ b/deny.toml @@ -48,6 +48,10 @@ allow-git = [ # RustFS fork carrying presigned expiry and constant-time authentication fixes. # owner: rustfs-maintainers review: 2026-10 "https://github.com/rustfs/s3s.git", + # RustFS Gateway HTTP boundary behind RUSTFS_S3_STACK=gateway (rustfs/backlog#1752), + # pinned to one reviewed git rev for every gateway crate. + # owner: rustfs-maintainers review: 2026-12 + "https://github.com/rustfs/gateway.git", ] [bans] diff --git a/docs/architecture/compat-cleanup-register.md b/docs/architecture/compat-cleanup-register.md index 41acc7227..ca4aedf1d 100644 --- a/docs/architecture/compat-cleanup-register.md +++ b/docs/architecture/compat-cleanup-register.md @@ -49,6 +49,7 @@ - `backlog-2097-tier-delete-journal-v6` tier-delete sole-owner recovery: v6 distinguishes transactions that may replace the xl.meta free-version owner and therefore require the all-pool live-source proof. v5-and-older readers reject and retain v6 records during rolling upgrades instead of performing an unsafe remote delete. Keep v1-v5 readers for upgrade recovery and keep the downgrade prohibition while any v6 record exists; retire the fence only after every supported rollback release understands and enforces v6 proof semantics. - `multipart-compression-default-off-window` staged multipart disk-compression rollout: releases before the resumable legacy decompressor fail transient reads of compressed objects under mid-payload suspension, so multipart uploads advertise the compression marker only when RUSTFS_COMPRESSION_MULTIPART_ENABLED is set in addition to RUSTFS_COMPRESSION_ENABLED, keeping rolling upgrades from creating new compressed multipart objects while pre-fix nodes may still serve reads. Flip the default to enabled (and retire the extra switch) after the minimum supported direct-upgrade release ships the resumable decompressor. - `s3gate-trailer-adapter` aws-chunked trailer handle bridge: s3s decodes aws-chunked request bodies and publishes their x-amz-checksum-* trailers through its own handle, while rio consumes trailers only through its framework-neutral TrailerSource trait. The RustFS application crate adapts the s3s handle at the object write path so rio carries no s3s dependency. Remove the adapter after the gateway stack replaces s3s as the request body decoder and publishes trailers through its own TrailerSource implementation. +- `backlog-1752-gateway-exchange` RustFS Gateway handler bridge: with RUSTFS_S3_STACK=gateway, gateway handlers receive the decoded input but neither the accepted request head nor the verified identity, which RustFS's access check and the s3s app bodies need. The stack entry captures the head, the gateway authorizer records the verified principal, and the handler rebuilds the s3s request from both through a request-scoped task-local that fails closed when missing or contradictory. The GetBucketLocation DTOs are mapped in RustFS because the gateway compat conversions are compiled against a different s3s revision than RustFS links. Remove the task-local and the local mapping after the pinned gateway rev carries the typed handler request context (rustfs/gateway#771, ADR-0022) and the handler builds the s3s request from it, and the gateway compat conversions target the s3s revision RustFS links. ## Review Checklist diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 2fa7e3247..5f8213111 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -348,6 +348,7 @@ pin-project-lite.workspace = true parking_lot = { workspace = true } rust-embed = { workspace = true, features = ["interpolate-folder-path"] } s3s = { workspace = true, features = ["minio"] } +rustfs-gateway = { workspace = true } shadow-rs = { workspace = true, default-features = false, features = ["build", "metadata"] } sysinfo = { workspace = true, features = ["multithread"] } thiserror = { workspace = true } diff --git a/rustfs/src/config/config_struct.rs b/rustfs/src/config/config_struct.rs index b8971c2c9..3cc418204 100644 --- a/rustfs/src/config/config_struct.rs +++ b/rustfs/src/config/config_struct.rs @@ -26,6 +26,65 @@ use rustfs_credentials::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY, Masked}; use std::collections::HashSet; use std::sync::{Mutex, OnceLock}; +/// The environment variable that selects the S3 HTTP stack (rustfs/backlog#1752). +pub const ENV_RUSTFS_S3_STACK: &str = "RUSTFS_S3_STACK"; + +/// Which HTTP boundary serves S3 requests. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum S3Stack { + /// The s3s service, byte-for-byte as before the switch. + #[default] + Legacy, + /// The RustFS Gateway pipeline for the operations it serves, the s3s service for the rest. + Gateway, +} + +impl S3Stack { + /// Parses the value of [`ENV_RUSTFS_S3_STACK`]. + /// + /// # Errors + /// + /// Anything other than exactly `legacy` or `gateway`, the empty string included: a typo must + /// stop the server instead of silently choosing a stack. + pub fn parse(value: &str) -> std::io::Result { + match value { + "legacy" => Ok(Self::Legacy), + "gateway" => Ok(Self::Gateway), + other => Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("invalid {ENV_RUSTFS_S3_STACK}={other:?}: expected \"legacy\" or \"gateway\""), + )), + } + } + + /// Reads [`ENV_RUSTFS_S3_STACK`]; unset selects [`S3Stack::Legacy`]. + /// + /// # Errors + /// + /// A set value that [`S3Stack::parse`] refuses, or one that is not valid Unicode. + pub fn from_env() -> std::io::Result { + match std::env::var_os(ENV_RUSTFS_S3_STACK) { + None => Ok(Self::Legacy), + Some(value) => match value.to_str() { + Some(value) => Self::parse(value), + None => Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("invalid {ENV_RUSTFS_S3_STACK}: the value is not valid Unicode"), + )), + }, + } + } + + /// The configuration spelling. + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Legacy => "legacy", + Self::Gateway => "gateway", + } + } +} + pub(crate) const LEGACY_ENV_RUSTFS_ROOT_USER: &str = "RUSTFS_ROOT_USER"; pub(crate) const LEGACY_ENV_RUSTFS_ROOT_PASSWORD: &str = "RUSTFS_ROOT_PASSWORD"; static LEGACY_CREDENTIAL_WARNED_KEYS: OnceLock>> = OnceLock::new(); @@ -148,6 +207,9 @@ pub struct Config { /// Workload profile for adaptive buffer sizing pub buffer_profile: String, + + /// S3 HTTP stack (`RUSTFS_S3_STACK`): `legacy` (default) or `gateway`. + pub s3_stack: S3Stack, } impl Config { @@ -179,6 +241,7 @@ impl Config { kms_allow_insecure_dev_defaults: false, buffer_profile_disable: false, buffer_profile: "GeneralPurpose".to_string(), + s3_stack: S3Stack::Legacy, } } @@ -256,6 +319,8 @@ impl Config { kms_allow_insecure_dev_defaults, buffer_profile_disable, buffer_profile, + // An invalid or empty value fails startup here instead of choosing a stack. + s3_stack: S3Stack::from_env()?, }) } @@ -298,6 +363,7 @@ impl std::fmt::Debug for Config { .field("kms_default_key_id", &self.kms_default_key_id) .field("kms_allow_insecure_dev_defaults", &self.kms_allow_insecure_dev_defaults) .field("buffer_profile_disable", &self.buffer_profile_disable) + .field("s3_stack", &self.s3_stack) .field("buffer_profile", &self.buffer_profile) .finish() } diff --git a/rustfs/src/config/mod.rs b/rustfs/src/config/mod.rs index 61c84bcbf..420a4a06e 100644 --- a/rustfs/src/config/mod.rs +++ b/rustfs/src/config/mod.rs @@ -73,7 +73,7 @@ pub use cli::{ConnectTopCaptureOpts, ConnectTopCommands}; pub use cli::{DiagnoseFormat, DiagnoseOpts}; pub use cli::{InspectBucketMetaOpts, InspectCommands, InspectOpts}; pub use cli::{TlsCommands, TlsInspectOpts, TlsOpts}; -pub use config_struct::Config; +pub use config_struct::{Config, ENV_RUSTFS_S3_STACK, S3Stack}; pub use info::execute_info; pub use opt::Opt; pub(crate) use snapshot::get_config_snapshot; diff --git a/rustfs/src/embedded.rs b/rustfs/src/embedded.rs index 430fad85a..200974d8b 100644 --- a/rustfs/src/embedded.rs +++ b/rustfs/src/embedded.rs @@ -275,6 +275,14 @@ impl RustFSServerBuilder { self } + /// Select the S3 HTTP stack (default: [`S3Stack::Legacy`](crate::config::S3Stack::Legacy)). + /// + /// The embedded server does not read `RUSTFS_S3_STACK`; this is its equivalent. + pub fn s3_stack(mut self, stack: crate::config::S3Stack) -> Self { + self.startup_args.set_s3_stack(stack); + self + } + /// Add a data volume path. /// /// If no volumes are added, a temporary directory with a single drive is diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 6caf84e2b..84c07caeb 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -30,6 +30,7 @@ use crate::server::{ }, rate_limit::{RateLimitLayer, api_rate_limit_layer_from_env}, ssec_transport::SsecTransportLayer, + stack::{GatewayFront, S3StackService}, strip_valid_port_suffix, tls_material::{ TlsAcceptFailure, TlsAcceptorHolder, TlsHandshakeFailureKind, accept_tls_with_deadline, build_acceptor_from_loaded, @@ -199,6 +200,7 @@ const EVENT_SOCKET_FALLBACK: &str = "socket_fallback"; const EVENT_HTTP_BIND_FAILED: &str = "http_bind_failed"; const EVENT_HTTP_STARTUP_ENDPOINTS: &str = "http_startup_endpoints"; const EVENT_HTTP_HOST_ROUTING: &str = "http_host_routing"; +const EVENT_S3_STACK_SELECTED: &str = "s3_stack_selected"; const EVENT_HTTP_COMPRESSION_STATE: &str = "http_compression_state"; const EVENT_API_RATE_LIMIT_STATE: &str = "api_rate_limit_state"; const EVENT_CONNECTION_CAP_STATE: &str = "connection_cap_state"; @@ -1302,6 +1304,34 @@ pub async fn start_http_server( b.build() }; + // RUSTFS_S3_STACK (rustfs/backlog#1752): `legacy` keeps the s3s service above as the whole S3 + // entry. `gateway` fronts it with the RustFS Gateway pipeline for the operations + // `stack::gateway_operation` classifies and hands it every other request. + let s3_service = S3StackService::assemble(config.s3_stack, s3_service, || { + let regions = vec![ + config + .region + .clone() + .unwrap_or_else(|| rustfs_config::RUSTFS_REGION.to_string()), + "us-east-1".to_string(), + ]; + let pipeline = storage::gateway::GatewayPipeline::build( + storage::ecfs::FS::with_server_ctx(server_ctx.clone()), + IAMAuth::with_server_context(config.access_key.clone(), config.secret_key.clone(), server_ctx.clone()), + ®ions, + )?; + info!( + event = EVENT_S3_STACK_SELECTED, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_HTTP, + stack = config.s3_stack.as_str(), + gateway_operations = "GetBucketLocation", + signing_regions = ?regions, + "S3 HTTP stack selected" + ); + Ok(GatewayFront::new(pipeline, s3_host_domains(config)?.unwrap_or_default())) + })?; + // Create shutdown channel let (shutdown_tx, mut shutdown_rx) = tokio::sync::broadcast::channel(1); // Create compression configuration from environment variables @@ -1672,7 +1702,7 @@ pub async fn start_http_server( #[derive(Clone)] struct ConnectionContext { http_server: Arc>, - s3_service: S3Service, + s3_service: S3StackService, compression_config: HttpCompressionConfig, is_console: bool, /// Whether `RUSTFS_SERVER_DOMAINS` is configured (i.e. s3s virtual-hosted-style routing is active). diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index 445c3ee9a..4bb54cf70 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -28,6 +28,7 @@ mod runtime; pub(crate) mod runtime_sources; mod service_state; mod ssec_transport; +mod stack; pub mod tls_material; use tracing::warn; diff --git a/rustfs/src/server/stack.rs b/rustfs/src/server/stack.rs new file mode 100644 index 000000000..c080b1316 --- /dev/null +++ b/rustfs/src/server/stack.rs @@ -0,0 +1,386 @@ +// 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. + +//! `RUSTFS_S3_STACK`: which HTTP boundary answers S3 requests (rustfs/backlog#1752). +//! +//! `legacy`, the default, serves every S3 request through the s3s service exactly as before the +//! switch existed: [`S3StackService::Legacy`] forwards each call to that service and does nothing +//! else. `gateway` puts the RustFS Gateway pipeline — wire acceptance, host resolution, routing, +//! SigV4 and the codecs — in front of the operations it is proven for, and hands every other +//! request to the same s3s service. +//! +//! # Fallback: route by operation before the body is read +//! +//! [`gateway_operation`] classifies a request from its method, path, query, `Host` and +//! authentication headers, before either stack has read a body byte. A request it does not +//! recognise exactly — another operation, a presigned or session-token request, a virtual-hosted +//! bucket, a Keystone token — goes to the legacy service untouched, so the fallback never replays +//! a body and every unrecognised shape keeps today's behaviour. A classified request is answered +//! by the gateway alone: a gateway refusal is never retried on s3s, which would let a request the +//! gateway rejected (a forged signature, say) be judged a second time by the other stack. +//! +//! The switch moves the wire boundary only. It selects no metadata writer: every persisted byte is +//! still produced by the unchanged s3s application layer, so going back to `legacy` is an +//! environment change and a restart. + +use std::future::Future; +use std::io; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; + +use bytes::Bytes; +use http::header::{AUTHORIZATION, HOST}; +use http::{HeaderMap, Method, Request as HttpRequest, Uri}; +use tower::Service; + +use super::{is_admin_path, strip_valid_port_suffix}; +use crate::config::S3Stack; +use crate::storage_api::server::http::gateway::{GatewayPipeline, HttpError, HttpResponse}; + +/// An operation the gateway stack serves. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum GatewayOperation { + GetBucketLocation, +} + +impl GatewayOperation { + /// The operation name both the s3s and the gateway router use. + pub(crate) const fn name(self) -> &'static str { + match self { + Self::GetBucketLocation => "GetBucketLocation", + } + } +} + +/// The operation the gateway stack answers `method uri` with, or `None` for the legacy stack. +/// +/// Deliberately narrower than the gateway router: it recognises only the exact request shapes this +/// slice has proven, and everything else stays on s3s. +pub(crate) fn gateway_operation( + method: &Method, + uri: &Uri, + headers: &HeaderMap, + vhost_domains: &[String], +) -> Option { + // `?location` and nothing else: a presigned query, `versionId` or a second subresource is + // another request shape that has not been compared yet. + if method != Method::GET || uri.query() != Some("location") { + return None; + } + let path = uri.path(); + let bucket = path.strip_prefix('/')?; + if bucket.is_empty() || bucket.contains('/') || is_admin_path(path) { + return None; + } + if !is_path_style_host(uri, headers, vhost_domains) { + return None; + } + // Session-token and Keystone requests authenticate through material the gateway credential + // bridge does not carry yet; header SigV4 and anonymous are the two proven shapes. + if headers.contains_key("x-amz-security-token") || headers.contains_key("x-auth-token") { + return None; + } + match headers.get(AUTHORIZATION) { + None => {} + Some(value) if value.as_bytes().starts_with(b"AWS4-HMAC-SHA256 ") => {} + Some(_) => return None, + } + Some(GatewayOperation::GetBucketLocation) +} + +/// Whether the request addresses its bucket in the path. +/// +/// The gateway stack resolves path-style only in this slice, so a host that is a label under a +/// configured `RUSTFS_SERVER_DOMAINS` domain — a virtual-hosted bucket — stays on s3s. +fn is_path_style_host(uri: &Uri, headers: &HeaderMap, vhost_domains: &[String]) -> bool { + let host = match headers.get(HOST) { + Some(value) => match value.to_str() { + Ok(host) => host, + Err(_) => return false, + }, + None => match uri.authority() { + Some(authority) => authority.as_str(), + None => return false, + }, + }; + if vhost_domains.is_empty() { + return true; + } + let host = strip_valid_port_suffix(host).trim_end_matches('.').to_ascii_lowercase(); + !vhost_domains.iter().any(|domain| { + let domain = strip_valid_port_suffix(domain).trim_end_matches('.').to_ascii_lowercase(); + host.strip_suffix(domain.as_str()) + .is_some_and(|label| label.len() > 1 && label.ends_with('.')) + }) +} + +/// The future both stacks answer with: the s3s service's own future type. +pub(crate) type StackFuture = Pin> + Send>>; + +/// The gateway half of a [`S3StackService::Gateway`] stack. +#[derive(Clone)] +pub(crate) struct GatewayFront { + pipeline: GatewayPipeline, + vhost_domains: Arc<[String]>, +} + +impl GatewayFront { + pub(crate) fn new(pipeline: GatewayPipeline, vhost_domains: Vec) -> Self { + Self { + pipeline, + vhost_domains: vhost_domains.into(), + } + } +} + +/// The S3 entry service for one listener. +#[derive(Clone)] +pub(crate) enum S3StackService { + /// Every request goes to the legacy service. + Legacy(L), + /// Classified requests go to the gateway, every other request to the legacy service. + Gateway { legacy: L, front: GatewayFront }, +} + +impl S3StackService { + /// Builds the entry for `stack` around the already-built legacy service. + /// + /// `gateway` runs only when `stack` is [`S3Stack::Gateway`], so the legacy stack constructs + /// nothing it did not construct before the switch. + /// + /// # Errors + /// + /// Whatever `gateway` returns: a gateway stack that cannot be assembled fails startup. + pub(crate) fn assemble(stack: S3Stack, legacy: L, gateway: impl FnOnce() -> io::Result) -> io::Result { + match stack { + S3Stack::Legacy => Ok(Self::Legacy(legacy)), + S3Stack::Gateway => Ok(Self::Gateway { + legacy, + front: gateway()?, + }), + } + } +} + +impl Service> for S3StackService +where + L: Service, Response = HttpResponse, Error = HttpError, Future = StackFuture>, + B: http_body::Body + Send + 'static, + B::Error: std::error::Error + Send + Sync + 'static, +{ + type Response = HttpResponse; + type Error = HttpError; + type Future = StackFuture; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + match self { + Self::Legacy(legacy) | Self::Gateway { legacy, .. } => legacy.poll_ready(cx), + } + } + + fn call(&mut self, request: HttpRequest) -> Self::Future { + match self { + Self::Legacy(legacy) => legacy.call(request), + Self::Gateway { legacy, front } => { + match gateway_operation(request.method(), request.uri(), request.headers(), &front.vhost_domains) { + Some(operation) => { + let pipeline = front.pipeline.clone(); + Box::pin(async move { Ok(pipeline.serve(operation.name(), request).await) }) + } + None => legacy.call(request), + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::ENV_RUSTFS_S3_STACK; + use crate::storage_api::server::http::gateway::Body; + use std::sync::Mutex; + + fn classify( + method: Method, + target: &str, + headers: &[(&'static str, &'static str)], + domains: &[&str], + ) -> Option { + let uri: Uri = target.parse().expect("test uri"); + let mut map = HeaderMap::new(); + map.insert(HOST, http::HeaderValue::from_static("127.0.0.1:9000")); + for (name, value) in headers { + map.insert(*name, http::HeaderValue::from_static(value)); + } + let domains: Vec = domains.iter().map(|domain| (*domain).to_string()).collect(); + gateway_operation(&method, &uri, &map, &domains) + } + + #[test] + fn unset_and_default_select_the_legacy_stack() { + assert_eq!(S3Stack::default(), S3Stack::Legacy); + temp_env::with_var_unset(ENV_RUSTFS_S3_STACK, || { + assert_eq!(S3Stack::from_env().expect("unset is legacy"), S3Stack::Legacy); + }); + } + + #[test] + fn both_spellings_parse_and_round_trip() { + for stack in [S3Stack::Legacy, S3Stack::Gateway] { + assert_eq!(S3Stack::parse(stack.as_str()).expect("own spelling"), stack); + } + temp_env::with_var(ENV_RUSTFS_S3_STACK, Some("gateway"), || { + assert_eq!(S3Stack::from_env().expect("gateway"), S3Stack::Gateway); + }); + } + + #[test] + fn an_invalid_or_empty_value_fails_instead_of_choosing_a_stack() { + for value in ["", "s3s", "Gateway", "legacy ", "gateway,legacy"] { + let error = S3Stack::parse(value).expect_err(value); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput, "{value:?}"); + assert!(error.to_string().contains(ENV_RUSTFS_S3_STACK), "{error}"); + } + temp_env::with_var(ENV_RUSTFS_S3_STACK, Some(""), || { + assert!(S3Stack::from_env().is_err(), "an empty value must fail startup"); + }); + } + + #[test] + fn the_default_stack_builds_the_legacy_service_and_never_the_gateway() { + let service = S3StackService::assemble(S3Stack::default(), "legacy-service", || { + panic!("the legacy stack must not assemble the gateway") + }) + .expect("legacy assembly cannot fail"); + assert!(matches!(service, S3StackService::Legacy("legacy-service"))); + } + + #[test] + fn a_gateway_assembly_failure_fails_startup() { + let result = S3StackService::assemble(S3Stack::Gateway, (), || Err(io::Error::other("assembly refused"))); + assert!(matches!(result, Err(error) if error.to_string() == "assembly refused")); + } + + /// A legacy service that records what it was handed and answers a fixed response. + #[derive(Clone, Default)] + struct RecordingLegacy { + seen: Arc>>, + } + + impl Service>> for RecordingLegacy { + type Response = HttpResponse; + type Error = HttpError; + type Future = StackFuture; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, request: HttpRequest>) -> Self::Future { + let (parts, _body) = request.into_parts(); + self.seen + .lock() + .expect("recording lock") + .push((parts.method, parts.uri, parts.headers)); + Box::pin(async { + let mut response = HttpResponse::new(Body::from(Bytes::from_static(b"legacy"))); + response.headers_mut().insert("x-legacy", http::HeaderValue::from_static("1")); + Ok(response) + }) + } + } + + #[tokio::test] + async fn the_legacy_stack_forwards_every_request_unchanged() { + let legacy = RecordingLegacy::default(); + let mut service = S3StackService::assemble(S3Stack::Legacy, legacy.clone(), || unreachable!()).expect("legacy"); + + // Even the request shape the gateway stack would take stays on the legacy service. + let request = HttpRequest::get("/bucket?location") + .header(HOST, "127.0.0.1:9000") + .header("x-amz-date", "20260914T000000Z") + .body(http_body_util::Full::new(Bytes::new())) + .expect("request"); + let response = service.call(request).await.expect("legacy answers"); + + assert_eq!(response.headers().get("x-legacy").map(|value| value.as_bytes()), Some(&b"1"[..])); + let seen = legacy.seen.lock().expect("recording lock"); + assert_eq!(seen.len(), 1); + let (method, uri, headers) = &seen[0]; + assert_eq!(method, Method::GET); + assert_eq!(uri, "/bucket?location"); + assert_eq!(headers.get("x-amz-date").map(|value| value.as_bytes()), Some(&b"20260914T000000Z"[..])); + } + + #[test] + fn get_bucket_location_is_the_only_gateway_shape() { + let signed = [("authorization", "AWS4-HMAC-SHA256 Credential=AK/20260914/us-east-1/s3/aws4_request")]; + assert_eq!( + classify(Method::GET, "/bucket?location", &signed, &[]), + Some(GatewayOperation::GetBucketLocation) + ); + assert_eq!( + classify(Method::GET, "/bucket?location", &[], &[]), + Some(GatewayOperation::GetBucketLocation), + "anonymous requests reach the gateway and RustFS access decides" + ); + } + + #[test] + fn every_other_shape_falls_back_to_legacy() { + type Case = (Method, &'static str, &'static [(&'static str, &'static str)]); + let cases: &[Case] = &[ + (Method::PUT, "/bucket/key", &[]), + (Method::GET, "/bucket", &[]), + (Method::HEAD, "/bucket?location", &[]), + (Method::GET, "/bucket/?location", &[]), + (Method::GET, "/bucket/key?location", &[]), + (Method::GET, "/?location", &[]), + (Method::GET, "/bucket?location=", &[]), + (Method::GET, "/bucket?location&versionId=1", &[]), + (Method::GET, "/bucket?location&X-Amz-Signature=00", &[]), + (Method::GET, "/rustfs/admin?location", &[]), + (Method::GET, "/bucket?location", &[("x-amz-security-token", "token")]), + (Method::GET, "/bucket?location", &[("x-auth-token", "token")]), + (Method::GET, "/bucket?location", &[("authorization", "AWS AK:signature")]), + ]; + for (method, target, headers) in cases { + assert_eq!(classify(method.clone(), target, headers, &[]), None, "{method} {target} {headers:?}"); + } + } + + #[test] + fn a_virtual_hosted_bucket_stays_on_legacy_but_the_base_domain_does_not() { + let uri: Uri = "/bucket?location".parse().expect("uri"); + let domains = vec!["s3.example.com".to_string()]; + let mut headers = HeaderMap::new(); + headers.insert(HOST, http::HeaderValue::from_static("photos.S3.example.com:9000")); + assert_eq!(gateway_operation(&Method::GET, &uri, &headers, &domains), None); + headers.insert(HOST, http::HeaderValue::from_static("s3.example.com:9000")); + assert_eq!( + gateway_operation(&Method::GET, &uri, &headers, &domains), + Some(GatewayOperation::GetBucketLocation) + ); + headers.insert(HOST, http::HeaderValue::from_static("nots3.example.com")); + assert_eq!( + gateway_operation(&Method::GET, &uri, &headers, &domains), + Some(GatewayOperation::GetBucketLocation), + "a suffix without a label boundary is not a virtual host" + ); + headers.remove(HOST); + assert_eq!(gateway_operation(&Method::GET, &uri, &headers, &domains), None, "no host stays legacy"); + } +} diff --git a/rustfs/src/startup_embedded.rs b/rustfs/src/startup_embedded.rs index 7738f85ba..8b3021e71 100644 --- a/rustfs/src/startup_embedded.rs +++ b/rustfs/src/startup_embedded.rs @@ -13,6 +13,7 @@ // limitations under the License. use crate::{ + config::S3Stack, server::ShutdownHandle, startup_lifecycle::{ EmbeddedStartupGuard, embedded_endpoint_address, log_embedded_server_ready, publish_embedded_startup_ready, @@ -37,6 +38,7 @@ pub(crate) struct EmbeddedStartupArgs { secret_key: String, volumes: Vec, region: String, + s3_stack: S3Stack, } impl EmbeddedStartupArgs { @@ -47,9 +49,14 @@ impl EmbeddedStartupArgs { secret_key: rustfs_credentials::DEFAULT_SECRET_KEY.to_string(), volumes: Vec::new(), region: rustfs_config::RUSTFS_REGION.to_string(), + s3_stack: S3Stack::Legacy, } } + pub(crate) fn set_s3_stack(&mut self, s3_stack: S3Stack) { + self.s3_stack = s3_stack; + } + pub(crate) fn set_address(&mut self, address: String) { self.address = address; } @@ -105,6 +112,7 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result Result S3Result<() Ok(()) } -#[async_trait::async_trait] -impl S3Access for FS { - async fn check(&self, cx: &mut S3AccessContext<'_>) -> S3Result<()> { +/// The request facts RustFS's access check reads. +/// +/// Implemented by the s3s access context and by the gateway bridge (rustfs/backlog#1752), so one +/// check body serves both stacks and the legacy stack reads exactly what it read before. +pub(crate) trait AccessCheckContext: Send + Sync { + fn credentials(&self) -> Option<&Credentials>; + fn uri(&self) -> &Uri; + fn headers(&self) -> &HeaderMap; + fn extensions_mut(&mut self) -> &mut Extensions; + fn operation_name(&self) -> &str; +} + +impl AccessCheckContext for S3AccessContext<'_> { + fn credentials(&self) -> Option<&Credentials> { + S3AccessContext::credentials(self) + } + + fn uri(&self) -> &Uri { + S3AccessContext::uri(self) + } + + fn headers(&self) -> &HeaderMap { + S3AccessContext::headers(self) + } + + fn extensions_mut(&mut self) -> &mut Extensions { + S3AccessContext::extensions_mut(self) + } + + fn operation_name(&self) -> &str { + self.s3_op().name() + } +} + +impl FS { + /// RustFS's request-level access check: credential and session validation, the `ReqInfo` + /// and server-context extensions every access hook and app body reads, the presigned + /// capability scoping, and the license check. The s3s `S3Access::check` hook and the gateway + /// bridge both run it before the operation's own hook. + pub(crate) async fn check_request_access(&self, cx: &mut C) -> S3Result<()> { // GHSA-g8w9-qw9q-fghr: a presigned URL only authorises the headers it // signed. Reject unsigned `x-amz-*` headers first, before the session // token lookup below or any handler reads a request header. @@ -1767,7 +1805,7 @@ impl S3Access for FS { // boundary so unsupported GET/HEAD/DELETE/bucket routes cannot silently // ignore a signed capability query. if parse_presigned_put_max_content_length(cx.headers(), cx.uri().query(), verified_presigned)?.is_some() - && cx.s3_op().name() != "PutObject" + && cx.operation_name() != "PutObject" { return Err(S3Error::with_message( S3ErrorCode::InvalidRequest, @@ -1775,7 +1813,7 @@ impl S3Access for FS { )); } if parse_presigned_multipart_max_total_object_size(cx.headers(), cx.uri().query(), verified_sigv4)?.is_some() - && cx.s3_op().name() != "CreateMultipartUpload" + && cx.operation_name() != "CreateMultipartUpload" { return Err(S3Error::with_message( S3ErrorCode::InvalidRequest, @@ -1794,6 +1832,13 @@ impl S3Access for FS { Ok(()) } +} + +#[async_trait::async_trait] +impl S3Access for FS { + async fn check(&self, cx: &mut S3AccessContext<'_>) -> S3Result<()> { + self.check_request_access(cx).await + } /// Checks whether the CreateBucket request has accesses to the resources. /// diff --git a/rustfs/src/storage/gateway_bridge.rs b/rustfs/src/storage/gateway_bridge.rs new file mode 100644 index 000000000..e3faac1bc --- /dev/null +++ b/rustfs/src/storage/gateway_bridge.rs @@ -0,0 +1,491 @@ +// 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. + +//! The gateway half of `RUSTFS_S3_STACK=gateway` (rustfs/backlog#1752): a RustFS Gateway service +//! whose handlers run the unchanged s3s application layer. +//! +//! Responsible for: assembling the gateway pipeline — SigV4 through the same IAM credential chain +//! the legacy stack uses, the authorization entry, one handler per served operation — and running +//! each request through RustFS's own access check and `S3` implementation, so neither the app +//! bodies nor the storage layer change. +//! NOT responsible for: choosing which requests come here (`crate::server::stack`), or making a +//! policy decision of its own. +//! +//! # One request +//! +//! 1. [`GatewayPipeline::serve`] captures the request head — method, URI, headers, and the +//! extensions RustFS's outer layers installed (`RequestContext`, `Option`, …) — +//! and runs the gateway service inside a request-scoped exchange. +//! 2. The gateway accepts the wire, resolves the host, routes, and verifies SigV4 against +//! [`IamCredentials`], which asks `IAMAuth` exactly as the s3s auth hook does. +//! 3. [`ExchangeAuthorizer`] records the identity and credential scope the pipeline verified and +//! admits the request. It decides nothing: the decision is step 4's. +//! 4. The handler rebuilds the s3s `S3Request` from the captured head and the recorded verdict, +//! runs `FS::check_request_access` (the body of the legacy `S3Access::check`) and the +//! operation's access hook, then calls the operation on `FS` — the order s3s uses. +//! +//! # Why the exchange is task-local +//! +//! A gateway handler receives `Req`: the decoded input and authorization proofs, but no +//! identity, scope or header map at the pinned rev, and RustFS's access check needs all three. +//! The head and verdict therefore cross from step 1 and 3 to step 4 through a task-local set for +//! this one request. It is never shared and never global, and a handler that finds it missing, +//! unverified or contradictory fails closed. rustfs/gateway#771 (ADR-0022) adds the typed +//! `req.context()` that replaces it once this crate pins a rev that carries it. +// RUSTFS_COMPAT_TODO(backlog-1752-gateway-exchange): the pinned gateway rev hands handlers neither the accepted request head nor the verified identity, so both reach the s3s app layer through a request-scoped task-local, and the GetBucketLocation DTOs are mapped here because the gateway compat conversions target another s3s revision than RustFS links. Remove after the pinned gateway rev carries the typed handler request context (rustfs/gateway#771) and the handler builds the s3s request from it, and the compat conversions target the s3s revision RustFS links. + +use std::cell::RefCell; +use std::io; +use std::sync::Arc; + +use http::{Extensions, HeaderMap, Method, Request as HttpRequest, StatusCode, Uri}; +use rustfs_gateway::dto::{GetBucketLocation, GetBucketLocationOutput, LocationConstraint}; +use rustfs_gateway::{ + Authorizer, AuthzRequest, BoxFuture, CredentialGuardConfig, CredentialLookup, CredentialProvider, + Credentials as GatewayCredentials, Decision, ErrorCode, Handler, HandlerError, HandlerErrorContext, HandlerResult, + InputAuthzRequest, InputDecisions, ProviderError, RegionSet, Req, RequestContext, Resp, SecurityFloor, ServiceBuilder, + SigV4Authenticator, +}; + +use super::access::AccessCheckContext; +use super::ecfs::FS; +use super::storage_api::gateway_bridge_consumer::{ + Body, Credentials, GetBucketLocationInput, HttpResponse, Region, S3, S3Access, S3Auth, S3Error, S3Request, +}; +use crate::auth::IAMAuth; + +tokio::task_local! { + static EXCHANGE: RefCell; +} + +/// What the s3s service would have handed the app layer, captured before the gateway reads the +/// request. +struct CapturedHead { + method: Method, + uri: Uri, + headers: HeaderMap, + extensions: Extensions, +} + +/// Who a request runs as, as the gateway verified it. +#[derive(Clone, Debug, PartialEq, Eq)] +struct Principal { + /// The access key the verified signature named; `None` for an anonymous request. + access_key: Option, + /// The verified credential-scope region, the value s3s puts in `S3Request::region`. + region: Option, + /// The verified credential-scope service, the value s3s puts in `S3Request::service`. + service: Option, +} + +#[derive(Debug)] +enum VerdictSlot { + Pending, + Recorded(Principal), + /// The authorizer saw another operation than the stack switch classified, or two verdicts. + Conflict, +} + +/// The per-request state shared by the stack entry, the authorizer and the handler. +struct Exchange { + operation: &'static str, + head: Option, + verdict: VerdictSlot, +} + +impl Exchange { + fn new(operation: &'static str, head: CapturedHead) -> Self { + Self { + operation, + head: Some(head), + verdict: VerdictSlot::Pending, + } + } + + /// Records the verified principal. `false` — and a poisoned exchange — when the gateway routed + /// another operation than the switch classified, or a second, different verdict arrives. + fn record(&mut self, operation: &str, principal: Principal) -> bool { + if operation != self.operation { + self.verdict = VerdictSlot::Conflict; + return false; + } + match &self.verdict { + VerdictSlot::Pending => { + self.verdict = VerdictSlot::Recorded(principal); + true + } + VerdictSlot::Recorded(recorded) if *recorded == principal => true, + VerdictSlot::Recorded(_) | VerdictSlot::Conflict => { + self.verdict = VerdictSlot::Conflict; + false + } + } + } + + fn is_recorded(&self) -> bool { + matches!(self.verdict, VerdictSlot::Recorded(_)) + } + + /// Hands the head and the verdict to the handler, once. + fn take(&mut self, operation: &str) -> Result<(CapturedHead, Principal), &'static str> { + if operation != self.operation { + return Err("the gateway dispatched another operation than the stack switch classified"); + } + let VerdictSlot::Recorded(principal) = &self.verdict else { + return Err("the request reached a handler without a recorded authorization verdict"); + }; + let principal = principal.clone(); + let head = self.head.take().ok_or("the request head was already consumed")?; + Ok((head, principal)) + } +} + +/// The RustFS Gateway service the gateway stack routes classified requests to. +#[derive(Clone)] +pub(crate) struct GatewayPipeline { + service: rustfs_gateway::S3Service, +} + +impl GatewayPipeline { + /// Assembles the pipeline over this server's `FS` and IAM credential chain. + /// + /// `regions` are the credential-scope regions a signature may name. + /// + /// # Errors + /// + /// An invalid region name, or a gateway assembly refusal. + pub(crate) fn build(fs: FS, auth: IAMAuth, regions: &[String]) -> io::Result { + let regions = RegionSet::new(regions).map_err(|error| { + io::Error::new( + io::ErrorKind::InvalidInput, + format!("gateway S3 stack: invalid credential-scope region set {regions:?}: {error}"), + ) + })?; + // The legacy stack looks every key up afresh, so an IAM user created a moment ago signs + // in at once; a negative cache would refuse that user until its entry aged out. + let guard = CredentialGuardConfig { + negative_entries: 0, + ..CredentialGuardConfig::default() + }; + let credentials: Arc = Arc::new(IamCredentials { auth: auth.clone() }); + let backend = Arc::new(GatewayBackend { fs, auth }); + let service = ServiceBuilder::new() + .register::(backend) + .authenticator(SigV4Authenticator::with_guard_config(credentials, regions, guard)) + .authorizer(ExchangeAuthorizer) + // Anonymous requests reach the authorizer and then RustFS's access check, which + // applies bucket policy to them exactly as the legacy stack does. + .security_floor(SecurityFloor::new().delegate_anonymous_to_authorizer_after_listing_in_the_posture_report()) + .build() + .map_err(|error| io::Error::other(format!("gateway S3 stack assembly failed: {error}")))?; + Ok(Self { service }) + } + + /// Answers one request the stack switch classified as `operation`. + pub(crate) async fn serve(&self, operation: &'static str, request: HttpRequest) -> HttpResponse + where + B: http_body::Body + Send + 'static, + B::Data: Send, + B::Error: Into>, + { + let (mut parts, body) = request.into_parts(); + let head = CapturedHead { + method: parts.method.clone(), + uri: parts.uri.clone(), + headers: parts.headers.clone(), + extensions: std::mem::take(&mut parts.extensions), + }; + let exchange = RefCell::new(Exchange::new(operation, head)); + // Through the tower adapter, not the inherent `call`: the adapter also writes + // `Connection: close` when the gateway refuses to reuse the connection (a refusal that + // left request bytes unread), which hyper needs to see to stop reading from it. + let mut service = self.service.clone(); + let response = EXCHANGE + .scope(exchange, tower::Service::call(&mut service, HttpRequest::from_parts(parts, body))) + .await; + let response = match response { + Ok(response) => response, + Err(never) => match never {}, + }; + response.map(Body::http_body_unsync) + } +} + +/// SigV4 secrets from the credential chain the legacy s3s auth hook uses. +struct IamCredentials { + auth: IAMAuth, +} + +impl CredentialProvider for IamCredentials { + fn lookup<'a>(&'a self, access_key_id: &'a str) -> BoxFuture<'a, Result> { + Box::pin(async move { + match S3Auth::get_secret_key(&self.auth, access_key_id).await { + // Keystone keys answer an empty secret: they authenticate by token, never by + // signature, so no signature over them may verify. + Ok(secret) if secret.expose().is_empty() => Ok(CredentialLookup::NotFound), + Ok(secret) => GatewayCredentials::new(access_key_id, secret.expose().as_bytes()) + .map(CredentialLookup::Found) + .map_err(|_| ProviderError::Backend), + Err(error) if error.code().as_str() == "InvalidAccessKeyId" => Ok(CredentialLookup::NotFound), + Err(_) => Err(ProviderError::Backend), + } + }) + } +} + +/// Records the verified principal for the handler and admits the request. +/// +/// The gateway requires an authorizer; RustFS's decision needs the headers and the typed s3s +/// input, which this stage does not see, so the decision is taken by RustFS's access check in the +/// handler, before any storage call. A request whose verdict cannot be recorded is denied here. +/// +/// Because the input stage admits every derived resource, register an operation here only when +/// its RustFS access hook authorizes each resource the operation derives (a copy source, say), as +/// the legacy stack relies on that hook for the same decision. +struct ExchangeAuthorizer; + +impl Authorizer for ExchangeAuthorizer { + fn authorize_route<'a>(&'a self, context: &'a RequestContext<'a>, request: &'a AuthzRequest<'a>) -> BoxFuture<'a, Decision> { + let scope = context.verified_scope(); + let principal = Principal { + access_key: request.identity.map(|identity| identity.access_key_id().to_owned()), + region: scope.map(|scope| scope.region().to_owned()), + service: scope.map(|scope| scope.service().to_owned()), + }; + let admitted = EXCHANGE + .try_with(|exchange| exchange.borrow_mut().record(request.operation, principal)) + .unwrap_or(false); + Box::pin(async move { if admitted { Decision::Allow } else { Decision::Deny } }) + } + + fn authorize_input<'a>( + &'a self, + _context: &'a RequestContext<'a>, + request: &'a InputAuthzRequest<'a>, + ) -> BoxFuture<'a, InputDecisions> { + let admitted = EXCHANGE.try_with(|exchange| exchange.borrow().is_recorded()).unwrap_or(false); + let stage = if admitted { Decision::Allow } else { Decision::Deny }; + let decisions = request.decide_all(stage, |_| stage); + Box::pin(async move { decisions }) + } +} + +/// The handlers: each runs one operation of the unchanged s3s app layer. +struct GatewayBackend { + fs: FS, + auth: IAMAuth, +} + +impl GatewayBackend { + /// The s3s request the legacy stack would have built for this request, around `input`. + async fn s3_request(&self, operation: &'static str, input: T) -> Result, HandlerError> { + let (head, principal) = EXCHANGE + .try_with(|exchange| exchange.borrow_mut().take(operation)) + .map_err(|_| HandlerError::internal_error("the gateway request exchange is missing"))? + .map_err(HandlerError::internal_error)?; + let credentials = match principal.access_key { + Some(access_key) => { + let secret_key = S3Auth::get_secret_key(&self.auth, &access_key).await.map_err(handler_error)?; + Some(Credentials { access_key, secret_key }) + } + None => None, + }; + let region = principal + .region + .map(|region| Region::new(region.into_boxed_str())) + .transpose() + .map_err(|_| HandlerError::internal_error("the verified credential-scope region is not an s3s region"))?; + Ok(S3Request { + input, + method: head.method, + uri: head.uri, + headers: head.headers, + extensions: head.extensions, + credentials, + region, + service: principal.service, + trailing_headers: None, + }) + } +} + +impl Handler for GatewayBackend { + async fn call(&self, request: Req) -> HandlerResult { + let input = request.into_input(); + let input = GetBucketLocationInput { + bucket: input.bucket.as_str().to_owned(), + expected_bucket_owner: input.expected_bucket_owner, + }; + let mut request = self.s3_request(GetBucketLocation::NAME, input).await?; + self.fs + .check_request_access(&mut BridgeAccessContext::new(&mut request, GetBucketLocation::NAME)) + .await + .map_err(handler_error)?; + S3Access::get_bucket_location(&self.fs, &mut request) + .await + .map_err(handler_error)?; + let output = S3::get_bucket_location(&self.fs, request) + .await + .map_err(handler_error)? + .output; + Ok(Resp::new(GetBucketLocationOutput { + location_constraint: output + .location_constraint + .map(|constraint| LocationConstraint::custom(constraint.as_str().to_owned())), + })) + } +} + +/// The facts RustFS's access check reads, borrowed from the rebuilt s3s request. +struct BridgeAccessContext<'a> { + credentials: Option<&'a Credentials>, + uri: &'a Uri, + headers: &'a HeaderMap, + extensions: &'a mut Extensions, + operation: &'static str, +} + +impl<'a> BridgeAccessContext<'a> { + fn new(request: &'a mut S3Request, operation: &'static str) -> Self { + Self { + credentials: request.credentials.as_ref(), + uri: &request.uri, + headers: &request.headers, + extensions: &mut request.extensions, + operation, + } + } +} + +impl AccessCheckContext for BridgeAccessContext<'_> { + fn credentials(&self) -> Option<&Credentials> { + self.credentials + } + + fn uri(&self) -> &Uri { + self.uri + } + + fn headers(&self) -> &HeaderMap { + self.headers + } + + fn extensions_mut(&mut self) -> &mut Extensions { + self.extensions + } + + fn operation_name(&self) -> &str { + self.operation + } +} + +/// The gateway error for an s3s error: the same code, status and message. +/// +/// A code the gateway declares with the same status keeps the declared row; any other pairing is +/// carried as a custom code with the status s3s would have rendered, so the answer on the wire does +/// not change with the stack. +/// +/// The gateway renders a few codes only from typed context (its `is_contextual` set) and answers +/// 500 for a bare one. `NoSuchBucket` is the one GetBucketLocation can produce; map each further +/// contextual code here before registering an operation that can return it. +fn handler_error(error: S3Error) -> HandlerError { + if error.code().as_str() == "NoSuchBucket" { + return HandlerError::from(HandlerErrorContext::missing_bucket()); + } + let name = error.code().as_str(); + let status = error + .status_code() + .or_else(|| error.code().status_code()) + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); + let code = match ErrorCode::known(name) { + Some(known) if known.default_status() == status => known, + _ => ErrorCode::custom(name.to_owned(), status), + }; + HandlerError::new(code, error.message().unwrap_or_default().to_owned()) +} + +#[cfg(test)] +mod tests { + use super::{CapturedHead, Exchange, Principal, handler_error}; + use crate::storage::storage_api::gateway_bridge_consumer::{S3Error, S3ErrorCode}; + use http::{Extensions, HeaderMap, Method, StatusCode}; + use rustfs_gateway::{ErrorCode, HandlerError, HandlerErrorContext}; + + fn head() -> CapturedHead { + CapturedHead { + method: Method::GET, + uri: "/bucket?location".parse().expect("uri"), + headers: HeaderMap::new(), + extensions: Extensions::new(), + } + } + + fn principal(access_key: Option<&str>) -> Principal { + Principal { + access_key: access_key.map(str::to_owned), + region: Some("us-east-1".to_owned()), + service: Some("s3".to_owned()), + } + } + + #[test] + fn a_recorded_verdict_is_handed_over_once() { + let mut exchange = Exchange::new("GetBucketLocation", head()); + assert!(exchange.take("GetBucketLocation").is_err(), "no verdict yet"); + assert!(exchange.record("GetBucketLocation", principal(Some("AK")))); + assert!( + exchange.record("GetBucketLocation", principal(Some("AK"))), + "the same verdict again is harmless" + ); + let (head, verdict) = exchange.take("GetBucketLocation").expect("recorded"); + assert_eq!(head.uri, "/bucket?location"); + assert_eq!(verdict, principal(Some("AK"))); + assert!(exchange.take("GetBucketLocation").is_err(), "the head is consumed"); + } + + #[test] + fn a_contradictory_or_misrouted_verdict_poisons_the_exchange() { + let mut exchange = Exchange::new("GetBucketLocation", head()); + assert!(!exchange.record("PutObject", principal(None)), "another operation than classified"); + assert!(!exchange.is_recorded()); + assert!(exchange.take("GetBucketLocation").is_err()); + + let mut exchange = Exchange::new("GetBucketLocation", head()); + assert!(exchange.record("GetBucketLocation", principal(Some("AK")))); + assert!(!exchange.record("GetBucketLocation", principal(Some("OTHER")))); + assert!(exchange.take("GetBucketLocation").is_err(), "a conflict never reaches a handler"); + + let mut exchange = Exchange::new("GetBucketLocation", head()); + assert!(exchange.record("GetBucketLocation", principal(None))); + assert!(exchange.take("PutObject").is_err(), "a handler for another operation is refused"); + } + + #[test] + fn s3_errors_keep_their_code_status_and_message() { + let error = handler_error(S3Error::with_message(S3ErrorCode::AccessDenied, "Access Denied.")); + assert_eq!(error.code(), &ErrorCode::ACCESS_DENIED); + assert_eq!(error.message(), "Access Denied."); + + // A bare NoSuchBucket would resolve to 500 in the gateway; it must carry the typed + // missing-bucket context, which renders 404 with the legacy message. + let error = handler_error(S3Error::with_message(S3ErrorCode::NoSuchBucket, "gone")); + assert_eq!(error, HandlerError::from(HandlerErrorContext::missing_bucket())); + + let mut overridden = S3Error::with_message(S3ErrorCode::AccessDenied, "denied"); + overridden.set_status_code(StatusCode::SERVICE_UNAVAILABLE); + let error = handler_error(overridden); + assert_eq!(error.code().as_str(), "AccessDenied"); + assert_eq!(error.code().default_status(), StatusCode::SERVICE_UNAVAILABLE); + } +} diff --git a/rustfs/src/storage/mod.rs b/rustfs/src/storage/mod.rs index af98b5bb3..c43a4778c 100644 --- a/rustfs/src/storage/mod.rs +++ b/rustfs/src/storage/mod.rs @@ -16,6 +16,7 @@ pub mod access; pub mod concurrency; pub mod deadlock_detector; pub mod ecfs; +pub(crate) mod gateway_bridge; pub(crate) mod helper; pub mod options; pub mod request_context; diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 1d3af56d1..51cafec2f 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -126,6 +126,20 @@ pub(crate) mod access_consumer { }; } +/// The s3s names the RustFS Gateway bridge hands to the unchanged app layer +/// (rustfs/backlog#1752), funneled here so the bridge stays off the direct s3s surface +/// (`scripts/check_s3s_footprint.sh`). +pub(crate) mod gateway_bridge_consumer { + pub(crate) use super::super::gateway_bridge::GatewayPipeline; + #[cfg(test)] + pub(crate) use s3s::S3ErrorCode; + pub(crate) use s3s::access::S3Access; + pub(crate) use s3s::auth::{Credentials, S3Auth}; + pub(crate) use s3s::dto::GetBucketLocationInput; + pub(crate) use s3s::region::Region; + pub(crate) use s3s::{Body, HttpError, HttpResponse, S3, S3Error, S3Request}; +} + pub(crate) mod concurrency_consumer { #[cfg(test)] pub(crate) use super::super::concurrency::SNOWBALL_MEMBER_COMMIT_LIMIT; diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index f4f898247..503a5bbcd 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -169,6 +169,12 @@ pub(crate) mod server { pub(crate) type FS = crate::storage::storage_api::FS; } + pub(crate) mod gateway { + #[cfg(test)] + pub(crate) use crate::storage::storage_api::gateway_bridge_consumer::Body; + pub(crate) use crate::storage::storage_api::gateway_bridge_consumer::{GatewayPipeline, HttpError, HttpResponse}; + } + pub(crate) mod metadata_route { pub(crate) fn with_metadata_route(admin: A, host: Option) -> impl s3s::route::S3Route where diff --git a/rustfs/tests/embedded_gateway_stack_test.rs b/rustfs/tests/embedded_gateway_stack_test.rs new file mode 100644 index 000000000..81867333e --- /dev/null +++ b/rustfs/tests/embedded_gateway_stack_test.rs @@ -0,0 +1,368 @@ +// 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. + +//! `RUSTFS_S3_STACK=gateway` end to end (rustfs/backlog#1752). +//! +//! Embedded servers run the legacy and the gateway stack; the tests prove that a signed PutObject +//! and GetBucketLocation succeed on the gateway stack, that the same requests answer equally on +//! both stacks, and that an unsigned, forged or unknown-key request is refused by the gateway. + +#![recursion_limit = "256"] + +use aws_sdk_s3::config::{Credentials, Region}; +use aws_sdk_s3::primitives::ByteStream; +use aws_sdk_s3::{Client, Config}; +use hmac::{Hmac, KeyInit, Mac}; +use rustfs::config::S3Stack; +use rustfs::embedded::{RustFSServer, RustFSServerBuilder, find_available_port}; +use sha2::{Digest, Sha256}; + +mod common; + +const ACCESS_KEY: &str = "gatewaystackaccess"; +const SECRET_KEY: &str = "gatewaystacksecret"; +const REGION: &str = "us-east-1"; +const BUCKET: &str = "gateway-stack-bucket"; +const MISSING_BUCKET: &str = "gateway-stack-missing"; +const OBJECT_KEY: &str = "greeting.txt"; +const OBJECT_BODY: &[u8] = b"hello from the gateway stack"; +const EMPTY_SHA256: &str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"; +const XML_DECLARATION: &str = ""; +/// The one message the gateway gives both credential rejections (rustfs-gateway `AuthError`). +const GATEWAY_CREDENTIAL_MESSAGE: &str = "the request was not authenticated"; + +/// A pinned difference between the stacks that has no `rd-` ruling yet. +/// +/// The ruled divergences on record (rd-put-0001..0008, rd-ctx-0001..0006, rustfs/gateway#759) do +/// not cover GetBucketLocation; every difference below was observed by this test and must gain a +/// ruling in rustfs/gateway's request-divergence register before the default stack changes. +/// Matching is exact, so a change on either side fails the test instead of widening the pin. +fn unruled_divergence(legacy: &Answer, gateway: &Answer) -> Option<&'static str> { + fn seen(answer: &Answer) -> (u16, Option<&str>, Option<&str>) { + (answer.status, answer.element("Code"), answer.element("Message")) + } + match (seen(legacy), seen(gateway)) { + // The gateway answers a signature mismatch as an unknown key (`render::from_auth`), so + // the wire never confirms the access key exists. + ((403, Some("SignatureDoesNotMatch"), Some(_)), (403, Some("InvalidAccessKeyId"), Some(GATEWAY_CREDENTIAL_MESSAGE))) => { + Some("signature-mismatch-answered-as-invalid-access-key") + } + // Same code; the gateway's message is its one credential-rejection sentence. + ( + (403, Some("InvalidAccessKeyId"), Some(message)), + (403, Some("InvalidAccessKeyId"), Some(GATEWAY_CREDENTIAL_MESSAGE)), + ) if message != GATEWAY_CREDENTIAL_MESSAGE => Some("credential-rejection-message"), + _ => None, + } +} + +async fn start(stack: S3Stack) -> Option { + let port = match find_available_port() { + Ok(port) => port, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None, + Err(err) => panic!("find free port: {err}"), + }; + let server = RustFSServerBuilder::new() + .address(format!("127.0.0.1:{port}")) + .access_key(ACCESS_KEY) + .secret_key(SECRET_KEY) + .s3_stack(stack) + .build() + .await + .expect("start embedded server"); + Some(server) +} + +fn sdk_client(server: &RustFSServer) -> Client { + let config = Config::builder() + .credentials_provider(Credentials::new(ACCESS_KEY, SECRET_KEY, None, None, "test")) + .region(Region::new(REGION)) + .endpoint_url(server.endpoint()) + .force_path_style(true) + .behavior_version_latest() + .build(); + Client::from_conf(config) +} + +/// Creates the bucket and stores one object with the SDK's signed PutObject. +async fn seed(client: &Client) { + client.create_bucket().bucket(BUCKET).send().await.expect("create bucket"); + client + .put_object() + .bucket(BUCKET) + .key(OBJECT_KEY) + .body(ByteStream::from_static(OBJECT_BODY)) + .send() + .await + .expect("signed PutObject"); +} + +#[derive(Clone, Copy, Debug)] +enum Auth { + Anonymous, + Signed, + /// A valid access key signed with the wrong secret. + Forged, + /// An access key IAM does not know. + UnknownKey, +} + +fn hmac(key: &[u8], data: &str) -> Vec { + let mut mac = Hmac::::new_from_slice(key).expect("any key length is valid for HMAC"); + mac.update(data.as_bytes()); + mac.finalize().into_bytes().to_vec() +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|byte| format!("{byte:02x}")).collect() +} + +/// Header SigV4 for a bodyless `GET {path}?{query_key}`, the shape GetBucketLocation uses. +fn sigv4_headers(host: &str, path: &str, query_key: &str, access_key: &str, secret_key: &str) -> Vec<(&'static str, String)> { + let amz_date = jiff::Timestamp::now().strftime("%Y%m%dT%H%M%SZ").to_string(); + let date = amz_date.get(..8).expect("amz date has a day part").to_owned(); + let scope = format!("{date}/{REGION}/s3/aws4_request"); + let signed_headers = "host;x-amz-content-sha256;x-amz-date"; + let canonical_request = format!( + "GET\n{path}\n{query_key}=\nhost:{host}\nx-amz-content-sha256:{EMPTY_SHA256}\nx-amz-date:{amz_date}\n\n{signed_headers}\n{EMPTY_SHA256}" + ); + let string_to_sign = format!( + "AWS4-HMAC-SHA256\n{amz_date}\n{scope}\n{}", + hex(&Sha256::digest(canonical_request.as_bytes())) + ); + let signing_key = hmac( + &hmac(&hmac(&hmac(format!("AWS4{secret_key}").as_bytes(), &date), REGION), "s3"), + "aws4_request", + ); + let signature = hex(&hmac(&signing_key, &string_to_sign)); + vec![ + ("x-amz-date", amz_date), + ("x-amz-content-sha256", EMPTY_SHA256.to_owned()), + ( + "authorization", + format!("AWS4-HMAC-SHA256 Credential={access_key}/{scope}, SignedHeaders={signed_headers}, Signature={signature}"), + ), + ] +} + +/// One observed response. +#[derive(Debug)] +struct Answer { + status: u16, + content_type: Option, + body: String, +} + +impl Answer { + /// The text of the first `` or `` element. + fn element(&self, name: &str) -> Option<&str> { + let open = format!("<{name}"); + let mut from = 0; + let tag = loop { + let at = self.body.get(from..)?.find(&open)? + from; + let after = at + open.len(); + match self.body.as_bytes().get(after) { + Some(b'>') | Some(b' ') => break after, + _ => from = after, + } + }; + let start = self.body.get(tag..)?.find('>')? + tag + 1; + let end = self.body.get(start..)?.find(&format!(""))? + start; + self.body.get(start..end) + } + + /// The body with the one newline the gateway writes after the XML declaration removed. + /// + /// UNRULED `xml-declaration-newline`: the gateway XML writer ends its declaration with `\n`, + /// s3s does not. Pinned (not ignored) by the differential test below; it needs an `rd-` entry + /// before the default stack changes. + fn document(&self) -> String { + match self + .body + .strip_prefix(XML_DECLARATION) + .and_then(|rest| rest.strip_prefix('\n')) + { + Some(rest) => format!("{XML_DECLARATION}{rest}"), + None => self.body.clone(), + } + } + + /// The part of an answer a client acts on. Per-response identifiers (`RequestId`, `HostId`, + /// `x-amz-request-id`, `Date`) differ between any two responses and are left out. + fn client_view(&self) -> (u16, Option<&str>, Option<&str>, Option<&str>, Option) { + let success_body = (self.status == 200).then(|| self.document()); + ( + self.status, + self.content_type.as_deref(), + self.element("Code"), + self.element("Message"), + success_body, + ) + } +} + +async fn get_bucket_location(server: &RustFSServer, bucket: &str, auth: Auth) -> Answer { + let endpoint = server.endpoint(); + let host = endpoint.trim_start_matches("http://").to_owned(); + let path = format!("/{bucket}"); + let headers = match auth { + Auth::Anonymous => Vec::new(), + Auth::Signed => sigv4_headers(&host, &path, "location", ACCESS_KEY, SECRET_KEY), + Auth::Forged => sigv4_headers(&host, &path, "location", ACCESS_KEY, "not-the-secret-key"), + Auth::UnknownKey => sigv4_headers(&host, &path, "location", "AKIDUNKNOWNGATEWAY", SECRET_KEY), + }; + let client = reqwest::Client::builder().no_proxy().build().expect("http client"); + let mut request = client.get(format!("{endpoint}{path}?location")); + for (name, value) in headers { + request = request.header(name, value); + } + let response = request.send().await.expect("GetBucketLocation round trip"); + let status = response.status().as_u16(); + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let body = response.text().await.expect("response body"); + Answer { + status, + content_type, + body, + } +} + +#[test] +fn gateway_stack_serves_signed_put_object_and_get_bucket_location() { + common::run_embedded_test(gateway_stack_serves_signed_put_object_and_get_bucket_location_body); +} + +async fn gateway_stack_serves_signed_put_object_and_get_bucket_location_body() { + let Some(server) = start(S3Stack::Gateway).await else { + return; + }; + let client = sdk_client(&server); + seed(&client).await; + + // The SDK's own signed GetBucketLocation goes through the gateway pipeline. + let location = client + .get_bucket_location() + .bucket(BUCKET) + .send() + .await + .expect("signed GetBucketLocation through the gateway stack"); + assert_eq!(location.location_constraint().map(|constraint| constraint.as_str()), Some(REGION)); + + let answer = get_bucket_location(&server, BUCKET, Auth::Signed).await; + assert_eq!(answer.status, 200, "{answer:?}"); + assert_eq!(answer.element("LocationConstraint"), Some(REGION), "{answer:?}"); + + // The PutObject the stack switch left on the legacy service stored the object. + let object = client + .get_object() + .bucket(BUCKET) + .key(OBJECT_KEY) + .send() + .await + .expect("GetObject"); + let stored = object.body.collect().await.expect("object body").into_bytes(); + assert_eq!(stored.as_ref(), OBJECT_BODY); + + server.shutdown().await; +} + +#[test] +fn legacy_and_gateway_stacks_answer_get_bucket_location_equally() { + common::run_embedded_test(legacy_and_gateway_stacks_answer_get_bucket_location_equally_body); +} + +async fn legacy_and_gateway_stacks_answer_get_bucket_location_equally_body() { + let Some(legacy) = start(S3Stack::Legacy).await else { + return; + }; + let Some(gateway) = start(S3Stack::Gateway).await else { + legacy.shutdown().await; + return; + }; + seed(&sdk_client(&legacy)).await; + seed(&sdk_client(&gateway)).await; + + // No ruled divergence applies: the rulings on record (rd-put-0001..0008, rd-ctx-0001..0006, + // rustfs/gateway#759) concern PutObject decoding and request shapes this slice keeps on the + // legacy stack (regional virtual hosts, absolute-form URIs, a bare `?`, non-UTF-8 headers, + // repeated metadata). Every answer must be equal apart from the pinned unruled differences. + let cases = [ + (BUCKET, Auth::Signed), + (MISSING_BUCKET, Auth::Signed), + (BUCKET, Auth::Anonymous), + (BUCKET, Auth::Forged), + (BUCKET, Auth::UnknownKey), + ]; + let mut mismatches = Vec::new(); + let mut unruled = Vec::new(); + for (bucket, auth) in cases { + let from_legacy = get_bucket_location(&legacy, bucket, auth).await; + let from_gateway = get_bucket_location(&gateway, bucket, auth).await; + if from_legacy.status == 200 { + // Pin `xml-declaration-newline` on both sides so it cannot drift unseen. + assert!(!from_legacy.body.starts_with(&format!("{XML_DECLARATION}\n")), "{from_legacy:?}"); + assert!(from_gateway.body.starts_with(&format!("{XML_DECLARATION}\n")), "{from_gateway:?}"); + } + if from_legacy.client_view() == from_gateway.client_view() { + continue; + } + match unruled_divergence(&from_legacy, &from_gateway) { + Some(id) => unruled.push(format!("{bucket} {auth:?}: {id}")), + None => mismatches.push(format!("{bucket} {auth:?}\n legacy: {from_legacy:?}\n gateway: {from_gateway:?}")), + } + } + assert!(mismatches.is_empty(), "stacks differ:\n{}", mismatches.join("\n")); + assert_eq!( + unruled, + [ + format!("{BUCKET} Forged: signature-mismatch-answered-as-invalid-access-key"), + format!("{BUCKET} UnknownKey: credential-rejection-message"), + ], + "the unruled differences changed; update the pin and the ruling request together" + ); + + gateway.shutdown().await; + legacy.shutdown().await; +} + +#[test] +fn gateway_stack_refuses_unsigned_forged_and_unknown_key_requests() { + common::run_embedded_test(gateway_stack_refuses_unsigned_forged_and_unknown_key_requests_body); +} + +async fn gateway_stack_refuses_unsigned_forged_and_unknown_key_requests_body() { + let Some(server) = start(S3Stack::Gateway).await else { + return; + }; + seed(&sdk_client(&server)).await; + + for (auth, code) in [ + (Auth::Anonymous, "AccessDenied"), + // The gateway answers a signature mismatch as InvalidAccessKeyId on purpose + // (rustfs-gateway `render::from_auth`), so the wire never confirms a key exists. + (Auth::Forged, "InvalidAccessKeyId"), + (Auth::UnknownKey, "InvalidAccessKeyId"), + ] { + let answer = get_bucket_location(&server, BUCKET, auth).await; + assert_eq!(answer.status, 403, "{auth:?}: {answer:?}"); + assert_eq!(answer.element("Code"), Some(code), "{auth:?}: {answer:?}"); + assert_eq!(answer.element("LocationConstraint"), None, "{auth:?} must not reveal the location"); + } + + server.shutdown().await; +}