TIKA-5006: Support web identity (IRSA) credentials in the S3 plugin

CHANGES.txt

@@ -1,4 +1,10 @@
-Release 4.2.1 - ???
+Release 4.2.1 - unreleased
+
+  SERVER, PIPES AND OPERATIONS
+
+   * The S3 fetcher, emitter and pipes iterator accept credentialsProvider "web_identity"
+     for EKS IAM roles for service accounts (IRSA); the plugin now bundles the AWS STS
+     module it needs (TIKA-5006).
 
   OTHER CHANGES
 

docs/modules/ROOT/pages/pipes/plugins/s3.adoc

@@ -44,10 +44,13 @@ The Amazon S3 plugin (`tika-pipes-s3`) provides fetcher, emitter, and iterator i
 All three components share the same `credentialsProvider` selector:
 
 * `profile` — reads credentials from the local AWS profile named by `profile` (e.g., `default`).
-* `instance` — uses the instance/container role attached to the host (EC2 IAM role, ECS task role, etc.). No additional fields needed.
+* `instance` — uses the IAM role of the EC2 instance the host runs on, read from the instance metadata service. No additional fields needed.
+* `web_identity` — assumes a role with a web identity token, as EKS IAM roles for service accounts (IRSA) provide. Reads the role from `AWS_ROLE_ARN` and the token from `AWS_WEB_IDENTITY_TOKEN_FILE`, which EKS sets on the pod. No additional fields needed.
 * `key_secret` — reads `accessKey` and `secretKey` from the config. Avoid checking these into source control; prefer environment-variable substitution or one of the other providers.
 
-The emitter's `validate()` enforces these values, but the fetcher and iterator do not — they fail later when the AWS SDK tries to resolve credentials.
+All three components reject any other value when they are created. If `credentialsProvider` is unset, the emitter's `validate()` rejects it, but the fetcher and iterator fall back to `instance`.
+
+ECS task roles and EKS Pod Identity are not supported — neither `instance` nor `web_identity` picks them up.
 
 [#s3-fetcher]
 == S3 Fetcher (`s3-fetcher`)
@@ -78,8 +81,8 @@ include::example$pipes-s3-fetcher.json[]
 |Optional key prefix, prepended verbatim to the fetch key — include the trailing `/` yourself. Leave it unset when the fetch keys already carry the prefix (see <<prefix-warning>>).
 
 |`credentialsProvider`
-|_required_
-|One of `profile`, `instance`, `key_secret`. See <<credentials>>.
+|`instance`
+|One of `profile`, `instance`, `web_identity`, `key_secret`. See <<credentials>>.
 
 |`profile` / `accessKey` / `secretKey`
 |_conditional_
@@ -141,7 +144,7 @@ include::example$pipes-s3-emitter.json[]
 
 |`credentialsProvider`
 |_required_
-|One of `profile`, `instance`, `key_secret` (validated). See <<credentials>>.
+|One of `profile`, `instance`, `web_identity`, `key_secret` (validated). See <<credentials>>.
 
 |`profile` / `accessKey` / `secretKey`
 |_conditional_
@@ -203,8 +206,8 @@ In addition to the required `fetcherId` / `emitterId` (see xref:pipes/iterators.
 |Key prefix to scope the listing.
 
 |`credentialsProvider`
-|_none_
-|One of `profile`, `instance`, `key_secret`. See <<credentials>>.
+|`instance`
+|One of `profile`, `instance`, `web_identity`, `key_secret`. See <<credentials>>.
 
 |`profile` / `accessKey` / `secretKey` / `endpointConfigurationService`
 |_conditional_

tika-pipes/tika-pipes-plugins/tika-pipes-s3/pom.xml

@@ -49,6 +49,11 @@
       <groupId>software.amazon.awssdk</groupId>
       <artifactId>apache-client</artifactId>
     </dependency>
+    <dependency>
+      <!-- WebIdentityTokenFileCredentialsProvider ('web_identity') loads STS reflectively -->
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>sts</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-slf4j2-impl</artifactId>

tika-pipes/tika-pipes-plugins/tika-pipes-s3/src/main/java/org/apache/tika/pipes/emitter/s3/S3Emitter.java

@@ -39,6 +39,7 @@
 import software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider;
 import software.amazon.awssdk.core.checksums.RequestChecksumCalculation;
 import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.core.sync.RequestBody;
@@ -140,6 +141,8 @@ private static AwsCredentialsProvider buildCredentialsProvider(S3EmitterConfig c
         switch (config.credentialsProvider()) {
             case "instance":
                 return InstanceProfileCredentialsProvider.builder().build();
+            case "web_identity":
+                return WebIdentityTokenFileCredentialsProvider.builder().build();
             case "profile":
                 return ProfileCredentialsProvider.builder()
                         .profileName(config.profile())
@@ -149,7 +152,7 @@ private static AwsCredentialsProvider buildCredentialsProvider(S3EmitterConfig c
                         config.accessKey(), config.secretKey());
                 return StaticCredentialsProvider.create(awsCreds);
             default:
-                throw new TikaConfigException("credentialsProvider must be 'instance', 'profile' or 'key_secret'");
+                throw new TikaConfigException("credentialsProvider must be 'instance', 'web_identity', 'profile' or 'key_secret'");
         }
     }
 

tika-pipes/tika-pipes-plugins/tika-pipes-s3/src/main/java/org/apache/tika/pipes/emitter/s3/S3EmitterConfig.java

@@ -67,13 +67,15 @@ public void validate() throws TikaConfigException {
             throw new TikaConfigException("'region' must not be empty");
         }
         if (credentialsProvider == null || credentialsProvider.isBlank()) {
-            throw new TikaConfigException("'credentialsProvider' must be set to 'profile', 'instance' or 'key_secret'");
+            throw new TikaConfigException("'credentialsProvider' must be set to 'profile', 'instance', 'web_identity' or 'key_secret'");
         }
         if (!credentialsProvider.equals("profile")
                 && !credentialsProvider.equals("instance")
+                && !credentialsProvider.equals("web_identity")
                 && !credentialsProvider.equals("key_secret")) {
             throw new TikaConfigException(
-                    "credentialsProvider must be 'profile', 'instance' or 'key_secret', but was: " + credentialsProvider);
+                    "credentialsProvider must be 'profile', 'instance', 'web_identity' or 'key_secret', but was: "
+                            + credentialsProvider);
         }
         if (credentialsProvider.equals("profile") && (profile == null || profile.isBlank())) {
             throw new TikaConfigException("'profile' must be set when credentialsProvider is 'profile'");

tika-pipes/tika-pipes-plugins/tika-pipes-s3/src/main/java/org/apache/tika/pipes/fetcher/s3/S3Fetcher.java

@@ -34,6 +34,7 @@
 import software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider;
 import software.amazon.awssdk.awscore.exception.AwsServiceException;
 import software.amazon.awssdk.core.ResponseInputStream;
 import software.amazon.awssdk.core.exception.SdkClientException;
@@ -106,25 +107,7 @@ private void initialize() throws TikaConfigException {
         ConfigValidator.mustNotBeEmpty("bucket", config.getBucket());
         ConfigValidator.mustNotBeEmpty("region", config.getRegion());
 
-        AwsCredentialsProvider provider;
-        String credentialsProvider = config.getCredentialsProvider();
-        if (credentialsProvider == null) {
-            credentialsProvider = "instance";
-        }
-        switch (credentialsProvider) {
-            case "instance":
-                provider = InstanceProfileCredentialsProvider.builder().build();
-                break;
-            case "profile":
-                provider = ProfileCredentialsProvider.builder().profileName(config.getProfile()).build();
-                break;
-            case "key_secret":
-                AwsBasicCredentials awsCreds = AwsBasicCredentials.create(config.getAccessKey(), config.getSecretKey());
-                provider = StaticCredentialsProvider.create(awsCreds);
-                break;
-            default:
-                throw new TikaConfigException("credentialsProvider must be set and must be either 'instance', 'profile' or 'key_secret'");
-        }
+        AwsCredentialsProvider provider = buildCredentialsProvider(config);
 
         int maxConnections = config.getMaxConnections();
         if (maxConnections <= 0) {
@@ -153,6 +136,27 @@ private void initialize() throws TikaConfigException {
         }
     }
 
+    static AwsCredentialsProvider buildCredentialsProvider(S3FetcherConfig config) throws TikaConfigException {
+        String credentialsProvider = config.getCredentialsProvider();
+        if (credentialsProvider == null) {
+            credentialsProvider = "instance";
+        }
+        switch (credentialsProvider) {
+            case "instance":
+                return InstanceProfileCredentialsProvider.builder().build();
+            case "web_identity":
+                return WebIdentityTokenFileCredentialsProvider.builder().build();
+            case "profile":
+                return ProfileCredentialsProvider.builder().profileName(config.getProfile()).build();
+            case "key_secret":
+                AwsBasicCredentials awsCreds = AwsBasicCredentials.create(config.getAccessKey(), config.getSecretKey());
+                return StaticCredentialsProvider.create(awsCreds);
+            default:
+                throw new TikaConfigException("credentialsProvider must be set and must be either "
+                        + "'instance', 'web_identity', 'profile' or 'key_secret'");
+        }
+    }
+
     @Override
     public TikaInputStream fetch(String fetchKey, Metadata metadata, ParseContext parseContext) throws TikaException, IOException {
         return fetch(fetchKey, -1, -1, metadata, parseContext);

tika-pipes/tika-pipes-plugins/tika-pipes-s3/src/main/java/org/apache/tika/pipes/iterator/s3/S3PipesIterator.java

@@ -31,6 +31,7 @@
 import software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider;
 import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.http.SdkHttpClient;
 import software.amazon.awssdk.http.apache.ApacheHttpClient;
@@ -84,6 +85,9 @@ private S3PipesIterator(S3PipesIteratorConfig config, ExtensionConfig extensionC
             case "instance":
                 provider = InstanceProfileCredentialsProvider.builder().build();
                 break;
+            case "web_identity":
+                provider = WebIdentityTokenFileCredentialsProvider.builder().build();
+                break;
             case "profile":
                 provider = ProfileCredentialsProvider.builder().profileName(config.getProfile()).build();
                 break;
@@ -92,7 +96,7 @@ private S3PipesIterator(S3PipesIteratorConfig config, ExtensionConfig extensionC
                 provider = StaticCredentialsProvider.create(awsCreds);
                 break;
             default:
-                throw new TikaConfigException("credentialsProvider must be set and " + "must be either 'instance', 'profile' or 'key_secret'");
+                throw new TikaConfigException("credentialsProvider must be set and must be either 'instance', 'web_identity', 'profile' or 'key_secret'");
         }
 
         SdkHttpClient httpClient = ApacheHttpClient.builder().maxConnections(config.getMaxConnections()).build();

tika-pipes/tika-pipes-plugins/tika-pipes-s3/src/test/java/org/apache/tika/pipes/emitter/s3/S3EmitterConfigTest.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.pipes.emitter.s3;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+
+import org.apache.tika.exception.TikaConfigException;
+
+public class S3EmitterConfigTest {
+
+    @Test
+    public void testWebIdentityIsValid() throws Exception {
+        S3EmitterConfig config = load("web_identity");
+        assertDoesNotThrow(config::validate);
+    }
+
+    @Test
+    public void testUnknownProviderIsInvalid() throws Exception {
+        S3EmitterConfig config = load("bogus");
+        TikaConfigException e = assertThrows(TikaConfigException.class, config::validate);
+        assertTrue(e.getMessage().contains("'web_identity'"), e.getMessage());
+    }
+
+    private static S3EmitterConfig load(String credentialsProvider) throws TikaConfigException {
+        return S3EmitterConfig.load("{\"bucket\": \"bucket\", \"region\": \"us-east-1\", "
+                + "\"credentialsProvider\": \"" + credentialsProvider + "\"}");
+    }
+}

tika-pipes/tika-pipes-plugins/tika-pipes-s3/src/test/java/org/apache/tika/pipes/fetcher/s3/S3FetcherCredentialsTest.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.pipes.fetcher.s3;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider;
+
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.pipes.fetcher.s3.config.S3FetcherConfig;
+import org.apache.tika.plugins.ExtensionConfig;
+
+public class S3FetcherCredentialsTest {
+
+    @Test
+    public void testWebIdentity() throws Exception {
+        assertInstanceOf(WebIdentityTokenFileCredentialsProvider.class,
+                S3Fetcher.buildCredentialsProvider(new S3FetcherConfig().setCredentialsProvider("web_identity")));
+    }
+
+    @Test
+    public void testDefaultsToInstance() throws Exception {
+        assertInstanceOf(InstanceProfileCredentialsProvider.class,
+                S3Fetcher.buildCredentialsProvider(new S3FetcherConfig()));
+    }
+
+    @Test
+    public void testKeySecret() throws Exception {
+        S3FetcherConfig config = new S3FetcherConfig()
+                .setCredentialsProvider("key_secret")
+                .setAccessKey("access")
+                .setSecretKey("secret");
+        assertInstanceOf(StaticCredentialsProvider.class, S3Fetcher.buildCredentialsProvider(config));
+    }
+
+    @Test
+    public void testUnknownProvider() {
+        TikaConfigException e = assertThrows(TikaConfigException.class,
+                () -> S3Fetcher.buildCredentialsProvider(new S3FetcherConfig().setCredentialsProvider("bogus")));
+        assertTrue(e.getMessage().contains("'web_identity'"), e.getMessage());
+    }
+
+    @Test
+    public void testBuildWithWebIdentity() throws Exception {
+        // Credentials resolve on the first request, so the fetcher builds without a web identity
+        // token in the environment.
+        String json = "{\"bucket\": \"bucket\", \"region\": \"us-east-1\", \"credentialsProvider\": \"web_identity\"}";
+        assertNotNull(S3Fetcher.build(new ExtensionConfig("s3", "s3-fetcher", json)));
+    }
+
+    @Test
+    public void testStsAvailable() {
+        // The web identity provider loads this from the sts module when resolving credentials.
+        // Without it, every request fails with "To use web identity tokens, the 'sts' service
+        // module must be on the class path."
+        assertDoesNotThrow(() -> Class.forName(
+                "software.amazon.awssdk.services.sts.internal.StsWebIdentityCredentialsProviderFactory"));
+    }
+}