feat: bootstrap event streams (#545)

* Bootstrap event streams (#537)

* fix event stream filtering

* add parsing of common headers

* refractor frame decoder into a Flow

* remove event stream operation filter from customizations

* refactore event stream parsing; implement rough deserialization codegen

* fix warning

* filter out event stream errors

* render deserialization for exception event stream messages

* inject http request signature into the execution context once known

* add support for chunked signing

* add encode transform for message stream

* inline signing config builder

* initial event stream serialize implementation

* fix compile issues

* disable wip integration tests

* suppress test; cleanup codegen

* Event Stream Codegen Tests (#542)

* Checkpoint Event Streams (#544)

* fix tests

* increase windows runner memory

Author: aajtodd

.github/workflows/continuous-integration.yml

@@ -72,7 +72,7 @@ jobs:
         # windows job runs out of memory with the defaults normally used
         shell: bash
         run: |
-          sed -i 's/org\.gradle\.jvmargs=.*$/org.gradle.jvmargs=-Xmx4g/' gradle.properties
+          sed -i 's/org\.gradle\.jvmargs=.*$/org.gradle.jvmargs=-Xmx5g/' gradle.properties
           cat gradle.properties
       - name: Build and Test ${{ env.PACKAGE_NAME }}
         run: |

aws-runtime/aws-core/build.gradle.kts

@@ -15,6 +15,9 @@ kotlin {
         commonMain {
             dependencies {
                 api("aws.smithy.kotlin:runtime-core:$smithyKotlinVersion")
+
+                // FIXME - should we just move these into core and get rid of aws-types at this point?
+                api(project(":aws-runtime:aws-types"))
                 implementation("aws.smithy.kotlin:logging:$smithyKotlinVersion")
             }
         }

aws-runtime/aws-core/common/src/aws/sdk/kotlin/runtime/execution/AuthAttributes.kt

@@ -5,8 +5,11 @@
 
 package aws.sdk.kotlin.runtime.execution
 
+import aws.sdk.kotlin.runtime.auth.credentials.CredentialsProvider
 import aws.smithy.kotlin.runtime.client.ClientOption
 import aws.smithy.kotlin.runtime.time.Instant
+import aws.smithy.kotlin.runtime.util.AttributeKey
+import aws.smithy.kotlin.runtime.util.InternalApi
 
 /**
  * Operation (execution) options related to authorization
@@ -31,7 +34,36 @@ public object AuthAttributes {
 
     /**
      * Override the date to complete the signing process with. Defaults to current time when not specified.
-     * NOTE: This is not a common option.
+     *
+     * NOTE: This is an advanced configuration option that does not normally need to be set manually.
      */
     public val SigningDate: ClientOption<Instant> = ClientOption("SigningDate")
+
+    /**
+     * The [CredentialsProvider] to complete the signing process with. Defaults to the provider configured
+     * on the service client.
+     *
+     * NOTE: This is an advanced configuration option that does not normally need to be set manually.
+     */
+    public val CredentialsProvider: ClientOption<CredentialsProvider> = ClientOption("CredentialsProvider")
+
+    /**
+     * The precomputed signed body value. See [aws.sdk.kotlin.runtime.auth.signing.AwsSigningConfig.signedBodyValue].
+     *
+     * NOTE: This is an advanced configuration option that does not normally need to be set manually.
+     */
+    public val SignedBodyValue: ClientOption<String> = ClientOption("SignedBodyValue")
+
+    /**
+     * The signed body header type. See [aws.sdk.kotlin.runtime.auth.signing.AwsSigningConfig.signedBodyHeaderType].
+     *
+     * NOTE: This is an advanced configuration option that does not normally need to be set manually.
+     */
+    public val SignedBodyHeaderType: ClientOption<String> = ClientOption("SignedBodyHeaderType")
+
+    /**
+     * The signature of the HTTP request. This will only exist after the request has been signed!
+     */
+    @InternalApi
+    public val RequestSignature: AttributeKey<ByteArray> = AttributeKey("AWS_HTTP_SIGNATURE")
 }

aws-runtime/aws-signing/common/src/aws/sdk/kotlin/runtime/auth/signing/AwsSigV4SigningMiddleware.kt

@@ -69,6 +69,15 @@ public class AwsSigV4SigningMiddleware(private val config: Config) : ModifyReque
          */
         public var omitSessionToken: Boolean = false
 
+        /**
+         * Optional string to use as the canonical request's body public value.
+         * If string is empty, a public value will be calculated from the payload during signing.
+         * Typically, this is the SHA-256 of the (request/chunk/event) payload, written as lowercase hex.
+         * If this has been precalculated, it can be set here. Special public values used by certain services can also be set
+         * (e.g. "UNSIGNED-PAYLOAD" "STREAMING-AWS4-HMAC-SHA256-PAYLOAD" "STREAMING-AWS4-HMAC-SHA256-EVENTS").
+         */
+        public val signedBodyValue: String? = null
+
         /**
          * Controls what body "hash" header, if any, should be added to the canonical request and the signed request.
          * Most services do not require this additional header.
@@ -117,6 +126,10 @@ public class AwsSigV4SigningMiddleware(private val config: Config) : ModifyReque
         // then we must decide how to compute the payload hash ourselves (defaults to unsigned payload)
         val isUnboundedStream = signableRequest.body == null && req.subject.body is HttpBody.Streaming
 
+        // favor attributes from the current request context
+        val precomputedSignedBodyValue = req.context.getOrNull(AuthAttributes.SignedBodyValue)
+        val precomputedHeaderType = req.context.getOrNull(AuthAttributes.SignedBodyHeaderType)?.let { AwsSignedBodyHeaderType.valueOf(it) }
+
         // operation signing config is baseConfig + operation specific config/overrides
         val opSigningConfig = AwsSigningConfig {
             region = req.context[AuthAttributes.SigningRegion]
@@ -131,8 +144,9 @@ public class AwsSigV4SigningMiddleware(private val config: Config) : ModifyReque
             useDoubleUriEncode = config.useDoubleUriEncode
             expiresAfter = config.expiresAfter
 
-            signedBodyHeader = config.signedBodyHeaderType
+            signedBodyHeader = precomputedHeaderType ?: config.signedBodyHeaderType
             signedBodyValue = when {
+                precomputedSignedBodyValue != null -> precomputedSignedBodyValue
                 isUnsignedRequest -> AwsSignedBodyValue.UNSIGNED_PAYLOAD
                 req.subject.body is HttpBody.Empty -> AwsSignedBodyValue.EMPTY_SHA256
                 isUnboundedStream -> {
@@ -144,7 +158,12 @@ public class AwsSigV4SigningMiddleware(private val config: Config) : ModifyReque
             }
         }
 
-        val signedRequest = AwsSigner.signRequest(signableRequest, opSigningConfig.toCrt())
+        val signingResult = AwsSigner.sign(signableRequest, opSigningConfig.toCrt())
+        val signedRequest = checkNotNull(signingResult.signedRequest) { "signing result must return a non-null HTTP request" }
+
+        // Add the signature to the request context
+        req.context[AuthAttributes.RequestSignature] = signingResult.signature
+
         req.subject.update(signedRequest)
         req.subject.body.resetStream()
 

aws-runtime/aws-signing/common/src/aws/sdk/kotlin/runtime/auth/signing/AwsSigning.kt

@@ -6,6 +6,7 @@
 package aws.sdk.kotlin.runtime.auth.signing
 
 import aws.sdk.kotlin.crt.auth.signing.AwsSigner
+import aws.sdk.kotlin.runtime.InternalSdkApi
 import aws.sdk.kotlin.runtime.crt.toSignableCrtRequest
 import aws.sdk.kotlin.runtime.crt.update
 import aws.smithy.kotlin.runtime.http.request.HttpRequest
@@ -61,3 +62,18 @@ public suspend fun sign(request: HttpRequest, config: AwsSigningConfig): Signing
     val output = builder.build()
     return SigningResult(output, crtResult.signature)
 }
+
+/**
+ * Sign a body [chunk] using the given signing [config]
+ *
+ * @param chunk the body chunk to sign
+ * @param prevSignature the signature of the previous component of the request (either the initial request signature
+ * itself for the first chunk or the previous chunk otherwise)
+ * @param config the signing configuration to use
+ * @return the signing result
+ */
+@InternalSdkApi
+public suspend fun sign(chunk: ByteArray, prevSignature: ByteArray, config: AwsSigningConfig): SigningResult<Unit> {
+    val crtResult = AwsSigner.signChunk(chunk, prevSignature, config.toCrt())
+    return SigningResult(Unit, crtResult.signature)
+}

aws-runtime/aws-signing/common/src/aws/sdk/kotlin/runtime/auth/signing/AwsSigningConfig.kt

@@ -5,6 +5,7 @@
 
 package aws.sdk.kotlin.runtime.auth.signing
 
+import aws.sdk.kotlin.runtime.InternalSdkApi
 import aws.sdk.kotlin.runtime.auth.credentials.Credentials
 import aws.sdk.kotlin.runtime.auth.credentials.CredentialsProvider
 import aws.smithy.kotlin.runtime.time.Instant
@@ -21,7 +22,7 @@ public typealias ShouldSignHeaderFn = (String) -> Boolean
  */
 public class AwsSigningConfig private constructor(builder: Builder) {
     public companion object {
-        public operator fun invoke(block: Builder.() -> Unit): AwsSigningConfig = Builder().apply(block).build()
+        public inline operator fun invoke(block: Builder.() -> Unit): AwsSigningConfig = Builder().apply(block).build()
     }
     /**
      * The region to sign against
@@ -119,6 +120,27 @@ public class AwsSigningConfig private constructor(builder: Builder) {
      */
     public val expiresAfter: Duration? = builder.expiresAfter
 
+    @InternalSdkApi
+    public fun toBuilder(): Builder {
+        val config = this
+        return Builder().apply {
+            region = config.region
+            service = config.service
+            date = config.date
+            algorithm = config.algorithm
+            shouldSignHeader = config.shouldSignHeader
+            signatureType = config.signatureType
+            useDoubleUriEncode = config.useDoubleUriEncode
+            normalizeUriPath = config.normalizeUriPath
+            omitSessionToken = config.omitSessionToken
+            signedBodyValue = config.signedBodyValue
+            signedBodyHeader = config.signedBodyHeaderType
+            credentials = config.credentials
+            credentialsProvider = config.credentialsProvider
+            expiresAfter = config.expiresAfter
+        }
+    }
+
     public class Builder {
         public var region: String? = null
         public var service: String? = null
@@ -135,7 +157,8 @@ public class AwsSigningConfig private constructor(builder: Builder) {
         public var credentialsProvider: CredentialsProvider? = null
         public var expiresAfter: Duration? = null
 
-        internal fun build(): AwsSigningConfig = AwsSigningConfig(this)
+        @InternalSdkApi
+        public fun build(): AwsSigningConfig = AwsSigningConfig(this)
     }
 }
 

aws-runtime/aws-signing/common/test/aws/sdk/kotlin/runtime/auth/signing/AwsSigningTest.kt

@@ -5,9 +5,14 @@
 
 package aws.sdk.kotlin.runtime.auth.signing
 
+import aws.sdk.kotlin.runtime.auth.credentials.Credentials
 import aws.smithy.kotlin.runtime.http.HttpMethod
+import aws.smithy.kotlin.runtime.http.Url
 import aws.smithy.kotlin.runtime.http.content.ByteArrayContent
+import aws.smithy.kotlin.runtime.http.request.HttpRequest
 import aws.smithy.kotlin.runtime.http.request.HttpRequestBuilder
+import aws.smithy.kotlin.runtime.http.request.headers
+import aws.smithy.kotlin.runtime.http.request.url
 import aws.smithy.kotlin.runtime.time.Instant
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.runTest
@@ -83,4 +88,102 @@ class AwsSigningTest {
         val authHeader = result.output.headers["Authorization"]!!
         assertTrue(authHeader.contains(expectedPrefix), "Sigv4A auth header: $authHeader")
     }
+
+    // based on: https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html#example-signature-calculations-streaming
+    private val CHUNKED_ACCESS_KEY_ID = "AKIAIOSFODNN7EXAMPLE"
+    private val CHUNKED_SECRET_ACCESS_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+    private val CHUNKED_TEST_CREDENTIALS = Credentials(CHUNKED_ACCESS_KEY_ID, CHUNKED_SECRET_ACCESS_KEY)
+    private val CHUNKED_TEST_REGION = "us-east-1"
+    private val CHUNKED_TEST_SERVICE = "s3"
+    private val CHUNKED_TEST_SIGNING_TIME = "2013-05-24T00:00:00Z"
+    private val CHUNK1_SIZE = 65536
+    private val CHUNK2_SIZE = 1024
+
+    private val EXPECTED_CHUNK_REQUEST_AUTHORIZATION_HEADER =
+        "AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request, " +
+            "SignedHeaders=content-encoding;content-length;host;x-amz-content-sha256;x-amz-date;x-amz-decoded-content-length;x-" +
+            "amz-storage-class, Signature=4f232c4386841ef735655705268965c44a0e4690baa4adea153f7db9fa80a0a9"
+
+    private val EXPECTED_REQUEST_SIGNATURE = "4f232c4386841ef735655705268965c44a0e4690baa4adea153f7db9fa80a0a9"
+    private val EXPECTED_FIRST_CHUNK_SIGNATURE = "ad80c730a21e5b8d04586a2213dd63b9a0e99e0e2307b0ade35a65485a288648"
+    private val EXPECTED_SECOND_CHUNK_SIGNATURE = "0055627c9e194cb4542bae2aa5492e3c1575bbb81b612b7d234b86a503ef5497"
+    private val EXPECTED_FINAL_CHUNK_SIGNATURE = "b6c6ea8a5354eaf15b3cb7646744f4275b71ea724fed81ceb9323e279d449df9"
+    private val EXPECTED_TRAILING_HEADERS_SIGNATURE = "df5735bd9f3295cd9386572292562fefc93ba94e80a0a1ddcbd652c4e0a75e6c"
+
+    private fun createChunkedRequestSigningConfig(): AwsSigningConfig = AwsSigningConfig {
+        algorithm = AwsSigningAlgorithm.SIGV4
+        signatureType = AwsSignatureType.HTTP_REQUEST_VIA_HEADERS
+        region = CHUNKED_TEST_REGION
+        service = CHUNKED_TEST_SERVICE
+        date = Instant.fromIso8601(CHUNKED_TEST_SIGNING_TIME)
+        useDoubleUriEncode = false
+        normalizeUriPath = true
+        signedBodyHeader = AwsSignedBodyHeaderType.X_AMZ_CONTENT_SHA256
+        signedBodyValue = AwsSignedBodyValue.STREAMING_AWS4_HMAC_SHA256_PAYLOAD
+        credentials = CHUNKED_TEST_CREDENTIALS
+    }
+
+    private fun createChunkedSigningConfig(): AwsSigningConfig = AwsSigningConfig {
+        algorithm = AwsSigningAlgorithm.SIGV4
+        signatureType = AwsSignatureType.HTTP_REQUEST_CHUNK
+        region = CHUNKED_TEST_REGION
+        service = CHUNKED_TEST_SERVICE
+        date = Instant.fromIso8601(CHUNKED_TEST_SIGNING_TIME)
+        useDoubleUriEncode = false
+        normalizeUriPath = true
+        signedBodyHeader = AwsSignedBodyHeaderType.NONE
+        credentials = CHUNKED_TEST_CREDENTIALS
+    }
+
+    private fun createChunkedTestRequest() = HttpRequest {
+        method = HttpMethod.PUT
+        url(Url.parse("https://s3.amazonaws.com/examplebucket/chunkObject.txt"))
+        headers {
+            set("Host", url.host)
+            set("x-amz-storage-class", "REDUCED_REDUNDANCY")
+            set("Content-Encoding", "aws-chunked")
+            set("x-amz-decoded-content-length", "66560")
+            set("Content-Length", "66824")
+        }
+    }
+
+    private fun chunk1(): ByteArray {
+        val chunk = ByteArray(CHUNK1_SIZE)
+        for (i in chunk.indices) {
+            chunk[i] = 'a'.code.toByte()
+        }
+        return chunk
+    }
+
+    private fun chunk2(): ByteArray {
+        val chunk = ByteArray(CHUNK2_SIZE)
+        for (i in chunk.indices) {
+            chunk[i] = 'a'.code.toByte()
+        }
+        return chunk
+    }
+
+    @Test
+    fun testSignChunks() = runTest {
+        val request = createChunkedTestRequest()
+        val chunkedRequestConfig = createChunkedRequestSigningConfig()
+        val requestResult = sign(request, chunkedRequestConfig)
+        assertEquals(EXPECTED_CHUNK_REQUEST_AUTHORIZATION_HEADER, requestResult.output.headers["Authorization"])
+        assertEquals(EXPECTED_REQUEST_SIGNATURE, requestResult.signature.decodeToString())
+
+        var prevSignature = requestResult.signature
+
+        val chunkedSigningConfig = createChunkedSigningConfig()
+        val chunk1Result = sign(chunk1(), prevSignature, chunkedSigningConfig)
+        assertEquals(EXPECTED_FIRST_CHUNK_SIGNATURE, chunk1Result.signature.decodeToString())
+        prevSignature = chunk1Result.signature
+
+        val chunk2Result = sign(chunk2(), prevSignature, chunkedSigningConfig)
+        assertEquals(EXPECTED_SECOND_CHUNK_SIGNATURE, chunk2Result.signature.decodeToString())
+        prevSignature = chunk2Result.signature
+
+        // TODO - do we want 0 byte data like this or just allow null?
+        val finalChunkResult = sign(ByteArray(0), prevSignature, chunkedSigningConfig)
+        assertEquals(EXPECTED_FINAL_CHUNK_SIGNATURE, finalChunkResult.signature.decodeToString())
+    }
 }

aws-runtime/protocols/aws-event-stream/build.gradle.kts

@@ -7,17 +7,27 @@ description = "Support for the vnd.amazon.event-stream content type"
 extra["displayName"] = "AWS :: SDK :: Kotlin :: Protocols :: Event Stream"
 extra["moduleName"] = "aws.sdk.kotlin.runtime.protocol.eventstream"
 
+val smithyKotlinVersion: String by project
+val coroutinesVersion: String by project
 kotlin {
     sourceSets {
         commonMain {
             dependencies {
                 api(project(":aws-runtime:aws-core"))
+                // exposes Buffer/MutableBuffer and SdkByteReadChannel
+                api("aws.smithy.kotlin:io:$smithyKotlinVersion")
+                // exposes Flow<T>
+                api("org.jetbrains.kotlinx:kotlinx-coroutines-core:$coroutinesVersion")
+
+                // exposes AwsSigningConfig
+                api(project(":aws-runtime:aws-signing"))
             }
         }
 
         commonTest {
             dependencies {
                 implementation(project(":aws-runtime:testing"))
+                api("org.jetbrains.kotlinx:kotlinx-coroutines-test:$coroutinesVersion")
             }
         }