feat(rt): add support for HTTP_REQUEST_EVENT signing (#644)

* feat(rt): add support for HTTP_REQUEST_EVENT signing

* fix: chunk signature uses raw signature bytes not hex encoding

* add sigv4 model dependency

* only add signer to context when signer is expected to exist

* simplify error check

Author: aajtodd

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

@@ -12,6 +12,7 @@ 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.decodeHexBytes
 import aws.smithy.kotlin.runtime.util.get
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
@@ -30,8 +31,6 @@ public fun Flow<Message>.sign(
 ): Flow<Message> = flow {
     val messages = this@sign
 
-    // FIXME Nothing actually populates this context attribute yet. It's possible we'll need some middleware or an
-    // alternate way of passing signers to this method.
     val signer = context.getOrNull(AwsSigningAttributes.Signer) ?: error("No signer was found in context")
 
     // NOTE: We need the signature of the initial HTTP request to seed the event stream signatures
@@ -70,10 +69,12 @@ internal suspend fun AwsSigner.signPayload(
 
     val result = signChunk(messagePayload, prevSignature, config)
     val signature = result.signature
+    // TODO - consider adding a direct Bytes -> Bytes hex decode rather than having to go through string
+    val binarySignature = signature.decodeToString().decodeHexBytes()
 
     val signedMessage = buildMessage {
         addHeader(":date", HeaderValue.Timestamp(dt))
-        addHeader(":chunk-signature", HeaderValue.ByteArray(signature))
+        addHeader(":chunk-signature", HeaderValue.ByteArray(binarySignature))
         payload = messagePayload
     }
 
@@ -92,7 +93,7 @@ private fun Instant.truncateSubsecs(): Instant = Instant.fromEpochSeconds(epochS
 @InternalApi
 public fun ExecutionContext.newEventStreamSigningConfig(): AwsSigningConfig = AwsSigningConfig {
     algorithm = AwsSigningAlgorithm.SIGV4
-    signatureType = AwsSignatureType.HTTP_REQUEST_CHUNK
+    signatureType = AwsSignatureType.HTTP_REQUEST_EVENT
     region = this@newEventStreamSigningConfig[AwsSigningAttributes.SigningRegion]
     service = this@newEventStreamSigningConfig[AwsSigningAttributes.SigningService]
     credentialsProvider = this@newEventStreamSigningConfig[AwsSigningAttributes.CredentialsProvider]

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

@@ -23,13 +23,8 @@ public class EventStreamFramingException(message: String, cause: Throwable? = nu
 public suspend fun decodeFrames(chan: SdkByteReadChannel): Flow<Message> = flow {
     while (!chan.isClosedForRead) {
         // get the prelude to figure out how much is left to read of the message
-        val preludeBytes = ByteArray(PRELUDE_BYTE_LEN_WITH_CRC)
-
-        try {
-            chan.readFully(preludeBytes)
-        } catch (ex: Exception) {
-            throw EventStreamFramingException("failed to read message prelude from channel", ex)
-        }
+        // null indicates the channel was closed and that no more messages are coming
+        val preludeBytes = readPrelude(chan) ?: return@flow
 
         val preludeBuf = SdkByteBuffer.of(preludeBytes).apply { advance(preludeBytes.size.toULong()) }
         val prelude = Prelude.decode(preludeBuf)
@@ -52,3 +47,27 @@ public suspend fun decodeFrames(chan: SdkByteReadChannel): Flow<Message> = flow
         emit(message)
     }
 }
+
+/**
+ * Read the message prelude from the channel.
+ * @return prelude bytes or null if the channel is closed and no additional prelude is coming
+ */
+private suspend fun readPrelude(chan: SdkByteReadChannel): ByteArray? {
+    val dest = ByteArray(PRELUDE_BYTE_LEN_WITH_CRC)
+    var remaining = dest.size
+    var offset = 0
+    while (remaining > 0 && !chan.isClosedForRead) {
+        val rc = chan.readAvailable(dest, offset, remaining)
+        if (rc == -1) break
+        offset += rc
+        remaining -= rc
+    }
+
+    // 0 bytes read and channel closed indicates no messages remaining -> null
+    if (remaining == PRELUDE_BYTE_LEN_WITH_CRC && chan.isClosedForRead) return null
+
+    // partial read -> failure
+    if (remaining > 0) throw EventStreamFramingException("failed to read event stream message prelude from channel: read: $offset bytes, expected $remaining more bytes")
+
+    return dest
+}

aws-runtime/protocols/aws-event-stream/common/test/aws/sdk/kotlin/runtime/protocol/eventstream/EventStreamSigningTest.kt

@@ -7,26 +7,27 @@ package aws.sdk.kotlin.runtime.protocol.eventstream
 
 import aws.smithy.kotlin.runtime.auth.awscredentials.Credentials
 import aws.smithy.kotlin.runtime.auth.awscredentials.CredentialsProvider
-import aws.smithy.kotlin.runtime.auth.awssigning.AwsSignatureType
-import aws.smithy.kotlin.runtime.auth.awssigning.AwsSigningConfig
-import aws.smithy.kotlin.runtime.auth.awssigning.DefaultAwsSigner
+import aws.smithy.kotlin.runtime.auth.awssigning.*
+import aws.smithy.kotlin.runtime.client.ExecutionContext
 import aws.smithy.kotlin.runtime.hashing.sha256
 import aws.smithy.kotlin.runtime.io.SdkByteBuffer
 import aws.smithy.kotlin.runtime.io.bytes
 import aws.smithy.kotlin.runtime.time.Instant
 import aws.smithy.kotlin.runtime.time.ManualClock
 import aws.smithy.kotlin.runtime.util.encodeToHex
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.test.runTest
-import kotlin.test.Ignore
 import kotlin.test.Test
 import kotlin.test.assertEquals
 
 @OptIn(ExperimentalCoroutinesApi::class)
 class EventStreamSigningTest {
+    private val testCredentialsProvider = object : CredentialsProvider {
+        override suspend fun getCredentials() = Credentials("fake access key", "fake secret key")
+    }
 
-    // FIXME - see https://github.com/awslabs/aws-sdk-kotlin/issues/543
-    @Ignore
     @Test
     fun testSignPayload() = runTest {
         val messageToSign = buildMessage {
@@ -37,12 +38,10 @@ class EventStreamSigningTest {
         val epoch = Instant.fromEpochSeconds(123_456_789L, 1234)
         val testClock = ManualClock(epoch)
         val signingConfig = AwsSigningConfig.Builder().apply {
-            credentialsProvider = object : CredentialsProvider {
-                override suspend fun getCredentials() = Credentials("fake access key", "fake secret key")
-            }
+            credentialsProvider = testCredentialsProvider
             region = "us-east-1"
             service = "testservice"
-            signatureType = AwsSignatureType.HTTP_REQUEST_CHUNK
+            signatureType = AwsSignatureType.HTTP_REQUEST_EVENT
         }
 
         val prevSignature = "last message sts".encodeToByteArray().sha256().encodeToHex().encodeToByteArray()
@@ -58,12 +57,32 @@ class EventStreamSigningTest {
         assertEquals(0, dateHeader.nanosecondsOfSecond)
 
         assertEquals(":chunk-signature", result.output.headers[1].name)
-        val expectedSignature = result.signature.encodeToHex()
+        // signature is hex encoded string bytes, the header value is the raw bytes
+        val expectedSignature = result.signature.decodeToString()
         val actualSignature = result.output.headers[1].value.expectByteArray().encodeToHex()
         assertEquals(expectedSignature, actualSignature)
 
-        // FIXME - based on Rust test: https://github.com/awslabs/smithy-rs/blob/v0.38.0/aws/rust-runtime/aws-sigv4/src/event_stream.rs#L166
         val expected = "1ea04a4f6becd85ae3e38e379ffaf4bb95042603f209512476cc6416868b31ee"
         assertEquals(expected, actualSignature)
     }
+
+    @Test
+    fun testEmptyEndFrameSent() = runTest {
+        val messageToSign = buildMessage {
+            addHeader("some-header", HeaderValue.String("value"))
+            payload = "test payload".encodeToByteArray()
+        }
+
+        val context = ExecutionContext()
+        context[AwsSigningAttributes.Signer] = DefaultAwsSigner
+        context[AwsSigningAttributes.RequestSignature] = HashSpecification.EmptyBody.hash.encodeToByteArray()
+        context[AwsSigningAttributes.SigningRegion] = "us-east-2"
+        context[AwsSigningAttributes.SigningService] = "test"
+        context[AwsSigningAttributes.CredentialsProvider] = testCredentialsProvider
+
+        val config = context.newEventStreamSigningConfig()
+        val signedEvents = flowOf(messageToSign).sign(context, config).toList()
+        // 1 message + empty signed frame
+        assertEquals(2, signedEvents.size)
+    }
 }

aws-runtime/protocols/aws-event-stream/common/test/aws/sdk/kotlin/runtime/protocol/eventstream/FrameDecoderTest.kt

@@ -6,12 +6,13 @@
 package aws.sdk.kotlin.runtime.protocol.eventstream
 
 import aws.smithy.kotlin.runtime.io.*
+import io.kotest.matchers.string.shouldContain
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.test.runTest
-import kotlin.test.Ignore
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
 
 @OptIn(ExperimentalCoroutinesApi::class)
 class FrameDecoderTest {
@@ -52,9 +53,14 @@ class FrameDecoderTest {
         assertEquals(expected3, actual[2])
     }
 
-    @Ignore
     @Test
     fun testChannelClosed() = runTest {
-        TODO("not implemented yet: need to add test for channel closed normally while waiting on prelude")
+        // contents don't matter
+        val partialPrelude = ByteArray(PRELUDE_BYTE_LEN_WITH_CRC - 4)
+        val chan = SdkByteReadChannel(partialPrelude)
+
+        assertFailsWith<EventStreamFramingException> {
+            decodeFrames(chan).collect()
+        }.message.shouldContain("failed to read event stream message prelude from channel: read: 8 bytes, expected 4 more bytes")
     }
 }

codegen/smithy-aws-kotlin-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/protocols/core/AwsHttpProtocolClientGenerator.kt

@@ -76,6 +76,7 @@ open class AwsHttpProtocolClientGenerator(
             if (AwsSignatureVersion4.isSupportedAuthentication(ctx.model, ctx.service)) {
                 val signingServiceName = AwsSignatureVersion4.signingServiceName(ctx.service)
                 write("ctx.#T(#T.SigningService, #S)", putIfAbsentSym, RuntimeTypes.Auth.Signing.AwsSigningCommon.AwsSigningAttributes, signingServiceName)
+                write("ctx.#T(#T.Signer, config.signer)", putIfAbsentSym, RuntimeTypes.Auth.Signing.AwsSigningCommon.AwsSigningAttributes)
             }
             write("ctx.#T(#T.SigningRegion, config.region)", putIfAbsentSym, RuntimeTypes.Auth.Signing.AwsSigningCommon.AwsSigningAttributes)
             write("ctx.#T(#T.CredentialsProvider, config.credentialsProvider)", putIfAbsentSym, RuntimeTypes.Auth.Signing.AwsSigningCommon.AwsSigningAttributes)

codegen/smithy-aws-kotlin-codegen/src/main/kotlin/aws/sdk/kotlin/codegen/protocols/eventstream/EventStreamSerializerGenerator.kt

@@ -63,9 +63,9 @@ class EventStreamSerializerGenerator(
 
         writer.write("val stream = input.#L ?: return #T.Empty", streamingMember.defaultName(), RuntimeTypes.Http.HttpBody)
         writer.write("val signingConfig = context.#T()", AwsRuntimeTypes.AwsEventStream.newEventStreamSigningConfig)
-        // FIXME - needs to be set on the operation for initial request
-        //   context[AwsSigningAttributes.SignedBodyHeader] = AwsSignedBodyHeader.X_AMZ_CONTENT_SHA256.name
-        //   context[AwsSigningAttributes.HashSpecification] = HashSpecification.StreamingAws4HmacSha256Events
+
+        // initial HTTP request should use an empty body hash since the actual body is the event stream
+        writer.write("context[#T.HashSpecification] = #T.EmptyBody", RuntimeTypes.Auth.Signing.AwsSigningCommon.AwsSigningAttributes, RuntimeTypes.Auth.Signing.AwsSigningCommon.HashSpecification)
 
         val encodeFn = encodeEventStreamMessage(ctx, op, streamShape)
         writer.withBlock("val messages = stream", "") {

tests/codegen/event-stream/event-stream-model-template.smithy

@@ -2,8 +2,10 @@ namespace aws.sdk.kotlin.test.eventstream
 
 use aws.protocols#{protocol-name}
 use aws.api#service
+use aws.auth#sigv4
 
 @{protocol-name}
+@sigv4(name: "event-stream-test")
 @service(sdkId: "EventStreamTest")
 service TestService { version: "123", operations: [TestStreamOp] }