Add operation customization for disabling payload signing (#3915)

## Motivation and Context
<!--- Why is this change required? What problem does it solve? -->
<!--- If it fixes an open issue, please link to the issue here -->
#3583

## Description
<!--- Describe your changes in detail -->
This PR adds the ability for users to disable payload signing with an
operation customization.

## Testing
<!--- Please describe in detail how you tested your changes -->
<!--- Include details of your testing environment, and the tests you ran
to -->
<!--- see how your change affects other areas of the code, etc. -->
This PR includes tests.

## Checklist
<!--- If a checkbox below is not applicable, then please DELETE it
rather than leaving it unchecked -->
- [x] For changes to the AWS SDK, generated SDK code, or SDK runtime
crates, I have created a changelog entry Markdown file in the
`.changelog` directory, specifying "aws-sdk-rust" in the `applies_to`
key.

----

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

Author: Zelda Hessler

.changelog/1732034799.md

@@ -0,0 +1,28 @@
+---
+applies_to: ["client"]
+authors: ["Velfi"]
+references: ["smithy-rs#3583"]
+breaking: false
+new_feature: true
+bug_fix: false
+---
+
+It is now possible to disable payload signing through an operation customization.
+
+```rust
+async fn put_example_object(client: &aws_sdk_s3::Client) {
+    let res = client
+        .put_object()
+        .bucket("test-bucket")
+        .key("test-key")
+        .body(ByteStream::from_static(b"Hello, world!"))
+        .customize()
+        // Setting this will disable payload signing.
+        .disable_payload_signing()
+        .send()
+        .await;
+}
+```
+
+Disabling payload signing will result in a small speedup at the cost of removing a data integrity check.
+However, this is an advanced feature and **may not be supported by all services/operations**.

aws/rust-runtime/Cargo.lock

@@ -7,9 +7,9 @@
 
 //! Interceptor for handling Smithy `@httpChecksum` request checksumming with AWS SigV4
 
+use aws_runtime::auth::PayloadSigningOverride;
+use aws_runtime::content_encoding::header_value::AWS_CHUNKED;
 use aws_runtime::content_encoding::{AwsChunkedBody, AwsChunkedBodyOptions};
-use aws_runtime::{auth::SigV4OperationSigningConfig, content_encoding::header_value::AWS_CHUNKED};
-use aws_sigv4::http_request::SignableBody;
 use aws_smithy_checksums::ChecksumAlgorithm;
 use aws_smithy_checksums::{body::calculate, http::HttpChecksum};
 use aws_smithy_runtime_api::box_error::BoxError;
@@ -190,11 +190,8 @@ fn add_checksum_for_request_body(
         // Body is streaming: wrap the body so it will emit a checksum as a trailer.
         None => {
             tracing::debug!("applying {checksum_algorithm:?} of the request body as a trailer");
-            if let Some(mut signing_config) = cfg.load::<SigV4OperationSigningConfig>().cloned() {
-                signing_config.signing_options.payload_override =
-                    Some(SignableBody::StreamingUnsignedPayloadTrailer);
-                cfg.interceptor_state().store_put(signing_config);
-            }
+            cfg.interceptor_state()
+                .store_put(PayloadSigningOverride::StreamingUnsignedPayloadTrailer);
             wrap_streaming_request_body_in_checksum_calculating_body(request, checksum_algorithm)?;
         }
     }

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

@@ -1,6 +1,6 @@
 [package]
 name = "aws-runtime"
-version = "1.4.4"
+version = "1.5.0"
 authors = ["AWS Rust SDK Team <[email protected]>"]
 description = "Runtime support code for the AWS SDK. This crate isn't intended to be used directly."
 edition = "2021"

aws/rust-runtime/aws-runtime/Cargo.toml

@@ -11,7 +11,8 @@ use aws_smithy_runtime_api::box_error::BoxError;
 use aws_smithy_runtime_api::client::auth::AuthSchemeEndpointConfig;
 use aws_smithy_runtime_api::client::identity::Identity;
 use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
-use aws_smithy_types::config_bag::{ConfigBag, Storable, StoreReplace};
+use aws_smithy_runtime_api::client::runtime_plugin::RuntimePlugin;
+use aws_smithy_types::config_bag::{ConfigBag, FrozenLayer, Layer, Storable, StoreReplace};
 use aws_smithy_types::Document;
 use aws_types::region::{Region, SigningRegion, SigningRegionSet};
 use aws_types::SigningName;
@@ -265,3 +266,70 @@ fn apply_signing_instructions(
     }
     Ok(())
 }
+
+/// When present in the config bag, this type will signal that the default
+/// payload signing should be overridden.
+#[non_exhaustive]
+#[derive(Clone, Debug)]
+pub enum PayloadSigningOverride {
+    /// An unsigned payload
+    ///
+    /// UnsignedPayload is used for streaming requests where the contents of the body cannot be
+    /// known prior to signing
+    UnsignedPayload,
+
+    /// A precomputed body checksum. The checksum should be a SHA256 checksum of the body,
+    /// lowercase hex encoded. Eg:
+    /// `e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855`
+    Precomputed(String),
+
+    /// Set when a streaming body has checksum trailers.
+    StreamingUnsignedPayloadTrailer,
+}
+
+impl PayloadSigningOverride {
+    /// Create a payload signing override that will prevent the payload from
+    /// being signed.
+    pub fn unsigned_payload() -> Self {
+        Self::UnsignedPayload
+    }
+
+    /// Convert this type into the type used by the signer to determine how a
+    /// request body should be signed.
+    pub fn to_signable_body(self) -> SignableBody<'static> {
+        match self {
+            Self::UnsignedPayload => SignableBody::UnsignedPayload,
+            Self::Precomputed(checksum) => SignableBody::Precomputed(checksum),
+            Self::StreamingUnsignedPayloadTrailer => SignableBody::StreamingUnsignedPayloadTrailer,
+        }
+    }
+}
+
+impl Storable for PayloadSigningOverride {
+    type Storer = StoreReplace<Self>;
+}
+
+/// A runtime plugin that, when set, will override how the signer signs request payloads.
+#[derive(Debug)]
+pub struct PayloadSigningOverrideRuntimePlugin {
+    inner: FrozenLayer,
+}
+
+impl PayloadSigningOverrideRuntimePlugin {
+    /// Create a new runtime plugin that will force the signer to skip signing
+    /// the request payload when signing an HTTP request.
+    pub fn unsigned() -> Self {
+        let mut layer = Layer::new("PayloadSigningOverrideRuntimePlugin");
+        layer.store_put(PayloadSigningOverride::UnsignedPayload);
+
+        Self {
+            inner: layer.freeze(),
+        }
+    }
+}
+
+impl RuntimePlugin for PayloadSigningOverrideRuntimePlugin {
+    fn config(&self) -> Option<FrozenLayer> {
+        Some(self.inner.clone())
+    }
+}

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

@@ -6,7 +6,8 @@
 use crate::auth;
 use crate::auth::{
     extract_endpoint_auth_scheme_signing_name, extract_endpoint_auth_scheme_signing_region,
-    SigV4OperationSigningConfig, SigV4SessionTokenNameOverride, SigV4SigningError,
+    PayloadSigningOverride, SigV4OperationSigningConfig, SigV4SessionTokenNameOverride,
+    SigV4SigningError,
 };
 use aws_credential_types::Credentials;
 use aws_sigv4::http_request::{
@@ -177,7 +178,7 @@ impl Sign for SigV4Signer {
         let (signing_instructions, _signature) = {
             // A body that is already in memory can be signed directly. A body that is not in memory
             // (any sort of streaming body or presigned request) will be signed via UNSIGNED-PAYLOAD.
-            let signable_body = operation_config
+            let mut signable_body = operation_config
                 .signing_options
                 .payload_override
                 .as_ref()
@@ -192,6 +193,15 @@ impl Sign for SigV4Signer {
                         .unwrap_or(SignableBody::UnsignedPayload)
                 });
 
+            // Sometimes it's necessary to override the payload signing scheme.
+            // If an override exists then fetch and apply it.
+            if let Some(payload_signing_override) = config_bag.load::<PayloadSigningOverride>() {
+                tracing::trace!(
+                    "payload signing was overridden, now set to {payload_signing_override:?}"
+                );
+                signable_body = payload_signing_override.clone().to_signable_body();
+            }
+
             let signable_request = SignableRequest::new(
                 request.method(),
                 request.uri(),

aws/rust-runtime/aws-runtime/src/auth/sigv4.rs

@@ -66,6 +66,7 @@ val DECORATORS: List<ClientCodegenDecorator> =
             TokenProvidersDecorator(),
             ServiceEnvConfigDecorator(),
             HttpRequestCompressionDecorator(),
+            DisablePayloadSigningDecorator(),
             // TODO(https://github.com/smithy-lang/smithy-rs/issues/3863): Comment in once the issue has been resolved
             // SmokeTestsDecorator(),
         ),

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

@@ -0,0 +1,71 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package software.amazon.smithy.rustsdk
+
+import software.amazon.smithy.model.shapes.OperationShape
+import software.amazon.smithy.model.shapes.ShapeId
+import software.amazon.smithy.rust.codegen.client.smithy.ClientCodegenContext
+import software.amazon.smithy.rust.codegen.client.smithy.customize.ClientCodegenDecorator
+import software.amazon.smithy.rust.codegen.client.smithy.generators.OperationCustomization
+import software.amazon.smithy.rust.codegen.client.smithy.generators.OperationSection
+import software.amazon.smithy.rust.codegen.core.rustlang.Writable
+import software.amazon.smithy.rust.codegen.core.rustlang.rustTemplate
+import software.amazon.smithy.rust.codegen.core.rustlang.writable
+import software.amazon.smithy.rust.codegen.core.smithy.RuntimeType.Companion.preludeScope
+
+internal val DISABLE_PAYLOAD_SIGNING_OPERATIONS by lazy {
+    listOf(
+        // S3
+        ShapeId.from("com.amazonaws.s3#PutObject"),
+        ShapeId.from("com.amazonaws.s3#UploadPart"),
+    )
+}
+
+class DisablePayloadSigningDecorator : ClientCodegenDecorator {
+    override val name: String = "DisablePayloadSigning"
+    override val order: Byte = 0
+
+    override fun operationCustomizations(
+        codegenContext: ClientCodegenContext,
+        operation: OperationShape,
+        baseCustomizations: List<OperationCustomization>,
+    ): List<OperationCustomization> {
+        return baseCustomizations +
+            object : OperationCustomization() {
+                private val runtimeConfig = codegenContext.runtimeConfig
+
+                override fun section(section: OperationSection): Writable {
+                    return writable {
+                        when (section) {
+                            is OperationSection.CustomizableOperationImpl -> {
+                                if (DISABLE_PAYLOAD_SIGNING_OPERATIONS.contains(operation.id)) {
+                                    rustTemplate(
+                                        """
+                                        /// Disable payload signing for this request.
+                                        ///
+                                        /// **WARNING:** This is an advanced feature that removes
+                                        /// the cost of signing a request payload by removing a data
+                                        /// integrity check. Not all services/operations support
+                                        /// this feature.
+                                        pub fn disable_payload_signing(self) -> Self {
+                                            self.runtime_plugin(#{PayloadSigningOverrideRuntimePlugin}::unsigned())
+                                        }
+                                        """,
+                                        *preludeScope,
+                                        "PayloadSigningOverrideRuntimePlugin" to
+                                            AwsRuntimeType.awsRuntime(runtimeConfig)
+                                                .resolve("auth::PayloadSigningOverrideRuntimePlugin"),
+                                    )
+                                }
+                            }
+
+                            else -> {}
+                        }
+                    }
+                }
+            }
+    }
+}

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

@@ -7,7 +7,9 @@
 
 use aws_credential_types::provider::SharedCredentialsProvider;
 use aws_sdk_s3::config::{Credentials, Region};
+use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::{Client, Config};
+use aws_smithy_runtime::client::http::test_util::capture_request;
 use aws_smithy_runtime::client::http::test_util::{ReplayEvent, StaticReplayClient};
 use aws_smithy_types::body::SdkBody;
 use http::header::AUTHORIZATION;
@@ -40,3 +42,80 @@ async fn test_signer() {
 
     http_client.assert_requests_match(&[AUTHORIZATION.as_str()]);
 }
+
+#[tokio::test]
+async fn disable_payload_signing_works() {
+    let (http_client, request) = capture_request(None);
+    let conf = aws_sdk_s3::Config::builder()
+        .with_test_defaults()
+        .behavior_version_latest()
+        .region(Region::new("us-east-1"))
+        .http_client(http_client)
+        .build();
+    let client = aws_sdk_s3::Client::from_conf(conf);
+    let _ = client
+        .put_object()
+        .bucket("XXXXXXXXXXX")
+        .key("test-key")
+        .body(ByteStream::from_static(b"Hello, world!"))
+        .customize()
+        .disable_payload_signing()
+        .send()
+        .await;
+
+    let request = request.expect_request();
+    let x_amz_content_sha256 = request
+        .headers()
+        .get("x-amz-content-sha256")
+        .expect("x-amz-content-sha256 is set")
+        .to_owned();
+    assert_eq!("UNSIGNED-PAYLOAD", x_amz_content_sha256);
+}
+
+// This test ensures that the request checksum interceptor payload signing
+// override takes priority over the runtime plugin's override. If it didn't,
+// then disabling payload signing would cause requests to incorrectly omit
+// trailers.
+#[tokio::test]
+async fn disable_payload_signing_works_with_checksums() {
+    let (http_client, request) = capture_request(None);
+    let conf = aws_sdk_s3::Config::builder()
+        .with_test_defaults()
+        .behavior_version_latest()
+        .region(Region::new("us-east-1"))
+        .http_client(http_client)
+        .build();
+    let client = aws_sdk_s3::Client::from_conf(conf);
+
+    // ByteStreams created from a file are streaming and have a known size
+    let mut file = tempfile::NamedTempFile::new().unwrap();
+    use std::io::Write;
+    file.write_all(b"Hello, world!").unwrap();
+
+    let body = aws_sdk_s3::primitives::ByteStream::read_from()
+        .path(file.path())
+        .buffer_size(1024)
+        .build()
+        .await
+        .unwrap();
+
+    let _ = client
+        .put_object()
+        .bucket("XXXXXXXXXXX")
+        .key("test-key")
+        .body(body)
+        .checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Crc32)
+        .customize()
+        .disable_payload_signing()
+        .send()
+        .await;
+
+    let request = request.expect_request();
+    let x_amz_content_sha256 = request
+        .headers()
+        .get("x-amz-content-sha256")
+        .expect("x-amz-content-sha256 is set")
+        .to_owned();
+    // The checksum interceptor sets this.
+    assert_eq!("STREAMING-UNSIGNED-PAYLOAD-TRAILER", x_amz_content_sha256);
+}

aws/sdk/integration-tests/s3/tests/signing-it.rs

@@ -11,6 +11,7 @@ async fn assume_role_signed() {
     let creds = Credentials::for_tests();
     let (http_client, request) = capture_request(None);
     let conf = aws_sdk_sts::Config::builder()
+        .behavior_version_latest()
         .credentials_provider(creds)
         .region(Region::new("us-east-1"))
         .http_client(http_client)
@@ -28,6 +29,7 @@ async fn assume_role_signed() {
 async fn web_identity_unsigned() {
     let (http_client, request) = capture_request(None);
     let conf = aws_sdk_sts::Config::builder()
+        .behavior_version_latest()
         .region(Region::new("us-east-1"))
         .http_client(http_client)
         .build();
@@ -44,6 +46,7 @@ async fn web_identity_unsigned() {
 async fn assume_role_saml_unsigned() {
     let (http_client, request) = capture_request(None);
     let conf = aws_sdk_sts::Config::builder()
+        .behavior_version_latest()
         .region(Region::new("us-east-1"))
         .http_client(http_client)
         .build();
@@ -60,6 +63,7 @@ async fn assume_role_saml_unsigned() {
 async fn web_identity_no_creds() {
     let (http_client, request) = capture_request(None);
     let conf = aws_sdk_sts::Config::builder()
+        .behavior_version_latest()
         .region(Region::new("us-east-1"))
         .http_client(http_client)
         .build();