Enable canned ACL for S3

constants.go

@@ -349,6 +349,9 @@ const (
 	// DisableServerSideEncryption is an optional switch to opt out of SSE in case the provider does not support it
 	DisableServerSideEncryption = "disablesse"
 
+	// ACL is the canned ACL to send to S3
+	ACL = "acl"
+
 	// SSEKMSKey is an optional switch to use an KMS CMK key for S3 SSE.
 	SSEKMSKey = "sse_kms_key"
 

docs/pages/setup/reference/backends.mdx

@@ -209,13 +209,14 @@ running on an EC2 instance with an IAM role.
 
 These optional `GET` parameters control how Teleport interacts with an S3 endpoint, including S3-compatible endpoints.
 
-`s3://bucket/path?region=us-east-1&endpoint=mys3.example.com&insecure=false&disablesse=false`
+`s3://bucket/path?region=us-east-1&endpoint=mys3.example.com&insecure=false&disablesse=false&acl=private`
 
 - `region=us-east-1` - set the Amazon region to use.
 - `endpoint=mys3.example.com` - connect to a custom S3 endpoint.
 - `insecure=true` - set to `true` or `false`. If `true`, TLS will be disabled.
 - `disablesse=true` - set to `true` or `false`. If `true`, S3 server-side encryption will be disabled. If `false`, aws:kms (Key Management Service) will be used for server-side encryption. Other SSE types are not supported at this time.
 - `sse_kms_key=kms_key_id` - If set to a valid AWS KMS CMK key ID all objects uploaded to S3 will be encrypted with this key. Details can be found below.
+- `acl=private` - set the [canned ACL](https://docs.aws.amazon.com/AmazonS3/latest/userguide/acl-overview.html#canned-acl) to use.
 
 ### S3 Server Side Encryption
 
@@ -329,6 +330,47 @@ This is needed to detect session recordings.
 }
 ```
 
+### ACL Example: Transferring Object Ownership
+
+If you are uploading from account `A` to a bucket owned by account `B` and want `A` to retain ownership of the objects, you can take one of two approaches.
+
+#### Without ACL
+
+If ACLs are disabled, object ownership will be set to `Bucket owner enforced` and no action is needed.
+
+#### With ACL
+
+- Set object ownership to `Bucket owner preferred` (under Permissions in the management console).
+- Add `acl=bucket-owner-full-control` to `audit_sessions_uri`.
+
+To enforce the ownership transfer, set `B`'s bucket's policy to only allow uploads that include the `bucket-owner-full-control` canned ACL.
+
+```json
+{
+    "Version": "2012-10-17",
+    "Id": "[id]",
+    "Statement": [
+        {
+            "Sid": "[sid]",
+            "Effect": "Allow",
+            "Principal": {
+                "AWS": "[ARN of account A]"
+            },
+            "Action": "s3:PutObject",
+            "Resource": "arn:aws:s3:::BucketName/*",
+            "Condition": {
+                "StringEquals": {
+                    "s3:x-amz-acl": "bucket-owner-full-control"
+                }
+            }
+        }
+    ]
+}
+
+```
+
+For more information, see the [AWS Docs](https://docs.aws.amazon.com/AmazonS3/latest/userguide/about-object-ownership.html).
+
 ## DynamoDB
 
 <Admonition

lib/events/s3sessions/s3handler.go

@@ -40,6 +40,23 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+// s3AllowedACL is the set of canned ACLs that S3 accepts
+var s3AllowedACL = map[string]struct{}{
+	"private":                   {},
+	"public-read":               {},
+	"public-read-write":         {},
+	"aws-exec-read":             {},
+	"authenticated-read":        {},
+	"bucket-owner-read":         {},
+	"bucket-owner-full-control": {},
+	"log-delivery-write":        {},
+}
+
+func isCannedACL(acl string) bool {
+	_, ok := s3AllowedACL[acl]
+	return ok
+}
+
 // Config is handler configuration
 type Config struct {
 	// Bucket is S3 bucket name
@@ -54,6 +71,8 @@ type Config struct {
 	Insecure bool
 	//DisableServerSideEncryption is an optional switch to opt out of SSE in case the provider does not support it
 	DisableServerSideEncryption bool
+	// ACL is the canned ACL to send to S3
+	ACL string
 	// Session is an optional existing AWS client session
 	Session *awssession.Session
 	// Credentials if supplied are used in tests
@@ -85,6 +104,12 @@ func (s *Config) SetFromURL(in *url.URL, inRegion string) error {
 		}
 		s.DisableServerSideEncryption = disableServerSideEncryption
 	}
+	if acl := in.Query().Get(teleport.ACL); acl != "" {
+		if !isCannedACL(acl) {
+			return trace.BadParameter("failed to parse URI %q flag %q - %q is not a valid canned ACL", in.String(), teleport.ACL, acl)
+		}
+		s.ACL = acl
+	}
 	if val := in.Query().Get(teleport.SSEKMSKey); val != "" {
 		s.SSEKMSKey = val
 	}
@@ -186,6 +211,9 @@ func (h *Handler) Upload(ctx context.Context, sessionID session.ID, reader io.Re
 			uploadInput.SSEKMSKeyId = aws.String(h.Config.SSEKMSKey)
 		}
 	}
+	if h.Config.ACL != "" {
+		uploadInput.ACL = aws.String(h.Config.ACL)
+	}
 	_, err = h.uploader.UploadWithContext(ctx, uploadInput)
 	if err != nil {
 		return "", ConvertS3Error(err)

lib/events/s3sessions/s3handler_test.go

@@ -58,3 +58,29 @@ func TestStreams(t *testing.T) {
 		test.DownloadNotFound(t, handler)
 	})
 }
+
+func TestACL(t *testing.T) {
+	t.Parallel()
+	baseUrl := "s3://mybucket/path"
+	for _, tc := range []struct {
+		desc, acl string
+		isError   bool
+	}{
+		{"no ACL", "", false},
+		{"correct ACL", "bucket-owner-full-control", false},
+		{"incorrect ACL", "something-else", true},
+	} {
+		t.Run(tc.desc, func(t *testing.T) {
+			url, err := url.Parse(fmt.Sprintf("%s?acl=%s", baseUrl, tc.acl))
+			require.Nil(t, err)
+			conf := Config{}
+			err = conf.SetFromURL(url, "")
+			if tc.isError {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, tc.acl, conf.ACL)
+			}
+		})
+	}
+}

lib/events/s3sessions/s3stream.go

@@ -52,6 +52,9 @@ func (h *Handler) CreateUpload(ctx context.Context, sessionID session.ID) (*even
 			input.SSEKMSKeyId = aws.String(h.Config.SSEKMSKey)
 		}
 	}
+	if h.Config.ACL != "" {
+		input.ACL = aws.String(h.Config.ACL)
+	}
 
 	resp, err := h.client.CreateMultipartUploadWithContext(ctx, input)
 	if err != nil {