Skip to content

AWS S3: Handle Illegal Headers in Copy Part #1246

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 4 commits into
base: main
Choose a base branch
from
Draft

AWS S3: Handle Illegal Headers in Copy Part #1246

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ad49f3a
Minimal update to demonstrate problem
Oct 16, 2025
272bc2a
update to structure discussed in PR
Oct 17, 2025
37c46d6
add functions to remove mima issues
Oct 20, 2025
e5f5c63
fix broken test
Oct 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions s3/src/main/resources/reference.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,4 +125,20 @@ pekko.connectors.s3 {

# Add signature headers to requests when aws.credentials.provider is anon
sign-anonymous-requests = true

additional-allowed-headers {
GetObject = []
HeadObject = []
PutObject = []
InitiateMultipartUpload = []
UploadPart = []
CopyPart = []
DeleteObject = []
ListBucket = []
MakeBucket = []
DeleteBucket = []
CheckBucket = []
PutBucketVersioning = []
GetBucketVersioning = []
}
}
6 changes: 4 additions & 2 deletions s3/src/main/scala/org/apache/pekko/stream/connectors/s3/S3Headers.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,8 +77,10 @@ final class S3Headers private (val cannedAcl: Option[CannedAcl] = None,
RawHeader(header._1, header._2)
}

@InternalApi private[s3] def headersFor(request: S3Request) =
headers ++ serverSideEncryption.toIndexedSeq.flatMap(_.headersFor(request))
@InternalApi private[s3] def headersFor(request: S3Request)(implicit s3Settings: S3Settings) =
headers.filter(header =>
request.allowedHeaders.contains(header.name()) || s3Settings.concereteAdditionalAllowedHeaders.getOrElse(request,
Set.empty).contains(header.name())) ++ serverSideEncryption.toIndexedSeq.flatMap(_.headersFor(request))

def withCannedAcl(cannedAcl: CannedAcl): S3Headers = copy(cannedAcl = Some(cannedAcl))
def withMetaHeaders(metaHeaders: MetaHeaders): S3Headers = copy(metaHeaders = Some(metaHeaders))
Expand Down
281 changes: 267 additions & 14 deletions s3/src/main/scala/org/apache/pekko/stream/connectors/s3/impl/S3Request.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,69 +18,322 @@ import org.apache.pekko.annotation.InternalApi
/**
* Internal Api
*/
@InternalApi private[s3] sealed trait S3Request
@InternalApi private[s3] sealed trait S3Request {
def allowedHeaders: Set[String]
}

/**
* Internal Api
*/
@InternalApi private[s3] case object GetObject extends S3Request
@InternalApi private[s3] object S3Request {
def fromString(str: String): Option[S3Request] = {
str match {
case "GetObject" => Some(GetObject)
case "HeadObject" => Some(HeadObject)
case "PutObject" => Some(PutObject)
case "InitiateMultipartUpload" => Some(InitiateMultipartUpload)
case "UploadPart" => Some(UploadPart)
case "CopyPart" => Some(CopyPart)
case "DeleteObject" => Some(DeleteObject)
case "ListBucket" => Some(ListBucket)
case "MakeBucket" => Some(MakeBucket)
case "DeleteBucket" => Some(DeleteBucket)
case "CheckBucket" => Some(CheckBucket)
case _ => None
}
}

val allRequests: List[S3Request] = List(GetObject, HeadObject, PutObject, InitiateMultipartUpload, UploadPart,
CopyPart, DeleteObject, ListBucket, MakeBucket, DeleteBucket, CheckBucket)
}

/**
* Internal Api
*/
@InternalApi private[s3] case object HeadObject extends S3Request
@InternalApi private[s3] case object GetObject extends S3Request {

override def allowedHeaders: Set[String] = Set(
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll add links to where in the docs I sourced this from

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll also double check these when I move this out of draft

"Host",
"If-Match",
"If-Modified-Since",
"If-None-Match",
"If-Unmodified-Since",
"Range",
"x-amz-server-side-encryption-customer-algorithm",
"x-amz-server-side-encryption-customer-key",
"x-amz-server-side-encryption-customer-key-MD5",
"x-amz-request-payer",
"x-amz-expected-bucket-owner",
"x-amz-checksum-mode"
)

override def toString() = "GetObject"
}

/**
* Internal Api
*/
@InternalApi private[s3] case object PutObject extends S3Request
@InternalApi private[s3] case object HeadObject extends S3Request {

override def allowedHeaders: Set[String] = Set(
"Host",
"If-Match",
"If-Modified-Since",
"If-None-Match",
"If-Unmodified-Since",
"Range",
"x-amz-server-side-encryption-customer-algorithm",
"x-amz-server-side-encryption-customer-key",
"x-amz-server-side-encryption-customer-key-MD5",
"x-amz-request-payer",
"x-amz-expected-bucket-owner",
"x-amz-checksum-mode"
)

override def toString() = "HeadObject"
}

/**
* Internal Api
*/
@InternalApi private[s3] case object InitiateMultipartUpload extends S3Request
@InternalApi private[s3] case object PutObject extends S3Request {

override def allowedHeaders: Set[String] = Set(
"Host",
"x-amz-acl",
"Cache-Control",
"Content-Disposition",
"Content-Encoding",
"Content-Language",
"Content-Length",
"Content-MD5",
"Content-Type",
"x-amz-sdk-checksum-algorithm",
"x-amz-checksum-crc32",
"x-amz-checksum-crc32c",
"x-amz-checksum-crc64nvme",
"x-amz-checksum-sha1",
"x-amz-checksum-sha256",
"Expires",
"If-Match",
"If-None-Match",
"x-amz-grant-full-control",
"x-amz-grant-read",
"x-amz-grant-read-acp",
"x-amz-grant-write-acp",
"x-amz-write-offset-bytes",
"x-amz-server-side-encryption",
"x-amz-storage-class",
"x-amz-website-redirect-location",
"x-amz-server-side-encryption-customer-algorithm",
"x-amz-server-side-encryption-customer-key",
"x-amz-server-side-encryption-customer-key-MD5",
"x-amz-server-side-encryption-aws-kms-key-id",
"x-amz-server-side-encryption-context",
"x-amz-server-side-encryption-bucket-key-enabled",
"x-amz-request-payer",
"x-amz-tagging",
"x-amz-object-lock-mode",
"x-amz-object-lock-retain-until-date",
"x-amz-object-lock-legal-hold",
"x-amz-expected-bucket-owner"
)

override def toString() = "PutObject"
}

/**
* Internal Api
*/
@InternalApi private[s3] case object UploadPart extends S3Request
@InternalApi private[s3] case object InitiateMultipartUpload extends S3Request {

override def allowedHeaders: Set[String] = Set(
"Host",
"x-amz-acl",
"Cache-Control",
"Content-Disposition",
"Content-Encoding",
"Content-Language",
"Content-Type",
"Expires",
"x-amz-grant-full-control",
"x-amz-grant-read",
"x-amz-grant-read-acp",
"x-amz-grant-write-acp",
"x-amz-server-side-encryption",
"x-amz-storage-class",
"x-amz-website-redirect-location",
"x-amz-server-side-encryption-customer-algorithm",
"x-amz-server-side-encryption-customer-key",
"x-amz-server-side-encryption-customer-key-MD5",
"x-amz-server-side-encryption-aws-kms-key-id",
"x-amz-server-side-encryption-context",
"x-amz-server-side-encryption-bucket-key-enabled",
"x-amz-request-payer",
"x-amz-tagging",
"x-amz-object-lock-mode",
"x-amz-object-lock-retain-until-date",
"x-amz-object-lock-legal-hold",
"x-amz-expected-bucket-owner",
"x-amz-checksum-algorithm",
"x-amz-checksum-type"
)

override def toString() = "InitiateMultipartUpload"
}

/**
* Internal Api
*/
@InternalApi private[s3] case object CopyPart extends S3Request
@InternalApi private[s3] case object UploadPart extends S3Request {

override def allowedHeaders: Set[String] = Set(
"Host",
"Content-Length",
"Content-MD5",
"x-amz-sdk-checksum-algorithm",
"x-amz-checksum-crc32",
"x-amz-checksum-crc32c",
"x-amz-checksum-crc64nvme",
"x-amz-checksum-sha1",
"x-amz-checksum-sha256",
"x-amz-server-side-encryption-customer-algorithm",
"x-amz-server-side-encryption-customer-key",
"x-amz-server-side-encryption-customer-key-MD5",
"x-amz-request-payer",
"x-amz-expected-bucket-owner"
)

override def toString() = "UploadPart"
}

/**
* Internal Api
*/
@InternalApi private[s3] case object DeleteObject extends S3Request
@InternalApi private[s3] case object CopyPart extends S3Request {

override def allowedHeaders: Set[String] = Set(
"Host",
"x-amz-copy-source",
"x-amz-copy-source-if-match",
"x-amz-copy-source-if-modified-since",
"x-amz-copy-source-if-none-match",
"x-amz-copy-source-if-unmodified-since",
"x-amz-copy-source-range",
"x-amz-server-side-encryption-customer-algorithm",
"x-amz-server-side-encryption-customer-key",
"x-amz-server-side-encryption-customer-key-MD5",
"x-amz-copy-source-server-side-encryption-customer-algorithm",
"x-amz-copy-source-server-side-encryption-customer-key",
"x-amz-copy-source-server-side-encryption-customer-key-MD5",
"x-amz-request-payer",
"x-amz-expected-bucket-owner",
"x-amz-source-expected-bucket-owner"
)

override def toString() = "CopyPart"
}

/**
* Internal Api
*/
@InternalApi private[s3] case object ListBucket extends S3Request
@InternalApi private[s3] case object DeleteObject extends S3Request {

override def allowedHeaders: Set[String] = Set(
"Host",
"x-amz-mfa",
"x-amz-request-payer",
"x-amz-bypass-governance-retention",
"x-amz-expected-bucket-owner",
"If-Match",
"x-amz-if-match-last-modified-time",
"x-amz-if-match-size"
)

override def toString() = "DeleteObject"
}

/**
* Internal Api
*/
@InternalApi private[s3] case object MakeBucket extends S3Request
@InternalApi private[s3] case object ListBucket extends S3Request {

override def allowedHeaders: Set[String] = Set("Host")

override def toString() = "ListBucket"
}

/**
* Internal Api
*/
@InternalApi private[s3] case object DeleteBucket extends S3Request
@InternalApi private[s3] case object MakeBucket extends S3Request {

override def allowedHeaders: Set[String] = Set(
"Host",
"x-amz-acl",
"x-amz-grant-full-control",
"x-amz-grant-read",
"x-amz-grant-read-acp",
"x-amz-grant-write",
"x-amz-grant-write-acp",
"x-amz-bucket-object-lock-enabled",
"x-amz-object-ownership"
)

override def toString() = "MakeBucket"
}

/**
* Internal Api
*/
@InternalApi private[s3] case object CheckBucket extends S3Request
@InternalApi private[s3] case object DeleteBucket extends S3Request {

override def allowedHeaders: Set[String] = Set(
"Host",
"x-amz-expected-bucket-owner"
)

override def toString() = "DeleteBucket"
}

/**
* Internal Api
*/
@InternalApi private[s3] case object PutBucketVersioning extends S3Request
@InternalApi private[s3] case object CheckBucket extends S3Request {

override def allowedHeaders: Set[String] = Set(
"Host",
"x-amz-expected-bucket-owner"
)

override def toString() = "CheckBucket"
}

/**
* Internal Api
*/
@InternalApi private[s3] case object GetBucketVersioning extends S3Request
@InternalApi private[s3] case object PutBucketVersioning extends S3Request {

override def allowedHeaders: Set[String] = Set(
"Host",
"Content-MD5",
"x-amz-sdk-checksum-algorithm",
"x-amz-mfa",
"x-amz-expected-bucket-owner"
)

override def toString() = "PutBucketVersioning"
}

/**
* Internal Api
*/
@InternalApi private[s3] case object GetBucketVersioning extends S3Request {

override def allowedHeaders: Set[String] = Set(
"Host",
"x-amz-expected-bucket-owner"
)

override def toString() = "GetBucketVersioning"
}
Loading