CVE-2026-42811

@@ -21,6 +21,7 @@
 import static org.apache.polaris.core.config.FeatureConfiguration.STORAGE_CREDENTIAL_DURATION_SECONDS;
 import static org.apache.polaris.core.storage.aws.AwsSessionTagsBuilder.buildSessionTags;
 
+import com.google.common.annotations.VisibleForTesting;
 import jakarta.annotation.Nonnull;
 import java.net.URI;
 import java.util.EnumSet;
@@ -38,6 +39,7 @@
 import org.apache.polaris.core.storage.InMemoryStorageIntegration;
 import org.apache.polaris.core.storage.StorageAccessConfig;
 import org.apache.polaris.core.storage.StorageAccessProperty;
+import org.apache.polaris.core.storage.StorageUri;
 import org.apache.polaris.core.storage.StorageUtil;
 import org.apache.polaris.core.storage.aws.StsClientProvider.StsDestination;
 import org.slf4j.Logger;
@@ -236,12 +238,14 @@ private IamPolicy policyString(
         .distinct()
         .forEach(
             location -> {
-              URI uri = URI.create(location);
+              StorageUri uri = StorageUri.parse(location);
+              String escapedObjectPrefix = escapeIamGlobLiteral(parseS3Path(uri));
               allowGetObjectStatementBuilder.addResource(
                   IamResource.create(
-                      arnPrefix + StorageUtil.concatFilePrefixes(parseS3Path(uri), "*", "/")));
-              final var bucket = arnPrefix + StorageUtil.getBucket(uri);
+                      arnPrefix + StorageUtil.concatFilePrefixes(escapedObjectPrefix, "*", "/")));
+              final var bucket = arnPrefix + uri.authority();
               if (allowList) {
+                String escapedListPrefix = escapeIamGlobLiteral(trimLeadingSlash(uri.rawPath()));
                 bucketListStatementBuilder
                     .computeIfAbsent(
                         bucket,
@@ -253,7 +257,7 @@ private IamPolicy policyString(
                     .addCondition(
                         IamConditionOperator.STRING_LIKE,
                         "s3:prefix",
-                        StorageUtil.concatFilePrefixes(trimLeadingSlash(uri.getPath()), "*", "/"));
+                        StorageUtil.concatFilePrefixes(escapedListPrefix, "*", "/"));
               }
               bucketGetLocationStatementBuilder.computeIfAbsent(
                   bucket,
@@ -273,10 +277,11 @@ private IamPolicy policyString(
               .addAction("s3:DeleteObject");
       writeLocations.forEach(
           location -> {
-            URI uri = URI.create(location);
+            StorageUri uri = StorageUri.parse(location);
+            String escapedObjectPrefix = escapeIamGlobLiteral(parseS3Path(uri));
             allowPutObjectStatementBuilder.addResource(
                 IamResource.create(
-                    arnPrefix + StorageUtil.concatFilePrefixes(parseS3Path(uri), "*", "/")));
+                    arnPrefix + StorageUtil.concatFilePrefixes(escapedObjectPrefix, "*", "/")));
           });
       policyBuilder.addStatement(allowPutObjectStatementBuilder.build());
     }
@@ -397,12 +402,31 @@ private static String arnPrefixForPartition(String awsPartition) {
     return String.format("arn:%s:s3:::", awsPartition != null ? awsPartition : "aws");
   }
 
-  private static @Nonnull String parseS3Path(URI uri) {
-    String bucket = StorageUtil.getBucket(uri);
-    String path = trimLeadingSlash(uri.getPath());
+  private static @Nonnull String parseS3Path(StorageUri uri) {
+    String bucket = uri.authority();
+    String path = trimLeadingSlash(uri.rawPath());
     return String.join("/", bucket, path);
   }
 
+  /**
+   * Escapes IAM pattern characters that must remain literal inside object resource ARNs and {@link
+   * IamConditionOperator#STRING_LIKE StringLike} conditions.
+   */
+  @VisibleForTesting
+  public static @Nonnull String escapeIamGlobLiteral(String value) {
+    StringBuilder escaped = new StringBuilder(value.length() + 8);
+    for (int i = 0; i < value.length(); i++) {
+      char c = value.charAt(i);
+      switch (c) {
+        case '*' -> escaped.append("${*}");
+        case '?' -> escaped.append("${?}");
+        case '$' -> escaped.append("${$}");
+        default -> escaped.append(c);
+      }
+    }
+    return escaped.toString();
+  }
+
   private static @Nonnull String trimLeadingSlash(String path) {
     if (path.startsWith("/")) {
       path = path.substring(1);

@@ -36,7 +36,6 @@
 import com.google.protobuf.Timestamp;
 import jakarta.annotation.Nonnull;
 import java.io.IOException;
-import java.net.URI;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Date;
@@ -55,7 +54,7 @@
 import org.apache.polaris.core.storage.PolarisStorageIntegration;
 import org.apache.polaris.core.storage.StorageAccessConfig;
 import org.apache.polaris.core.storage.StorageAccessProperty;
-import org.apache.polaris.core.storage.StorageUtil;
+import org.apache.polaris.core.storage.StorageUri;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -209,30 +208,21 @@ public static CredentialAccessBoundary generateAccessBoundaryRules(
         .distinct()
         .forEach(
             location -> {
-              URI uri = URI.create(location);
-              String bucket = StorageUtil.getBucket(uri);
+              StorageUri uri = StorageUri.parse(location);
+              String bucket = uri.authority();
               readBuckets.add(bucket);
-              String path = uri.getPath().substring(1);
+              String path = uri.rawPath().substring(1);
               List<String> resourceExpressions =
                   readConditionsMap.computeIfAbsent(bucket, key -> new ArrayList<>());
-              resourceExpressions.add(
-                  String.format(
-                      "resource.name.startsWith('projects/_/buckets/%s/objects/%s')",
-                      bucket, path));
+              resourceExpressions.add(resourceNameStartsWithExpression(bucket, path));
               if (allowListOperation) {
-                resourceExpressions.add(
-                    String.format(
-                        "api.getAttribute('storage.googleapis.com/objectListPrefix', '').startsWith('%s')",
-                        path));
+                resourceExpressions.add(objectListPrefixStartsWithExpression(path));
               }
               if (allowedWriteLocations.contains(location)) {
                 writeBuckets.add(bucket);
                 List<String> writeExpressions =
                     writeConditionsMap.computeIfAbsent(bucket, key -> new ArrayList<>());
-                writeExpressions.add(
-                    String.format(
-                        "resource.name.startsWith('projects/_/buckets/%s/objects/%s')",
-                        bucket, path));
+                writeExpressions.add(resourceNameStartsWithExpression(bucket, path));
               }
             });
     CredentialAccessBoundary.Builder accessBoundaryBuilder = CredentialAccessBoundary.newBuilder();
@@ -288,6 +278,55 @@ protected IamCredentialsClient createIamCredentialsClient(GoogleCredentials cred
             .build());
   }
 
+  @VisibleForTesting
+  static String resourceNameStartsWithExpression(String bucket, String path) {
+    return String.format(
+        "resource.name.startsWith('projects/_/buckets/%s/objects/%s')",
+        escapeCelLiteral(bucket), escapeCelLiteral(path));
+  }
+
+  @VisibleForTesting
+  static String objectListPrefixStartsWithExpression(String path) {
+    return String.format(
+        "api.getAttribute('storage.googleapis.com/objectListPrefix', '').startsWith('%s')",
+        escapeCelLiteral(path));
+  }
+
+  @VisibleForTesting
+  public static String escapeCelLiteral(String value) {
+    StringBuilder escaped = new StringBuilder(value.length() * 3 / 2);
+    for (int i = 0; i < value.length(); i++) {
+      char c = value.charAt(i);
+      switch (c) {
+        case '\'' -> escaped.append("\\'");
+        case '"' -> escaped.append("\\\"");
+        case '\\' -> escaped.append("\\\\");
+        case '\b' -> escaped.append("\\b");
+        case '\f' -> escaped.append("\\f");
+        case '\n' -> escaped.append("\\n");
+        case '\r' -> escaped.append("\\r");
+        case '\t' -> escaped.append("\\t");
+        default -> {
+          if (Character.isSurrogate(c)) {
+            if (!Character.isHighSurrogate(c)
+                || i + 1 >= value.length()
+                || !Character.isLowSurrogate(value.charAt(i + 1))) {
+              throw new IllegalArgumentException(
+                  "Unsupported unpaired surrogate in GCS credential access boundary input");
+            }
+            escaped.append(c).append(value.charAt(++i));
+          } else if (Character.isISOControl(c)) {
+            throw new IllegalArgumentException(
+                "Unsupported control character in GCS credential access boundary input");
+          } else {
+            escaped.append(c);
+          }
+        }
+      }
+    }
+    return escaped.toString();
+  }
+
   private static String bucketResource(String bucket) {
     return "//storage.googleapis.com/projects/_/buckets/" + bucket;
   }

@@ -0,0 +1,56 @@
+/*
+ * 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;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import org.jspecify.annotations.NonNull;
+
+/**
+ * Minimal URI parser for storage locations that preserves the full raw path once the path has
+ * started, rather than treating {@code ?} or {@code #} inside object keys as query or fragment
+ * delimiters.
+ *
+ * <p>This is an internal utility class and not meant for use outside of this package.
+ */
+public record StorageUri(String scheme, String authority, String rawPath) {
+
+  @SuppressWarnings("ConstantValue")
+  public static @NonNull StorageUri parse(@NonNull String location) {
+    checkArgument(location != null, "Invalid storage location uri %s", location);
+    int schemeSeparator = location.indexOf("://");
+    checkArgument(schemeSeparator > 0, "Invalid storage location uri %s", location);
+
+    int authorityStart = schemeSeparator + 3;
+    int pathStart = location.indexOf('/', authorityStart);
+    int authorityEnd = pathStart >= 0 ? pathStart : location.length();
+
+    String authority = location.substring(authorityStart, authorityEnd);
+    checkArgument(
+        authority
+            .codePoints()
+            .noneMatch(c -> Character.isWhitespace(c) || Character.isSpaceChar(c)),
+        "Invalid storage location uri %s",
+        location);
+
+    String rawPath = pathStart >= 0 ? location.substring(pathStart) : "";
+    return new StorageUri(
+        location.substring(0, schemeSeparator), authority.isEmpty() ? null : authority, rawPath);
+  }
+}

@@ -55,16 +55,21 @@ public class StorageUtil {
    * @return The bucket/authority of the path
    */
   public static @Nonnull String getBucket(String path) {
-    URI uri = URI.create(path);
-    return getBucket(uri);
+    int schemeSeparator = path.indexOf("://");
+    if (schemeSeparator <= 0) {
+      return null;
+    }
+    return StorageUri.parse(path).authority();
   }
 
   /**
    * Given a URI, extract the bucket (authority).
    *
    * @param uri A path to parse
    * @return The bucket/authority of the URI
+   * @deprecated Use {@link StorageUri#parse(String)} instead.
    */
+  @Deprecated(forRemoval = true)
   public static @Nonnull String getBucket(URI uri) {
     return uri.getAuthority();
   }

@@ -0,0 +1,76 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.stream.Stream;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class StorageUriTest {
+
+  @ParameterizedTest
+  @MethodSource("validStorageLocations")
+  void testParsePreservesRawPath(
+      String location, String expectedAuthority, String expectedRawPath) {
+    StorageUri uri = StorageUri.parse(location);
+
+    assertThat(uri.authority()).isEqualTo(expectedAuthority);
+    assertThat(uri.rawPath()).isEqualTo(expectedRawPath);
+  }
+
+  private static Stream<Arguments> validStorageLocations() {
+    return Stream.of(
+        Arguments.of(
+            "s3://bucket/path/with?question#and-fragment",
+            "bucket",
+            "/path/with?question#and-fragment"),
+        Arguments.of("s3://bucket/path/ns*%3F$/tb$%3F*", "bucket", "/path/ns*%3F$/tb$%3F*"),
+        Arguments.of("gs://bucket", "bucket", ""),
+        Arguments.of("abfs://container", "container", ""));
+  }
+
+  @ParameterizedTest
+  @MethodSource("invalidStorageLocations")
+  void testParseRejects(String location) {
+    assertThatThrownBy(() -> StorageUri.parse(location))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageStartingWith("Invalid storage location uri");
+  }
+
+  private static Stream<String> invalidStorageLocations() {
+    return Stream.of(
+        "s3://buck et/path",
+        "s3://buck\tet/path",
+        "s3://buck\net/path",
+        "s3://buck\ret/path",
+        "s3://buck\fet/path",
+        "s3://buck\u00A0et/path",
+        "s3://buck\u2003et/path",
+        "s3://buck\u2028et/path",
+        "s3://buck\u3000et/path",
+        "s3://bucket /path",
+        null,
+        "://",
+        "/foo/bar");
+  }
+}

@@ -43,6 +43,7 @@
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
@@ -358,6 +359,196 @@ public void testGetSubscopedCredsInlinePolicy(String awsPartition) {
     }
   }
 
+  @Test
+  public void testGetSubscopedCredsInlinePolicyEscapesIamSpecialCharacters() {
+    StsClient stsClient = Mockito.mock(StsClient.class);
+    String roleARN = "arn:aws:iam::012345678901:role/jdoe";
+    String externalId = "externalId";
+    String bucket = "bucket";
+    String warehouseKeyPrefix = "path/to/warehouse";
+    String specialLocation = "s3://bucket/" + warehouseKeyPrefix + "/ns*?$/tb$?*";
+    String escapedSpecialPath = "path/to/warehouse/ns${*}${?}${$}/tb${$}${?}${*}";
+
+    Mockito.when(stsClient.assumeRole(Mockito.isA(AssumeRoleRequest.class)))
+        .thenAnswer(
+            invocation -> {
+              AssumeRoleRequest request = invocation.getArgument(0);
+              IamPolicy policy = IamPolicy.fromJson(request.policy());
+
+              assertThat(policy.statements())
+                  .anySatisfy(
+                      statement ->
+                          assertThat(statement)
+                              .returns(
+                                  List.of(
+                                      IamAction.create("s3:PutObject"),
+                                      IamAction.create("s3:DeleteObject")),
+                                  IamStatement::actions)
+                              .returns(
+                                  List.of(
+                                      IamResource.create(
+                                          s3Arn(AWS_PARTITION, bucket, escapedSpecialPath))),
+                                  IamStatement::resources));
+
+              assertThat(policy.statements())
+                  .anySatisfy(
+                      statement ->
+                          assertThat(statement)
+                              .returns(
+                                  List.of(IamAction.create("s3:ListBucket")), IamStatement::actions)
+                              .returns(
+                                  List.of(IamResource.create(s3Arn(AWS_PARTITION, bucket, null))),
+                                  IamStatement::resources)
+                              .satisfies(
+                                  st ->
+                                      assertThat(st.conditions())
+                                          .containsExactly(
+                                              IamCondition.builder()
+                                                  .operator(IamConditionOperator.STRING_LIKE)
+                                                  .key("s3:prefix")
+                                                  .value(escapedSpecialPath + "/*")
+                                                  .build())));
+
+              assertThat(policy.statements())
+                  .anySatisfy(
+                      statement ->
+                          assertThat(statement)
+                              .returns(
+                                  List.of(
+                                      IamAction.create("s3:GetObject"),
+                                      IamAction.create("s3:GetObjectVersion")),
+                                  IamStatement::actions)
+                              .returns(
+                                  List.of(
+                                      IamResource.create(
+                                          s3Arn(AWS_PARTITION, bucket, escapedSpecialPath))),
+                                  IamStatement::resources));
+
+              return ASSUME_ROLE_RESPONSE;
+            });
+
+    new AwsCredentialsStorageIntegration(
+            AwsStorageConfigurationInfo.builder()
+                .addAllowedLocation(s3Path(bucket, warehouseKeyPrefix))
+                .roleARN(roleARN)
+                .externalId(externalId)
+                .region("us-east-1")
+                .build(),
+            stsClient)
+        .getSubscopedCreds(
+            EMPTY_REALM_CONFIG,
+            true,
+            Set.of(specialLocation),
+            Set.of(specialLocation),
+            POLARIS_PRINCIPAL,
+            Optional.empty(),
+            CredentialVendingContext.empty());
+  }
+
+  @Test
+  public void testGetSubscopedCredsInlinePolicyPreservesLiteralQuestionMarksInLocation() {
+    StsClient stsClient = Mockito.mock(StsClient.class);
+    String roleARN = "arn:aws:iam::012345678901:role/jdoe";
+    String externalId = "externalId";
+    String bucket = "bucket";
+    String warehouseKeyPrefix = "path/to/warehouse";
+    String specialLocation = "s3://bucket/" + warehouseKeyPrefix + "/ns?/tb?*";
+    String escapedSpecialPath = "path/to/warehouse/ns${?}/tb${?}${*}";
+
+    Mockito.when(stsClient.assumeRole(Mockito.isA(AssumeRoleRequest.class)))
+        .thenAnswer(
+            invocation -> {
+              AssumeRoleRequest request = invocation.getArgument(0);
+              IamPolicy policy = IamPolicy.fromJson(request.policy());
+
+              assertThat(policy.statements())
+                  .anySatisfy(
+                      statement ->
+                          assertThat(statement)
+                              .returns(
+                                  List.of(
+                                      IamAction.create("s3:PutObject"),
+                                      IamAction.create("s3:DeleteObject")),
+                                  IamStatement::actions)
+                              .returns(
+                                  List.of(
+                                      IamResource.create(
+                                          s3Arn(AWS_PARTITION, bucket, escapedSpecialPath))),
+                                  IamStatement::resources));
+
+              assertThat(policy.statements())
+                  .anySatisfy(
+                      statement ->
+                          assertThat(statement)
+                              .returns(
+                                  List.of(IamAction.create("s3:ListBucket")), IamStatement::actions)
+                              .returns(
+                                  List.of(IamResource.create(s3Arn(AWS_PARTITION, bucket, null))),
+                                  IamStatement::resources)
+                              .satisfies(
+                                  st ->
+                                      assertThat(st.conditions())
+                                          .containsExactly(
+                                              IamCondition.builder()
+                                                  .operator(IamConditionOperator.STRING_LIKE)
+                                                  .key("s3:prefix")
+                                                  .value(escapedSpecialPath + "/*")
+                                                  .build())));
+
+              assertThat(policy.statements())
+                  .anySatisfy(
+                      statement ->
+                          assertThat(statement)
+                              .returns(
+                                  List.of(
+                                      IamAction.create("s3:GetObject"),
+                                      IamAction.create("s3:GetObjectVersion")),
+                                  IamStatement::actions)
+                              .returns(
+                                  List.of(
+                                      IamResource.create(
+                                          s3Arn(AWS_PARTITION, bucket, escapedSpecialPath))),
+                                  IamStatement::resources));
+
+              return ASSUME_ROLE_RESPONSE;
+            });
+
+    new AwsCredentialsStorageIntegration(
+            AwsStorageConfigurationInfo.builder()
+                .addAllowedLocation(s3Path(bucket, warehouseKeyPrefix))
+                .roleARN(roleARN)
+                .externalId(externalId)
+                .region("us-east-1")
+                .build(),
+            stsClient)
+        .getSubscopedCreds(
+            EMPTY_REALM_CONFIG,
+            true,
+            Set.of(specialLocation),
+            Set.of(specialLocation),
+            POLARIS_PRINCIPAL,
+            Optional.empty(),
+            CredentialVendingContext.empty());
+  }
+
+  @ParameterizedTest
+  @CsvSource({
+    "plain, plain",
+    "'*', '${*}'",
+    "'?', '${?}'",
+    "'$', '${$}'",
+    "'abc $ def * ghi ? jkl', 'abc ${$} def ${*} ghi ${?} jkl'",
+    "'path/*/file', 'path/${*}/file'",
+    "'path/?/file', 'path/${?}/file'",
+    "'path/$/file', 'path/${$}/file'",
+    "'*?$', '${*}${?}${$}'",
+    "'path/*?$/$?*/file', 'path/${*}${?}${$}/${$}${?}${*}/file'",
+  })
+  public void testEscapeIamGlobLiteral(String input, String expectedOutput) {
+    assertThat(AwsCredentialsStorageIntegration.escapeIamGlobLiteral(input))
+        .isEqualTo(expectedOutput);
+  }
+
   @Test
   public void testGetSubscopedCredsInlinePolicyWithoutList() {
     StsClient stsClient = Mockito.mock(StsClient.class);

@@ -51,6 +51,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.stream.Stream;
 import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.storage.BaseStorageIntegrationTest;
 import org.apache.polaris.core.storage.CredentialVendingContext;
@@ -63,6 +64,9 @@
 import org.assertj.core.api.recursive.comparison.RecursiveComparisonConfiguration;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.Mockito;
 
@@ -274,6 +278,128 @@ public void testGenerateAccessBoundaryWithoutWrites() throws IOException {
         .isEqualTo(refRules);
   }
 
+  @Test
+  public void testGenerateAccessBoundaryQuotesCelLiteralCharacters() {
+    String path = "a'b\"c\\d";
+    CredentialAccessBoundary credentialAccessBoundary =
+        GcpCredentialsStorageIntegration.generateAccessBoundaryRules(
+            true, Set.of("gs://bucket1/" + path), Set.of("gs://bucket1/" + path));
+
+    ObjectMapper mapper = JsonMapper.builder().build();
+    JsonNode parsedRules = mapper.convertValue(credentialAccessBoundary, JsonNode.class);
+    assertThat(parsedRules.path("accessBoundaryRules")).hasSize(2);
+
+    assertThat(expressionAt(parsedRules, 0))
+        .isEqualTo(
+            "resource.name.startsWith('projects/_/buckets/bucket1/objects/"
+                + "a\\'b\\\"c\\\\d"
+                + "') || api.getAttribute('storage.googleapis.com/objectListPrefix', '').startsWith('"
+                + "a\\'b\\\"c\\\\d"
+                + "')");
+    assertThat(expressionAt(parsedRules, 1))
+        .isEqualTo(
+            "resource.name.startsWith('projects/_/buckets/bucket1/objects/a\\'b\\\"c\\\\d')");
+  }
+
+  @ParameterizedTest
+  @CsvSource(
+      delimiter = '|',
+      quoteCharacter = '`',
+      emptyValue = "",
+      value = {
+        "``|``",
+        "plain|plain",
+        "'|\\'",
+        "\\\"|\\\\\\\"",
+        "\\\\|\\\\\\\\",
+        "a'b\\\\c|a\\'b\\\\\\\\c",
+        "a\tb|a\\tb",
+      })
+  public void testescapeCelLiteral(String input, String expectedOutput) {
+    assertThat(GcpCredentialsStorageIntegration.escapeCelLiteral(input)).isEqualTo(expectedOutput);
+  }
+
+  @ParameterizedTest
+  @MethodSource("handledControlCharacterCases")
+  public void testescapeCelLiteralEscapesHandledControlCharacters(
+      String input, String expectedOutput) {
+    assertThat(GcpCredentialsStorageIntegration.escapeCelLiteral(input)).isEqualTo(expectedOutput);
+  }
+
+  private static Stream<Arguments> handledControlCharacterCases() {
+    return Stream.of(
+        Arguments.of("\b", "\\b"),
+        Arguments.of("a\bb", "a\\bb"),
+        Arguments.of("\f", "\\f"),
+        Arguments.of("a\fb", "a\\fb"),
+        Arguments.of("\n", "\\n"),
+        Arguments.of("a\nb", "a\\nb"),
+        Arguments.of("\r", "\\r"),
+        Arguments.of("a\rb", "a\\rb"));
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"\uD83D\uDE00", "prefix-\uD834\uDD1E-suffix"})
+  public void testescapeCelLiteralPreservesValidSurrogatePairs(String input) {
+    assertThat(GcpCredentialsStorageIntegration.escapeCelLiteral(input)).isEqualTo(input);
+  }
+
+  @ParameterizedTest
+  @MethodSource("unpairedSurrogateCases")
+  public void testescapeCelLiteralRejectsUnpairedSurrogates(String input) {
+    Assertions.assertThatThrownBy(() -> GcpCredentialsStorageIntegration.escapeCelLiteral(input))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Unsupported unpaired surrogate");
+  }
+
+  private static Stream<String> unpairedSurrogateCases() {
+    return Stream.of(
+        String.valueOf((char) 0xD83D),
+        String.valueOf((char) 0xDE00),
+        new String(new char[] {(char) 0xD83D, 'a'}),
+        new String(new char[] {'a', (char) 0xDE00}));
+  }
+
+  @ParameterizedTest
+  @ValueSource(ints = {0x0000, 0x0001, 0x0007, 0x000B, 0x001F, 0x007F})
+  public void testescapeCelLiteralRejectsUnsupportedControlCharacters(int codePoint) {
+    String input = "a" + (char) codePoint + "b";
+
+    Assertions.assertThatThrownBy(() -> GcpCredentialsStorageIntegration.escapeCelLiteral(input))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Unsupported control character");
+  }
+
+  @Test
+  public void testGenerateAccessBoundaryRejectsUnsupportedCelLiteralCharacters() {
+    Assertions.assertThatThrownBy(
+            () ->
+                GcpCredentialsStorageIntegration.generateAccessBoundaryRules(
+                    true, Set.of("gs://bucket1/a\u001fb"), Set.of()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Unsupported control character");
+  }
+
+  @Test
+  public void testGenerateAccessBoundaryPreservesLiteralQuestionMarksInPath() {
+    CredentialAccessBoundary credentialAccessBoundary =
+        GcpCredentialsStorageIntegration.generateAccessBoundaryRules(
+            true, Set.of("gs://bucket1/path/to/data?with?question"), Set.of());
+
+    ObjectMapper mapper = JsonMapper.builder().build();
+    JsonNode parsedRules = mapper.convertValue(credentialAccessBoundary, JsonNode.class);
+
+    assertThat(
+            parsedRules
+                .path("accessBoundaryRules")
+                .get(0)
+                .path("availabilityCondition")
+                .path("expression")
+                .asText())
+        .contains("projects/_/buckets/bucket1/objects/path/to/data?with?question")
+        .contains("startsWith('path/to/data?with?question')");
+  }
+
   /**
    * Custom comparator as ObjectNodes are compared by field indexes as opposed to field names. They
    * also don't equate a field that is present and set to null with a field that is omitted
@@ -387,4 +513,13 @@ protected AccessToken refreshAccessToken(DownscopedCredentials credentials) {
   private boolean isNotNull(JsonNode node) {
     return node != null && !node.isNull();
   }
+
+  private static String expressionAt(JsonNode parsedRules, int ruleIndex) {
+    return parsedRules
+        .path("accessBoundaryRules")
+        .path(ruleIndex)
+        .path("availabilityCondition")
+        .path("expression")
+        .asText();
+  }
 }

@@ -51,6 +51,7 @@
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.OutputFile;
@@ -265,6 +266,92 @@ public void testCreateTableVendedCredentials(boolean pathStyle) throws IOExcepti
     assertThat(response.credentials()).hasSize(1);
   }
 
+  @Test
+  public void testCreateTableVendedCredentialsWithPartialAwsShapePasses() throws IOException {
+    try (RESTCatalog restCatalog =
+        createCatalog(
+            Optional.of(endpoint),
+            Optional.of(endpoint),
+            true,
+            Optional.empty(),
+            Optional.of(VENDED_CREDENTIALS),
+            true)) {
+      TableIdentifier id = createTableAndVerifyMetadata(restCatalog);
+      try {
+        assertLoadTableWithVendedCredentialsSucceeds(id);
+      } finally {
+        catalogApi.dropTable(catalogName, id);
+      }
+    }
+  }
+
+  @Test
+  public void testVendedCredentialsFailClosedForWildcardTableIdentifiers() throws IOException {
+    try (RESTCatalog restCatalog =
+        createCatalog(
+            Optional.of(endpoint),
+            Optional.of(endpoint),
+            true,
+            Optional.empty(),
+            Optional.of(VENDED_CREDENTIALS),
+            true)) {
+      Namespace targetNamespace = Namespace.of("foo");
+      Namespace checkedNamespace = Namespace.of("*");
+      TableIdentifier targetId = TableIdentifier.of(targetNamespace, "target");
+      TableIdentifier checkedId = TableIdentifier.of(checkedNamespace, "*");
+      restCatalog.createNamespace(targetNamespace);
+      restCatalog.createNamespace(checkedNamespace);
+      restCatalog.createTable(targetId, SCHEMA);
+
+      try {
+        assertThatThrownBy(() -> restCatalog.createTable(checkedId, SCHEMA))
+            .hasMessageContaining("Access Denied");
+      } finally {
+        if (restCatalog.tableExists(checkedId)) {
+          restCatalog.dropTable(checkedId);
+        }
+        if (restCatalog.tableExists(targetId)) {
+          restCatalog.dropTable(targetId);
+        }
+      }
+    }
+  }
+
+  private TableIdentifier createTableAndVerifyMetadata(RESTCatalog restCatalog) {
+    catalogApi.createNamespace(catalogName, "test-ns");
+    TableIdentifier id = TableIdentifier.of("test-ns", "t1");
+    Table table = restCatalog.createTable(id, SCHEMA);
+    assertThat(table).isNotNull();
+
+    TableOperations ops = ((HasTableOperations) table).operations();
+    URI location = URI.create(ops.current().metadataFileLocation());
+
+    GetObjectResponse response =
+        s3Client
+            .getObject(
+                GetObjectRequest.builder()
+                    .bucket(location.getAuthority())
+                    .key(location.getPath().substring(1)) // drop leading slash
+                    .build())
+            .response();
+    assertThat(response.contentLength()).isGreaterThan(0);
+    return id;
+  }
+
+  private void assertLoadTableWithVendedCredentialsSucceeds(TableIdentifier id) {
+    LoadTableResponse response =
+        catalogApi.loadTable(
+            catalogName,
+            id,
+            "ALL",
+            Map.of("X-Iceberg-Access-Delegation", VENDED_CREDENTIALS.protocolValue()));
+    assertThat(response.config())
+        .containsEntry(
+            REFRESH_CREDENTIALS_ENDPOINT,
+            "v1/" + catalogName + "/namespaces/test-ns/tables/t1/credentials");
+    assertThat(response.credentials()).hasSize(1);
+  }
+
   private LoadTableResponse doTestCreateTable(
       boolean pathStyle, Optional<AccessDelegationMode> dm, boolean stsEnabled) throws IOException {
     try (RESTCatalog restCatalog =