diff --git a/.changes/6aef179b-d710-40a5-bd6e-37078f07dfa4.json b/.changes/6aef179b-d710-40a5-bd6e-37078f07dfa4.json
new file mode 100644
index 00000000000..68c24754d0d
--- /dev/null
+++ b/.changes/6aef179b-d710-40a5-bd6e-37078f07dfa4.json
@@ -0,0 +1,5 @@
+{
+    "id": "6aef179b-d710-40a5-bd6e-37078f07dfa4",
+    "type": "feature",
+    "description": "Add support for S3 Express One Zone"
+}
\ No newline at end of file
diff --git a/aws-runtime/aws-http/build.gradle.kts b/aws-runtime/aws-http/build.gradle.kts
index c91aa409876..21a560d31a8 100644
--- a/aws-runtime/aws-http/build.gradle.kts
+++ b/aws-runtime/aws-http/build.gradle.kts
@@ -15,6 +15,7 @@ kotlin {
                 api(project(":aws-runtime:aws-endpoint"))
                 api(libs.smithy.kotlin.aws.signing.common)
                 api(libs.smithy.kotlin.http.client)
+                api(libs.smithy.kotlin.http.auth.aws)
             }
         }
 
diff --git a/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/flexiblechecksums/FlexibleChecksumsRequest.kt b/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/flexiblechecksums/FlexibleChecksumsRequest.kt
index d13aa4eb3c2..8799ac5364e 100644
--- a/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/flexiblechecksums/FlexibleChecksumsRequest.kt
+++ b/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/flexiblechecksums/FlexibleChecksumsRequest.kt
@@ -8,7 +8,6 @@ import software.amazon.smithy.aws.traits.HttpChecksumTrait
 import software.amazon.smithy.kotlin.codegen.KotlinSettings
 import software.amazon.smithy.kotlin.codegen.core.KotlinWriter
 import software.amazon.smithy.kotlin.codegen.core.RuntimeTypes
-import software.amazon.smithy.kotlin.codegen.core.defaultName
 import software.amazon.smithy.kotlin.codegen.core.withBlock
 import software.amazon.smithy.kotlin.codegen.integration.KotlinIntegration
 import software.amazon.smithy.kotlin.codegen.model.*
@@ -43,8 +42,8 @@ class FlexibleChecksumsRequest : KotlinIntegration {
         }
 
         override fun render(ctx: ProtocolGenerator.GenerationContext, op: OperationShape, writer: KotlinWriter) {
-            val inputSymbol = ctx.symbolProvider.toSymbol(ctx.model.expectShape(op.inputShape))
             val interceptorSymbol = RuntimeTypes.HttpClient.Interceptors.FlexibleChecksumsRequestInterceptor
+            val inputSymbol = ctx.symbolProvider.toSymbol(ctx.model.expectShape(op.inputShape))
 
             val httpChecksumTrait = op.getTrait<HttpChecksumTrait>()!!
 
@@ -52,13 +51,13 @@ class FlexibleChecksumsRequest : KotlinIntegration {
                 .members()
                 .first { it.memberName == httpChecksumTrait.requestAlgorithmMember.get() }
 
-            writer.withBlock(
-                "op.interceptors.add(#T<#T> {",
-                "})",
-                interceptorSymbol,
-                inputSymbol,
-            ) {
-                writer.write("it.#L?.value", requestAlgorithmMember.defaultName())
+            val requestAlgorithmMemberName = ctx.symbolProvider.toMemberName(requestAlgorithmMember)
+
+            writer.withBlock("op.interceptors.add(#T<#T>() {", "})", interceptorSymbol, inputSymbol) {
+                writer.write("input.#L?.value", requestAlgorithmMemberName)
+            }
+            writer.withBlock("input.#L?.let {", "}", requestAlgorithmMemberName) {
+                writer.write("op.context[#T.ChecksumAlgorithm] = it.value", RuntimeTypes.HttpClient.Operation.HttpOperationContext)
             }
         }
     }
diff --git a/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/s3/express/S3ExpressIntegration.kt b/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/s3/express/S3ExpressIntegration.kt
new file mode 100644
index 00000000000..ce5ded2221a
--- /dev/null
+++ b/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/s3/express/S3ExpressIntegration.kt
@@ -0,0 +1,190 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.codegen.customization.s3.express
+
+import SigV4S3ExpressAuthTrait
+import aws.sdk.kotlin.codegen.customization.s3.isS3
+import software.amazon.smithy.aws.traits.HttpChecksumTrait
+import software.amazon.smithy.kotlin.codegen.KotlinSettings
+import software.amazon.smithy.kotlin.codegen.core.*
+import software.amazon.smithy.kotlin.codegen.integration.KotlinIntegration
+import software.amazon.smithy.kotlin.codegen.lang.KotlinTypes
+import software.amazon.smithy.kotlin.codegen.model.*
+import software.amazon.smithy.kotlin.codegen.rendering.protocol.ProtocolGenerator
+import software.amazon.smithy.kotlin.codegen.rendering.protocol.ProtocolMiddleware
+import software.amazon.smithy.kotlin.codegen.rendering.util.ConfigProperty
+import software.amazon.smithy.kotlin.codegen.rendering.util.ConfigPropertyType
+import software.amazon.smithy.kotlin.codegen.utils.dq
+import software.amazon.smithy.kotlin.codegen.utils.getOrNull
+import software.amazon.smithy.model.Model
+import software.amazon.smithy.model.shapes.*
+import software.amazon.smithy.model.traits.*
+import software.amazon.smithy.model.transform.ModelTransformer
+
+/**
+ * An integration which handles codegen for S3 Express, such as:
+ * 1. Configure auth scheme by applying a synthetic shape and trait
+ * 2. Add ExpressClient and Bucket to execution context
+ * 3. Override checksums to use CRC32 instead of MD5
+ * 4. Disable all checksums for s3:UploadPart
+ */
+class S3ExpressIntegration : KotlinIntegration {
+    companion object {
+        val DisableExpressSessionAuth: ConfigProperty = ConfigProperty {
+            name = "disableS3ExpressSessionAuth"
+            useSymbolWithNullableBuilder(KotlinTypes.Boolean, "false")
+            documentation = """
+                Flag to disable S3 Express One Zone's bucket-level session authentication method.  
+            """.trimIndent()
+        }
+
+        val ExpressCredentialsProvider: ConfigProperty = ConfigProperty {
+            name = "expressCredentialsProvider"
+            symbol = RuntimeTypes.Auth.Credentials.AwsCredentials.CredentialsProvider
+            documentation = """
+                Credentials provider to be used for making requests to S3 Express.   
+            """.trimIndent()
+
+            propertyType = ConfigPropertyType.Custom(
+                render = { _, writer ->
+                    writer.write(
+                        "public val #1L: #2T = builder.#1L ?: #3T()",
+                        name,
+                        symbol,
+                        buildSymbol {
+                            name = "DefaultS3ExpressCredentialsProvider"
+                            namespace = "aws.sdk.kotlin.services.s3.express"
+                        },
+                    )
+                },
+                renderBuilder = { prop, writer ->
+                    prop.documentation?.let(writer::dokka)
+                    writer.write("public var #L: #T? = null", name, symbol)
+                },
+            )
+        }
+    }
+
+    override fun enabledForService(model: Model, settings: KotlinSettings) =
+        model.expectShape<ServiceShape>(settings.service).isS3
+
+    /**
+     * Add a synthetic SigV4 S3 Express auth trait and shape
+     */
+    override fun preprocessModel(model: Model, settings: KotlinSettings): Model {
+        val transformer = ModelTransformer.create()
+
+        // AuthIndex.getAuthSchemes looks for shapes with an AuthDefinitionTrait, so need to make one for SigV4 S3Express
+        val authDefinitionTrait = AuthDefinitionTrait.builder().addTrait(SigV4S3ExpressAuthTrait.ID).build()
+        val sigV4S3ExpressAuthShape = StructureShape.builder()
+            .addTrait(authDefinitionTrait)
+            .id(SigV4S3ExpressAuthTrait.ID)
+            .build()
+
+        val serviceShape = settings.getService(model)
+        val serviceShapeBuilder = serviceShape.toBuilder()
+
+        serviceShapeBuilder.addTrait(SigV4S3ExpressAuthTrait())
+
+        val authTrait = AuthTrait(serviceShape.expectTrait(AuthTrait::class.java).valueSet + mutableSetOf(SigV4S3ExpressAuthTrait.ID))
+        serviceShapeBuilder.addTrait(authTrait)
+
+        // Add the new shape and update the service shape's AuthTrait
+        return transformer.replaceShapes(model, listOf(sigV4S3ExpressAuthShape, serviceShapeBuilder.build()))
+    }
+
+    override fun customizeMiddleware(ctx: ProtocolGenerator.GenerationContext, resolved: List<ProtocolMiddleware>) =
+        resolved + listOf(
+            AddClientToExecutionContext,
+            AddBucketToExecutionContext,
+            UseCrc32Checksum,
+            UploadPartDisableChecksum,
+        )
+
+    private val S3AttributesSymbol = buildSymbol {
+        name = "S3Attributes"
+        namespace = "aws.sdk.kotlin.services.s3"
+    }
+
+    private val AddClientToExecutionContext = object : ProtocolMiddleware {
+        override val name: String = "AddClientToExecutionContext"
+
+        override fun isEnabledFor(ctx: ProtocolGenerator.GenerationContext, op: OperationShape): Boolean =
+            ctx.model.expectShape<ServiceShape>(ctx.settings.service).isS3
+
+        override fun render(ctx: ProtocolGenerator.GenerationContext, op: OperationShape, writer: KotlinWriter) {
+            writer.write("op.context[#T.ExpressClient] = this", S3AttributesSymbol)
+        }
+    }
+
+    private val AddBucketToExecutionContext = object : ProtocolMiddleware {
+        override val name: String = "AddBucketToExecutionContext"
+
+        override fun isEnabledFor(ctx: ProtocolGenerator.GenerationContext, op: OperationShape): Boolean =
+            ctx.model.expectShape<StructureShape>(op.input.get())
+                .members()
+                .any { it.memberName == "Bucket" }
+
+        override fun render(ctx: ProtocolGenerator.GenerationContext, op: OperationShape, writer: KotlinWriter) {
+            writer.write("input.bucket?.let { op.context[#T.Bucket] = it }", S3AttributesSymbol)
+        }
+    }
+
+    /**
+     * For any operations that require a checksum, set CRC32 if the user has not already configured a checksum.
+     */
+    private val UseCrc32Checksum = object : ProtocolMiddleware {
+        override val name: String = "UseCrc32Checksum"
+
+        override val order: Byte = -1 // Render before flexible checksums
+
+        override fun isEnabledFor(ctx: ProtocolGenerator.GenerationContext, op: OperationShape): Boolean = !op.isS3UploadPart &&
+            (op.hasTrait<HttpChecksumRequiredTrait>() || (op.hasTrait<HttpChecksumTrait>() && op.expectTrait<HttpChecksumTrait>().isRequestChecksumRequired))
+
+        override fun render(ctx: ProtocolGenerator.GenerationContext, op: OperationShape, writer: KotlinWriter) {
+            val interceptorSymbol = buildSymbol {
+                namespace = "aws.sdk.kotlin.services.s3.express"
+                name = "S3ExpressCrc32ChecksumInterceptor"
+            }
+
+            val httpChecksumTrait = op.getTrait<HttpChecksumTrait>()
+
+            val checksumAlgorithmMember = ctx.model.expectShape<StructureShape>(op.input.get())
+                .members()
+                .firstOrNull { it.memberName == httpChecksumTrait?.requestAlgorithmMember?.getOrNull() }
+
+            // S3 models a header name x-amz-sdk-checksum-algorithm representing the name of the checksum algorithm used
+            val checksumHeaderName = checksumAlgorithmMember?.getTrait<HttpHeaderTrait>()?.value
+
+            writer.write("op.interceptors.add(#T(${checksumHeaderName?.dq() ?: ""}))", interceptorSymbol)
+        }
+    }
+
+    /**
+     * Disable all checksums for s3:UploadPart
+     */
+    private val UploadPartDisableChecksum = object : ProtocolMiddleware {
+        override val name: String = "UploadPartDisableChecksum"
+
+        override fun isEnabledFor(ctx: ProtocolGenerator.GenerationContext, op: OperationShape): Boolean =
+            op.isS3UploadPart
+
+        override fun render(ctx: ProtocolGenerator.GenerationContext, op: OperationShape, writer: KotlinWriter) {
+            val interceptorSymbol = buildSymbol {
+                namespace = "aws.sdk.kotlin.services.s3.express"
+                name = "S3ExpressDisableChecksumInterceptor"
+            }
+            writer.addImport(interceptorSymbol)
+            writer.write("op.interceptors.add(#T())", interceptorSymbol)
+        }
+    }
+
+    private val OperationShape.isS3UploadPart: Boolean get() = id.name == "UploadPart"
+
+    override fun additionalServiceConfigProps(ctx: CodegenContext): List<ConfigProperty> = listOf(
+        DisableExpressSessionAuth,
+        ExpressCredentialsProvider,
+    )
+}
diff --git a/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/s3/express/SigV4S3ExpressAuthSchemeIntegration.kt b/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/s3/express/SigV4S3ExpressAuthSchemeIntegration.kt
new file mode 100644
index 00000000000..17d249bc063
--- /dev/null
+++ b/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/s3/express/SigV4S3ExpressAuthSchemeIntegration.kt
@@ -0,0 +1,142 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.codegen.customization.s3.express
+
+import aws.sdk.kotlin.codegen.customization.s3.isS3
+import software.amazon.smithy.aws.traits.auth.UnsignedPayloadTrait
+import software.amazon.smithy.codegen.core.Symbol
+import software.amazon.smithy.codegen.core.SymbolReference
+import software.amazon.smithy.kotlin.codegen.KotlinSettings
+import software.amazon.smithy.kotlin.codegen.core.*
+import software.amazon.smithy.kotlin.codegen.integration.AppendingSectionWriter
+import software.amazon.smithy.kotlin.codegen.integration.AuthSchemeHandler
+import software.amazon.smithy.kotlin.codegen.integration.KotlinIntegration
+import software.amazon.smithy.kotlin.codegen.integration.SectionWriterBinding
+import software.amazon.smithy.kotlin.codegen.model.buildSymbol
+import software.amazon.smithy.kotlin.codegen.model.expectShape
+import software.amazon.smithy.kotlin.codegen.model.hasTrait
+import software.amazon.smithy.kotlin.codegen.model.knowledge.AwsSignatureVersion4
+import software.amazon.smithy.kotlin.codegen.rendering.endpoints.EndpointCustomization
+import software.amazon.smithy.kotlin.codegen.rendering.endpoints.EndpointPropertyRenderer
+import software.amazon.smithy.kotlin.codegen.rendering.endpoints.ExpressionRenderer
+import software.amazon.smithy.kotlin.codegen.rendering.protocol.*
+import software.amazon.smithy.kotlin.codegen.utils.getOrNull
+import software.amazon.smithy.model.Model
+import software.amazon.smithy.model.node.Node
+import software.amazon.smithy.model.shapes.OperationShape
+import software.amazon.smithy.model.shapes.ServiceShape
+import software.amazon.smithy.model.shapes.ShapeId
+import software.amazon.smithy.rulesengine.language.syntax.expressions.Expression
+import java.util.*
+
+/**
+ * Register support for the `sigv4-s3express` auth scheme.
+ */
+class SigV4S3ExpressAuthSchemeIntegration : KotlinIntegration {
+    // Needs to run after `SigV4AuthSchemeIntegration`
+    override val order: Byte = -51
+
+    override fun enabledForService(model: Model, settings: KotlinSettings): Boolean = model.expectShape<ServiceShape>(settings.service).isS3
+
+    override fun authSchemes(ctx: ProtocolGenerator.GenerationContext): List<AuthSchemeHandler> = listOf(SigV4S3ExpressAuthSchemeHandler())
+
+    override fun customizeEndpointResolution(ctx: ProtocolGenerator.GenerationContext): EndpointCustomization = SigV4S3ExpressEndpointCustomization
+
+    override val sectionWriters: List<SectionWriterBinding>
+        get() = listOf(SectionWriterBinding(HttpProtocolClientGenerator.ClientInitializer, renderClientInitializer))
+
+    // add S3 Express credentials provider to managed resources in the service client initializer
+    private val renderClientInitializer = AppendingSectionWriter { writer ->
+        writer.write("managedResources.#T(config.expressCredentialsProvider)", RuntimeTypes.Core.IO.addIfManaged)
+    }
+}
+
+internal val sigV4S3ExpressSymbol = buildSymbol {
+    name = "sigV4S3Express"
+    namespace = "aws.sdk.kotlin.services.s3.express"
+}
+
+internal val SigV4S3ExpressAuthSchemeSymbol = buildSymbol {
+    name = "SigV4S3ExpressAuthScheme"
+    namespace = "aws.sdk.kotlin.services.s3.express"
+}
+
+private object SigV4S3ExpressEndpointCustomization : EndpointCustomization {
+    override val propertyRenderers: Map<String, EndpointPropertyRenderer> = mapOf(
+        "authSchemes" to ::renderAuthScheme,
+    )
+}
+
+class SigV4S3ExpressAuthSchemeHandler : AuthSchemeHandler {
+    override val authSchemeId: ShapeId = ShapeId.from("aws.auth#sigv4s3express")
+
+    override val authSchemeIdSymbol: Symbol = buildSymbol {
+        name = "AuthSchemeId(\"aws.auth#sigv4s3express\")"
+        val ref = RuntimeTypes.Auth.Identity.AuthSchemeId
+        objectRef = ref
+        namespace = ref.namespace
+        reference(ref, SymbolReference.ContextOption.USE)
+    }
+
+    override fun identityProviderAdapterExpression(writer: KotlinWriter) {
+        writer.write("config.#L", S3ExpressIntegration.ExpressCredentialsProvider.propertyName)
+    }
+
+    override fun authSchemeProviderInstantiateAuthOptionExpr(
+        ctx: ProtocolGenerator.GenerationContext,
+        op: OperationShape?,
+        writer: KotlinWriter,
+    ) {
+        val expr = if (op?.hasTrait<UnsignedPayloadTrait>() == true) {
+            "#T(unsignedPayload = true)"
+        } else {
+            "#T()"
+        }
+        writer.write(expr, sigV4S3ExpressSymbol)
+    }
+
+    override fun instantiateAuthSchemeExpr(ctx: ProtocolGenerator.GenerationContext, writer: KotlinWriter) {
+        val signingService = AwsSignatureVersion4.signingServiceName(ctx.service)
+        writer.write("#T(#T, #S)", SigV4S3ExpressAuthSchemeSymbol, RuntimeTypes.Auth.Signing.AwsSigningStandard.DefaultAwsSigner, signingService)
+    }
+}
+
+private fun renderAuthScheme(writer: KotlinWriter, authSchemes: Expression, expressionRenderer: ExpressionRenderer) {
+    val expressScheme = authSchemes.toNode().expectArrayNode().find {
+        it.expectObjectNode().expectStringMember("name").value == "sigv4-s3express"
+    }?.expectObjectNode()
+
+    expressScheme?.let {
+        writer.writeInline("#T to ", RuntimeTypes.SmithyClient.Endpoints.SigningContextAttributeKey)
+        writer.withBlock("listOf(", ")") {
+            withBlock("#T(", "),", sigV4S3ExpressSymbol) {
+                // we delegate back to the expression visitor for each of these fields because it's possible to
+                // encounter template strings throughout
+
+                writeInline("serviceName = ")
+                renderOrElse(expressionRenderer, expressScheme.getStringMember("signingName"), "null")
+
+                writeInline("disableDoubleUriEncode = ")
+                renderOrElse(expressionRenderer, expressScheme.getBooleanMember("disableDoubleEncoding"), "false")
+
+                writeInline("signingRegion = ")
+                renderOrElse(expressionRenderer, expressScheme.getStringMember("signingRegion"), "null")
+            }
+        }
+    }
+}
+
+private fun KotlinWriter.renderOrElse(
+    expressionRenderer: ExpressionRenderer,
+    optionalNode: Optional<out Node>,
+    whenNullValue: String,
+) {
+    val nullableNode = optionalNode.getOrNull()
+    when (nullableNode) {
+        null -> writeInline(whenNullValue)
+        else -> expressionRenderer.renderExpression(Expression.fromNode(nullableNode))
+    }
+    write(",")
+}
diff --git a/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/s3/express/SigV4S3ExpressAuthTrait.kt b/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/s3/express/SigV4S3ExpressAuthTrait.kt
new file mode 100644
index 00000000000..f6f9ba29faf
--- /dev/null
+++ b/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/customization/s3/express/SigV4S3ExpressAuthTrait.kt
@@ -0,0 +1,18 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+import software.amazon.smithy.model.node.Node
+import software.amazon.smithy.model.shapes.ShapeId
+import software.amazon.smithy.model.traits.AbstractTrait
+
+/**
+ * Synthetic auth trait applied to S3's model to enable SigV4 S3 Express auth scheme.
+ */
+internal class SigV4S3ExpressAuthTrait : AbstractTrait(ID, Node.objectNode()) {
+    companion object {
+        val ID = ShapeId.from("aws.auth#sigv4s3express")
+    }
+    override fun createNode(): Node = Node.objectNode()
+    override fun isSynthetic(): Boolean = true
+}
diff --git a/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/endpoints/AwsBuiltins.kt b/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/endpoints/AwsBuiltins.kt
index 6d90b7b8f14..c6e1924ea47 100644
--- a/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/endpoints/AwsBuiltins.kt
+++ b/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/endpoints/AwsBuiltins.kt
@@ -11,6 +11,7 @@ object AwsBuiltins {
     const val S3_ACCELERATE = "AWS::S3::Accelerate"
     const val S3_FORCE_PATH_STYLE = "AWS::S3::ForcePathStyle"
     const val S3_DISABLE_MRAP = "AWS::S3::DisableMultiRegionAccessPoints"
+    const val S3_DISABLE_EXPRESS_SESSION_AUTH = "AWS::S3::DisableS3ExpressSessionAuth"
     const val S3_USE_ARN_REGION = "AWS::S3::UseArnRegion"
     const val S3_CONTROL_USE_ARN_REGION = "AWS::S3Control::UseArnRegion"
     const val S3_USE_GLOBAL_ENDPOINT = "AWS::S3::UseGlobalEndpoint"
diff --git a/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/endpoints/BindAwsEndpointBuiltins.kt b/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/endpoints/BindAwsEndpointBuiltins.kt
index 6643d3f8e77..5c418be243e 100644
--- a/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/endpoints/BindAwsEndpointBuiltins.kt
+++ b/codegen/aws-sdk-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/endpoints/BindAwsEndpointBuiltins.kt
@@ -6,6 +6,7 @@ package aws.sdk.kotlin.codegen.endpoints
 
 import aws.sdk.kotlin.codegen.AwsRuntimeTypes
 import aws.sdk.kotlin.codegen.customization.AccountIdEndpointBuiltinCustomization
+import aws.sdk.kotlin.codegen.customization.s3.express.S3ExpressIntegration
 import software.amazon.smithy.codegen.core.Symbol
 import software.amazon.smithy.kotlin.codegen.core.CodegenContext
 import software.amazon.smithy.kotlin.codegen.core.KotlinWriter
@@ -90,6 +91,7 @@ fun renderBindAwsBuiltins(ctx: ProtocolGenerator.GenerationContext, writer: Kotl
                 AwsBuiltins.S3_ACCELERATE -> renderBasicConfigBinding(writer, it, S3ClientConfigIntegration.EnableAccelerateProp.propertyName)
                 AwsBuiltins.S3_FORCE_PATH_STYLE -> renderBasicConfigBinding(writer, it, S3ClientConfigIntegration.ForcePathStyleProp.propertyName)
                 AwsBuiltins.S3_DISABLE_MRAP -> renderBasicConfigBinding(writer, it, S3ClientConfigIntegration.DisableMrapProp.propertyName)
+                AwsBuiltins.S3_DISABLE_EXPRESS_SESSION_AUTH -> renderBasicConfigBinding(writer, it, S3ExpressIntegration.DisableExpressSessionAuth.propertyName)
                 AwsBuiltins.S3_USE_ARN_REGION -> renderBasicConfigBinding(writer, it, S3ClientConfigIntegration.UseArnRegionProp.propertyName)
                 AwsBuiltins.S3_CONTROL_USE_ARN_REGION -> renderBasicConfigBinding(writer, it, S3ControlClientConfigIntegration.UseArnRegionProp.propertyName)
 
diff --git a/codegen/aws-sdk-codegen/src/main/resources/META-INF/services/software.amazon.smithy.kotlin.codegen.integration.KotlinIntegration b/codegen/aws-sdk-codegen/src/main/resources/META-INF/services/software.amazon.smithy.kotlin.codegen.integration.KotlinIntegration
index 1eeb8f5a1d3..cbcf623fe0b 100644
--- a/codegen/aws-sdk-codegen/src/main/resources/META-INF/services/software.amazon.smithy.kotlin.codegen.integration.KotlinIntegration
+++ b/codegen/aws-sdk-codegen/src/main/resources/META-INF/services/software.amazon.smithy.kotlin.codegen.integration.KotlinIntegration
@@ -39,3 +39,5 @@ aws.sdk.kotlin.codegen.customization.RemoveDefaults
 aws.sdk.kotlin.codegen.customization.s3.UnsupportedSigningAlgorithmIntegration
 aws.sdk.kotlin.codegen.customization.SigV4AsymmetricTraitCustomization
 aws.sdk.kotlin.codegen.customization.cloudfrontkeyvaluestore.BackfillSigV4ACustomization
+aws.sdk.kotlin.codegen.customization.s3.express.SigV4S3ExpressAuthSchemeIntegration
+aws.sdk.kotlin.codegen.customization.s3.express.S3ExpressIntegration
diff --git a/services/s3/common/src/aws/sdk/kotlin/services/s3/S3Attributes.kt b/services/s3/common/src/aws/sdk/kotlin/services/s3/S3Attributes.kt
new file mode 100644
index 00000000000..13ce1311198
--- /dev/null
+++ b/services/s3/common/src/aws/sdk/kotlin/services/s3/S3Attributes.kt
@@ -0,0 +1,26 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.services.s3
+
+import aws.sdk.kotlin.runtime.InternalSdkApi
+import aws.smithy.kotlin.runtime.client.SdkClient
+import aws.smithy.kotlin.runtime.collections.AttributeKey
+
+/**
+ * Execution context attributes specific to S3
+ */
+@InternalSdkApi
+public object S3Attributes {
+    /**
+     * The name of the bucket requests are being made to
+     */
+    public val Bucket: AttributeKey<String> = AttributeKey("aws.sdk.kotlin#S3Bucket")
+
+    /**
+     * The S3 client being used to make requests to directory buckets. This client will be used to call s3:CreateSession
+     * to obtain directory bucket credentials.
+     */
+    public val ExpressClient: AttributeKey<SdkClient> = AttributeKey("aws.sdk.kotlin#S3ExpressClient")
+}
diff --git a/services/s3/common/src/aws/sdk/kotlin/services/s3/express/DefaultS3ExpressCredentialsProvider.kt b/services/s3/common/src/aws/sdk/kotlin/services/s3/express/DefaultS3ExpressCredentialsProvider.kt
new file mode 100644
index 00000000000..d8936f28a3b
--- /dev/null
+++ b/services/s3/common/src/aws/sdk/kotlin/services/s3/express/DefaultS3ExpressCredentialsProvider.kt
@@ -0,0 +1,89 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.services.s3.express
+
+import aws.sdk.kotlin.services.s3.*
+import aws.sdk.kotlin.services.s3.S3Attributes
+import aws.sdk.kotlin.services.s3.S3Client
+import aws.smithy.kotlin.runtime.ExperimentalApi
+import aws.smithy.kotlin.runtime.auth.awscredentials.CloseableCredentialsProvider
+import aws.smithy.kotlin.runtime.auth.awscredentials.Credentials
+import aws.smithy.kotlin.runtime.collections.Attributes
+import aws.smithy.kotlin.runtime.collections.get
+import aws.smithy.kotlin.runtime.io.SdkManagedBase
+import aws.smithy.kotlin.runtime.telemetry.logging.getLogger
+import aws.smithy.kotlin.runtime.time.Clock
+import aws.smithy.kotlin.runtime.time.until
+import kotlinx.coroutines.*
+import kotlin.coroutines.CoroutineContext
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.minutes
+import kotlin.time.TimeSource
+
+/**
+ * The default implementation of a credentials provider for S3 Express One Zone. Performs best-effort asynchronous refresh
+ * if the cached credentials are expiring within a [refreshBuffer] during a call to [resolve].
+ * Otherwise, performs synchronous refresh.
+ *
+ * @param timeSource the time source to use. defaults to [TimeSource.Monotonic]
+ * @param clock the clock to use. defaults to [Clock.System]. note: the clock is only used to get an initial [Duration]
+ * until credentials expiration, [timeSource] is used as the source of truth for credentials expiration.
+ * @param credentialsCache an [S3ExpressCredentialsCache] to be used for caching session credentials, defaults to
+ * [S3ExpressCredentialsCache].
+ * @param refreshBuffer an optional [Duration] representing the duration before expiration that [Credentials]
+ * are considered refreshable, defaults to 1 minute.
+ */
+internal class DefaultS3ExpressCredentialsProvider(
+    private val timeSource: TimeSource.WithComparableMarks = TimeSource.Monotonic,
+    private val clock: Clock = Clock.System,
+    private val credentialsCache: S3ExpressCredentialsCache = S3ExpressCredentialsCache(),
+    private val refreshBuffer: Duration = 1.minutes,
+) : CloseableCredentialsProvider, SdkManagedBase(), CoroutineScope {
+    override val coroutineContext: CoroutineContext = Job() + CoroutineName("DefaultS3ExpressCredentialsProvider")
+
+    override suspend fun resolve(attributes: Attributes): Credentials {
+        val client = attributes[S3Attributes.ExpressClient] as S3Client
+
+        val key = S3ExpressCredentialsCacheKey(attributes[S3Attributes.Bucket], client.config.credentialsProvider.resolve(attributes))
+
+        return credentialsCache.get(key)
+            ?.takeIf { !it.expiringCredentials.isExpired }
+            ?.also {
+                if (it.expiringCredentials.isExpiringWithin(refreshBuffer)) {
+                    client.logger.trace { "Credentials for ${key.bucket} are expiring in ${it.expiringCredentials.expiresAt} and are within their refresh window, performing asynchronous refresh..." }
+                    launch(coroutineContext) {
+                        try {
+                            it.sfg.singleFlight { createSessionCredentials(key, client) }
+                        } catch (e: Exception) {
+                            client.logger.warn(e) { "Asynchronous refresh for ${key.bucket} failed." }
+                        }
+                    }
+                }
+            }
+            ?.expiringCredentials
+            ?.value
+            ?: createSessionCredentials(key, client).value
+    }
+
+    override fun close() {
+        coroutineContext.cancel(null)
+    }
+
+    /**
+     * Create a new set of session credentials by calling s3:CreateSession and then store them in the cache.
+     */
+    internal suspend fun createSessionCredentials(key: S3ExpressCredentialsCacheKey, client: S3Client): ExpiringValue<Credentials> =
+        client.createSession { bucket = key.bucket }.credentials!!.let {
+            ExpiringValue(
+                Credentials(it.accessKeyId, it.secretAccessKey, it.sessionToken, it.expiration),
+                expiresAt = timeSource.markNow() + clock.now().until(it.expiration),
+            )
+        }.also {
+            credentialsCache.put(key, S3ExpressCredentialsCacheValue(it))
+        }
+
+    @OptIn(ExperimentalApi::class)
+    internal val S3Client.logger get() = config.telemetryProvider.loggerProvider.getLogger<DefaultS3ExpressCredentialsProvider>()
+}
diff --git a/services/s3/common/src/aws/sdk/kotlin/services/s3/express/S3ExpressCrc32ChecksumInterceptor.kt b/services/s3/common/src/aws/sdk/kotlin/services/s3/express/S3ExpressCrc32ChecksumInterceptor.kt
new file mode 100644
index 00000000000..bad493a2fbd
--- /dev/null
+++ b/services/s3/common/src/aws/sdk/kotlin/services/s3/express/S3ExpressCrc32ChecksumInterceptor.kt
@@ -0,0 +1,49 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.services.s3.express
+
+import aws.smithy.kotlin.runtime.client.ProtocolRequestInterceptorContext
+import aws.smithy.kotlin.runtime.collections.AttributeKey
+import aws.smithy.kotlin.runtime.http.interceptors.HttpInterceptor
+import aws.smithy.kotlin.runtime.http.operation.HttpOperationContext
+import aws.smithy.kotlin.runtime.http.request.HttpRequest
+import aws.smithy.kotlin.runtime.http.request.toBuilder
+import aws.smithy.kotlin.runtime.telemetry.logging.logger
+import kotlin.coroutines.coroutineContext
+
+internal const val S3_EXPRESS_ENDPOINT_PROPERTY_KEY = "backend"
+internal const val S3_EXPRESS_ENDPOINT_PROPERTY_VALUE = "S3Express"
+private const val CRC32_ALGORITHM_NAME = "CRC32"
+
+internal class S3ExpressCrc32ChecksumInterceptor(
+    val checksumAlgorithmHeaderName: String? = null,
+) : HttpInterceptor {
+    override suspend fun modifyBeforeSigning(context: ProtocolRequestInterceptorContext<Any, HttpRequest>): HttpRequest {
+        if (context.executionContext.getOrNull(AttributeKey(S3_EXPRESS_ENDPOINT_PROPERTY_KEY)) != S3_EXPRESS_ENDPOINT_PROPERTY_VALUE) {
+            return context.protocolRequest
+        }
+
+        val logger = coroutineContext.logger<S3ExpressCrc32ChecksumInterceptor>()
+        val req = context.protocolRequest.toBuilder()
+
+        if (!context.executionContext.contains(HttpOperationContext.ChecksumAlgorithm)) {
+            logger.debug { "Checksum is required and not already configured, enabling CRC32 for S3 Express" }
+
+            // Update the execution context so flexible checksums uses CRC32
+            context.executionContext[HttpOperationContext.ChecksumAlgorithm] = CRC32_ALGORITHM_NAME
+
+            // Most checksum headers are handled by the flexible checksums feature. But, S3 models an HTTP header binding for the
+            // checksum algorithm, which also needs to be overwritten and set to CRC32.
+            //
+            // The header is already set by the time this interceptor runs, so it needs to be overwritten and can't be set
+            // through the normal path.
+            checksumAlgorithmHeaderName?.let {
+                req.headers[it] = CRC32_ALGORITHM_NAME
+            }
+        }
+
+        return req.build()
+    }
+}
diff --git a/services/s3/common/src/aws/sdk/kotlin/services/s3/express/S3ExpressCredentialsCache.kt b/services/s3/common/src/aws/sdk/kotlin/services/s3/express/S3ExpressCredentialsCache.kt
new file mode 100644
index 00000000000..054322c57a3
--- /dev/null
+++ b/services/s3/common/src/aws/sdk/kotlin/services/s3/express/S3ExpressCredentialsCache.kt
@@ -0,0 +1,50 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.services.s3.express
+
+import aws.smithy.kotlin.runtime.auth.awscredentials.Credentials
+import aws.smithy.kotlin.runtime.collections.LruCache
+import aws.smithy.kotlin.runtime.util.SingleFlightGroup
+import kotlin.time.ComparableTimeMark
+import kotlin.time.Duration
+import kotlin.time.TimeMark
+
+private const val DEFAULT_S3_EXPRESS_CACHE_SIZE: Int = 100
+
+internal typealias S3ExpressCredentialsCache = LruCache<S3ExpressCredentialsCacheKey, S3ExpressCredentialsCacheValue>
+internal fun S3ExpressCredentialsCache() = S3ExpressCredentialsCache(DEFAULT_S3_EXPRESS_CACHE_SIZE)
+
+internal data class S3ExpressCredentialsCacheKey(
+    /**
+     * The directory bucket requests are being made to
+     */
+    val bucket: String,
+    /**
+     * The base credentials used to resolve session credentials
+     */
+    val baseCredentials: Credentials,
+)
+
+internal data class S3ExpressCredentialsCacheValue(
+    /**
+     * The expiring session [Credentials]
+     */
+    val expiringCredentials: ExpiringValue<Credentials>,
+    /**
+     * A [SingleFlightGroup] used to de-duplicate asynchronous refresh attempts
+     */
+    val sfg: SingleFlightGroup<ExpiringValue<Credentials>> = SingleFlightGroup(),
+)
+
+/**
+ * A value with an expiration [TimeMark]
+ */
+internal data class ExpiringValue<T>(val value: T, val expiresAt: ComparableTimeMark)
+
+internal val ExpiringValue<Credentials>.isExpired: Boolean get() = expiresAt.hasPassedNow()
+
+internal fun ExpiringValue<Credentials>.isExpiringWithin(duration: Duration) = (expiresAt - duration).hasPassedNow()
+
+internal typealias S3ExpressCredentialsCacheEntry = Map.Entry<S3ExpressCredentialsCacheKey, S3ExpressCredentialsCacheValue>
diff --git a/services/s3/common/src/aws/sdk/kotlin/services/s3/express/S3ExpressDisableChecksumInterceptor.kt b/services/s3/common/src/aws/sdk/kotlin/services/s3/express/S3ExpressDisableChecksumInterceptor.kt
new file mode 100644
index 00000000000..3b10ad3fa69
--- /dev/null
+++ b/services/s3/common/src/aws/sdk/kotlin/services/s3/express/S3ExpressDisableChecksumInterceptor.kt
@@ -0,0 +1,35 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.services.s3.express
+
+import aws.smithy.kotlin.runtime.client.ProtocolRequestInterceptorContext
+import aws.smithy.kotlin.runtime.collections.AttributeKey
+import aws.smithy.kotlin.runtime.http.interceptors.HttpInterceptor
+import aws.smithy.kotlin.runtime.http.operation.HttpOperationContext
+import aws.smithy.kotlin.runtime.http.request.HttpRequest
+import aws.smithy.kotlin.runtime.telemetry.logging.logger
+import kotlin.coroutines.coroutineContext
+
+/**
+ * Disable checksums entirely for s3:UploadPart requests.
+ */
+internal class S3ExpressDisableChecksumInterceptor : HttpInterceptor {
+    override suspend fun modifyBeforeSigning(context: ProtocolRequestInterceptorContext<Any, HttpRequest>): HttpRequest {
+        if (context.executionContext.getOrNull(AttributeKey(S3_EXPRESS_ENDPOINT_PROPERTY_KEY)) != S3_EXPRESS_ENDPOINT_PROPERTY_VALUE) {
+            return context.protocolRequest
+        }
+
+        val logger = coroutineContext.logger<S3ExpressDisableChecksumInterceptor>()
+
+        val configuredChecksumAlgorithm = context.executionContext.getOrNull(HttpOperationContext.ChecksumAlgorithm)
+
+        configuredChecksumAlgorithm?.let {
+            logger.warn { "Disabling configured checksum $it for S3 Express UploadPart" }
+            context.executionContext.remove(HttpOperationContext.ChecksumAlgorithm)
+        }
+
+        return context.protocolRequest
+    }
+}
diff --git a/services/s3/common/src/aws/sdk/kotlin/services/s3/express/S3ExpressHttpSigner.kt b/services/s3/common/src/aws/sdk/kotlin/services/s3/express/S3ExpressHttpSigner.kt
new file mode 100644
index 00000000000..288ed5e6f69
--- /dev/null
+++ b/services/s3/common/src/aws/sdk/kotlin/services/s3/express/S3ExpressHttpSigner.kt
@@ -0,0 +1,46 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.services.s3.express
+
+import aws.smithy.kotlin.runtime.auth.awscredentials.Credentials
+import aws.smithy.kotlin.runtime.auth.awssigning.AwsSigningAttributes
+import aws.smithy.kotlin.runtime.collections.toMutableAttributes
+import aws.smithy.kotlin.runtime.http.auth.HttpSigner
+import aws.smithy.kotlin.runtime.http.auth.SignHttpRequest
+import aws.smithy.kotlin.runtime.http.request.header
+
+internal const val S3_EXPRESS_SESSION_TOKEN_HEADER = "X-Amz-S3session-Token"
+private const val SESSION_TOKEN_HEADER = "X-Amz-Security-Token"
+
+/**
+ * An [HttpSigner] used for S3 Express requests. It has identical behavior with the given [httpSigner] except for two differences:
+ *    1. Adds an `X-Amz-S3Session-Token` header, with a value of the credentials' sessionToken
+ *    2. Removes the `X-Amz-Security-Token` header, which must not be sent for S3 Express requests.
+ * @param httpSigner An instance of [HttpSigner]
+ */
+internal class S3ExpressHttpSigner(
+    private val httpSigner: HttpSigner,
+) : HttpSigner {
+    /**
+     * Sign the request, adding `X-Amz-S3Session-Token` header and removing `X-Amz-Security-Token` header.
+     */
+    override suspend fun sign(signingRequest: SignHttpRequest) {
+        val sessionToken = (signingRequest.identity as? Credentials)?.sessionToken
+            ?: error("No session token found on identity, required for S3 Express")
+
+        // 1. add the S3 Express Session Token header
+        signingRequest.httpRequest.header(S3_EXPRESS_SESSION_TOKEN_HEADER, sessionToken)
+
+        // 2. enable omitSessionToken for awsHttpSigner to disable signing session token header
+        val mutAttrs = signingRequest.signingAttributes.toMutableAttributes()
+        mutAttrs[AwsSigningAttributes.OmitSessionToken] = true
+
+        // 3. call main signer
+        httpSigner.sign(signingRequest.copy(signingAttributes = mutAttrs))
+
+        // 4. remove session token header
+        signingRequest.httpRequest.headers.remove(SESSION_TOKEN_HEADER)
+    }
+}
diff --git a/services/s3/common/src/aws/sdk/kotlin/services/s3/express/SigV4S3ExpressAuthScheme.kt b/services/s3/common/src/aws/sdk/kotlin/services/s3/express/SigV4S3ExpressAuthScheme.kt
new file mode 100644
index 00000000000..c14502ab2a3
--- /dev/null
+++ b/services/s3/common/src/aws/sdk/kotlin/services/s3/express/SigV4S3ExpressAuthScheme.kt
@@ -0,0 +1,60 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.services.s3.express
+
+import aws.smithy.kotlin.runtime.InternalApi
+import aws.smithy.kotlin.runtime.auth.AuthOption
+import aws.smithy.kotlin.runtime.auth.AuthSchemeId
+import aws.smithy.kotlin.runtime.auth.awssigning.AwsSigner
+import aws.smithy.kotlin.runtime.auth.awssigning.AwsSigningAlgorithm
+import aws.smithy.kotlin.runtime.http.auth.AuthScheme
+import aws.smithy.kotlin.runtime.http.auth.AwsHttpSigner
+import aws.smithy.kotlin.runtime.http.auth.HttpSigner
+import aws.smithy.kotlin.runtime.http.auth.sigV4
+
+public val AuthSchemeId.Companion.AwsSigV4S3Express: AuthSchemeId
+    get() = AuthSchemeId("aws.auth#sigv4s3express")
+
+/**
+ * HTTP auth scheme for S3 Express One Zone authentication
+ */
+public class SigV4S3ExpressAuthScheme(
+    httpSigner: HttpSigner,
+) : AuthScheme {
+    public constructor(awsSigner: AwsSigner, serviceName: String? = null) : this(
+        AwsHttpSigner(
+            AwsHttpSigner.Config().apply {
+                signer = awsSigner
+                service = serviceName
+                algorithm = AwsSigningAlgorithm.SIGV4 // Note: There is no new signing algorithm for S3 Express
+            },
+        ),
+    )
+
+    override val schemeId: AuthSchemeId = AuthSchemeId.AwsSigV4S3Express
+    override val signer: HttpSigner = S3ExpressHttpSigner(httpSigner)
+}
+
+/**
+ * Create a new [AuthOption] for the [SigV4S3ExpressAuthScheme]
+ * @param unsignedPayload set the signing attribute to indicate the signer should use unsigned payload.
+ * @param serviceName override the service name to sign for
+ * @param signingRegion override the signing region to sign for
+ * @param disableDoubleUriEncode disable double URI encoding
+ * @param normalizeUriPath flag indicating if the URI path should be normalized when forming the canonical request
+ * @return auth scheme option representing the [SigV4S3ExpressAuthScheme]
+ */
+@InternalApi
+public fun sigV4S3Express(
+    unsignedPayload: Boolean = false,
+    serviceName: String? = null,
+    signingRegion: String? = null,
+    disableDoubleUriEncode: Boolean? = null,
+    normalizeUriPath: Boolean? = null,
+): AuthOption {
+    // Note: SigV4-S3Express has the same attributes as SigV4
+    val sigV4AuthOption = sigV4(unsignedPayload, serviceName, signingRegion, disableDoubleUriEncode, normalizeUriPath)
+    return AuthOption(AuthSchemeId.AwsSigV4S3Express, sigV4AuthOption.attributes)
+}
diff --git a/services/s3/common/src/aws/sdk/kotlin/services/s3/internal/FinalizeS3Config.kt b/services/s3/common/src/aws/sdk/kotlin/services/s3/internal/FinalizeS3Config.kt
index cf75cc2bafd..f3e2f2ae43b 100644
--- a/services/s3/common/src/aws/sdk/kotlin/services/s3/internal/FinalizeS3Config.kt
+++ b/services/s3/common/src/aws/sdk/kotlin/services/s3/internal/FinalizeS3Config.kt
@@ -20,6 +20,7 @@ internal suspend fun finalizeS3Config(
     val activeProfile = sharedConfig.get().activeProfile
     builder.config.useArnRegion = builder.config.useArnRegion ?: S3Setting.UseArnRegion.resolve(provider) ?: activeProfile.useArnRegion
     builder.config.disableMrap = builder.config.disableMrap ?: S3Setting.DisableMultiRegionAccessPoints.resolve(provider) ?: activeProfile.disableMrap
+    builder.config.disableS3ExpressSessionAuth = builder.config.disableS3ExpressSessionAuth ?: S3Setting.DisableS3ExpressSessionAuth.resolve(provider) ?: activeProfile.disableS3ExpressSessionAuth
 }
 
 private val AwsProfile.useArnRegion: Boolean?
@@ -27,3 +28,6 @@ private val AwsProfile.useArnRegion: Boolean?
 
 private val AwsProfile.disableMrap: Boolean?
     get() = getBooleanOrNull("s3_disable_multiregion_access_points")
+
+private val AwsProfile.disableS3ExpressSessionAuth: Boolean?
+    get() = getBooleanOrNull("s3_disable_express_session_auth")
diff --git a/services/s3/common/src/aws/sdk/kotlin/services/s3/internal/S3Setting.kt b/services/s3/common/src/aws/sdk/kotlin/services/s3/internal/S3Setting.kt
index bd90cd01200..57c4d63bd79 100644
--- a/services/s3/common/src/aws/sdk/kotlin/services/s3/internal/S3Setting.kt
+++ b/services/s3/common/src/aws/sdk/kotlin/services/s3/internal/S3Setting.kt
@@ -21,4 +21,9 @@ internal object S3Setting {
      * See [Amazon S3 Multi-Region Access Points](https://docs.aws.amazon.com/sdkref/latest/guide/feature-s3-mrap.html)
      */
     public val DisableMultiRegionAccessPoints: EnvironmentSetting<Boolean> = boolEnvSetting("aws.s3DisableMultiRegionAccessPoints", "AWS_S3_DISABLE_MULTIREGION_ACCESS_POINTS")
+
+    /**
+     * Configure whether requests made to S3 Express One Zone should use bucket-level session authentication or the default S3 authentication method.
+     */
+    public val DisableS3ExpressSessionAuth: EnvironmentSetting<Boolean> = boolEnvSetting("aws.s3DisableExpressSessionAuth", "AWS_S3_DISABLE_EXPRESS_SESSION_AUTH")
 }
diff --git a/services/s3/common/test/aws/sdk/kotlin/services/s3/express/DefaultS3ExpressCredentialsProviderTest.kt b/services/s3/common/test/aws/sdk/kotlin/services/s3/express/DefaultS3ExpressCredentialsProviderTest.kt
new file mode 100644
index 00000000000..861c77ea149
--- /dev/null
+++ b/services/s3/common/test/aws/sdk/kotlin/services/s3/express/DefaultS3ExpressCredentialsProviderTest.kt
@@ -0,0 +1,279 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.services.s3.express
+
+import aws.sdk.kotlin.runtime.auth.credentials.StaticCredentialsProvider
+import aws.sdk.kotlin.services.s3.S3Attributes
+import aws.sdk.kotlin.services.s3.S3Client
+import aws.sdk.kotlin.services.s3.model.CreateSessionRequest
+import aws.sdk.kotlin.services.s3.model.CreateSessionResponse
+import aws.sdk.kotlin.services.s3.model.SessionCredentials
+import aws.smithy.kotlin.runtime.auth.awscredentials.Credentials
+import aws.smithy.kotlin.runtime.io.use
+import aws.smithy.kotlin.runtime.operation.ExecutionContext
+import aws.smithy.kotlin.runtime.time.ManualClock
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.runTest
+import kotlin.test.*
+import kotlin.time.ComparableTimeMark
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.minutes
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.TestTimeSource
+
+class DefaultS3ExpressCredentialsProviderTest {
+    private val DEFAULT_BASE_CREDENTIALS = Credentials("accessKeyId", "secretAccessKey", "sessionToken")
+
+    @Test
+    fun testCreateSessionCredentials() = runTest {
+        val timeSource = TestTimeSource()
+        val clock = ManualClock()
+
+        val expectedCredentials = SessionCredentials {
+            accessKeyId = "access"
+            secretAccessKey = "secret"
+            sessionToken = "session"
+            expiration = clock.now() + 5.minutes
+        }
+
+        val client = TestS3Client(expectedCredentials)
+
+        DefaultS3ExpressCredentialsProvider(timeSource, clock).use { provider ->
+            val credentials = provider.createSessionCredentials(
+                S3ExpressCredentialsCacheKey("bucket", DEFAULT_BASE_CREDENTIALS),
+                client,
+            )
+            assertFalse(credentials.isExpired)
+            assertEquals(timeSource.markNow() + 5.minutes, credentials.expiresAt)
+        }
+    }
+
+    @Test
+    fun testSyncRefresh() = runTest {
+        val timeSource = TestTimeSource()
+        val clock = ManualClock()
+
+        // Entry expired 30 seconds ago, next `resolve` call should trigger a sync refresh
+        val cache = S3ExpressCredentialsCache()
+        val entry = getCacheEntry(timeSource.markNow() - 30.seconds)
+        cache.put(entry.key, entry.value)
+
+        val expectedCredentials = SessionCredentials {
+            accessKeyId = "access"
+            secretAccessKey = "secret"
+            sessionToken = "session"
+            expiration = clock.now() + 5.minutes
+        }
+
+        val testClient = TestS3Client(expectedCredentials)
+        DefaultS3ExpressCredentialsProvider(timeSource, clock, cache, refreshBuffer = 1.minutes).use { provider ->
+            val attributes = ExecutionContext.build {
+                this.attributes[S3Attributes.ExpressClient] = testClient
+                this.attributes[S3Attributes.Bucket] = "bucket"
+            }
+
+            provider.resolve(attributes)
+        }
+        assertEquals(1, testClient.numCreateSession)
+    }
+
+    @Test
+    fun testAsyncRefresh() = runTest {
+        val timeSource = TestTimeSource()
+        val clock = ManualClock()
+
+        // Entry expires in 30 seconds, refresh buffer is 1 minute. Next `resolve` call should trigger the async refresh
+        val cache = S3ExpressCredentialsCache()
+        val entry = getCacheEntry(timeSource.markNow() + 30.seconds)
+        cache.put(entry.key, entry.value)
+
+        val expectedCredentials = SessionCredentials {
+            accessKeyId = "access"
+            secretAccessKey = "secret"
+            sessionToken = "session"
+            expiration = clock.now() + 5.minutes
+        }
+
+        val testClient = TestS3Client(expectedCredentials)
+
+        val provider = DefaultS3ExpressCredentialsProvider(timeSource, clock, cache, refreshBuffer = 1.minutes)
+
+        val attributes = ExecutionContext.build {
+            this.attributes[S3Attributes.ExpressClient] = testClient
+            this.attributes[S3Attributes.Bucket] = "bucket"
+        }
+        provider.resolve(attributes)
+
+        // allow the async refresh to initiate before closing the provider
+        runBlocking { delay(50.milliseconds) }
+
+        // close the provider, make sure all async refreshes are complete...
+        provider.close()
+        runBlocking { delay(50.milliseconds) }
+
+        assertEquals(1, testClient.numCreateSession)
+    }
+
+    @Test
+    fun testAsyncRefreshDebounce() = runTest {
+        val timeSource = TestTimeSource()
+        val clock = ManualClock()
+
+        // Entry expires in 30 seconds, refresh buffer is 1 minute. Next `resolve` call should trigger the async refresh
+        val cache = S3ExpressCredentialsCache()
+        val entry = getCacheEntry(expiration = timeSource.markNow() + 30.seconds)
+        cache.put(entry.key, entry.value)
+
+        val expectedCredentials = SessionCredentials {
+            accessKeyId = "access"
+            secretAccessKey = "secret"
+            sessionToken = "session"
+            expiration = clock.now() + 5.minutes
+        }
+
+        val testClient = TestS3Client(expectedCredentials)
+
+        val provider = DefaultS3ExpressCredentialsProvider(timeSource, clock, cache, refreshBuffer = 1.minutes)
+
+        val attributes = ExecutionContext.build {
+            this.attributes[S3Attributes.ExpressClient] = testClient
+            this.attributes[S3Attributes.Bucket] = "bucket"
+        }
+        val calls = (1..5).map {
+            async { provider.resolve(attributes) }
+        }
+        calls.awaitAll()
+
+        // allow the async refresh to initiate before closing the provider
+        runBlocking { delay(50.milliseconds) }
+
+        // close the provider, make sure all async refreshes are complete...
+        provider.close()
+        runBlocking { delay(50.milliseconds) }
+
+        assertEquals(1, testClient.numCreateSession)
+    }
+
+    @Test
+    fun testAsyncRefreshHandlesFailures() = runTest {
+        val timeSource = TestTimeSource()
+        val clock = ManualClock()
+
+        // Entry expires in 30 seconds, refresh buffer is 1 minute. Next `resolve` call should trigger the async refresh
+        val cache = S3ExpressCredentialsCache()
+        val successEntry = getCacheEntry(timeSource.markNow() + 30.seconds, bucket = "SuccessfulBucket", bootstrapCredentials = Credentials("1", "1", "1"))
+        val failedEntry = getCacheEntry(timeSource.markNow() + 30.seconds, bucket = "ExceptionBucket", bootstrapCredentials = Credentials("1", "1", "1"))
+        cache.put(successEntry.key, successEntry.value)
+        cache.put(failedEntry.key, failedEntry.value)
+
+        val expectedCredentials = SessionCredentials {
+            accessKeyId = "access"
+            secretAccessKey = "secret"
+            sessionToken = "session"
+            expiration = clock.now() + 5.minutes
+        }
+
+        // client will throw an exception when `ExceptionBucket` credentials are fetched,
+        // but there should be no crash
+        val testClient = TestS3Client(expectedCredentials, throwExceptionOnBucketNamed = "ExceptionBucket", baseCredentials = Credentials("1", "1", "1"))
+
+        val provider = DefaultS3ExpressCredentialsProvider(timeSource, clock, cache, refreshBuffer = 1.minutes)
+        val attributes = ExecutionContext.build {
+            this.attributes[S3Attributes.ExpressClient] = testClient
+            this.attributes[S3Attributes.Bucket] = "ExceptionBucket"
+        }
+        provider.resolve(attributes)
+
+        attributes[S3Attributes.Bucket] = "SuccessfulBucket"
+        provider.resolve(attributes)
+
+        // allow the async refresh to initiate before closing the provider
+        runBlocking { delay(50.milliseconds) }
+
+        // close the provider, make sure all async refreshes are complete...
+        provider.close()
+        runBlocking { delay(50.milliseconds) }
+
+        assertEquals(2, testClient.numCreateSession)
+    }
+
+    @Test
+    fun testAsyncRefreshClosesImmediately() = runTest {
+        val timeSource = TestTimeSource()
+        val clock = ManualClock()
+
+        // Entry expires in 30 seconds, refresh buffer is 1 minute. Next `resolve` call should trigger the async refresh
+        val cache = S3ExpressCredentialsCache()
+        val entry = getCacheEntry(timeSource.markNow() + 30.seconds)
+        cache.put(entry.key, entry.value)
+
+        val expectedCredentials = SessionCredentials {
+            accessKeyId = "access"
+            secretAccessKey = "secret"
+            sessionToken = "session"
+            expiration = clock.now() + 5.minutes
+        }
+
+        val provider = DefaultS3ExpressCredentialsProvider(timeSource, clock, cache, refreshBuffer = 1.minutes)
+
+        val blockingTestS3Client = object : TestS3Client(expectedCredentials) {
+            override suspend fun createSession(input: CreateSessionRequest): CreateSessionResponse {
+                delay(10.seconds)
+                numCreateSession += 1
+                return CreateSessionResponse { credentials = expectedCredentials }
+            }
+        }
+
+        val attributes = ExecutionContext.build {
+            this.attributes[S3Attributes.ExpressClient] = blockingTestS3Client
+            this.attributes[S3Attributes.Bucket] = "bucket"
+        }
+
+        withTimeout(5.seconds) {
+            provider.resolve(attributes)
+            provider.close()
+        }
+        assertEquals(0, blockingTestS3Client.numCreateSession)
+    }
+
+    /**
+     * Get an instance of [Map.Entry<S3ExpressCredentialsCacheKey, S3ExpressCredentialsCacheValue>] using the given [expiration],
+     * [bucket], and optional [bootstrapCredentials] and [sessionCredentials].
+     */
+    private fun getCacheEntry(
+        expiration: ComparableTimeMark,
+        bucket: String = "bucket",
+        bootstrapCredentials: Credentials = Credentials(accessKeyId = "accessKeyId", secretAccessKey = "secretAccessKey", sessionToken = "sessionToken"),
+        sessionCredentials: Credentials = Credentials(accessKeyId = "s3AccessKeyId", secretAccessKey = "s3SecretAccessKey", sessionToken = "s3SessionToken"),
+    ): S3ExpressCredentialsCacheEntry = mapOf(
+        S3ExpressCredentialsCacheKey(bucket, bootstrapCredentials) to S3ExpressCredentialsCacheValue(ExpiringValue(sessionCredentials, expiration)),
+    ).entries.first()
+
+    /**
+     * A test S3Client used to mock calls to s3:CreateSession.
+     * @param expectedCredentials the expected session credentials returned from s3:CreateSession
+     * @param client the base S3 client used to implement other operations, though they are unused.
+     * @param throwExceptionOnBucketNamed an optional bucket name, which when specified and present in the [CreateSessionRequest], will
+     * cause the client to throw an exception instead of returning credentials. Used for testing s3:CreateSession failures.
+     */
+    private open inner class TestS3Client(
+        val expectedCredentials: SessionCredentials,
+        val baseCredentials: Credentials = DEFAULT_BASE_CREDENTIALS,
+        val client: S3Client = S3Client { credentialsProvider = StaticCredentialsProvider(baseCredentials) },
+        val throwExceptionOnBucketNamed: String? = null,
+    ) : S3Client by client {
+        var numCreateSession = 0
+
+        override suspend fun createSession(input: CreateSessionRequest): CreateSessionResponse {
+            numCreateSession += 1
+            throwExceptionOnBucketNamed?.let {
+                if (input.bucket == it) {
+                    throw Exception("Failed to create session credentials for bucket: $throwExceptionOnBucketNamed")
+                }
+            }
+            return CreateSessionResponse { credentials = expectedCredentials }
+        }
+    }
+}
diff --git a/services/s3/common/test/aws/sdk/kotlin/services/s3/express/S3ExpressCredentialsCacheTest.kt b/services/s3/common/test/aws/sdk/kotlin/services/s3/express/S3ExpressCredentialsCacheTest.kt
new file mode 100644
index 00000000000..2387496056f
--- /dev/null
+++ b/services/s3/common/test/aws/sdk/kotlin/services/s3/express/S3ExpressCredentialsCacheTest.kt
@@ -0,0 +1,75 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.services.s3.express
+
+import aws.smithy.kotlin.runtime.auth.awscredentials.Credentials
+import kotlinx.coroutines.test.runTest
+import kotlin.test.*
+import kotlin.time.Duration.Companion.minutes
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.TestTimeSource
+
+public class S3ExpressCredentialsCacheTest {
+    @Test
+    fun testCacheKeyEquality() = runTest {
+        val bucket = "bucket"
+        val testCredentials = Credentials("accessKeyId", "secretAccessKey", "sessionToken")
+
+        // Different keys with the same bucket and credentials should be considered equal
+        val key1 = S3ExpressCredentialsCacheKey(bucket, testCredentials)
+        val key2 = S3ExpressCredentialsCacheKey(bucket, testCredentials)
+
+        assertEquals(key1, key2)
+    }
+
+    @Test
+    fun testCacheOperations() = runTest {
+        val cache = S3ExpressCredentialsCache()
+
+        val bucket = "bucket"
+        val bootstrapCredentials = Credentials("accessKeyId", "secretAccessKey", "sessionToken")
+        val key = S3ExpressCredentialsCacheKey(bucket, bootstrapCredentials)
+
+        val sessionCredentials = Credentials("superFastAccessKey", "superSecretSecretKey", "s3SessionToken")
+        val expiringSessionCredentials = ExpiringValue(sessionCredentials, TestTimeSource().markNow())
+        val value = S3ExpressCredentialsCacheValue(expiringSessionCredentials)
+
+        cache.put(key, value) // put
+        assertEquals(expiringSessionCredentials, cache.get(key)?.expiringCredentials) // get
+        assertEquals(1, cache.size) // size
+        assertContains(cache.entries.map { it.key }, key) // entries
+        assertContains(cache.entries.map { it.value }, value) // entries
+
+        cache.remove(key)
+        assertEquals(0, cache.size)
+        assertNull(cache.get(key))
+    }
+
+    @Test
+    fun testIsExpired() = runTest {
+        val timeSource = TestTimeSource()
+
+        val sessionCredentials = Credentials("superFastAccessKey", "superSecretSecretKey", "s3SessionToken")
+
+        val expiringSessionCredentials = ExpiringValue(sessionCredentials, timeSource.markNow() + 5.minutes)
+        assertFalse(expiringSessionCredentials.isExpired)
+
+        timeSource += 5.minutes + 1.seconds // advance just past the expiration time
+        assertTrue(expiringSessionCredentials.isExpired)
+    }
+
+    @Test
+    fun testIsWithin() = runTest {
+        val timeSource = TestTimeSource()
+
+        val sessionCredentials = Credentials("superFastAccessKey", "superSecretSecretKey", "s3SessionToken")
+
+        val expiringSessionCredentials = ExpiringValue(sessionCredentials, timeSource.markNow() + 1.minutes)
+        assertFalse(expiringSessionCredentials.isExpiringWithin(30.seconds))
+
+        timeSource += 31.seconds
+        assertTrue(expiringSessionCredentials.isExpiringWithin(30.seconds))
+    }
+}
diff --git a/services/s3/e2eTest/src/S3ExpressTest.kt b/services/s3/e2eTest/src/S3ExpressTest.kt
new file mode 100644
index 00000000000..84538539521
--- /dev/null
+++ b/services/s3/e2eTest/src/S3ExpressTest.kt
@@ -0,0 +1,159 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.e2etest
+
+import aws.sdk.kotlin.services.s3.S3Client
+import aws.sdk.kotlin.services.s3.express.S3_EXPRESS_SESSION_TOKEN_HEADER
+import aws.sdk.kotlin.services.s3.model.*
+import aws.sdk.kotlin.services.s3.presigners.presignPutObject
+import aws.sdk.kotlin.services.s3.putObject
+import aws.sdk.kotlin.services.s3.withConfig
+import aws.smithy.kotlin.runtime.client.ProtocolRequestInterceptorContext
+import aws.smithy.kotlin.runtime.content.ByteStream
+import aws.smithy.kotlin.runtime.content.decodeToString
+import aws.smithy.kotlin.runtime.http.interceptors.HttpInterceptor
+import aws.smithy.kotlin.runtime.http.request.HttpRequest
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.AfterAll
+import org.junit.jupiter.api.BeforeAll
+import org.junit.jupiter.api.TestInstance
+import kotlin.test.*
+import kotlin.time.Duration.Companion.minutes
+
+/**
+ * Tests for S3 Express operations
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class S3ExpressTest {
+    private val client = S3Client {
+        region = S3TestUtils.DEFAULT_REGION
+    }
+
+    private val testBuckets: MutableList<String> = mutableListOf()
+
+    @BeforeAll
+    fun setup(): Unit = runBlocking {
+        val suffix = "--usw2-az1--x-s3" // us-west-2 availability zone 1
+
+        // create a few test buckets to test the credentials cache
+        testBuckets.add(S3TestUtils.getTestDirectoryBucket(client, suffix))
+        testBuckets.add(S3TestUtils.getTestDirectoryBucket(client, suffix))
+        testBuckets.add(S3TestUtils.getTestDirectoryBucket(client, suffix))
+    }
+
+    @AfterAll
+    fun cleanup(): Unit = runBlocking {
+        testBuckets.forEach { bucket ->
+            S3TestUtils.deleteBucketAndAllContents(client, bucket)
+        }
+        client.close()
+    }
+
+    @Test
+    fun testPutObject() = runTest {
+        val content = "30 minutes, or it's free!"
+        val keyName = "express.txt"
+
+        testBuckets.forEach { bucketName ->
+            val trackingInterceptor = S3ExpressInvocationTrackingInterceptor()
+            client.withConfig {
+                interceptors += trackingInterceptor
+            }.use { trackingClient ->
+                trackingClient.putObject {
+                    bucket = bucketName
+                    key = keyName
+                    body = ByteStream.fromString(content)
+                }
+
+                val req = GetObjectRequest {
+                    bucket = bucketName
+                    key = keyName
+                }
+
+                val respContent = client.getObject(req) {
+                    it.body?.decodeToString()
+                }
+
+                assertEquals(content, respContent)
+                assertEquals(1, trackingInterceptor.s3ExpressInvocations)
+            }
+        }
+    }
+
+    @Test
+    fun testPresignedPutObject() = runTest {
+        val content = "Presign this!"
+        val keyName = "express-presigned.txt"
+
+        testBuckets.forEach { bucketName ->
+            val presigned = client.presignPutObject(
+                PutObjectRequest {
+                    bucket = bucketName
+                    key = keyName
+                    body = ByteStream.fromString(content)
+                },
+                5.minutes,
+            )
+
+            assertTrue(presigned.url.parameters.decodedParameters.contains("X-Amz-Security-Token"))
+
+            // FIXME Presigned requests should use S3 Express Auth Scheme resulting in `X-Amz-S3session-Token`
+            // https://github.com/awslabs/aws-sdk-kotlin/issues/1236
+            assertFalse(presigned.url.parameters.decodedParameters.contains(S3_EXPRESS_SESSION_TOKEN_HEADER))
+        }
+    }
+
+    @Test
+    fun testChecksums() = runTest {
+        val bucketName = testBuckets.first() // only need one bucket for this test
+
+        val keysToDelete = listOf("checksums.txt", "delete-me.txt", "dont-forget-about-me.txt")
+        keysToDelete.forEach {
+            client.putObject {
+                bucket = bucketName
+                key = it
+                body = ByteStream.fromString("Check out these sums!")
+            }
+        }
+
+        client.withConfig {
+            interceptors += CRC32ChecksumValidatingInterceptor()
+        }.use { validatingClient ->
+            // s3:DeleteObjects requires a checksum, even if the user doesn't specify one.
+            // normally the SDK would default to MD5, but S3 Express must default to CRC32 instead.
+            val req = DeleteObjectsRequest {
+                bucket = bucketName
+                delete = Delete {
+                    objects = keysToDelete.map {
+                        ObjectIdentifier { key = it }
+                    }
+                }
+            }
+
+            validatingClient.deleteObjects(req)
+        }
+    }
+
+    private class S3ExpressInvocationTrackingInterceptor : HttpInterceptor {
+        var s3ExpressInvocations = 0
+
+        override fun readAfterSigning(context: ProtocolRequestInterceptorContext<Any, HttpRequest>) {
+            if (context.protocolRequest.headers.contains(S3_EXPRESS_SESSION_TOKEN_HEADER)) {
+                s3ExpressInvocations += 1
+            }
+        }
+    }
+
+    private class CRC32ChecksumValidatingInterceptor : HttpInterceptor {
+        override fun readAfterSigning(context: ProtocolRequestInterceptorContext<Any, HttpRequest>) {
+            val headers = context.protocolRequest.headers
+            if (headers.contains(S3_EXPRESS_SESSION_TOKEN_HEADER)) {
+                assertTrue(headers.contains("x-amz-checksum-crc32"), "Failed to find x-amz-checksum-crc32 header")
+                assertFalse(headers.contains("Content-MD5"), "Unexpectedly found Content-MD5 header")
+            }
+        }
+    }
+}
diff --git a/services/s3/e2eTest/src/S3TestUtils.kt b/services/s3/e2eTest/src/S3TestUtils.kt
index c6497cc74f2..c709fffe01f 100644
--- a/services/s3/e2eTest/src/S3TestUtils.kt
+++ b/services/s3/e2eTest/src/S3TestUtils.kt
@@ -17,6 +17,7 @@ import aws.sdk.kotlin.services.s3control.*
 import aws.sdk.kotlin.services.s3control.model.*
 import aws.sdk.kotlin.services.sts.StsClient
 import aws.smithy.kotlin.runtime.http.request.HttpRequest
+import aws.smithy.kotlin.runtime.text.ensurePrefix
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.withTimeout
@@ -35,6 +36,9 @@ object S3TestUtils {
     // The E2E test account only has permission to operate on buckets with the prefix
     private const val TEST_BUCKET_PREFIX = "s3-test-bucket-"
 
+    private const val S3_MAX_BUCKET_NAME_LENGTH = 63 // https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html
+    private const val S3_EXPRESS_DIRECTORY_BUCKET_SUFFIX = "--x-s3"
+
     suspend fun getTestBucket(
         client: S3Client,
         region: String? = null,
@@ -94,6 +98,41 @@ object S3TestUtils {
         testBucket
     }
 
+    suspend fun getTestDirectoryBucket(client: S3Client, suffix: String) = withTimeout(60.seconds) {
+        var testBucket = client.listBuckets()
+            .buckets
+            ?.mapNotNull { it.name }
+            ?.firstOrNull { it.startsWith(TEST_BUCKET_PREFIX) && it.endsWith(S3_EXPRESS_DIRECTORY_BUCKET_SUFFIX) }
+
+        if (testBucket == null) {
+            // Adding S3 Express suffix surpasses the bucket name length limit... trim the UUID if needed
+            testBucket = TEST_BUCKET_PREFIX +
+                UUID.randomUUID().toString().subSequence(0 until (S3_MAX_BUCKET_NAME_LENGTH - TEST_BUCKET_PREFIX.length - suffix.ensurePrefix("--").length)) +
+                suffix.ensurePrefix("--")
+
+            println("Creating S3 Express directory bucket: $testBucket")
+
+            val availabilityZone = testBucket // s3-test-bucket-UUID--use1-az4--x-s3
+                .removeSuffix(S3_EXPRESS_DIRECTORY_BUCKET_SUFFIX) // s3-test-bucket-UUID--use1-az4
+                .substringAfterLast("--") // use1-az4
+
+            client.createBucket {
+                bucket = testBucket
+                createBucketConfiguration {
+                    location = LocationInfo {
+                        type = LocationType.AvailabilityZone
+                        name = availabilityZone
+                    }
+                    bucket = BucketInfo {
+                        type = BucketType.Directory
+                        dataRedundancy = DataRedundancy.SingleAvailabilityZone
+                    }
+                }
+            }
+        }
+        testBucket
+    }
+
     @OptIn(ExperimentalCoroutinesApi::class)
     suspend fun deleteBucketAndAllContents(client: S3Client, bucketName: String): Unit = coroutineScope {
         val scope = this
@@ -234,21 +273,25 @@ object S3TestUtils {
         operation: String,
     ) {
         withTimeout(timeoutAfter) {
+            var status: String? = null
             while (true) {
-                val status = s3ControlClient.describeMultiRegionAccessPointOperation {
+                val latestStatus = s3ControlClient.describeMultiRegionAccessPointOperation {
                     accountId = testAccountId
                     requestTokenArn = request
                 }.asyncOperation?.requestStatus
 
-                println("Waiting on $operation operation. Status: $status ")
-
-                if (status == "SUCCEEDED") {
-                    println("$operation operation succeeded.")
-                    return@withTimeout
+                when (latestStatus) {
+                    "SUCCEEDED" -> {
+                        println("$operation operation succeeded.")
+                        return@withTimeout
+                    }
+                    "FAILED" -> throw IllegalStateException("$operation operation failed")
+                    else -> { if (status == null || latestStatus != status) {
+                        println("Waiting on $operation operation. Status: $latestStatus ")
+                        status = latestStatus
+                    } }
                 }
 
-                check(status != "FAILED") { "$operation operation failed" }
-
                 delay(10.seconds) // Avoid constant status checks
             }
         }
diff --git a/tests/benchmarks/service-benchmarks/README.md b/tests/benchmarks/service-benchmarks/README.md
index 5d13740fd5e..fd34d54ed41 100644
--- a/tests/benchmarks/service-benchmarks/README.md
+++ b/tests/benchmarks/service-benchmarks/README.md
@@ -51,6 +51,19 @@ The following benchmark run serves as a baseline for future runs:
 |   —GetEndpoint        |               |  555 | 0.220 | 0.401 | 0.406 | 0.452 |  0.506 |  6.606 |
 |   —PutEvents          |               |  415 | 0.242 | 0.400 | 0.420 | 0.466 |  0.619 |  2.762 |
 
+### S3 Express
+S3 Express benchmarks were ran separately.
+
+| Hardware type  | Operating system | SDK version     |
+|----------------|------------------|-----------------|
+| EC2 m5.4xlarge | Amazon Linux 2023   | 1.0.66 |
+
+|               | E2E Duration (ms) |    n |   min |   avg |   med |   p90 |   p99 |    max |
+| :---          |              ---: | ---: |  ---: |  ---: |  ---: |  ---: |  ---: |   ---: |
+| **S3Express** |                   |      |       |       |       |       |       |        |
+|   —PutObject  |                   | 1950 | 7.240 | 7.487 | 7.455 | 7.617 | 7.886 | 21.096 |
+|   —GetObject  |                   | 3402 | 4.049 | 4.188 | 4.141 | 4.243 | 4.470 | 20.537 |
+
 ## Methodology
 
 This section describes how the benchmarks actually work at a high level:
diff --git a/tests/benchmarks/service-benchmarks/jvm/src/aws/sdk/kotlin/benchmarks/service/BenchmarkHarness.kt b/tests/benchmarks/service-benchmarks/jvm/src/aws/sdk/kotlin/benchmarks/service/BenchmarkHarness.kt
index 4f523a93f2d..ae60c0a2437 100644
--- a/tests/benchmarks/service-benchmarks/jvm/src/aws/sdk/kotlin/benchmarks/service/BenchmarkHarness.kt
+++ b/tests/benchmarks/service-benchmarks/jvm/src/aws/sdk/kotlin/benchmarks/service/BenchmarkHarness.kt
@@ -22,6 +22,7 @@ private val benchmarks = setOf(
     CloudwatchEventsBenchmark(),
     DynamoDbBenchmark(),
     PinpointBenchmark(),
+    S3ExpressBenchmark(),
 ).map {
     @Suppress("UNCHECKED_CAST")
     it as ServiceBenchmark<SdkClient>
diff --git a/tests/benchmarks/service-benchmarks/jvm/src/aws/sdk/kotlin/benchmarks/service/definitions/S3ExpressBenchmark.kt b/tests/benchmarks/service-benchmarks/jvm/src/aws/sdk/kotlin/benchmarks/service/definitions/S3ExpressBenchmark.kt
new file mode 100644
index 00000000000..73df450e939
--- /dev/null
+++ b/tests/benchmarks/service-benchmarks/jvm/src/aws/sdk/kotlin/benchmarks/service/definitions/S3ExpressBenchmark.kt
@@ -0,0 +1,100 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package aws.sdk.kotlin.benchmarks.service.definitions
+
+import aws.sdk.kotlin.benchmarks.service.Common
+import aws.sdk.kotlin.services.s3.*
+import aws.sdk.kotlin.services.s3.model.*
+import aws.smithy.kotlin.runtime.ExperimentalApi
+import aws.smithy.kotlin.runtime.content.ByteStream
+
+/**
+ * Benchmarks for S3 Express One Zone.
+ * Note: This benchmark must be run from an EC2 host in the same AZ as the bucket (usw2-az1).
+ */
+class S3ExpressBenchmark : ServiceBenchmark<S3Client> {
+    private val regionAz = "usw2-az1" // FIXME Use IMDS to dynamically create a bucket in the EC2 host's AZ
+    private val bucketName = Common.random("sdk-benchmark-bucket-")
+        .substring(0 until 47) + // truncate to prevent "bucket name too long" errors
+        "--$regionAz--x-s3"
+
+    private val KEY = "64kb-object"
+    private val CONTENTS = "a".repeat(65536) // 64KB
+
+    @OptIn(ExperimentalApi::class)
+    override suspend fun client() = S3Client.fromEnvironment {
+        clientName = "S3Express"
+        retryStrategy = Common.noRetries
+        telemetryProvider = Common.telemetryProvider
+        httpClient {
+            telemetryProvider = Common.telemetryProvider
+        }
+    }
+
+    override suspend fun setup(client: S3Client) {
+        client.createBucket {
+            bucket = bucketName
+            createBucketConfiguration {
+                location {
+                    type = LocationType.AvailabilityZone
+                    this.name = regionAz
+                }
+                bucket {
+                    type = BucketType.Directory
+                    dataRedundancy = DataRedundancy.SingleAvailabilityZone
+                }
+            }
+        }
+    }
+
+    override val operations get() = listOf(putObjectBenchmark, getObjectBenchmark)
+
+    override suspend fun tearDown(client: S3Client) {
+        client.deleteBucket { bucket = bucketName }
+    }
+
+    private val putObjectBenchmark = object : AbstractOperationBenchmark<S3Client>("PutObject") {
+        override suspend fun transact(client: S3Client) {
+            client.putObject {
+                bucket = bucketName
+                key = KEY
+                body = ByteStream.fromString(CONTENTS)
+            }
+        }
+
+        override suspend fun tearDown(client: S3Client) {
+            client.deleteObject {
+                bucket = bucketName
+                key = KEY
+            }
+        }
+    }
+
+    private val getObjectBenchmark = object : AbstractOperationBenchmark<S3Client>("GetObject") {
+        override suspend fun setup(client: S3Client) {
+            client.putObject {
+                bucket = bucketName
+                key = KEY
+                body = ByteStream.fromString(CONTENTS)
+            }
+        }
+
+        override suspend fun transact(client: S3Client) {
+            client.getObject(
+                GetObjectRequest {
+                    bucket = bucketName
+                    key = KEY
+                },
+            ) { }
+        }
+
+        override suspend fun tearDown(client: S3Client) {
+            client.deleteObject {
+                bucket = bucketName
+                key = KEY
+            }
+        }
+    }
+}
diff --git a/tests/benchmarks/service-benchmarks/jvm/src/aws/sdk/kotlin/benchmarks/service/telemetry/BenchmarkTelemetryProvider.kt b/tests/benchmarks/service-benchmarks/jvm/src/aws/sdk/kotlin/benchmarks/service/telemetry/BenchmarkTelemetryProvider.kt
index e05c7a29e54..5a031679b83 100644
--- a/tests/benchmarks/service-benchmarks/jvm/src/aws/sdk/kotlin/benchmarks/service/telemetry/BenchmarkTelemetryProvider.kt
+++ b/tests/benchmarks/service-benchmarks/jvm/src/aws/sdk/kotlin/benchmarks/service/telemetry/BenchmarkTelemetryProvider.kt
@@ -14,12 +14,12 @@ import aws.smithy.kotlin.runtime.telemetry.metrics.*
 import aws.smithy.kotlin.runtime.telemetry.trace.TracerProvider
 
 private val capturedMetrics = mapOf(
-    "smithy.client.attempt_overhead_duration" to "Overhead",
+    "smithy.client.call.attempt_overhead_duration" to "Overhead",
     // "smithy.client.http.time_to_first_byte" to "TTFB",
-    // "smithy.client.attempt_duration" to "Call",
-    // "smithy.client.serialization_duration" to "Serlz",
-    // "smithy.client.deserialization_duration" to "Deserlz",
-    // "smithy.client.resolve_endpoint_duration" to "EPR",
+    // "smithy.client.call.attempt_duration" to "Call",
+    // "smithy.client.call.serialization_duration" to "Serlz",
+    // "smithy.client.call.deserialization_duration" to "Deserlz",
+    // "smithy.client.call.resolve_endpoint_duration" to "EPR",
 )
 
 @ExperimentalApi