This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new fe51527f2 S3: Allow custom non-standard reserved characters (#1971)
fe51527f2 is described below
commit fe51527f2b8b5e4558e5d0d21b2c6e2652a5e0fe
Author: liloew <[email protected]>
AuthorDate: Tue Oct 20 03:19:07 2026 +0800
S3: Allow custom non-standard reserved characters (#1971)
* S3: Allow custom non-standard reserved characters
* Fix `Status code: 403 Forbidden` error when using pekko-connectors-s3
with Huawei OBS and Tencent COS.
* refactor: adopt idiomatic Scala syntax and unify imports
* refactor: optimize and unify imports
* S3: add configurable path encoding for compatible services
Add `signing.path-encoded-characters` to support S3-compatible
services whose signature validation differs from AWS S3.
Document the Huawei OBS and Tencent COS compatibility setting, validate
that `/` is not included, and add tests covering configuration parsing,
custom character sets, empty values, and canonical path encoding.
* S3: address review comments on path-encoded-characters
Motivation:
Remaining review comments on the configurable path encoding: the key in
reference.conf made the setting always present, the docs example value
did the opposite of what was described, and the Signer import was out
of order.
Modification:
Comment out `signing.path-encoded-characters` in reference.conf so an
unset value really means "use the RFC 3986 default". Correct the docs
and reference.conf to recommend ":?#[]$'()*+=" for Huawei OBS and
Tencent COS based on the reported test matrix. Fix the Signer import
grouping and minor typos.
Result:
The default-settings spec passes and the documented value matches the
observed vendor behaviour.
Tests:
sbt "s3/testOnly *CanonicalRequestSpec *SignerSpec *S3SettingsSpec"
sbt "s3/mimaReportBinaryIssues"
---------
Co-authored-by: PJ Fanning <[email protected]>
---
docs/src/main/paradox/s3.md | 43 +++++++++++++++
s3/src/main/resources/reference.conf | 19 +++++++
.../connectors/s3/impl/auth/CanonicalRequest.scala | 12 +++--
.../stream/connectors/s3/impl/auth/Signer.scala | 3 +-
.../pekko/stream/connectors/s3/settings.scala | 55 ++++++++++++++-----
.../stream/connectors/s3/S3SettingsSpec.scala | 31 +++++++++++
.../s3/impl/auth/CanonicalRequestSpec.scala | 63 ++++++++++++++++++++++
.../connectors/s3/impl/auth/SignerSpec.scala | 3 ++
8 files changed, 212 insertions(+), 17 deletions(-)
diff --git a/docs/src/main/paradox/s3.md b/docs/src/main/paradox/s3.md
index ef02678f0..74fe0e245 100644
--- a/docs/src/main/paradox/s3.md
+++ b/docs/src/main/paradox/s3.md
@@ -196,6 +196,49 @@ Java
: @@snip [snip](/s3/src/test/java/docs/javadsl/S3Test.java) {
#check-if-bucket-exists }
+## Compatibility
+
+The connector is mainly written for [Amazon S3](https://aws.amazon.com/s3/),
but many other object stores expose an S3-compatible API and can be used with
it.
+However, these compatible services do not always behave exactly like AWS S3,
and a few differences affect how requests have to be signed or sent.
+This section describes the settings that exist for such services and the
behaviour that is known so far.
+It is based on testing against specific services and not on a guarantee from
the connector or the vendors,
+so if you hit a difference that is not covered here, please report it.
+
+### Services that sign the path differently (Huawei OBS, Tencent COS and
others)
+
+AWS S3 re-encodes the request path before checking the signature. Some
S3-compatible services do not,
+and reject requests with `403 Forbidden` (`SignatureDoesNotMatch`) when an
object key contains certain
+characters, for example `@`, `;`, `,`, `&` or `!`. Plain keys are not affected.
+
+By default the [RFC 3986 reserved
characters](https://tools.ietf.org/html/rfc3986#section-2.2) are
+percent-encoded in the path when the signature is calculated:
+
+```
+:?#[]@!$&'()*+,;=
+```
+
+`signing.path-encoded-characters` replaces that set. `/` is deliberately not
in it: it must stay literal
+in the signed path, so a value containing `/` is rejected.
+
+In testing against Huawei OBS and Tencent COS, keys containing `@`, `!`, `&`,
`,` or `;` only worked when
+those characters were removed from the set, while keys containing `=` (Tencent
COS) and `+` (both services)
+only worked when they stayed in it. Based on those results, the following
value (the default without `@!&,;`)
+is the one to try for both services:
+
+```hocon
+pekko.connectors.s3 {
+ signing.path-encoded-characters = ":?#[]$'()*+="
+}
+```
+
+Notes:
+
+* The value must be a quoted string. Unquoted, HOCON treats `#` as the start
of a comment and rejects `[`, `]` and `$`.
+* This is based on observed behaviour, not on vendor documentation. Characters
that were not tested
+ (`$`, `'`, `(`, `)`, `*` and `:`) are left as in the default.
+* Programmatic equivalent:
`S3Settings(...).withReservedCharacters(":?#[]$'()*+=")`.
+
+
## Running the example code
The code in this guide is part of runnable tests of this project. You are
welcome to edit the code and run it in sbt.
diff --git a/s3/src/main/resources/reference.conf
b/s3/src/main/resources/reference.conf
index 7045c80ed..c95cbe910 100644
--- a/s3/src/main/resources/reference.conf
+++ b/s3/src/main/resources/reference.conf
@@ -342,4 +342,23 @@ pekko.connectors.s3 {
PutBucketVersioning = []
GetBucketVersioning = []
}
+
+ # Characters that are percent-encoded in the request path when the AWS
+ # Signature V4 canonical request is calculated. When this setting is not set,
+ # the RFC 3986 reserved characters (excluding `/`) are encoded, as AWS S3
expects:
+ # https://tools.ietf.org/html/rfc3986#section-2.2
+ #
+ # Only change this for S3-compatible services (e.g. Huawei OBS, Tencent COS)
whose
+ # signature validation encodes the path differently from AWS S3. The symptom
is
+ # `403 Forbidden` / `SignatureDoesNotMatch` for object keys containing some
of
+ # these characters, while plain keys work. For Huawei OBS and Tencent COS,
try
+ # ":?#[]$'()*+=" (the default without `@`, `!`, `&`, `,` and `;`); see the S3
+ # documentation for the observed behaviour.
+ #
+ # `/` must stay literal in the signed path, so a value containing `/` is
rejected.
+ #
+ # The value MUST be a quoted string. Unquoted, HOCON treats `#` as the start
of
+ # a comment and rejects `[`, `]` and `$`, so the override either fails to
parse
+ # or is silently cut short.
+ # signing.path-encoded-characters = ":?#[]@!$&'()*+,;="
}
diff --git
a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/auth/CanonicalRequest.scala
b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/auth/CanonicalRequest.scala
index ad7212e78..5bf196a2d 100644
---
a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/auth/CanonicalRequest.scala
+++
b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/auth/CanonicalRequest.scala
@@ -18,6 +18,7 @@ import pekko.annotation.InternalApi
import pekko.http.scaladsl.model.Uri.{ Path, Query }
import pekko.http.scaladsl.model.headers.{ `Raw-Request-URI`,
`Timeout-Access`, `Tls-Session-Info`, `X-Forwarded-For` }
import pekko.http.scaladsl.model.{ HttpHeader, HttpRequest }
+import pekko.stream.connectors.s3.S3Settings
// Documentation:
http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
@InternalApi private[impl] final case class CanonicalRequest(
@@ -38,7 +39,7 @@ import pekko.http.scaladsl.model.{ HttpHeader, HttpRequest }
`Timeout-Access`.lowercaseName,
`Tls-Session-Info`.lowercaseName)
- def from(request: HttpRequest): CanonicalRequest = {
+ def from(request: HttpRequest)(implicit conf: S3Settings): CanonicalRequest
= {
val hashedBody =
request.headers
.collectFirst { case header if header.is("x-amz-content-sha256") =>
header.value }
@@ -63,8 +64,11 @@ import pekko.http.scaladsl.model.{ HttpHeader, HttpRequest }
// Excludes "/" as it is an exception according to spec.
val reservedCharacters: String = ":?#[]@!$&'()*+,;="
- def isReservedCharacter(c: Char): Boolean =
- reservedCharacters.contains(c)
+ def isReservedCharacter(c: Char)(implicit conf: S3Settings): Boolean =
+ conf.reservedCharacters match {
+ case Some(chars) => chars.contains(c)
+ case None => reservedCharacters.contains(c)
+ }
def canonicalQueryString(query: Query): String = {
def uriEncode(s: String): String = s.flatMap {
@@ -96,7 +100,7 @@ import pekko.http.scaladsl.model.{ HttpHeader, HttpRequest }
def signedHeadersString(headers: Seq[HttpHeader]): String =
headers.map(_.lowercaseName).distinct.sorted.mkString(";")
- def pathEncode(path: Path): String =
+ def pathEncode(path: Path)(implicit conf: S3Settings): String =
if (path.isEmpty) "/"
else {
path.toString.flatMap {
diff --git
a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/auth/Signer.scala
b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/auth/Signer.scala
index ac26cb15a..0eac2d276 100644
---
a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/auth/Signer.scala
+++
b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/auth/Signer.scala
@@ -22,6 +22,7 @@ import pekko.NotUsed
import pekko.annotation.InternalApi
import pekko.http.scaladsl.model.headers.RawHeader
import pekko.http.scaladsl.model.{ HttpHeader, HttpRequest }
+import pekko.stream.connectors.s3.S3Settings
import pekko.stream.scaladsl.Source
@InternalApi private[impl] object Signer {
@@ -29,7 +30,7 @@ import pekko.stream.scaladsl.Source
def signedRequest(request: HttpRequest,
key: SigningKey,
- signAnonymousRequests: Boolean): Source[HttpRequest, NotUsed] =
+ signAnonymousRequests: Boolean)(implicit conf: S3Settings):
Source[HttpRequest, NotUsed] =
if (!signAnonymousRequests && key.anonymous) Source.single(request)
else {
val hashedBody = request.entity.dataBytes.via(digest()).map(hash =>
encodeHex(hash))
diff --git
a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/settings.scala
b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/settings.scala
index 691d69183..1b4b9c44f 100644
--- a/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/settings.scala
+++ b/s3/src/main/scala/org/apache/pekko/stream/connectors/s3/settings.scala
@@ -365,9 +365,13 @@ final class S3Settings private (
val retrySettings: RetrySettings,
val multipartUploadSettings: MultipartUploadSettings,
val signAnonymousRequests: Boolean,
- val allowedHeaders: Map[String, Set[String]]
+ val allowedHeaders: Map[String, Set[String]],
+ val reservedCharacters: Option[String]
) {
+ /** Java API */
+ def getReservedCharacters: java.util.Optional[String] =
reservedCharacters.toJava
+
/** Java API */
def getBufferType: BufferType = bufferType
@@ -427,6 +431,12 @@ final class S3Settings private (
def withSignAnonymousRequests(value: Boolean): S3Settings =
if (signAnonymousRequests == value) this else copy(signAnonymousRequests =
value)
+ def withReservedCharacters(value: String): S3Settings =
+ reservedCharacters match {
+ case Some(`value`) => this
+ case _ => copy(reservedCharacters =
Option(S3Settings.validateReservedCharacters(value)))
+ }
+
private[s3] val concreteAllowedHeaders: Map[S3Request, Set[String]] = {
allowedHeaders.foldLeft(Map.empty[S3Request, Set[String]]) {
case (acc, (header, value)) =>
@@ -450,7 +460,8 @@ final class S3Settings private (
retrySettings: RetrySettings = retrySettings,
multipartUploadSettings: MultipartUploadSettings =
multipartUploadSettings,
signAnonymousRequests: Boolean = signAnonymousRequests,
- allowedHeaders: Map[String, Set[String]] = allowedHeaders
+ allowedHeaders: Map[String, Set[String]] = allowedHeaders,
+ reservedCharacters: Option[String] = reservedCharacters
): S3Settings = new S3Settings(
bufferType,
credentialsProvider,
@@ -463,7 +474,8 @@ final class S3Settings private (
retrySettings,
multipartUploadSettings,
signAnonymousRequests,
- allowedHeaders
+ allowedHeaders,
+ reservedCharacters
)
override def toString: String =
@@ -475,16 +487,17 @@ final class S3Settings private (
s"endpointUrl=$endpointUrl," +
s"listBucketApiVersion=$listBucketApiVersion," +
s"forwardProxy=$forwardProxy," +
- s"validateObjectKey=$validateObjectKey" +
- s"retrySettings=$retrySettings" +
- s"multipartUploadSettings=$multipartUploadSettings" +
- s"signAnonymousRequests=$signAnonymousRequests" +
+ s"validateObjectKey=$validateObjectKey," +
+ s"retrySettings=$retrySettings," +
+ s"multipartUploadSettings=$multipartUploadSettings," +
+ s"signAnonymousRequests=$signAnonymousRequests," +
s"allowedHeaders=${
val entries = allowedHeaders.toSeq.sortBy(_._1).map { case (key,
values) =>
s"$key -> Set(${values.mkString(", ")})"
}.mkString(", ")
s"Map($entries)"
- }" +
+ }," +
+ s"reservedCharacters=$reservedCharacters" +
")"
override def equals(other: Any): Boolean = other match {
@@ -500,7 +513,8 @@ final class S3Settings private (
Objects.equals(this.retrySettings, that.retrySettings) &&
Objects.equals(this.multipartUploadSettings, multipartUploadSettings) &&
this.signAnonymousRequests == that.signAnonymousRequests &&
- this.allowedHeaders == that.allowedHeaders
+ this.allowedHeaders == that.allowedHeaders &&
+ Objects.equals(this.reservedCharacters, that.reservedCharacters)
case _ => false
}
@@ -516,13 +530,23 @@ final class S3Settings private (
Boolean.box(validateObjectKey),
retrySettings,
multipartUploadSettings,
- Boolean.box(signAnonymousRequests))
+ Boolean.box(signAnonymousRequests),
+ allowedHeaders,
+ reservedCharacters)
}
object S3Settings {
private final val log = LoggerFactory.getLogger(getClass)
val ConfigPath = "pekko.connectors.s3"
+ private val PathEncodedCharacterPath = "signing.path-encoded-characters"
+
+ private def validateReservedCharacters(value: String): String = {
+ require(!value.contains("/"),
+ s"'$PathEncodedCharacterPath' must not contain '/': it has to stay
literal in the signed path. Got: [$value]")
+ value
+ }
+
/**
* Reads from the given config.
*/
@@ -689,6 +713,11 @@ object S3Settings {
k -> (v ++ allowedHeadersBase.getOrElse(k, Set.empty[String]))
}
+ val reservedCharacters =
+ Option.when(c.hasPath(PathEncodedCharacterPath)) {
+ validateReservedCharacters(c.getString(PathEncodedCharacterPath))
+ }
+
new S3Settings(
bufferType,
credentialsProvider,
@@ -701,7 +730,8 @@ object S3Settings {
retrySettings,
multipartUploadSettings,
signAnonymousRequests,
- finalAllowedHeaders
+ finalAllowedHeaders,
+ reservedCharacters
)
}
@@ -743,7 +773,8 @@ object S3Settings {
RetrySettings.default,
MultipartUploadSettings(RetrySettings.default),
signAnonymousRequests = true,
- allowedHeaders = allowedHeaders
+ allowedHeaders = allowedHeaders,
+ reservedCharacters = None
)
/** Java API */
diff --git
a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/S3SettingsSpec.scala
b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/S3SettingsSpec.scala
index 935fc0cbe..075bfb7f0 100644
---
a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/S3SettingsSpec.scala
+++
b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/S3SettingsSpec.scala
@@ -335,4 +335,35 @@ class S3SettingsSpec extends S3WireMockBase with
S3ClientIntegrationSpec with Op
}
}
+ it should "have no signing.path-encoded-characters configured by default" in
{
+ val settings = mkSettings("")
+ settings.reservedCharacters shouldBe empty
+ settings.getReservedCharacters.isPresent shouldBe false
+ }
+
+ it should "parse signing.path-encoded-characters" in {
+ val settings = mkSettings(
+ """
+ |signing.path-encoded-characters = "!()"
+ |""".stripMargin)
+ settings.reservedCharacters shouldBe Some("!()")
+ settings.getReservedCharacters.get shouldBe "!()"
+ }
+
+ it should "parse an empty signing.path-encoded-characters value" in {
+ val settings = mkSettings(
+ """
+ |signing.path-encoded-characters = ""
+ |""".stripMargin)
+ settings.reservedCharacters shouldBe Some("")
+ }
+
+ it should "reject signing.path-encoded-characters containing '/'" in {
+ an[IllegalArgumentException] should be thrownBy
+ mkSettings(
+ """
+ |signing.path-encoded-characters = "/!"
+ |""".stripMargin)
+ }
+
}
diff --git
a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/auth/CanonicalRequestSpec.scala
b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/auth/CanonicalRequestSpec.scala
index f17877cd0..e29e06a24 100644
---
a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/auth/CanonicalRequestSpec.scala
+++
b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/auth/CanonicalRequestSpec.scala
@@ -17,11 +17,31 @@ import org.apache.pekko
import pekko.http.scaladsl.model.Uri.Query
import pekko.http.scaladsl.model._
import pekko.http.scaladsl.model.headers._
+import pekko.stream.connectors.s3.{ ApiVersion, BufferType, MemoryBufferType,
S3Settings }
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
+import software.amazon.awssdk.auth.credentials.{ AnonymousCredentialsProvider,
AwsCredentialsProvider }
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.regions.providers.AwsRegionProvider
class CanonicalRequestSpec extends AnyFlatSpec with Matchers {
+ def getSettings(
+ bufferType: BufferType = MemoryBufferType,
+ awsCredentials: AwsCredentialsProvider =
AnonymousCredentialsProvider.create(),
+ s3Region: Region = Region.US_EAST_1,
+ listBucketApiVersion: ApiVersion = ApiVersion.ListBucketVersion2,
+ reservedCharacters: Option[String] = None) = {
+ val regionProvider = new AwsRegionProvider {
+ def getRegion = s3Region
+ }
+
+ val base = S3Settings(bufferType, awsCredentials, regionProvider,
listBucketApiVersion, Map.empty)
+ reservedCharacters.fold(base)(base.withReservedCharacters)
+ }
+
+ implicit val settings: S3Settings = getSettings()
+
it should "correctly build a canonicalString for eu-central-1" in {
val req = HttpRequest(
HttpMethods.GET,
@@ -153,6 +173,49 @@ class CanonicalRequestSpec extends AnyFlatSpec with
Matchers {
}
}
+ private def pathRequest(path: Uri.Path): HttpRequest =
+ HttpRequest(
+ HttpMethods.GET,
+ Uri(s"https://mytestbucket.s3.us-east-1.amazonaws.com";)
+ .withPath(path)).withHeaders(
+ RawHeader("x-amz-content-sha256", "testhash"),
+ `Content-Type`(ContentTypes.`application/json`))
+
+ private def expectedCanonical(encodedPath: String): String =
+ s"""GET
+ |$encodedPath
+ |
+ |content-type:application/json
+ |x-amz-content-sha256:testhash
+ |
+ |content-type;x-amz-content-sha256
+ |testhash""".stripMargin
+
+ it should "encode the default reserved characters when none are configured"
in {
+ val req = pathRequest(Uri.Path.Empty / "file-(1)!.txt")
+ CanonicalRequest.from(req).canonicalString should equal(
+ expectedCanonical("/file-%281%29%21.txt")
+ )
+ }
+
+ it should "only encode the configured reserved characters, leaving other
default ones unencoded" in {
+ // '(' and ')' are in the default set, but not in the custom one
+ val customSettings = getSettings(reservedCharacters = Some("!"))
+ val req = pathRequest(Uri.Path.Empty / "file-(1)!.txt")
+ val canonicalRequest = CanonicalRequest.from(req)(customSettings)
+ canonicalRequest.canonicalString should
equal(expectedCanonical("/file-(1)%21.txt"))
+
+ // sanity check: the same request is encoded differently with the default
settings
+ (CanonicalRequest.from(req).canonicalString should
not).equal(canonicalRequest.canonicalString)
+ }
+
+ it should "not encode any path characters when reserved character is empty"
in {
+ val customSettings = getSettings(reservedCharacters = Some(""))
+ val req = pathRequest(Uri.Path.Empty / "file-(1)!.txt")
+
+ CanonicalRequest.from(req)(customSettings).canonicalString should
equal(expectedCanonical("/file-(1)!.txt"))
+ }
+
it should "correctly build a canonicalString when synthetic headers are
present" in {
val req = HttpRequest(
HttpMethods.GET,
diff --git
a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/auth/SignerSpec.scala
b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/auth/SignerSpec.scala
index 8463aa227..94133a4e9 100644
---
a/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/auth/SignerSpec.scala
+++
b/s3/src/test/scala/org/apache/pekko/stream/connectors/s3/impl/auth/SignerSpec.scala
@@ -19,6 +19,7 @@ import org.apache.pekko
import pekko.actor.ActorSystem
import pekko.http.scaladsl.model.headers.{ `Raw-Request-URI`, Host, RawHeader }
import pekko.http.scaladsl.model.{ HttpMethods, HttpRequest }
+import pekko.stream.connectors.s3.S3Settings
import pekko.stream.connectors.testkit.scaladsl.LogCapturing
import pekko.stream.scaladsl.Sink
import pekko.testkit.TestKit
@@ -45,6 +46,8 @@ class SignerSpec(_system: ActorSystem)
implicit val defaultPatience: PatienceConfig =
PatienceConfig(timeout = Span(2, Seconds), interval = Span(5, Millis))
+ implicit val settings: S3Settings = S3Settings()
+
val credentials = StaticCredentialsProvider.create(
AwsBasicCredentials.create("AKIDEXAMPLE",
"wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
