Bootstrap event streams

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
@@ -34,4 +37,17 @@ public object AuthAttributes {
      * NOTE: This is not a common option.
      */
     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 not a common option.
+     */
+    public val CredentialsProvider: ClientOption<CredentialsProvider> = ClientOption("CredentialsProvider")
+
+    /**
+     * 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

@@ -144,7 +144,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")
             }
         }
 

aws-runtime/protocols/aws-event-stream/common/src/aws/sdk/kotlin/runtime/protocol/eventstream/EventStreamSigning.kt

@@ -0,0 +1,99 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+package aws.sdk.kotlin.runtime.protocol.eventstream
+
+import aws.sdk.kotlin.runtime.auth.signing.*
+import aws.sdk.kotlin.runtime.execution.AuthAttributes
+import aws.smithy.kotlin.runtime.client.ExecutionContext
+import aws.smithy.kotlin.runtime.io.SdkByteBuffer
+import aws.smithy.kotlin.runtime.io.bytes
+import aws.smithy.kotlin.runtime.time.Clock
+import aws.smithy.kotlin.runtime.time.Instant
+import aws.smithy.kotlin.runtime.util.InternalApi
+import aws.smithy.kotlin.runtime.util.get
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
+
+/**
+ * Creates a flow that signs each event stream message with the given signing config.
+ *
+ * Each message's signature incorporates the signature of the previous message.
+ * The very first message incorporates the signature of the initial-request for
+ * both HTTP2 and WebSockets. The initial signature comes from the execution context.
+ */
+@InternalApi
+public fun Flow<Message>.sign(
+    context: ExecutionContext,
+    config: AwsSigningConfig,
+): Flow<Message> = flow {
+    val messages = this@sign
+
+    // NOTE: We need the signature of the initial HTTP request to seed the event stream signatures
+    // This is a bit of a chicken and egg problem since the event stream is constructed before the request
+    // is signed. The body of the stream shouldn't start being consumed though until after the entire request
+    // is built. Thus, by the time we get here the signature will exist in the context.
+    var prevSignature = context.getOrNull(AuthAttributes.RequestSignature) ?: error("expected initial HTTP signature to be set before message signing commences")
+
+    // signature date is updated per event message
+    val configBuilder = config.toBuilder()
+
+    messages.collect { message ->
+        // FIXME - can we get an estimate here on size?
+        val buffer = SdkByteBuffer(0U)
+        message.encode(buffer)
+
+        // the entire message is wrapped as the payload of the signed message
+        val result = signPayload(configBuilder, prevSignature, buffer.bytes())
+        prevSignature = result.signature
+        emit(result.output)
+    }
+}
+
+internal suspend fun signPayload(
+    configBuilder: AwsSigningConfig.Builder,
+    prevSignature: ByteArray,
+    messagePayload: ByteArray,
+    clock: Clock = Clock.System
+): SigningResult<Message> {
+    val dt = clock.now().truncateSubsecs()
+    val config = configBuilder.apply { date = dt }.build()
+
+    val result = sign(messagePayload, prevSignature, config)
+    val signature = result.signature
+
+    val signedMessage = buildMessage {
+        addHeader(":date", HeaderValue.Timestamp(dt))
+        addHeader(":chunk-signature", HeaderValue.ByteArray(signature))
+        payload = messagePayload
+    }
+
+    return SigningResult(signedMessage, signature)
+}
+
+/**
+ * Truncate the sub-seconds from the current time
+ */
+private fun Instant.truncateSubsecs(): Instant = Instant.fromEpochSeconds(epochSeconds, 0)
+
+/**
+ * Create a new signing config for an event stream using the current context to set the operation/service specific
+ * configuration (e.g. region, signing service, credentials, etc)
+ */
+@InternalApi
+public fun ExecutionContext.newEventStreamSigningConfig(): AwsSigningConfig = AwsSigningConfig {
+    algorithm = AwsSigningAlgorithm.SIGV4
+    signatureType = AwsSignatureType.HTTP_REQUEST_CHUNK
+    region = this@newEventStreamSigningConfig[AuthAttributes.SigningRegion]
+    service = this@newEventStreamSigningConfig[AuthAttributes.SigningService]
+    credentialsProvider = this@newEventStreamSigningConfig[AuthAttributes.CredentialsProvider]
+    useDoubleUriEncode = false
+    normalizeUriPath = true
+    signedBodyHeader = AwsSignedBodyHeaderType.NONE
+
+    // FIXME - needs to be set on the operation for initial request
+    // signedBodyHeader = AwsSignedBodyHeaderType.X_AMZ_CONTENT_SHA256
+    // signedBodyValue = AwsSignedBodyValue.STREAMING_AWS4_HMAC_SHA256_PAYLOAD
+}