Exception when using Flow#toByteStream

[felixscheinost]

Describe the bug

For example a putObject fails when using flow { ... }.toByteStream()

 val bytes = "abc".toByteArray(Charsets.UTF_8)
 client.putObject {
      bucket = "test"
      key = "object_non_repeatable_small"
      checksumAlgorithm = ChecksumAlgorithm.Sha256
      checksumSha256 = sha256sum(bytes)
      contentLength = bytes.size.toLong()
      body = flow {
        emit(bytes)
      }.toByteStream(this@runBlocking)
    }
}

This looks similar to #1130

Expected behavior

The object should be put successfully.

Current behavior

An exception is thrown"

Exception in thread "main" java.lang.IllegalArgumentException: Stream must be replayable to calculate a body hash
	at aws.smithy.kotlin.runtime.auth.awssigning.DefaultCanonicalizer.calculateHash(Canonicalizer.kt:155)
	at aws.smithy.kotlin.runtime.auth.awssigning.DefaultCanonicalizer.canonicalRequest(Canonicalizer.kt:79)
	at aws.smithy.kotlin.runtime.auth.awssigning.DefaultAwsSignerImpl.sign(DefaultAwsSigner.kt:27)
	at aws.smithy.kotlin.runtime.http.auth.AwsHttpSigner.sign(AwsHttpSigner.kt:176)
	at aws.smithy.kotlin.runtime.http.operation.AuthHandler.call(SdkOperationExecution.kt:307)
	at aws.smithy.kotlin.runtime.http.operation.AuthHandler.call(SdkOperationExecution.kt:259)
	at aws.sdk.kotlin.runtime.http.middleware.AwsRetryHeaderMiddleware.handle(AwsRetryHeaderMiddleware.kt:39)
	at aws.sdk.kotlin.runtime.http.middleware.AwsRetryHeaderMiddleware.handle(AwsRetryHeaderMiddleware.kt:26)
	at aws.smithy.kotlin.runtime.io.middleware.DecoratedHandler.call(Middleware.kt:44)
	at aws.smithy.kotlin.runtime.io.middleware.Phase.handle(Phase.kt:67)
	at aws.smithy.kotlin.runtime.io.middleware.DecoratedHandler.call(Middleware.kt:44)
	at aws.smithy.kotlin.runtime.http.middleware.RetryMiddleware.tryAttempt-BWLJW6A(RetryMiddleware.kt:78)
	at aws.smithy.kotlin.runtime.http.middleware.RetryMiddleware.access$tryAttempt-BWLJW6A(RetryMiddleware.kt:31)
	at aws.smithy.kotlin.runtime.http.middleware.RetryMiddleware$handle$$inlined$withSpan$default$1.invokeSuspend(CoroutineContextTraceExt.kt:126)
	at aws.smithy.kotlin.runtime.http.middleware.RetryMiddleware$handle$$inlined$withSpan$default$1.invoke(CoroutineContextTraceExt.kt)
	at aws.smithy.kotlin.runtime.http.middleware.RetryMiddleware$handle$$inlined$withSpan$default$1.invoke(CoroutineContextTraceExt.kt)
	at kotlinx.coroutines.intrinsics.UndispatchedKt.startUndispatchedOrReturn(Undispatched.kt:78)
	at kotlinx.coroutines.BuildersKt__Builders_commonKt.withContext(Builders.common.kt:167)
	at kotlinx.coroutines.BuildersKt.withContext(Unknown Source)
	at aws.smithy.kotlin.runtime.http.middleware.RetryMiddleware.handle(RetryMiddleware.kt:135)
	at aws.smithy.kotlin.runtime.http.middleware.RetryMiddleware.handle(RetryMiddleware.kt:31)
	at aws.smithy.kotlin.runtime.io.middleware.DecoratedHandler.call(Middleware.kt:44)
	at aws.smithy.kotlin.runtime.http.operation.MutateHandler.call(SdkOperationExecution.kt:256)
	at aws.smithy.kotlin.runtime.http.operation.MutateHandler.call(SdkOperationExecution.kt:253)
	at aws.smithy.kotlin.runtime.io.middleware.ModifyRequestMiddleware.handle(ModifyRequest.kt:26)
	at aws.smithy.kotlin.runtime.io.middleware.DecoratedHandler.call(Middleware.kt:44)
	at aws.smithy.kotlin.runtime.io.middleware.ModifyRequestMiddleware.handle(ModifyRequest.kt:26)
	at aws.smithy.kotlin.runtime.io.middleware.DecoratedHandler.call(Middleware.kt:44)
	at aws.smithy.kotlin.runtime.io.middleware.Phase.handle(Phase.kt:67)
	at aws.smithy.kotlin.runtime.io.middleware.DecoratedHandler.call(Middleware.kt:44)
	at aws.smithy.kotlin.runtime.http.operation.SerializeHandler.call(SdkOperationExecution.kt:249)
	at aws.smithy.kotlin.runtime.http.operation.SerializeHandler.call(SdkOperationExecution.kt:231)
	at aws.smithy.kotlin.runtime.http.operation.InitializeHandler.call(SdkOperationExecution.kt:228)
	at aws.smithy.kotlin.runtime.io.middleware.Phase.handle(Phase.kt:63)
	at aws.smithy.kotlin.runtime.io.middleware.DecoratedHandler.call(Middleware.kt:44)
	at aws.smithy.kotlin.runtime.http.operation.OperationHandler.call(SdkOperationExecution.kt:208)
	at aws.smithy.kotlin.runtime.http.operation.OperationHandler.call(SdkOperationExecution.kt:200)
	at aws.smithy.kotlin.runtime.http.operation.SdkHttpOperationKt$execute$$inlined$withSpan$1.invokeSuspend(CoroutineContextTraceExt.kt:126)
	at aws.smithy.kotlin.runtime.http.operation.SdkHttpOperationKt$execute$$inlined$withSpan$1.invoke(CoroutineContextTraceExt.kt)
	at aws.smithy.kotlin.runtime.http.operation.SdkHttpOperationKt$execute$$inlined$withSpan$1.invoke(CoroutineContextTraceExt.kt)
	at kotlinx.coroutines.intrinsics.UndispatchedKt.startUndispatchedOrReturn(Undispatched.kt:78)
	at kotlinx.coroutines.BuildersKt__Builders_commonKt.withContext(Builders.common.kt:167)
	at kotlinx.coroutines.BuildersKt.withContext(Unknown Source)
	at aws.smithy.kotlin.runtime.http.operation.SdkHttpOperationKt.execute(SdkHttpOperation.kt:179)
	at aws.smithy.kotlin.runtime.http.operation.SdkHttpOperationKt.roundTrip(SdkHttpOperation.kt:86)
	at aws.sdk.kotlin.services.s3.DefaultS3Client.putObject(DefaultS3Client.kt:4380)

Steps to Reproduce

I put up a example project to reproduce the issue here: https://github.com/felixscheinost/reproduce-aws-kotlin-s3-bytestream-oneshot-issue

The example requires Docker to be installed as it spins up a MinIO container.

Run ./gradlew :run to reproduce the issue.

Possible Solution

I included a workaround in the example project. Uncomment the lines doing an interceptors.add when constructing the S3Client.

I think it is correct that the payload can't be signed when the stream is not replayable and when the SHA-256 isn't known in advance.

But I think the library could contain a special case similar to my workaround: When the stream is not replayable AND the SHA256 is known, use that for the signature.

(SIdenote: This might be a nice little optimization in general? Even when the stream is replayable? Users of the library might often set checksumSha256)

Context

No response

AWS Kotlin SDK version used

1.0.19

Platform (JVM/JS/Native)

JVM

Operating System and version

macOS 13.6

[lauzadis]

Hi, thanks for the report. It is indeed a bug and we will work on a fix. In the short term, you can work around this by overriding the AwsSigningAttributes.HashSpecification with an interceptor. Here is a full example:

    fun hashSpecificationInterceptor(hash: String) = object : HttpInterceptor {
        override suspend fun modifyBeforeSigning(context: ProtocolRequestInterceptorContext<Any, HttpRequest>): HttpRequest {
            context.executionContext[AwsSigningAttributes.HashSpecification] = HashSpecification.Precalculated(hash)
            return super.modifyBeforeSigning(context)
        }
    }

    val bytes = "abc".toByteArray(Charsets.UTF_8)

    client.withConfig {
        interceptors = mutableListOf(hashSpecificationInterceptor(bytes.sha256().encodeToHex()))
    }.putObject {
        bucket = "test"
        key = "object_non_repeatable_small"
        checksumAlgorithm = ChecksumAlgorithm.Sha256
        contentLength = bytes.size.toLong()
        body = flow {
            emit(bytes)
        }.toByteStream(this@runBlocking)
    }

[lauzadis]

In my local testing, this is fixed by passing the contentLength to the .toByteStream(this@runBlocking) function instead of setting it on the request. A full usage might look like this:

client.putObject {
    bucket = <bucket>
    key = "object_non_repeatable_small"
    checksumAlgorithm = ChecksumAlgorithm.Sha256
    checksumSha256 = <your_checksum> // optional, request succeeds with or without this
//    contentLength = bytes.size.toLong() // don't set contentLength here
    body = flow {
        emit(bytes)
    }.toByteStream(this@runBlocking, contentLength = bytes.size.toLong()) // instead, set contentLength here
}

Does this meet your expectations? The optimization to reuse a provided SHA-256 checksum is a great suggestion and we will continue tracking that.

[z0mb1ek]

i can get it working by setting only checksumAlgorithm = ChecksumAlgorithm.Sha256 without setting byte stream length. But i use Flux<DataBuffer> from spring web flux

[felixscheinost]

Thanks @lauzadis setting contentLength in toByteStream totally makes sense, not sure why I missed that 😅

[github-actions]

⚠️COMMENT VISIBILITY WARNING⚠️

Comments on closed issues are hard for our team to see.
If you need more assistance, please either tag a team member or open a new issue that references this one.
If you wish to keep having a conversation with other community members under this issue feel free to do so.