Add SSE-C option on native-filesystem security mapping

docs/src/main/sphinx/object-storage/file-system-s3.md

@@ -174,6 +174,9 @@ The security mapping must provide one or more configuration settings:
 - `kmsKeyId`: ID of KMS-managed key to be used for client-side encryption.
 - `allowedKmsKeyIds`: KMS-managed key IDs that are allowed to be specified as an extra
   credential. If list cotains `*`, then any key can be specified via extra credential.
+- `sseCustomerKey`: The customer provided key (SSE-C) for server-side encryption.
+- `allowedSseCustomerKey`: The SSE-C keys that are allowed to be specified as an extra
+  credential. If list cotains `*`, then any key can be specified via extra credential.
 - `endpoint`: The S3 storage endpoint server. This optional property can be used
   to override S3 endpoints on a per-bucket basis.
 - `region`: The S3 region to connect to. This optional property can be used
@@ -262,6 +265,8 @@ Example JSON configuration:
   - The name of the *extra credential* used to provide the IAM role.
 * - `s3.security-mapping.kms-key-id-credential-name`
   - The name of the *extra credential* used to provide the KMS-managed key ID.
+* - `s3.security-mapping.sse-customer-key-credential-name`
+  - The name of the *extra credential* used to provide the server-side encryption with customer-provided keys (SSE-C).
 * - `s3.security-mapping.refresh-period`
   - How often to refresh the security mapping configuration, specified as a
     {ref}`prop-type-duration`. By default, the configuration is not refreshed.

lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3Context.java

@@ -25,6 +25,7 @@
 import java.util.Optional;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static io.trino.filesystem.s3.S3FileSystemConfig.S3SseType.CUSTOMER;
 import static io.trino.filesystem.s3.S3FileSystemConfig.S3SseType.KMS;
 import static io.trino.filesystem.s3.S3FileSystemConstants.EXTRA_CREDENTIALS_ACCESS_KEY_PROPERTY;
 import static io.trino.filesystem.s3.S3FileSystemConstants.EXTRA_CREDENTIALS_SECRET_KEY_PROPERTY;
@@ -70,6 +71,11 @@ public S3Context withCredentials(ConnectorIdentity identity)
         return this;
     }
 
+    public S3Context withSseCustomerKey(String key)
+    {
+        return new S3Context(partSize, requesterPays, S3SseContext.withSseCustomerKey(key), credentialsProviderOverride, cannedAcl, exclusiveWriteSupported);
+    }
+
     public S3Context withCredentialsProviderOverride(AwsCredentialsProvider credentialsProviderOverride)
     {
         return new S3Context(
@@ -109,5 +115,10 @@ public static S3SseContext withKmsKeyId(String kmsKeyId)
         {
             return new S3SseContext(KMS, Optional.ofNullable(kmsKeyId), Optional.empty());
         }
+
+        public static S3SseContext withSseCustomerKey(String key)
+        {
+            return new S3SseContext(CUSTOMER, Optional.empty(), Optional.ofNullable(key).map(S3SseCustomerKey::onAes256));
+        }
     }
 }

lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3FileSystemLoader.java

@@ -46,6 +46,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.function.Function;
 
+import static com.google.common.base.Preconditions.checkState;
 import static io.airlift.concurrent.Threads.daemonThreadsNamed;
 import static io.trino.filesystem.s3.S3FileSystemConfig.RetryMode.getRetryStrategy;
 import static java.lang.Math.toIntExact;
@@ -108,9 +109,14 @@ public TrinoFileSystemFactory apply(Location location)
             S3Context context = this.context.withCredentials(identity);
 
             if (mapping.isPresent() && mapping.get().kmsKeyId().isPresent()) {
+                checkState(mapping.get().sseCustomerKey().isEmpty(), "Both SSE-C and KMS-managed keys cannot be used at the same time");
                 context = context.withKmsKeyId(mapping.get().kmsKeyId().get());
             }
 
+            if (mapping.isPresent() && mapping.get().sseCustomerKey().isPresent()) {
+                context = context.withSseCustomerKey(mapping.get().sseCustomerKey().get());
+            }
+
             return new S3FileSystem(uploadExecutor, client, preSigner, context);
         };
     }

lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMapping.java

@@ -42,6 +42,8 @@ public final class S3SecurityMapping
     private final Set<String> allowedIamRoles;
     private final Optional<String> kmsKeyId;
     private final Set<String> allowedKmsKeyIds;
+    private final Optional<String> sseCustomerKey;
+    private final Set<String> allowedSseCustomerKeys;
     private final Optional<AwsCredentials> credentials;
     private final boolean useClusterDefault;
     private final Optional<String> endpoint;
@@ -57,6 +59,8 @@ public S3SecurityMapping(
             @JsonProperty("allowedIamRoles") Optional<List<String>> allowedIamRoles,
             @JsonProperty("kmsKeyId") Optional<String> kmsKeyId,
             @JsonProperty("allowedKmsKeyIds") Optional<List<String>> allowedKmsKeyIds,
+            @JsonProperty("sseCustomerKey") Optional<String> sseCustomerKey,
+            @JsonProperty("allowedSseCustomerKeys") Optional<List<String>> allowedSseCustomerKeys,
             @JsonProperty("accessKey") Optional<String> accessKey,
             @JsonProperty("secretKey") Optional<String> secretKey,
             @JsonProperty("useClusterDefault") Optional<Boolean> useClusterDefault,
@@ -86,6 +90,10 @@ public S3SecurityMapping(
 
         this.allowedKmsKeyIds = ImmutableSet.copyOf(allowedKmsKeyIds.orElse(ImmutableList.of()));
 
+        this.sseCustomerKey = requireNonNull(sseCustomerKey, "sseCustomerKey is null");
+
+        this.allowedSseCustomerKeys = allowedSseCustomerKeys.map(ImmutableSet::copyOf).orElse(ImmutableSet.of());
+
         requireNonNull(accessKey, "accessKey is null");
         requireNonNull(secretKey, "secretKey is null");
         checkArgument(accessKey.isPresent() == secretKey.isPresent(), "accessKey and secretKey must be provided together");
@@ -96,6 +104,8 @@ public S3SecurityMapping(
         checkArgument(this.useClusterDefault != roleOrCredentialsArePresent, "must either allow useClusterDefault role or provide role and/or credentials");
 
         checkArgument(!this.useClusterDefault || this.kmsKeyId.isEmpty(), "KMS key ID cannot be provided together with useClusterDefault");
+        checkArgument(!this.useClusterDefault || this.sseCustomerKey.isEmpty(), "SSE Customer key cannot be provided together with useClusterDefault");
+        checkArgument(this.kmsKeyId.isEmpty() || this.sseCustomerKey.isEmpty(), "SSE Customer key cannot be provided together with KMS key ID");
 
         this.endpoint = requireNonNull(endpoint, "endpoint is null");
         this.region = requireNonNull(region, "region is null");
@@ -133,6 +143,16 @@ public Set<String> allowedKmsKeyIds()
         return allowedKmsKeyIds;
     }
 
+    public Optional<String> sseCustomerKey()
+    {
+        return sseCustomerKey;
+    }
+
+    public Set<String> allowedSseCustomerKeys()
+    {
+        return allowedSseCustomerKeys;
+    }
+
     public Optional<AwsCredentials> credentials()
     {
         return credentials;

lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingConfig.java

@@ -31,6 +31,7 @@ public class S3SecurityMappingConfig
     private String jsonPointer = "";
     private String roleCredentialName;
     private String kmsKeyIdCredentialName;
+    private String sseCustomerKeyCredentialName;
     private Duration refreshPeriod;
     private String colonReplacement;
 
@@ -100,6 +101,19 @@ public S3SecurityMappingConfig setKmsKeyIdCredentialName(String kmsKeyIdCredenti
         return this;
     }
 
+    public Optional<String> getSseCustomerKeyCredentialName()
+    {
+        return Optional.ofNullable(sseCustomerKeyCredentialName);
+    }
+
+    @Config("s3.security-mapping.sse-customer-key-credential-name")
+    @ConfigDescription("Name of the extra credential used to provide SSE Customer key")
+    public S3SecurityMappingConfig setSseCustomerKeyCredentialName(String sseCustomerKeyCredentialName)
+    {
+        this.sseCustomerKeyCredentialName = sseCustomerKeyCredentialName;
+        return this;
+    }
+
     public Optional<Duration> getRefreshPeriod()
     {
         return Optional.ofNullable(refreshPeriod);

lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingProvider.java

@@ -33,6 +33,7 @@ final class S3SecurityMappingProvider
     private final Supplier<S3SecurityMappings> mappingsProvider;
     private final Optional<String> roleCredentialName;
     private final Optional<String> kmsKeyIdCredentialName;
+    private final Optional<String> sseCustomerKeyCredentialName;
     private final Optional<String> colonReplacement;
 
     @Inject
@@ -41,18 +42,21 @@ public S3SecurityMappingProvider(S3SecurityMappingConfig config, Supplier<S3Secu
         this(mappingsProvider(mappingsProvider, config.getRefreshPeriod()),
                 config.getRoleCredentialName(),
                 config.getKmsKeyIdCredentialName(),
+                config.getSseCustomerKeyCredentialName(),
                 config.getColonReplacement());
     }
 
     public S3SecurityMappingProvider(
             Supplier<S3SecurityMappings> mappingsProvider,
             Optional<String> roleCredentialName,
             Optional<String> kmsKeyIdCredentialName,
+            Optional<String> sseCustomerKeyCredentialName,
             Optional<String> colonReplacement)
     {
         this.mappingsProvider = requireNonNull(mappingsProvider, "mappingsProvider is null");
         this.roleCredentialName = requireNonNull(roleCredentialName, "roleCredentialName is null");
         this.kmsKeyIdCredentialName = requireNonNull(kmsKeyIdCredentialName, "kmsKeyIdCredentialName is null");
+        this.sseCustomerKeyCredentialName = requireNonNull(sseCustomerKeyCredentialName, "customerKeyCredentialName is null");
         this.colonReplacement = requireNonNull(colonReplacement, "colonReplacement is null");
     }
 
@@ -70,6 +74,7 @@ public Optional<S3SecurityMappingResult> getMapping(ConnectorIdentity identity,
                 selectRole(mapping, identity),
                 mapping.roleSessionName().map(name -> name.replace("${USER}", identity.getUser())),
                 selectKmsKeyId(mapping, identity),
+                getSseCustomerKey(mapping, identity),
                 mapping.endpoint(),
                 mapping.region()));
     }
@@ -131,6 +136,32 @@ private Optional<String> getKmsKeyIdFromExtraCredential(ConnectorIdentity identi
         return kmsKeyIdCredentialName.map(name -> identity.getExtraCredentials().get(name));
     }
 
+    private Optional<String> getSseCustomerKey(S3SecurityMapping mapping, ConnectorIdentity identity)
+    {
+        Optional<String> providedKey = getSseCustomerKeyFromExtraCredential(identity);
+
+        if (providedKey.isEmpty()) {
+            return mapping.sseCustomerKey();
+        }
+        if (mapping.sseCustomerKey().isPresent() && mapping.allowedSseCustomerKeys().isEmpty()) {
+            throw new AccessDeniedException("allowedSseCustomerKeys must be set if sseCustomerKey is provided");
+        }
+
+        String selected = providedKey.get();
+
+        if (selected.equals(mapping.sseCustomerKey().orElse(null)) ||
+                mapping.allowedSseCustomerKeys().contains(selected) ||
+                mapping.allowedSseCustomerKeys().contains("*")) {
+            return providedKey;
+        }
+        throw new AccessDeniedException("Provided SSE Customer Key is not allowed");
+    }
+
+    private Optional<String> getSseCustomerKeyFromExtraCredential(ConnectorIdentity identity)
+    {
+        return sseCustomerKeyCredentialName.map(name -> identity.getExtraCredentials().get(name));
+    }
+
     private static Supplier<S3SecurityMappings> mappingsProvider(Supplier<S3SecurityMappings> supplier, Optional<Duration> refreshPeriod)
     {
         return refreshPeriod

lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingResult.java

@@ -26,6 +26,7 @@ record S3SecurityMappingResult(
         Optional<String> iamRole,
         Optional<String> roleSessionName,
         Optional<String> kmsKeyId,
+        Optional<String> sseCustomerKey,
         Optional<String> endpoint,
         Optional<String> region)
 {
@@ -35,6 +36,7 @@ record S3SecurityMappingResult(
         requireNonNull(iamRole, "iamRole is null");
         requireNonNull(roleSessionName, "roleSessionName is null");
         requireNonNull(kmsKeyId, "kmsKeyId is null");
+        requireNonNull(sseCustomerKey, "sseCustomerKey is null");
         requireNonNull(endpoint, "endpoint is null");
         requireNonNull(region, "region is null");
     }