[WIP][File system] Allow to obtain temporary credentials via STS AssumeRoleWithWebIdentity in S3

fluss-filesystems/fluss-fs-s3/src/main/java/com/alibaba/fluss/fs/s3/token/S3DelegationTokenProvider.java

@@ -21,6 +21,7 @@
 
 import com.amazonaws.auth.AWSStaticCredentialsProvider;
 import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.client.builder.AwsClientBuilder;
 import com.amazonaws.services.securitytoken.AWSSecurityTokenService;
 import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceClientBuilder;
 import com.amazonaws.services.securitytoken.model.Credentials;
@@ -46,18 +47,39 @@ public class S3DelegationTokenProvider {
     private static final String REGION_KEY = "fs.s3a.region";
     private static final String ENDPOINT_KEY = "fs.s3a.endpoint";
 
+    private static final String STS_REGION_KEY = "fs.s3a.sts.region";
+    private static final String STS_ENDPOINT_KEY = "fs.s3a.sts.endpoint";
+
+    private final AWSSecurityTokenService stsClient;
     private final String scheme;
-    private final String region;
-    private final String accessKey;
-    private final String secretKey;
     private final Map<String, String> additionInfos;
 
     public S3DelegationTokenProvider(String scheme, Configuration conf) {
         this.scheme = scheme;
-        this.region = conf.get(REGION_KEY);
+
+        String region = conf.get(REGION_KEY);
         checkNotNull(region, "Region is not set.");
-        this.accessKey = conf.get(ACCESS_KEY_ID);
-        this.secretKey = conf.get(ACCESS_KEY_SECRET);
+        String accessKey = conf.get(ACCESS_KEY_ID);
+        String secretKey = conf.get(ACCESS_KEY_SECRET);
+
+        AWSSecurityTokenServiceClientBuilder stsClientBuilder =
+                AWSSecurityTokenServiceClientBuilder.standard()
+                        .withCredentials(
+                                new AWSStaticCredentialsProvider(
+                                        new BasicAWSCredentials(accessKey, secretKey)));
+        String stsEndpoint = conf.get(STS_ENDPOINT_KEY);
+        String stsRegion = conf.get(STS_REGION_KEY, REGION_KEY);
+        if (stsEndpoint != null) {
+            LOG.debug("Building STS client with endpoint {} and region {}", stsEndpoint, stsRegion);
+            AwsClientBuilder.EndpointConfiguration endpointConfiguration =
+                    new AwsClientBuilder.EndpointConfiguration(stsEndpoint, stsRegion);
+            stsClientBuilder.withEndpointConfiguration(endpointConfiguration);
+        } else {
+            LOG.debug("Building STS client with default endpoint and region {}", stsRegion);
+            stsClientBuilder.withRegion(stsRegion);
+        }
+        this.stsClient = stsClientBuilder.build();
+
         this.additionInfos = new HashMap<>();
         for (String key : Arrays.asList(REGION_KEY, ENDPOINT_KEY)) {
             if (conf.get(key) != null) {
@@ -67,15 +89,8 @@ public S3DelegationTokenProvider(String scheme, Configuration conf) {
     }
 
     public ObtainedSecurityToken obtainSecurityToken() {
-        LOG.info("Obtaining session credentials token with access key: {}", accessKey);
+        LOG.info("Obtaining session credentials token");
 
-        AWSSecurityTokenService stsClient =
-                AWSSecurityTokenServiceClientBuilder.standard()
-                        .withRegion(region)
-                        .withCredentials(
-                                new AWSStaticCredentialsProvider(
-                                        new BasicAWSCredentials(accessKey, secretKey)))
-                        .build();
         GetSessionTokenResult sessionTokenResult = stsClient.getSessionToken();
         Credentials credentials = sessionTokenResult.getCredentials();
 

website/docs/maintenance/filesystems/s3.md

@@ -23,17 +23,42 @@ sidebar_position: 4
 
 [Amazon Simple Storage Service](http://aws.amazon.com/s3/) (Amazon S3) is cloud object storage with industry-leading scalability, data availability, security, and performance.
 
-## Configurations setup
+## Configurations Setup
 
-To enabled S3 as remote storage, there are some required configurations that must be added to Fluss' `server.yaml`:
+To enable S3 as remote storage, set the following configuration options in Fluss' `server.yaml`:
 
 ```yaml
 # The dir that used to be as the remote storage of Fluss
 remote.data.dir: s3://<your-bucket>/path/to/remote/storage
+# region
+s3.region: <your-s3-region>
+```
+
+In addition, you need configure how temporary security credentials should be obtained from the Security Token Service (STS) API.
+
+List of supported STS API endpoints:
+- [`GetSessionToken`](#getsessiontoken)
+
+:::info
+When using an S3 compatible storage, first check if it supports at least one of the listed STS API endpoints!
+:::
+
+### `GetSessionToken`
+
+The following configuration options are mandatory.
+
+```yaml
 # access key
 s3.access-key: <your-access-key>
 # secret key
 s3.secret-key: <your-secret-key>
-# region
-s3.region: <your-s3-region>
+```
+
+Additionally, you may set the following configuration options to change the defaults.
+
+```yaml
+# STS service endpoint; for AWS S3 has to match the region
+s3.sts.endpoint: <your-sts-service-endpoint>
+# STS signing region; defaults to `s3.region`
+s3.sts.region: <your-sts-signing-region>
 ```