mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 14:27:10 +00:00
feat(app): add the RustFS-owned RequestEnvelope and its s3s constructor (#8437)
* feat(app): add the RustFS-owned RequestEnvelope and its s3s constructor * ci: classify the s3s edge adapters in storage/s3_compat as interface
This commit is contained in:
@@ -10,7 +10,7 @@
|
||||
# intentionally removed. Regenerate with:
|
||||
# scripts/check_test_count_ratchet.sh --update
|
||||
e2e_test 890 18
|
||||
rustfs 5490 62
|
||||
rustfs 5498 62
|
||||
rustfs-audit 70 0
|
||||
rustfs-checksums 26 0
|
||||
rustfs-common 23 0
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
//! This module introduces explicit dependency injection entry points
|
||||
//! for storage, IAM, and KMS handles.
|
||||
|
||||
mod envelope;
|
||||
mod global;
|
||||
mod handles;
|
||||
mod interfaces;
|
||||
@@ -23,6 +24,8 @@ mod runtime_sources;
|
||||
mod server_slot;
|
||||
mod startup;
|
||||
|
||||
pub(crate) use envelope::RequestHead;
|
||||
pub use envelope::{RequestCredentials, RequestEnvelope};
|
||||
pub use global::*;
|
||||
pub use handles::*;
|
||||
pub use interfaces::*;
|
||||
|
||||
@@ -0,0 +1,314 @@
|
||||
// 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 request facts the app layer reads, owned by RustFS rather than by the
|
||||
//! HTTP framework that decoded the request (rustfs/backlog#2734 task T1.7).
|
||||
//!
|
||||
//! Every extension payload that app, storage and admin code reads from a
|
||||
//! request is a named field of [`RequestEnvelope`]. An edge decodes the facts
|
||||
//! it owns into a [`RequestHead`] and hands over the request's extensions;
|
||||
//! [`RequestEnvelope::from_parts`] moves each named payload into its field and
|
||||
//! keeps whatever is left in a residual bag. The bag exists only while call
|
||||
//! sites migrate and must be empty once they have.
|
||||
//!
|
||||
//! This module names no HTTP-framework type. The legacy s3s edge that builds an
|
||||
//! envelope lives in `storage/s3_compat/envelope.rs`.
|
||||
|
||||
use super::ServerContextSlot;
|
||||
use crate::app::metadata_route::WebsiteRead;
|
||||
use crate::app::object::request_body::BodyReadControl;
|
||||
use crate::app::storage_api::context::{
|
||||
BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, ObjectTagConditions,
|
||||
OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectRequestMarker, ReqInfo, RequestContext,
|
||||
TableDataPlaneListAccess, TableDataPlanePublicationGuards,
|
||||
};
|
||||
use crate::auth::{VerifiedPresignedRequest, VerifiedSigV4Request};
|
||||
use crate::shared_types::RemoteAddr;
|
||||
use http::{Extensions, HeaderMap, Method, Uri};
|
||||
use rustfs_trusted_proxies::ClientInfo;
|
||||
use std::fmt;
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::Arc;
|
||||
use zeroize::Zeroizing;
|
||||
|
||||
/// The identity a request was authenticated with.
|
||||
///
|
||||
/// The secret key and the session token are bearer material: the type has no
|
||||
/// `PartialEq`, and its `Debug` output redacts both.
|
||||
#[derive(Clone)]
|
||||
pub struct RequestCredentials {
|
||||
access_key: String,
|
||||
secret_key: Zeroizing<String>,
|
||||
session_token: Option<Zeroizing<String>>,
|
||||
}
|
||||
|
||||
impl RequestCredentials {
|
||||
pub(crate) fn new(access_key: String, secret_key: String, session_token: Option<String>) -> Self {
|
||||
Self {
|
||||
access_key,
|
||||
secret_key: Zeroizing::new(secret_key),
|
||||
session_token: session_token.map(Zeroizing::new),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn access_key(&self) -> &str {
|
||||
&self.access_key
|
||||
}
|
||||
|
||||
/// Never log this value or compare it with `==`.
|
||||
pub fn secret_key(&self) -> &str {
|
||||
&self.secret_key
|
||||
}
|
||||
|
||||
pub fn session_token(&self) -> Option<&str> {
|
||||
self.session_token.as_deref().map(String::as_str)
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Debug for RequestCredentials {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.debug_struct("RequestCredentials")
|
||||
.field("access_key", &self.access_key)
|
||||
.field("secret_key", &"<redacted>")
|
||||
.field("session_token", &self.session_token.as_ref().map(|_| "<redacted>"))
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
/// The facts an edge decodes from the request itself, as opposed to the
|
||||
/// payloads that middleware attached to it.
|
||||
pub(crate) struct RequestHead {
|
||||
pub(crate) method: Method,
|
||||
pub(crate) uri: Uri,
|
||||
pub(crate) headers: HeaderMap,
|
||||
pub(crate) credentials: Option<RequestCredentials>,
|
||||
pub(crate) region: Option<String>,
|
||||
pub(crate) service: Option<String>,
|
||||
}
|
||||
|
||||
/// Every request fact the app layer reads.
|
||||
///
|
||||
/// There is deliberately no generic extension lookup: a payload the app layer
|
||||
/// needs gets a named field here, so an edge that builds an envelope can see
|
||||
/// everything it has to supply.
|
||||
#[cfg_attr(
|
||||
not(test),
|
||||
expect(
|
||||
dead_code,
|
||||
reason = "fields read only by the crate-private accessors stay unread until call sites move in T1.8 (rustfs/backlog#2749)"
|
||||
)
|
||||
)]
|
||||
pub struct RequestEnvelope {
|
||||
method: Method,
|
||||
uri: Uri,
|
||||
headers: HeaderMap,
|
||||
credentials: Option<RequestCredentials>,
|
||||
region: Option<String>,
|
||||
service: Option<String>,
|
||||
// Attached by the connection and request layers in `server/`.
|
||||
remote_addr: Option<SocketAddr>,
|
||||
client_info: Option<ClientInfo>,
|
||||
request_context: Option<RequestContext>,
|
||||
server_context: Option<Arc<ServerContextSlot>>,
|
||||
body_read_control: Option<BodyReadControl>,
|
||||
// Attached by routing and signature verification.
|
||||
verified_sigv4: Option<VerifiedSigV4Request>,
|
||||
verified_presigned: Option<VerifiedPresignedRequest>,
|
||||
website_read: Option<WebsiteRead>,
|
||||
post_object: Option<PostObjectRequestMarker>,
|
||||
// Attached by the authorization phase (`storage/access.rs`) for the use case.
|
||||
req_info: Option<ReqInfo>,
|
||||
bucket_generation: Option<BucketGenerationGuard>,
|
||||
pending_delete_bucket_generation: Option<PendingDeleteBucketGenerationGuard>,
|
||||
copy_source_bucket_generation: Option<CopySourceBucketGenerationGuard>,
|
||||
odm_read_generation: Option<OdmReadGenerationGuard>,
|
||||
bucket_config_mutation: Option<BucketConfigMutationSnapshot>,
|
||||
table_publication_guards: Option<TableDataPlanePublicationGuards>,
|
||||
table_list_access: Option<TableDataPlaneListAccess>,
|
||||
object_tag_conditions: Option<ObjectTagConditions>,
|
||||
residual_extensions: Extensions,
|
||||
}
|
||||
|
||||
impl RequestEnvelope {
|
||||
/// Builds an envelope from what an edge decoded and the request's
|
||||
/// extensions. A payload the request does not carry leaves its field
|
||||
/// `None`; it is never replaced by a default.
|
||||
pub(crate) fn from_parts(head: RequestHead, mut extensions: Extensions) -> Self {
|
||||
// Each payload is removed under exactly the type its readers fetch, so
|
||||
// a named type never lingers in the residual bag. The connection layer
|
||||
// stores the peer as `Option<RemoteAddr>`; `None` there means the same
|
||||
// as no peer at all.
|
||||
Self {
|
||||
method: head.method,
|
||||
uri: head.uri,
|
||||
headers: head.headers,
|
||||
credentials: head.credentials,
|
||||
region: head.region,
|
||||
service: head.service,
|
||||
remote_addr: extensions.remove::<Option<RemoteAddr>>().flatten().map(|addr| addr.0),
|
||||
client_info: extensions.remove(),
|
||||
request_context: extensions.remove(),
|
||||
server_context: extensions.remove(),
|
||||
body_read_control: extensions.remove(),
|
||||
verified_sigv4: extensions.remove(),
|
||||
verified_presigned: extensions.remove(),
|
||||
website_read: extensions.remove(),
|
||||
post_object: extensions.remove(),
|
||||
req_info: extensions.remove(),
|
||||
bucket_generation: extensions.remove(),
|
||||
pending_delete_bucket_generation: extensions.remove(),
|
||||
copy_source_bucket_generation: extensions.remove(),
|
||||
odm_read_generation: extensions.remove(),
|
||||
bucket_config_mutation: extensions.remove(),
|
||||
table_publication_guards: extensions.remove(),
|
||||
table_list_access: extensions.remove(),
|
||||
object_tag_conditions: extensions.remove(),
|
||||
residual_extensions: extensions,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn method(&self) -> &Method {
|
||||
&self.method
|
||||
}
|
||||
|
||||
pub fn uri(&self) -> &Uri {
|
||||
&self.uri
|
||||
}
|
||||
|
||||
pub fn headers(&self) -> &HeaderMap {
|
||||
&self.headers
|
||||
}
|
||||
|
||||
/// `None` means the request is anonymous.
|
||||
pub fn credentials(&self) -> Option<&RequestCredentials> {
|
||||
self.credentials.as_ref()
|
||||
}
|
||||
|
||||
pub fn region(&self) -> Option<&str> {
|
||||
self.region.as_deref()
|
||||
}
|
||||
|
||||
pub fn service(&self) -> Option<&str> {
|
||||
self.service.as_deref()
|
||||
}
|
||||
|
||||
/// The TCP peer of the connection, which is a proxy when one is in front;
|
||||
/// [`Self::client_info`] carries the client address a trusted proxy reported.
|
||||
pub fn remote_addr(&self) -> Option<SocketAddr> {
|
||||
self.remote_addr
|
||||
}
|
||||
|
||||
pub fn client_info(&self) -> Option<&ClientInfo> {
|
||||
self.client_info.as_ref()
|
||||
}
|
||||
|
||||
pub fn request_context(&self) -> Option<&RequestContext> {
|
||||
self.request_context.as_ref()
|
||||
}
|
||||
|
||||
pub fn server_context(&self) -> Option<&Arc<ServerContextSlot>> {
|
||||
self.server_context.as_ref()
|
||||
}
|
||||
|
||||
/// Extension types no field names yet. Must be empty once call sites have
|
||||
/// migrated; nothing should read it in the meantime.
|
||||
pub fn residual_extensions(&self) -> &Extensions {
|
||||
&self.residual_extensions
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(
|
||||
not(test),
|
||||
expect(
|
||||
dead_code,
|
||||
reason = "call sites move to the envelope in T1.8 (rustfs/backlog#2749); remove once every accessor has a caller"
|
||||
)
|
||||
)]
|
||||
impl RequestEnvelope {
|
||||
pub(crate) fn body_read_control(&self) -> Option<&BodyReadControl> {
|
||||
self.body_read_control.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn verified_sigv4(&self) -> Option<&VerifiedSigV4Request> {
|
||||
self.verified_sigv4.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn verified_presigned(&self) -> Option<&VerifiedPresignedRequest> {
|
||||
self.verified_presigned.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn website_read(&self) -> Option<&WebsiteRead> {
|
||||
self.website_read.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn post_object(&self) -> Option<&PostObjectRequestMarker> {
|
||||
self.post_object.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn req_info(&self) -> Option<&ReqInfo> {
|
||||
self.req_info.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn bucket_generation(&self) -> Option<&BucketGenerationGuard> {
|
||||
self.bucket_generation.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn pending_delete_bucket_generation(&self) -> Option<&PendingDeleteBucketGenerationGuard> {
|
||||
self.pending_delete_bucket_generation.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn copy_source_bucket_generation(&self) -> Option<&CopySourceBucketGenerationGuard> {
|
||||
self.copy_source_bucket_generation.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn odm_read_generation(&self) -> Option<&OdmReadGenerationGuard> {
|
||||
self.odm_read_generation.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn bucket_config_mutation(&self) -> Option<&BucketConfigMutationSnapshot> {
|
||||
self.bucket_config_mutation.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn table_publication_guards(&self) -> Option<&TableDataPlanePublicationGuards> {
|
||||
self.table_publication_guards.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn table_list_access(&self) -> Option<&TableDataPlaneListAccess> {
|
||||
self.table_list_access.as_ref()
|
||||
}
|
||||
|
||||
pub(crate) fn object_tag_conditions(&self) -> Option<&ObjectTagConditions> {
|
||||
self.object_tag_conditions.as_ref()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn request_credentials_debug_redacts_secret_and_session_token() {
|
||||
let credentials = RequestCredentials::new(
|
||||
"AKIAENVELOPE".to_owned(),
|
||||
"envelope-secret-value".to_owned(),
|
||||
Some("envelope-session-token".to_owned()),
|
||||
);
|
||||
|
||||
let debug = format!("{credentials:?}");
|
||||
|
||||
assert!(debug.contains("AKIAENVELOPE"), "{debug}");
|
||||
assert!(!debug.contains("envelope-secret-value"), "{debug}");
|
||||
assert!(!debug.contains("envelope-session-token"), "{debug}");
|
||||
}
|
||||
}
|
||||
@@ -31,6 +31,11 @@ pub(crate) mod storage_api;
|
||||
pub(crate) mod table_list_isolation;
|
||||
pub(crate) mod trailer_adapter;
|
||||
|
||||
// Request-scoped facts are not AppContext resolution, so consumers import them
|
||||
// from here rather than through `app::context`.
|
||||
pub(crate) use context::RequestHead;
|
||||
pub use context::{RequestCredentials, RequestEnvelope};
|
||||
|
||||
#[cfg(test)]
|
||||
mod capacity_dirty_scope_test;
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -28,7 +28,7 @@ pub(crate) fn EndpointServerPools(
|
||||
/// the direct s3s surface (s3s footprint ratchet, `scripts/check_s3s_footprint.sh`).
|
||||
pub(crate) mod s3 {
|
||||
#[cfg(test)]
|
||||
pub(crate) use s3s::auth::SimpleAuth;
|
||||
pub(crate) use s3s::auth::{Credentials, SimpleAuth};
|
||||
#[cfg(test)]
|
||||
pub(crate) use s3s::config::{S3Config, StaticConfigProvider};
|
||||
#[cfg(test)]
|
||||
@@ -1271,6 +1271,12 @@ pub(crate) mod context {
|
||||
pub(crate) use super::EndpointServerPools;
|
||||
pub(crate) use super::bucket;
|
||||
pub(crate) use super::runtime;
|
||||
pub(crate) use crate::storage::storage_api::access_consumer::{
|
||||
BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, ObjectTagConditions,
|
||||
OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectRequestMarker, ReqInfo, TableDataPlaneListAccess,
|
||||
TableDataPlanePublicationGuards,
|
||||
};
|
||||
pub(crate) use crate::storage::storage_api::request_context_consumer::RequestContext;
|
||||
pub(crate) use crate::storage::storage_api::{BootstrapLocalTarget, ECStore, EndpointServerPools, InstanceContext};
|
||||
#[cfg(test)]
|
||||
pub(crate) use crate::storage::storage_api::{Endpoint, Endpoints, PoolEndpoints};
|
||||
|
||||
@@ -404,7 +404,7 @@ async fn authorize_bucket_config_mutation<T>(fs: &FS, req: &mut S3Request<T>, ac
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
struct CopySourceBucketGenerationGuard {
|
||||
pub(crate) struct CopySourceBucketGenerationGuard {
|
||||
bucket: String,
|
||||
incarnation_id: uuid::Uuid,
|
||||
}
|
||||
@@ -447,7 +447,7 @@ async fn wait_for_object_generation_test_hook(bucket: &str) {
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
enum PendingDeleteBucketGenerationGuard {
|
||||
pub(crate) enum PendingDeleteBucketGenerationGuard {
|
||||
Ready(BucketGenerationGuard),
|
||||
Failed { code: S3ErrorCode, message: String },
|
||||
}
|
||||
@@ -495,7 +495,7 @@ async fn load_bucket_generation<T>(fs: &FS, req: &S3Request<T>, bucket: &str) ->
|
||||
/// A read may consult only the source admitted before its authorization.
|
||||
/// Capture failures are deferred until a local miss actually needs that source.
|
||||
#[derive(Clone, Debug)]
|
||||
enum OdmReadGenerationGuard {
|
||||
pub(crate) enum OdmReadGenerationGuard {
|
||||
Unavailable,
|
||||
Ready(BucketGenerationGuard),
|
||||
Failed { code: S3ErrorCode, message: String },
|
||||
@@ -680,6 +680,64 @@ impl ObjectTagConditions {
|
||||
}
|
||||
}
|
||||
|
||||
/// Authorization-phase payloads for request-envelope tests, which cannot reach
|
||||
/// the private fields. Production code builds them only through the
|
||||
/// authorization paths in this module.
|
||||
#[cfg(test)]
|
||||
pub(crate) mod envelope_payloads {
|
||||
use super::{
|
||||
BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, HashMap, ObjectTagConditions,
|
||||
OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, TableDataPlaneListAccess,
|
||||
};
|
||||
|
||||
fn incarnation() -> uuid::Uuid {
|
||||
uuid::Uuid::from_u128(0x7107)
|
||||
}
|
||||
|
||||
pub(crate) fn bucket_generation(bucket: &str) -> BucketGenerationGuard {
|
||||
BucketGenerationGuard {
|
||||
bucket: bucket.to_owned(),
|
||||
incarnation_id: incarnation(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn pending_delete_bucket_generation(bucket: &str) -> PendingDeleteBucketGenerationGuard {
|
||||
PendingDeleteBucketGenerationGuard::Ready(bucket_generation(bucket))
|
||||
}
|
||||
|
||||
pub(crate) fn copy_source_bucket_generation(bucket: &str) -> CopySourceBucketGenerationGuard {
|
||||
CopySourceBucketGenerationGuard {
|
||||
bucket: bucket.to_owned(),
|
||||
incarnation_id: incarnation(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn odm_read_generation(bucket: &str) -> OdmReadGenerationGuard {
|
||||
OdmReadGenerationGuard::Ready(bucket_generation(bucket))
|
||||
}
|
||||
|
||||
pub(crate) fn bucket_config_mutation(bucket: &str) -> BucketConfigMutationSnapshot {
|
||||
BucketConfigMutationSnapshot {
|
||||
bucket: bucket.to_owned(),
|
||||
incarnation_id: incarnation(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn table_list_access(bucket: &str) -> TableDataPlaneListAccess {
|
||||
TableDataPlaneListAccess {
|
||||
bucket: bucket.to_owned(),
|
||||
principal_fingerprint: String::new(),
|
||||
snapshot_fingerprint: String::new(),
|
||||
cursor_key: [0; 32],
|
||||
resources: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn object_tag_conditions(bucket: &str) -> ObjectTagConditions {
|
||||
ObjectTagConditions::new(bucket, "object", None, HashMap::new())
|
||||
}
|
||||
}
|
||||
|
||||
const AMZ_WRITE_OFFSET_BYTES_HEADER: &str = "x-amz-write-offset-bytes";
|
||||
|
||||
fn has_write_offset_bytes_header(headers: &http::HeaderMap) -> bool {
|
||||
|
||||
@@ -21,6 +21,7 @@ pub mod options;
|
||||
pub mod request_context;
|
||||
pub mod rpc;
|
||||
pub(crate) mod s3_api;
|
||||
pub(crate) mod s3_compat;
|
||||
pub(crate) mod sse;
|
||||
pub mod timeout_wrapper;
|
||||
pub mod tonic_service;
|
||||
|
||||
@@ -0,0 +1,303 @@
|
||||
// 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.
|
||||
|
||||
//! Builds a [`RequestEnvelope`] from an s3s request at the legacy edge.
|
||||
//!
|
||||
//! The request is only borrowed: until its call site moves to the envelope
|
||||
//! (T1.8), the legacy handler keeps reading the same request.
|
||||
|
||||
use crate::app::storage_api::s3::S3Request;
|
||||
use crate::app::{RequestCredentials, RequestEnvelope, RequestHead};
|
||||
use crate::auth::get_session_token;
|
||||
|
||||
impl RequestEnvelope {
|
||||
/// Copies every request fact of `req` the app layer reads.
|
||||
pub fn from_s3s<T>(req: &S3Request<T>) -> Self {
|
||||
let credentials = req.credentials.as_ref().map(|credentials| {
|
||||
RequestCredentials::new(
|
||||
credentials.access_key.clone(),
|
||||
credentials.secret_key.expose().to_owned(),
|
||||
get_session_token(&req.uri, &req.headers).map(str::to_owned),
|
||||
)
|
||||
});
|
||||
let head = RequestHead {
|
||||
method: req.method.clone(),
|
||||
uri: req.uri.clone(),
|
||||
headers: req.headers.clone(),
|
||||
credentials,
|
||||
region: req.region.as_ref().map(|region| region.as_str().to_owned()),
|
||||
service: req.service.clone(),
|
||||
};
|
||||
Self::from_parts(head, req.extensions.clone())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{RequestCredentials, RequestEnvelope, S3Request};
|
||||
use crate::app::metadata_route::WebsiteRead;
|
||||
use crate::app::object::request_body::BodyReadControl;
|
||||
use crate::app::storage_api::s3::Credentials as S3sCredentials;
|
||||
use crate::auth::{VerifiedPresignedRequest, VerifiedSigV4Request};
|
||||
use crate::runtime_sources::ServerContextSlot;
|
||||
use crate::shared_types::RemoteAddr;
|
||||
use crate::storage::access::envelope_payloads;
|
||||
use crate::storage::access::{PostObjectRequestMarker, ReqInfo, TableDataPlanePublicationGuards};
|
||||
use crate::storage::request_context::RequestContext;
|
||||
use http::{Extensions, HeaderMap, HeaderValue, Method};
|
||||
use rustfs_trusted_proxies::ClientInfo;
|
||||
use std::fmt::Debug;
|
||||
use std::net::{IpAddr, SocketAddr};
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
|
||||
const PEER: ([u8; 4], u16) = ([10, 0, 0, 7], 41000);
|
||||
const CLIENT: [u8; 4] = [203, 0, 113, 9];
|
||||
const REQUEST_ID: &str = "envelope-request-id";
|
||||
|
||||
/// A type no envelope field names.
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
struct UnrelatedPayload(u8);
|
||||
|
||||
fn request(extensions: Extensions) -> S3Request<()> {
|
||||
S3Request {
|
||||
input: (),
|
||||
method: Method::PUT,
|
||||
uri: "/bucket/key".parse().expect("static uri"),
|
||||
headers: HeaderMap::new(),
|
||||
extensions,
|
||||
credentials: None,
|
||||
region: None,
|
||||
service: None,
|
||||
trailing_headers: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn credentials() -> Option<S3sCredentials> {
|
||||
Some(S3sCredentials {
|
||||
access_key: "AKIAENVELOPE".to_owned(),
|
||||
secret_key: "envelope-secret".into(),
|
||||
})
|
||||
}
|
||||
|
||||
fn request_context() -> RequestContext {
|
||||
RequestContext {
|
||||
request_id: REQUEST_ID.to_owned(),
|
||||
x_amz_request_id: REQUEST_ID.to_owned(),
|
||||
trace_id: None,
|
||||
span_id: None,
|
||||
start_time: Instant::now(),
|
||||
}
|
||||
}
|
||||
|
||||
/// One value of every payload type the envelope names, each with a value
|
||||
/// that identifies it.
|
||||
fn every_named_payload(server_context: &Arc<ServerContextSlot>) -> Extensions {
|
||||
let mut extensions = Extensions::new();
|
||||
extensions.insert(Some(RemoteAddr(SocketAddr::from(PEER))));
|
||||
extensions.insert(ClientInfo::direct(SocketAddr::new(IpAddr::from(CLIENT), 443)));
|
||||
extensions.insert(request_context());
|
||||
extensions.insert(Arc::clone(server_context));
|
||||
extensions.insert(BodyReadControl::default());
|
||||
extensions.insert(VerifiedSigV4Request);
|
||||
extensions.insert(VerifiedPresignedRequest);
|
||||
extensions.insert(WebsiteRead);
|
||||
extensions.insert(PostObjectRequestMarker);
|
||||
extensions.insert(ReqInfo {
|
||||
bucket: Some("req-info-bucket".to_owned()),
|
||||
..ReqInfo::default()
|
||||
});
|
||||
extensions.insert(envelope_payloads::bucket_generation("generation-bucket"));
|
||||
extensions.insert(envelope_payloads::pending_delete_bucket_generation("pending-delete-bucket"));
|
||||
extensions.insert(envelope_payloads::copy_source_bucket_generation("copy-source-bucket"));
|
||||
extensions.insert(envelope_payloads::odm_read_generation("odm-read-bucket"));
|
||||
extensions.insert(envelope_payloads::bucket_config_mutation("config-mutation-bucket"));
|
||||
extensions.insert(TableDataPlanePublicationGuards::default());
|
||||
extensions.insert(envelope_payloads::table_list_access("table-list-bucket"));
|
||||
extensions.insert(envelope_payloads::object_tag_conditions("object-tag-bucket"));
|
||||
extensions
|
||||
}
|
||||
|
||||
/// The payload is present and is the instance built for `bucket`.
|
||||
fn assert_payload_for<T: Debug>(name: &str, payload: Option<&T>, bucket: &str) {
|
||||
let debug = format!("{payload:?}");
|
||||
assert!(payload.is_some() && debug.contains(&format!("\"{bucket}\"")), "{name}: {debug}");
|
||||
}
|
||||
|
||||
fn assert_every_named_payload_absent(envelope: &RequestEnvelope) {
|
||||
assert_eq!(envelope.remote_addr(), None, "remote_addr");
|
||||
assert!(envelope.client_info().is_none(), "client_info");
|
||||
assert!(envelope.request_context().is_none(), "request_context");
|
||||
assert!(envelope.server_context().is_none(), "server_context");
|
||||
assert!(envelope.body_read_control().is_none(), "body_read_control");
|
||||
assert!(envelope.verified_sigv4().is_none(), "verified_sigv4");
|
||||
assert!(envelope.verified_presigned().is_none(), "verified_presigned");
|
||||
assert!(envelope.website_read().is_none(), "website_read");
|
||||
assert!(envelope.post_object().is_none(), "post_object");
|
||||
assert!(envelope.req_info().is_none(), "req_info");
|
||||
assert!(envelope.bucket_generation().is_none(), "bucket_generation");
|
||||
assert!(envelope.pending_delete_bucket_generation().is_none(), "pending_delete_bucket_generation");
|
||||
assert!(envelope.copy_source_bucket_generation().is_none(), "copy_source_bucket_generation");
|
||||
assert!(envelope.odm_read_generation().is_none(), "odm_read_generation");
|
||||
assert!(envelope.bucket_config_mutation().is_none(), "bucket_config_mutation");
|
||||
assert!(envelope.table_publication_guards().is_none(), "table_publication_guards");
|
||||
assert!(envelope.table_list_access().is_none(), "table_list_access");
|
||||
assert!(envelope.object_tag_conditions().is_none(), "object_tag_conditions");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn from_s3s_copies_every_named_extension() {
|
||||
let server_context = ServerContextSlot::new();
|
||||
let mut req = request(every_named_payload(&server_context));
|
||||
req.method = Method::POST;
|
||||
req.uri = "/bucket/key?uploads".parse().expect("static uri");
|
||||
req.headers.insert("x-envelope-probe", HeaderValue::from_static("probe"));
|
||||
req.headers
|
||||
.insert("x-amz-security-token", HeaderValue::from_static("header-token"));
|
||||
req.credentials = credentials();
|
||||
req.region = Some("eu-west-3".parse().expect("valid region"));
|
||||
req.service = Some("s3".to_owned());
|
||||
|
||||
let envelope = RequestEnvelope::from_s3s(&req);
|
||||
|
||||
assert_eq!(envelope.method(), Method::POST);
|
||||
assert_eq!(envelope.uri(), &req.uri);
|
||||
assert_eq!(envelope.headers(), &req.headers);
|
||||
let credentials = envelope.credentials().expect("credentials");
|
||||
assert_eq!(credentials.access_key(), "AKIAENVELOPE");
|
||||
assert_eq!(credentials.secret_key(), "envelope-secret");
|
||||
assert_eq!(credentials.session_token(), Some("header-token"));
|
||||
assert_eq!(envelope.region(), Some("eu-west-3"));
|
||||
assert_eq!(envelope.service(), Some("s3"));
|
||||
|
||||
assert_eq!(envelope.remote_addr(), Some(SocketAddr::from(PEER)));
|
||||
assert_eq!(envelope.client_info().map(|info| info.real_ip), Some(IpAddr::from(CLIENT)));
|
||||
assert_eq!(envelope.request_context().map(|context| context.request_id.as_str()), Some(REQUEST_ID));
|
||||
assert!(
|
||||
envelope
|
||||
.server_context()
|
||||
.is_some_and(|slot| Arc::ptr_eq(slot, &server_context))
|
||||
);
|
||||
assert!(envelope.body_read_control().is_some(), "body_read_control");
|
||||
assert!(envelope.verified_sigv4().is_some(), "verified_sigv4");
|
||||
assert!(envelope.verified_presigned().is_some(), "verified_presigned");
|
||||
assert!(envelope.website_read().is_some(), "website_read");
|
||||
assert!(envelope.post_object().is_some(), "post_object");
|
||||
assert_eq!(envelope.req_info().and_then(|info| info.bucket.as_deref()), Some("req-info-bucket"));
|
||||
assert_payload_for("bucket_generation", envelope.bucket_generation(), "generation-bucket");
|
||||
assert_payload_for(
|
||||
"pending_delete_bucket_generation",
|
||||
envelope.pending_delete_bucket_generation(),
|
||||
"pending-delete-bucket",
|
||||
);
|
||||
assert_payload_for(
|
||||
"copy_source_bucket_generation",
|
||||
envelope.copy_source_bucket_generation(),
|
||||
"copy-source-bucket",
|
||||
);
|
||||
assert_payload_for("odm_read_generation", envelope.odm_read_generation(), "odm-read-bucket");
|
||||
assert_payload_for("bucket_config_mutation", envelope.bucket_config_mutation(), "config-mutation-bucket");
|
||||
assert!(envelope.table_publication_guards().is_some(), "table_publication_guards");
|
||||
assert!(envelope.table_list_access().is_some(), "table_list_access");
|
||||
assert_payload_for("object_tag_conditions", envelope.object_tag_conditions(), "object-tag-bucket");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn from_s3s_leaves_absent_payloads_none() {
|
||||
let mut req = request(Extensions::new());
|
||||
// A token on an anonymous request must not conjure credentials.
|
||||
req.headers
|
||||
.insert("x-amz-security-token", HeaderValue::from_static("anonymous-token"));
|
||||
|
||||
let envelope = RequestEnvelope::from_s3s(&req);
|
||||
|
||||
assert!(envelope.credentials().is_none(), "credentials");
|
||||
assert_eq!(envelope.region(), None, "region");
|
||||
assert_eq!(envelope.service(), None, "service");
|
||||
assert_every_named_payload_absent(&envelope);
|
||||
assert!(envelope.residual_extensions().is_empty(), "residual bag");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn from_s3s_treats_a_peerless_remote_addr_as_absent() {
|
||||
let mut extensions = Extensions::new();
|
||||
extensions.insert(None::<RemoteAddr>);
|
||||
|
||||
let envelope = RequestEnvelope::from_s3s(&request(extensions));
|
||||
|
||||
assert_eq!(envelope.remote_addr(), None);
|
||||
assert!(envelope.residual_extensions().is_empty(), "the peerless marker is consumed, not kept");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn from_s3s_keeps_unrelated_types_out_of_named_fields() {
|
||||
// Types adjacent to named payloads: the bare and raw peer addresses the
|
||||
// connection layer also inserts, wrapped variants of named types, and an
|
||||
// unrelated type. Today's readers never see any of them.
|
||||
let peer = SocketAddr::from(PEER);
|
||||
let mut extensions = Extensions::new();
|
||||
extensions.insert(RemoteAddr(peer));
|
||||
extensions.insert(peer);
|
||||
extensions.insert(Arc::new(request_context()));
|
||||
extensions.insert(Some(ClientInfo::direct(peer)));
|
||||
extensions.insert(UnrelatedPayload(7));
|
||||
|
||||
let envelope = RequestEnvelope::from_s3s(&request(extensions));
|
||||
|
||||
assert_every_named_payload_absent(&envelope);
|
||||
let residual = envelope.residual_extensions();
|
||||
assert_eq!(residual.len(), 5, "every unrelated type stays in the residual bag");
|
||||
assert_eq!(residual.get::<RemoteAddr>().map(|addr| addr.0), Some(peer));
|
||||
assert_eq!(residual.get::<SocketAddr>(), Some(&peer));
|
||||
assert!(residual.get::<Arc<RequestContext>>().is_some());
|
||||
assert!(residual.get::<Option<ClientInfo>>().is_some());
|
||||
assert_eq!(residual.get::<UnrelatedPayload>(), Some(&UnrelatedPayload(7)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn from_s3s_moves_named_payloads_out_of_the_residual_bag() {
|
||||
let server_context = ServerContextSlot::new();
|
||||
let mut extensions = every_named_payload(&server_context);
|
||||
extensions.insert(UnrelatedPayload(9));
|
||||
|
||||
let envelope = RequestEnvelope::from_s3s(&request(extensions));
|
||||
|
||||
let residual = envelope.residual_extensions();
|
||||
assert_eq!(residual.len(), 1, "only the unrelated type may remain");
|
||||
assert_eq!(residual.get::<UnrelatedPayload>(), Some(&UnrelatedPayload(9)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn from_s3s_takes_the_session_token_from_the_query_without_a_header() {
|
||||
let mut req = request(Extensions::new());
|
||||
req.uri = "/bucket/key?X-Amz-Security-Token=query-token".parse().expect("static uri");
|
||||
req.credentials = credentials();
|
||||
|
||||
let envelope = RequestEnvelope::from_s3s(&req);
|
||||
|
||||
assert_eq!(envelope.credentials().and_then(RequestCredentials::session_token), Some("query-token"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn from_s3s_leaves_the_session_token_none_when_the_request_carries_none() {
|
||||
let mut req = request(Extensions::new());
|
||||
req.credentials = credentials();
|
||||
|
||||
let envelope = RequestEnvelope::from_s3s(&req);
|
||||
|
||||
let credentials = envelope.credentials().expect("credentials");
|
||||
assert_eq!(credentials.access_key(), "AKIAENVELOPE");
|
||||
assert_eq!(credentials.session_token(), None);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
// 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 legacy s3s edge: conversions from s3s request types into the app
|
||||
//! layer's own types (rustfs/backlog#2734). Everything here goes away with the
|
||||
//! s3s edge itself (task T4.1).
|
||||
|
||||
mod envelope;
|
||||
@@ -124,11 +124,12 @@ pub(crate) use super::sse::{
|
||||
|
||||
pub(crate) mod access_consumer {
|
||||
pub(crate) use super::super::access::{
|
||||
PostObjectRequestMarker, ReqInfo, TABLE_DATA_PLANE_LIST_CURSOR_PREFIX, TableDataPlaneListAccess,
|
||||
TableDataPlaneListCursorPosition, TableDataPlanePublicationGuards, apply_bucket_generation_guard,
|
||||
apply_copy_source_bucket_generation_guard, authorize_internal_object_request, authorize_request,
|
||||
bucket_config_mutation_incarnation, delete_object_authorize_action, has_bypass_governance_header,
|
||||
load_bucket_generation_from_store, log_list_buckets_iam_implicit_deny, odm_read_generation,
|
||||
BucketConfigMutationSnapshot, BucketGenerationGuard, CopySourceBucketGenerationGuard, ObjectTagConditions,
|
||||
OdmReadGenerationGuard, PendingDeleteBucketGenerationGuard, PostObjectRequestMarker, ReqInfo,
|
||||
TABLE_DATA_PLANE_LIST_CURSOR_PREFIX, TableDataPlaneListAccess, TableDataPlaneListCursorPosition,
|
||||
TableDataPlanePublicationGuards, apply_bucket_generation_guard, apply_copy_source_bucket_generation_guard,
|
||||
authorize_internal_object_request, authorize_request, bucket_config_mutation_incarnation, delete_object_authorize_action,
|
||||
has_bypass_governance_header, load_bucket_generation_from_store, log_list_buckets_iam_implicit_deny, odm_read_generation,
|
||||
prepare_list_buckets_iam_authorization, prepare_odm_read_generation, recursive_force_delete_has_authenticated_caller,
|
||||
replication_request_authorized, req_info_mut, req_info_ref, retained_table_data_plane_publication_guards,
|
||||
};
|
||||
|
||||
@@ -22,7 +22,8 @@ classify_source_layer() {
|
||||
printf 'composition'
|
||||
elif [[ "$file" == rustfs/src/app/* ]]; then
|
||||
printf 'app'
|
||||
elif [[ "$file" == rustfs/src/admin/* ]] || [[ "$file" == rustfs/src/storage/ecfs.rs ]] || [[ "$file" == rustfs/src/storage/s3_api/* ]]; then
|
||||
elif [[ "$file" == rustfs/src/admin/* ]] || [[ "$file" == rustfs/src/storage/ecfs.rs ]] || [[ "$file" == rustfs/src/storage/s3_api/* ]] ||
|
||||
[[ "$file" == rustfs/src/storage/s3_compat/* ]]; then
|
||||
printf 'interface'
|
||||
elif [[ "$file" == rustfs/src/* ]]; then
|
||||
printf 'infra'
|
||||
@@ -54,7 +55,8 @@ classify_target_layer() {
|
||||
storage)
|
||||
storage_path="${path#storage::}"
|
||||
if [[ "$storage_path" == "ecfs" ]] || [[ "$storage_path" == ecfs::* ]] ||
|
||||
[[ "$storage_path" == "s3_api" ]] || [[ "$storage_path" == s3_api::* ]]; then
|
||||
[[ "$storage_path" == "s3_api" ]] || [[ "$storage_path" == s3_api::* ]] ||
|
||||
[[ "$storage_path" == "s3_compat" ]] || [[ "$storage_path" == s3_compat::* ]]; then
|
||||
printf 'interface'
|
||||
else
|
||||
printf 'infra'
|
||||
@@ -112,6 +114,7 @@ assert_dependency_direction() {
|
||||
|
||||
run_layer_model_self_tests() {
|
||||
local server_source app_source storage_source server_target admin_target app_target storage_target
|
||||
local s3_compat_source s3_compat_target
|
||||
|
||||
server_source="$(classify_source_layer rustfs/src/server/http.rs)"
|
||||
app_source="$(classify_source_layer rustfs/src/app/bucket_usecase.rs)"
|
||||
@@ -120,6 +123,9 @@ run_layer_model_self_tests() {
|
||||
admin_target="$(classify_target_layer admin::router)"
|
||||
app_target="$(classify_target_layer app::bucket_usecase)"
|
||||
storage_target="$(classify_target_layer storage::rpc)"
|
||||
# The s3s edge adapters convert wire requests into app types, as ecfs.rs does.
|
||||
s3_compat_source="$(classify_source_layer rustfs/src/storage/s3_compat/envelope.rs)"
|
||||
s3_compat_target="$(classify_target_layer storage::s3_compat::envelope)"
|
||||
|
||||
assert_dependency_direction 'allowed' "$server_source" "$admin_target"
|
||||
assert_dependency_direction 'allowed' "$server_source" "$app_target"
|
||||
@@ -128,6 +134,9 @@ run_layer_model_self_tests() {
|
||||
assert_dependency_direction 'reverse' "$storage_source" "$server_target"
|
||||
assert_dependency_direction 'reverse' "$app_source" "$admin_target"
|
||||
assert_dependency_direction 'reverse' "$storage_source" "$admin_target"
|
||||
assert_dependency_direction 'allowed' "$s3_compat_source" "$app_target"
|
||||
assert_dependency_direction 'reverse' "$app_source" "$s3_compat_target"
|
||||
assert_dependency_direction 'reverse' "$storage_source" "$s3_compat_target"
|
||||
}
|
||||
|
||||
normalize_import_group_item() {
|
||||
|
||||
Reference in New Issue
Block a user