Sanitize principal names in AWS STS role session names

polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsCredentialsStorageIntegration.java

@@ -103,17 +103,15 @@ public StorageAccessConfig getSubscopedCreds(
 
     String roleSessionName =
         includePrincipalNameInSubscopedCredential
-            ? "polaris-" + polarisPrincipal.getName()
+            ? AwsRoleSessionNameSanitizer.sanitize("polaris-" + polarisPrincipal.getName())
             : "PolarisAwsCredentialsStorageIntegration";
-    String cappedRoleSessionName =
-        roleSessionName.substring(0, Math.min(roleSessionName.length(), 64));
 
     if (shouldUseSts(storageConfig)) {
       AssumeRoleRequest.Builder request =
           AssumeRoleRequest.builder()
               .externalId(storageConfig.getExternalId())
               .roleArn(storageConfig.getRoleARN())
-              .roleSessionName(cappedRoleSessionName)
+              .roleSessionName(roleSessionName)
               .policy(
                   policyString(
                           storageConfig,

polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsRoleSessionNameSanitizer.java

@@ -0,0 +1,98 @@
+/*
+ * 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.polaris.core.storage.aws;
+
+import jakarta.annotation.Nonnull;
+import java.util.regex.Pattern;
+
+/**
+ * Utility class for sanitizing AWS STS role session names.
+ *
+ * <p>AWS STS role session names must conform to the pattern {@code [\w+=,.@-]*} and have a maximum
+ * length of 64 characters. This class provides methods to sanitize arbitrary strings (such as
+ * principal names) into valid role session names.
+ *
+ * @see <a href="https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html">AWS STS
+ *     AssumeRole API</a>
+ */
+public final class AwsRoleSessionNameSanitizer {
+
+  /**
+   * AWS STS role session name maximum length. While the AssumedRoleId can be up to 193 characters,
+   * the roleSessionName parameter itself is limited to 64 characters.
+   */
+  static final int MAX_ROLE_SESSION_NAME_LENGTH = 64;
+
+  /**
+   * Pattern matching characters that are NOT allowed in AWS STS role session names. AWS allows:
+   * alphanumeric characters (a-z, A-Z, 0-9), underscore (_), plus (+), equals (=), comma (,),
+   * period (.), at sign (@), and hyphen (-).
+   *
+   * <p>This pattern matches any character outside this allowed set.
+   */
+  private static final Pattern INVALID_ROLE_SESSION_NAME_CHARS =
+      Pattern.compile("[^a-zA-Z0-9_+=,.@-]");
+
+  /** Default replacement character for invalid characters. */
+  private static final String DEFAULT_REPLACEMENT = "_";
+
+  private AwsRoleSessionNameSanitizer() {
+    // Utility class to prevent instantiation
+  }
+
+  /**
+   * Sanitizes a string for use as an AWS STS role session name.
+   *
+   * <p>This method:
+   *
+   * <ol>
+   *   <li>Replaces any characters not matching {@code [\w+=,.@-]} with underscores
+   *   <li>Truncates the result to 64 characters (AWS maximum)
+   * </ol>
+   *
+   * <p>The underscore replacement character was chosen because:
+   *
+   * <ul>
+   *   <li>It is always valid in role session names
+   *   <li>It is visually distinct and indicates a substitution occurred
+   *   <li>It does not introduce ambiguity (unlike hyphen which is common in names)
+   * </ul>
+   *
+   * @param input the string to sanitize (typically a principal name)
+   * @return a sanitized string safe for use as an AWS STS role session name
+   */
+  public static @Nonnull String sanitize(@Nonnull String input) {
+    String sanitized =
+        INVALID_ROLE_SESSION_NAME_CHARS.matcher(input).replaceAll(DEFAULT_REPLACEMENT);
+    return truncate(sanitized);
+  }
+
+  /**
+   * Truncates a string to the maximum allowed role session name length.
+   *
+   * @param input the string to truncate
+   * @return the truncated string, or the original if already within limits
+   */
+  static @Nonnull String truncate(@Nonnull String input) {
+    if (input.length() <= MAX_ROLE_SESSION_NAME_LENGTH) {
+      return input;
+    }
+    return input.substring(0, MAX_ROLE_SESSION_NAME_LENGTH);
+  }
+}

polaris-core/src/test/java/org/apache/polaris/core/storage/aws/AwsRoleSessionNameSanitizerTest.java

@@ -0,0 +1,70 @@
+/*
+ * 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.polaris.core.storage.aws;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.regex.Pattern;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+class AwsRoleSessionNameSanitizerTest {
+
+  /** AWS STS role session name validation pattern. */
+  private static final Pattern AWS_ROLE_SESSION_NAME_PATTERN = Pattern.compile("[\\w+=,.@-]*");
+
+  @ParameterizedTest
+  @CsvSource({
+    "polaris-Invalid (local),polaris-Invalid__local_",
+    "service/account:readonly,service_account_readonly",
+    "user name,user_name",
+    "polaris-test-principal,polaris-test-principal",
+    "user@domain.com,user@domain.com",
+    "key=value,key=value"
+  })
+  void testSanitize(String input, String expected) {
+    assertThat(AwsRoleSessionNameSanitizer.sanitize(input)).isEqualTo(expected);
+  }
+
+  @Test
+  void testSanitizeTruncatesToMaxLength() {
+    String longInput = "a".repeat(100);
+    String result = AwsRoleSessionNameSanitizer.sanitize(longInput);
+    assertThat(result).hasSize(AwsRoleSessionNameSanitizer.MAX_ROLE_SESSION_NAME_LENGTH);
+  }
+
+  @Test
+  void testSanitizeOutputMatchesAwsPattern() {
+    String[] inputs = {
+      "polaris-Invalid (local)",
+      "special!@#$%chars",
+      "path/to/resource",
+      "very-long-name-" + "x".repeat(100)
+    };
+
+    for (String input : inputs) {
+      String sanitized = AwsRoleSessionNameSanitizer.sanitize(input);
+      assertThat(AWS_ROLE_SESSION_NAME_PATTERN.matcher(sanitized).matches())
+          .as("Sanitized '%s' should match AWS pattern", sanitized)
+          .isTrue();
+      assertThat(sanitized.length()).isLessThanOrEqualTo(64);
+    }
+  }
+}

polaris-core/src/test/java/org/apache/polaris/service/storage/aws/AwsCredentialsStorageIntegrationTest.java

@@ -969,6 +969,84 @@ public void testGetSubscopedCredsLongPrincipalName() {
             CredentialVendingContext.empty());
   }
 
+  @Test
+  public void testGetSubscopedCredsPrincipalNameWithInvalidCharacters() {
+    StsClient stsClient = Mockito.mock(StsClient.class);
+    String roleARN = "arn:aws:iam::012345678901:role/jdoe";
+    String externalId = "externalId";
+    // Principal name with spaces and parentheses - invalid for AWS STS
+    PolarisPrincipal polarisPrincipalWithInvalidChars =
+        PolarisPrincipal.of("Invalid Principal (local)", Map.of(), Set.of());
+
+    Mockito.when(stsClient.assumeRole(Mockito.isA(AssumeRoleRequest.class)))
+        .thenAnswer(
+            invocation -> {
+              assertThat(invocation.getArguments()[0])
+                  .isInstanceOf(AssumeRoleRequest.class)
+                  .asInstanceOf(InstanceOfAssertFactories.type(AssumeRoleRequest.class))
+                  .returns(externalId, AssumeRoleRequest::externalId)
+                  .returns(roleARN, AssumeRoleRequest::roleArn)
+                  // Spaces and parentheses should be replaced with underscores
+                  .returns("polaris-Invalid_Principal__local_", AssumeRoleRequest::roleSessionName);
+              return ASSUME_ROLE_RESPONSE;
+            });
+    String warehouseDir = "s3://bucket/path/to/warehouse";
+    new AwsCredentialsStorageIntegration(
+            AwsStorageConfigurationInfo.builder()
+                .addAllowedLocation(warehouseDir)
+                .roleARN(roleARN)
+                .externalId(externalId)
+                .build(),
+            stsClient)
+        .getSubscopedCreds(
+            PRINCIPAL_INCLUDER_REALM_CONFIG,
+            true,
+            Set.of(warehouseDir + "/namespace/table"),
+            Set.of(warehouseDir + "/namespace/table"),
+            polarisPrincipalWithInvalidChars,
+            Optional.of("/namespace/table/credentials"),
+            CredentialVendingContext.empty());
+  }
+
+  @Test
+  public void testGetSubscopedCredsPrincipalNameWithSpecialCharacters() {
+    StsClient stsClient = Mockito.mock(StsClient.class);
+    String roleARN = "arn:aws:iam::012345678901:role/jdoe";
+    String externalId = "externalId";
+    // Principal name with slashes and colons
+    PolarisPrincipal polarisPrincipalWithSpecialChars =
+        PolarisPrincipal.of("service/account:readonly", Map.of(), Set.of());
+
+    Mockito.when(stsClient.assumeRole(Mockito.isA(AssumeRoleRequest.class)))
+        .thenAnswer(
+            invocation -> {
+              assertThat(invocation.getArguments()[0])
+                  .isInstanceOf(AssumeRoleRequest.class)
+                  .asInstanceOf(InstanceOfAssertFactories.type(AssumeRoleRequest.class))
+                  .returns(externalId, AssumeRoleRequest::externalId)
+                  .returns(roleARN, AssumeRoleRequest::roleArn)
+                  // Slashes and colons should be replaced with underscores
+                  .returns("polaris-service_account_readonly", AssumeRoleRequest::roleSessionName);
+              return ASSUME_ROLE_RESPONSE;
+            });
+    String warehouseDir = "s3://bucket/path/to/warehouse";
+    new AwsCredentialsStorageIntegration(
+            AwsStorageConfigurationInfo.builder()
+                .addAllowedLocation(warehouseDir)
+                .roleARN(roleARN)
+                .externalId(externalId)
+                .build(),
+            stsClient)
+        .getSubscopedCreds(
+            PRINCIPAL_INCLUDER_REALM_CONFIG,
+            true,
+            Set.of(warehouseDir + "/namespace/table"),
+            Set.of(warehouseDir + "/namespace/table"),
+            polarisPrincipalWithSpecialChars,
+            Optional.of("/namespace/table/credentials"),
+            CredentialVendingContext.empty());
+  }
+
   private static @Nonnull String s3Arn(String partition, String bucket, String keyPrefix) {
     String bucketArn = "arn:" + partition + ":s3:::" + bucket;
     if (keyPrefix == null) {