Fix S3 presigning in the orchestrator (#2738)

## Motivation and Context
This PR fixes S3 presigning in the client orchestrator implementation. A
Polly presigning fix will come in a separate PR.

This PR also:
- Renames `ClientProtocolGenerator` to `OperationGenerator` to better
represent what its generating.
- Moves `OperationCustomization` into `codegen-client` since it is
client specific and unused by the server codegen.
- Deletes the `ProtocolGenerator` base class in `codegen-core` since it
wasn't adding any value.
- Adds the concept of stop points to the orchestrator so that
orchestration can be halted before transmitting a request.
- Adds `DisableInvocationIdInterceptor`,
`DisableRequestInfoInterceptor`, `DisableUserAgentInterceptor`, and
`HeaderSerializationSettings` config bag configs to facilitate
presigning requirements.
- Fixes compilation of all S3 and Polly tests when in orchestrator mode.

----

_By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice._

Author: jdisanti

aws/rust-runtime/aws-inlineable/src/lib.rs

@@ -28,6 +28,12 @@ pub mod no_credentials;
 /// Support types required for adding presigning to an operation in a generated service.
 pub mod presigning;
 
+/// Presigning tower service
+pub mod presigning_service;
+
+/// Presigning interceptors
+pub mod presigning_interceptors;
+
 /// Special logic for extracting request IDs from S3's responses.
 pub mod s3_request_id;
 
@@ -49,6 +55,10 @@ pub mod http_body_checksum;
 #[allow(dead_code)]
 pub mod endpoint_discovery;
 
+// This module is symlinked in from the smithy-rs rust-runtime inlineables so that
+// the `presigning_interceptors` module can refer to it.
+mod serialization_settings;
+
 // just so docs work
 #[allow(dead_code)]
 /// allow docs to work

aws/rust-runtime/aws-inlineable/src/presigning.rs

@@ -215,60 +215,3 @@ impl From<PresignedRequest> for http::request::Builder {
         builder
     }
 }
-
-/// Tower middleware service for creating presigned requests
-#[allow(dead_code)]
-pub(crate) mod service {
-    use super::PresignedRequest;
-    use aws_smithy_http::operation;
-    use http::header::USER_AGENT;
-    use std::future::{ready, Ready};
-    use std::marker::PhantomData;
-    use std::task::{Context, Poll};
-
-    /// Tower [`Service`](tower::Service) for generated a [`PresignedRequest`] from the AWS middleware.
-    #[derive(Default, Debug)]
-    #[non_exhaustive]
-    pub(crate) struct PresignedRequestService<E> {
-        _phantom: PhantomData<E>,
-    }
-
-    // Required because of the derive Clone on MapRequestService.
-    // Manually implemented to avoid requiring errors to implement Clone.
-    impl<E> Clone for PresignedRequestService<E> {
-        fn clone(&self) -> Self {
-            Self {
-                _phantom: Default::default(),
-            }
-        }
-    }
-
-    impl<E> PresignedRequestService<E> {
-        /// Creates a new `PresignedRequestService`
-        pub(crate) fn new() -> Self {
-            Self {
-                _phantom: Default::default(),
-            }
-        }
-    }
-
-    impl<E> tower::Service<operation::Request> for PresignedRequestService<E> {
-        type Response = PresignedRequest;
-        type Error = E;
-        type Future = Ready<Result<PresignedRequest, E>>;
-
-        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-            Poll::Ready(Ok(()))
-        }
-
-        fn call(&mut self, req: operation::Request) -> Self::Future {
-            let (mut req, _) = req.into_parts();
-
-            // Remove user agent headers since the request will not be executed by the AWS Rust SDK.
-            req.headers_mut().remove(USER_AGENT);
-            req.headers_mut().remove("X-Amz-User-Agent");
-
-            ready(Ok(PresignedRequest::new(req.map(|_| ()))))
-        }
-    }
-}

aws/rust-runtime/aws-inlineable/src/presigning_interceptors.rs

@@ -0,0 +1,103 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#![allow(dead_code)]
+
+use crate::presigning::PresigningConfig;
+use crate::serialization_settings::HeaderSerializationSettings;
+use aws_runtime::auth::sigv4::{HttpSignatureType, SigV4OperationSigningConfig};
+use aws_runtime::invocation_id::DisableInvocationIdInterceptor;
+use aws_runtime::request_info::DisableRequestInfoInterceptor;
+use aws_runtime::user_agent::DisableUserAgentInterceptor;
+use aws_smithy_async::time::{SharedTimeSource, StaticTimeSource};
+use aws_smithy_runtime_api::client::interceptors::{
+    BeforeSerializationInterceptorContextMut, BeforeTransmitInterceptorContextMut, BoxError,
+    Interceptor, InterceptorRegistrar, SharedInterceptor,
+};
+use aws_smithy_runtime_api::client::orchestrator::ConfigBagAccessors;
+use aws_smithy_runtime_api::client::runtime_plugin::RuntimePlugin;
+use aws_smithy_types::config_bag::ConfigBag;
+
+/// Interceptor that tells the SigV4 signer to add the signature to query params,
+/// and sets the request expiration time from the presigning config.
+#[derive(Debug)]
+pub(crate) struct SigV4PresigningInterceptor {
+    config: PresigningConfig,
+}
+
+impl SigV4PresigningInterceptor {
+    pub(crate) fn new(config: PresigningConfig) -> Self {
+        Self { config }
+    }
+}
+
+impl Interceptor for SigV4PresigningInterceptor {
+    fn modify_before_serialization(
+        &self,
+        _context: &mut BeforeSerializationInterceptorContextMut<'_>,
+        cfg: &mut ConfigBag,
+    ) -> Result<(), BoxError> {
+        cfg.put::<HeaderSerializationSettings>(
+            HeaderSerializationSettings::new()
+                .omit_default_content_length()
+                .omit_default_content_type(),
+        );
+        cfg.set_request_time(SharedTimeSource::new(StaticTimeSource::new(
+            self.config.start_time(),
+        )));
+        Ok(())
+    }
+
+    fn modify_before_signing(
+        &self,
+        _context: &mut BeforeTransmitInterceptorContextMut<'_>,
+        cfg: &mut ConfigBag,
+    ) -> Result<(), BoxError> {
+        if let Some(mut config) = cfg.get::<SigV4OperationSigningConfig>().cloned() {
+            config.signing_options.expires_in = Some(self.config.expires());
+            config.signing_options.signature_type = HttpSignatureType::HttpRequestQueryParams;
+            config.signing_options.payload_override =
+                Some(aws_sigv4::http_request::SignableBody::UnsignedPayload);
+            cfg.put::<SigV4OperationSigningConfig>(config);
+            Ok(())
+        } else {
+            Err(
+                "SigV4 presigning requires the SigV4OperationSigningConfig to be in the config bag. \
+                This is a bug. Please file an issue.".into(),
+            )
+        }
+    }
+}
+
+/// Runtime plugin that registers the SigV4PresigningInterceptor.
+#[derive(Debug)]
+pub(crate) struct SigV4PresigningRuntimePlugin {
+    interceptor: SharedInterceptor,
+}
+
+impl SigV4PresigningRuntimePlugin {
+    pub(crate) fn new(config: PresigningConfig) -> Self {
+        Self {
+            interceptor: SharedInterceptor::new(SigV4PresigningInterceptor::new(config)),
+        }
+    }
+}
+
+impl RuntimePlugin for SigV4PresigningRuntimePlugin {
+    fn configure(
+        &self,
+        cfg: &mut ConfigBag,
+        interceptors: &mut InterceptorRegistrar,
+    ) -> Result<(), BoxError> {
+        // Disable some SDK interceptors that shouldn't run for presigning
+        cfg.put(DisableInvocationIdInterceptor::new("presigning"));
+        cfg.put(DisableRequestInfoInterceptor::new("presigning"));
+        cfg.put(DisableUserAgentInterceptor::new("presigning"));
+
+        // Register the presigning interceptor
+        interceptors.register(self.interceptor.clone());
+        Ok(())
+    }
+}

aws/rust-runtime/aws-inlineable/src/presigning_service.rs

@@ -0,0 +1,61 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#![allow(dead_code)]
+
+//! Tower middleware service for creating presigned requests
+
+use crate::presigning::PresignedRequest;
+use aws_smithy_http::operation;
+use http::header::USER_AGENT;
+use std::future::{ready, Ready};
+use std::marker::PhantomData;
+use std::task::{Context, Poll};
+
+/// Tower [`Service`](tower::Service) for generated a [`PresignedRequest`] from the AWS middleware.
+#[derive(Default, Debug)]
+#[non_exhaustive]
+pub(crate) struct PresignedRequestService<E> {
+    _phantom: PhantomData<E>,
+}
+
+// Required because of the derive Clone on MapRequestService.
+// Manually implemented to avoid requiring errors to implement Clone.
+impl<E> Clone for PresignedRequestService<E> {
+    fn clone(&self) -> Self {
+        Self {
+            _phantom: Default::default(),
+        }
+    }
+}
+
+impl<E> PresignedRequestService<E> {
+    /// Creates a new `PresignedRequestService`
+    pub(crate) fn new() -> Self {
+        Self {
+            _phantom: Default::default(),
+        }
+    }
+}
+
+impl<E> tower::Service<operation::Request> for PresignedRequestService<E> {
+    type Response = PresignedRequest;
+    type Error = E;
+    type Future = Ready<Result<PresignedRequest, E>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, req: operation::Request) -> Self::Future {
+        let (mut req, _) = req.into_parts();
+
+        // Remove user agent headers since the request will not be executed by the AWS Rust SDK.
+        req.headers_mut().remove(USER_AGENT);
+        req.headers_mut().remove("X-Amz-User-Agent");
+
+        ready(Ok(PresignedRequest::new(req.map(|_| ()))))
+    }
+}

aws/rust-runtime/aws-inlineable/src/serialization_settings.rs

@@ -0,0 +1 @@
+../../../../rust-runtime/inlineable/src/serialization_settings.rs

aws/rust-runtime/aws-runtime/src/invocation_id.rs

@@ -18,6 +18,23 @@ pub use test_util::{NoInvocationIdGenerator, PredefinedInvocationIdGenerator};
 #[allow(clippy::declare_interior_mutable_const)] // we will never mutate this
 const AMZ_SDK_INVOCATION_ID: HeaderName = HeaderName::from_static("amz-sdk-invocation-id");
 
+/// Config marker that disables the invocation ID interceptor.
+#[doc(hidden)]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct DisableInvocationIdInterceptor {
+    why: &'static str,
+}
+
+impl DisableInvocationIdInterceptor {
+    /// Creates a new `DisableInvocationIdInterceptor`.
+    ///
+    /// Takes a human readable string for the `Debug` impl to state why it is being disabled.
+    /// This is to assist with debugging issues with requests.
+    pub fn new(why: &'static str) -> Self {
+        Self { why }
+    }
+}
+
 /// A generator for returning new invocation IDs on demand.
 pub trait InvocationIdGenerator: Debug + Send + Sync {
     /// Call this function to receive a new [`InvocationId`] or an error explaining why one couldn't

aws/rust-runtime/aws-runtime/src/request_info.rs

@@ -20,6 +20,23 @@ use std::time::{Duration, SystemTime};
 #[allow(clippy::declare_interior_mutable_const)] // we will never mutate this
 const AMZ_SDK_REQUEST: HeaderName = HeaderName::from_static("amz-sdk-request");
 
+/// Config marker that disables the invocation ID interceptor.
+#[doc(hidden)]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct DisableRequestInfoInterceptor {
+    why: &'static str,
+}
+
+impl DisableRequestInfoInterceptor {
+    /// Creates a new `DisableRequestInfoInterceptor`.
+    ///
+    /// Takes a human readable string for the `Debug` impl to state why it is being disabled.
+    /// This is to assist with debugging issues with requests.
+    pub fn new(why: &'static str) -> Self {
+        Self { why }
+    }
+}
+
 /// Generates and attaches a request header that communicates request-related metadata.
 /// Examples include:
 ///

aws/rust-runtime/aws-runtime/src/user_agent.rs

@@ -49,6 +49,23 @@ impl From<InvalidHeaderValue> for UserAgentInterceptorError {
     }
 }
 
+/// Config marker that disables the user agent interceptor.
+#[doc(hidden)]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct DisableUserAgentInterceptor {
+    why: &'static str,
+}
+
+impl DisableUserAgentInterceptor {
+    /// Creates a new `DisableUserAgentInterceptor`.
+    ///
+    /// Takes a human readable string for the `Debug` impl to state why it is being disabled.
+    /// This is to assist with debugging issues with requests.
+    pub fn new(why: &'static str) -> Self {
+        Self { why }
+    }
+}
+
 /// Generates and attaches the AWS SDK's user agent to a HTTP request
 #[non_exhaustive]
 #[derive(Debug, Default)]

aws/sdk-codegen/src/main/kotlin/software/amazon/smithy/rustsdk/AwsFluentClientDecorator.kt

@@ -64,7 +64,7 @@ class AwsFluentClientDecorator : ClientCodegenDecorator {
             reexportSmithyClientBuilder = false,
             generics = generics,
             customizations = listOf(
-                AwsPresignedFluentBuilderMethod(runtimeConfig),
+                AwsPresignedFluentBuilderMethod(codegenContext),
                 AwsFluentClientDocs(codegenContext),
             ),
             retryClassifier = AwsRuntimeType.awsHttp(runtimeConfig).resolve("retry::AwsResponseRetryClassifier"),