fix(auth): bind internal JWTs to principal secret generation

CHANGELOG.md

@@ -79,6 +79,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 - `TableCleanupTaskHandler` now clamps `TABLE_METADATA_CLEANUP_BATCH_SIZE` to at least 1. Previously a non-positive realm override caused an infinite loop (0) or `IllegalArgumentException` (<0) when splitting metadata files for cleanup.
 - `ManifestFileCleanupTaskHandler` now handles Iceberg v2 delete manifests in addition to data manifests. Previously, `DROP TABLE PURGE` on a v2 table that had been updated via merge-on-read DML left position-delete files and their manifests as orphans in object storage; the cleanup task failed silently because `ManifestFiles.read()` rejects delete manifests.
 - OPA authorizer now includes the realm identifier in the authorization context sent to OPA (`input.context.realm`). This ensures OPA policies can enforce tenant isolation across realms, preventing potential collisions if identical principal or resource names exist in different realms.
+- Internal JWTs are bound to a salted credentials-version fingerprint (`polaris-cv` claim, derived from secret hashes plus the principal salt so raw hashes are never exposed in the token). Verify accepts the current main or secondary generation so dual-secret rotate grace matches client-secret matching; a further rotate/reset invalidates older bound tokens. Tokens minted before this change carry no claim and remain valid until expiry for upgrade compatibility.
 
 ## [1.6.0]
 

polaris-core/src/main/java/org/apache/polaris/core/DigestUtils.java

@@ -20,6 +20,8 @@
 
 import com.google.common.hash.Hashing;
 import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.Objects;
 
 public final class DigestUtils {
   private DigestUtils() {
@@ -29,4 +31,14 @@ private DigestUtils() {
   public static String sha256Hex(String input) {
     return Hashing.sha256().hashString(input, StandardCharsets.UTF_8).toString();
   }
+
+  /**
+   * Constant-time equality check for secret material. Both arguments must be non-null; callers are
+   * expected to guard nullable inputs before comparing.
+   */
+  public static boolean constantTimeEquals(String a, String b) {
+    return MessageDigest.isEqual(
+        Objects.requireNonNull(a).getBytes(StandardCharsets.UTF_8),
+        Objects.requireNonNull(b).getBytes(StandardCharsets.UTF_8));
+  }
 }

polaris-core/src/main/java/org/apache/polaris/core/entity/PolarisPrincipalSecrets.java

@@ -185,6 +185,55 @@ public String getMainSecretHash() {
     return mainSecretHash;
   }
 
+  /**
+   * Credentials-generation fingerprint for tokens minted against the <em>current</em> main secret.
+   * Derived from existing fields with the principal's secret salt, so no schema change is required.
+   * The value is not itself a credential and is safe to embed in signed (but not encrypted) JWTs.
+   *
+   * @return the main credentials version, or {@code null} when no main secret hash is set
+   */
+  @Nullable
+  public String getCredentialsVersion() {
+    return credentialsVersionForHash(mainSecretHash);
+  }
+
+  /**
+   * Returns true when {@code credentialsVersion} matches the salted fingerprint of the current main
+   * secret hash <em>or</em> the secondary secret hash. After a single rotate, the previous main
+   * hash is kept as secondary so client secrets and bound JWTs both remain valid for that
+   * generation; a further rotate/reset advances secondary and invalidates the older fingerprint.
+   *
+   * <p>Comparisons are constant-time against each candidate fingerprint.
+   */
+  public boolean matchesCredentialsVersion(@Nullable String credentialsVersion) {
+    if (credentialsVersion == null || credentialsVersion.isEmpty()) {
+      return false;
+    }
+    boolean matches = false;
+    String mainVersion = credentialsVersionForHash(mainSecretHash);
+    if (mainVersion != null) {
+      matches |= DigestUtils.constantTimeEquals(credentialsVersion, mainVersion);
+    }
+    String secondaryVersion = credentialsVersionForHash(secondarySecretHash);
+    if (secondaryVersion != null) {
+      matches |= DigestUtils.constantTimeEquals(credentialsVersion, secondaryVersion);
+    }
+    return matches;
+  }
+
+  /**
+   * Salted, non-reversible fingerprint of a secret-verification hash for embedding in JWTs. Uses
+   * the same per-principal {@link #secretSalt} already stored for secret hashing, which is always
+   * set by the constructors.
+   */
+  @Nullable
+  private String credentialsVersionForHash(@Nullable String secretHash) {
+    if (secretHash == null || secretHash.isEmpty()) {
+      return null;
+    }
+    return DigestUtils.sha256Hex(secretHash + ":" + secretSalt);
+  }
+
   public String getSecondarySecretHash() {
     return secondarySecretHash;
   }

polaris-core/src/test/java/org/apache/polaris/core/DigestUtilsTest.java

@@ -0,0 +1,45 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.junit.jupiter.api.Test;
+
+public class DigestUtilsTest {
+
+  @Test
+  void constantTimeEquals() {
+    assertThat(DigestUtils.constantTimeEquals("abc", "abc")).isTrue();
+    assertThat(DigestUtils.constantTimeEquals("abc", "abd")).isFalse();
+    assertThat(DigestUtils.constantTimeEquals("abc", "abcd")).isFalse();
+    assertThat(DigestUtils.constantTimeEquals("", "")).isTrue();
+  }
+
+  @Test
+  void constantTimeEqualsRejectsNull() {
+    assertThatThrownBy(() -> DigestUtils.constantTimeEquals(null, "a"))
+        .isInstanceOf(NullPointerException.class);
+    assertThatThrownBy(() -> DigestUtils.constantTimeEquals("a", null))
+        .isInstanceOf(NullPointerException.class);
+    assertThatThrownBy(() -> DigestUtils.constantTimeEquals(null, null))
+        .isInstanceOf(NullPointerException.class);
+  }
+}

runtime/service/src/main/java/org/apache/polaris/service/auth/internal/broker/JWTBroker.java

@@ -22,20 +22,23 @@
 import com.auth0.jwt.algorithms.Algorithm;
 import com.auth0.jwt.interfaces.DecodedJWT;
 import com.auth0.jwt.interfaces.JWTVerifier;
+import com.google.common.annotations.VisibleForTesting;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import org.apache.iceberg.exceptions.NotAuthorizedException;
 import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.entity.PolarisPrincipalSecrets;
 import org.apache.polaris.core.entity.PrincipalEntity;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.dao.entity.PrincipalSecretsResult;
 import org.apache.polaris.service.auth.DefaultAuthenticator;
 import org.apache.polaris.service.auth.PolarisCredential;
 import org.apache.polaris.service.auth.internal.service.OAuthError;
 import org.apache.polaris.service.types.TokenType;
+import org.jspecify.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -49,6 +52,14 @@ public class JWTBroker implements TokenBroker {
   private static final String CLAIM_KEY_PRINCIPAL_ID = "principalId";
   private static final String CLAIM_KEY_SCOPE = "scope";
 
+  /**
+   * Credentials-generation fingerprint bound to the principal's current credentials version (see
+   * {@link PolarisPrincipalSecrets#getCredentialsVersion()}). When secrets are rotated or reset,
+   * the version changes and tokens carrying a prior value are rejected on verify. The claim name is
+   * collision-resistant per RFC 7519.
+   */
+  @VisibleForTesting static final String CLAIM_KEY_CREDENTIALS_VERSION = "polaris-cv";
+
   private final PolarisMetaStoreManager metaStoreManager;
   private final PolarisCallContext polarisCallContext;
   private final int maxTokenGenerationInSeconds;
@@ -78,24 +89,62 @@ static JWTVerifier buildVerifier(Algorithm algorithm) {
 
   @Override
   public PolarisCredential verify(String token) {
-    return verifyInternal(token);
+    return verifyInternal(token).token();
   }
 
-  private InternalPolarisToken verifyInternal(String token) {
+  private VerifiedToken verifyInternal(String token) {
     try {
       DecodedJWT decodedJWT = verifier.verify(token);
-      return InternalPolarisToken.of(
-          decodedJWT.getSubject(),
-          decodedJWT.getClaim(CLAIM_KEY_PRINCIPAL_ID).asLong(),
-          decodedJWT.getClaim(CLAIM_KEY_CLIENT_ID).asString(),
-          decodedJWT.getClaim(CLAIM_KEY_SCOPE).asString());
+      String clientId = decodedJWT.getClaim(CLAIM_KEY_CLIENT_ID).asString();
+      String credentialsVersion = decodedJWT.getClaim(CLAIM_KEY_CREDENTIALS_VERSION).asString();
+      PolarisPrincipalSecrets secrets =
+          assertCredentialsVersionCurrent(clientId, credentialsVersion);
+      return new VerifiedToken(
+          InternalPolarisToken.of(
+              decodedJWT.getSubject(),
+              decodedJWT.getClaim(CLAIM_KEY_PRINCIPAL_ID).asLong(),
+              clientId,
+              decodedJWT.getClaim(CLAIM_KEY_SCOPE).asString()),
+          secrets);
 
+    } catch (NotAuthorizedException e) {
+      throw e;
     } catch (Exception e) {
       throw (NotAuthorizedException)
           new NotAuthorizedException("Failed to verify the token").initCause(e);
     }
   }
 
+  /**
+   * Rejects tokens that were minted against a credentials generation that is no longer current.
+   * Accepts the main or secondary generation fingerprint so dual-secret rotate grace matches
+   * client-secret matching. Tokens minted before credentials binding (no claim) are accepted so
+   * existing clients are not broken on upgrade; they remain unbound until expiry.
+   *
+   * @return the loaded principal secrets when the claim was checked, or {@code null} for legacy
+   *     tokens without the claim (callers can reuse the secrets to avoid a second load)
+   */
+  @Nullable
+  private PolarisPrincipalSecrets assertCredentialsVersionCurrent(
+      String clientId, String credentialsVersion) {
+    if (credentialsVersion == null) {
+      // Legacy token minted before credentials binding existed.
+      return null;
+    }
+    if (clientId == null || clientId.isBlank() || credentialsVersion.isBlank()) {
+      throw new NotAuthorizedException("Failed to verify the token");
+    }
+    PrincipalSecretsResult secretsResult =
+        metaStoreManager.loadPrincipalSecrets(polarisCallContext, clientId);
+    if (!secretsResult.isSuccess() || secretsResult.getPrincipalSecrets() == null) {
+      throw new NotAuthorizedException("Failed to verify the token");
+    }
+    if (!secretsResult.getPrincipalSecrets().matchesCredentialsVersion(credentialsVersion)) {
+      throw new NotAuthorizedException("Failed to verify the token");
+    }
+    return secretsResult.getPrincipalSecrets();
+  }
+
   @Override
   public TokenResponse generateFromToken(
       TokenType subjectTokenType,
@@ -112,13 +161,15 @@ public TokenResponse generateFromToken(
     if (subjectToken == null || subjectToken.isBlank()) {
       return TokenResponse.of(OAuthError.invalid_request);
     }
-    InternalPolarisToken decodedToken;
+    VerifiedToken verified;
     try {
-      decodedToken = verifyInternal(subjectToken);
+      // verifyInternal enforces credentials-version binding (rejects after rotate/reset).
+      verified = verifyInternal(subjectToken);
     } catch (NotAuthorizedException e) {
       LOGGER.error("Failed to verify the token", e.getCause());
       return TokenResponse.of(OAuthError.invalid_client);
     }
+    InternalPolarisToken decodedToken = verified.token();
     Optional<PrincipalEntity> principalLookup =
         metaStoreManager.findPrincipalById(polarisCallContext, decodedToken.getPrincipalId());
     if (principalLookup.isEmpty()) {
@@ -138,12 +189,28 @@ public TokenResponse generateFromToken(
       }
       tokenScope = scope;
     }
+    String credentialsVersion;
+    if (verified.principalSecrets() != null) {
+      // Secrets were already loaded during verify; reuse them instead of a second read.
+      credentialsVersion = verified.principalSecrets().getCredentialsVersion();
+      if (credentialsVersion == null) {
+        return TokenResponse.of(OAuthError.unauthorized_client);
+      }
+    } else {
+      // Legacy token without the claim: load once to bind the re-minted token.
+      Optional<String> loaded = loadCurrentCredentialsVersion(decodedToken.getClientId());
+      if (loaded.isEmpty()) {
+        return TokenResponse.of(OAuthError.unauthorized_client);
+      }
+      credentialsVersion = loaded.get();
+    }
     String tokenString =
         generateTokenString(
             decodedToken.getPrincipalName(),
             decodedToken.getPrincipalId(),
             decodedToken.getClientId(),
-            tokenScope);
+            tokenScope,
+            credentialsVersion);
     return TokenResponse.of(
         tokenString, TokenType.ACCESS_TOKEN.getValue(), maxTokenGenerationInSeconds);
   }
@@ -163,30 +230,45 @@ public TokenResponse generateFromClientSecrets(
       return TokenResponse.of(initialValidationResponse.get());
     }
 
-    Optional<PrincipalEntity> principal = findPrincipalEntity(clientId, clientSecret);
-    if (principal.isEmpty()) {
+    Optional<AuthenticatedPrincipal> authenticated =
+        authenticateWithClientSecrets(clientId, clientSecret);
+    if (authenticated.isEmpty()) {
       return TokenResponse.of(OAuthError.unauthorized_client);
     }
+    AuthenticatedPrincipal principal = authenticated.get();
     String tokenString =
-        generateTokenString(principal.get().getName(), principal.get().getId(), clientId, scope);
+        generateTokenString(
+            principal.entity().getName(),
+            principal.entity().getId(),
+            clientId,
+            scope,
+            principal.credentialsVersion());
     return TokenResponse.of(
         tokenString, TokenType.ACCESS_TOKEN.getValue(), maxTokenGenerationInSeconds);
   }
 
   private String generateTokenString(
-      String principalName, long principalId, String clientId, String scope) {
+      String principalName,
+      long principalId,
+      String clientId,
+      String scope,
+      String credentialsVersion) {
     Instant now = Instant.now();
-    return JWT.create()
-        .withIssuer(ISSUER_KEY)
-        .withSubject(principalName)
-        .withIssuedAt(now)
-        .withExpiresAt(now.plus(maxTokenGenerationInSeconds, ChronoUnit.SECONDS))
-        .withJWTId(UUID.randomUUID().toString())
-        .withClaim(CLAIM_KEY_ACTIVE, true)
-        .withClaim(CLAIM_KEY_CLIENT_ID, clientId)
-        .withClaim(CLAIM_KEY_PRINCIPAL_ID, principalId)
-        .withClaim(CLAIM_KEY_SCOPE, scopes(scope))
-        .sign(algorithm);
+    var builder =
+        JWT.create()
+            .withIssuer(ISSUER_KEY)
+            .withSubject(principalName)
+            .withIssuedAt(now)
+            .withExpiresAt(now.plus(maxTokenGenerationInSeconds, ChronoUnit.SECONDS))
+            .withJWTId(UUID.randomUUID().toString())
+            .withClaim(CLAIM_KEY_ACTIVE, true)
+            .withClaim(CLAIM_KEY_CLIENT_ID, clientId)
+            .withClaim(CLAIM_KEY_PRINCIPAL_ID, principalId)
+            .withClaim(CLAIM_KEY_SCOPE, scopes(scope));
+    if (credentialsVersion != null) {
+      builder.withClaim(CLAIM_KEY_CREDENTIALS_VERSION, credentialsVersion);
+    }
+    return builder.sign(algorithm);
   }
 
   @Override
@@ -203,17 +285,41 @@ private String scopes(String scope) {
     return scope == null || scope.isBlank() ? DefaultAuthenticator.PRINCIPAL_ROLE_ALL : scope;
   }
 
-  private Optional<PrincipalEntity> findPrincipalEntity(String clientId, String clientSecret) {
-    // Validate the principal is present and secrets match
+  private Optional<String> loadCurrentCredentialsVersion(String clientId) {
     PrincipalSecretsResult principalSecrets =
         metaStoreManager.loadPrincipalSecrets(polarisCallContext, clientId);
-    if (!principalSecrets.isSuccess()) {
+    if (!principalSecrets.isSuccess() || principalSecrets.getPrincipalSecrets() == null) {
       return Optional.empty();
     }
-    if (!principalSecrets.getPrincipalSecrets().matchesSecret(clientSecret)) {
+    return Optional.ofNullable(principalSecrets.getPrincipalSecrets().getCredentialsVersion());
+  }
+
+  private Optional<AuthenticatedPrincipal> authenticateWithClientSecrets(
+      String clientId, String clientSecret) {
+    PrincipalSecretsResult principalSecrets =
+        metaStoreManager.loadPrincipalSecrets(polarisCallContext, clientId);
+    if (!principalSecrets.isSuccess() || principalSecrets.getPrincipalSecrets() == null) {
       return Optional.empty();
     }
-    return metaStoreManager.findPrincipalById(
-        polarisCallContext, principalSecrets.getPrincipalSecrets().getPrincipalId());
+    PolarisPrincipalSecrets secrets = principalSecrets.getPrincipalSecrets();
+    if (!secrets.matchesSecret(clientSecret)) {
+      return Optional.empty();
+    }
+    Optional<PrincipalEntity> principal =
+        metaStoreManager.findPrincipalById(polarisCallContext, secrets.getPrincipalId());
+    if (principal.isEmpty()) {
+      return Optional.empty();
+    }
+    return Optional.of(
+        new AuthenticatedPrincipal(principal.get(), secrets.getCredentialsVersion()));
   }
+
+  private record AuthenticatedPrincipal(PrincipalEntity entity, String credentialsVersion) {}
+
+  /**
+   * Result of token verification. {@code principalSecrets} carries the secrets loaded while
+   * checking the credentials-version claim, or {@code null} for legacy tokens without the claim.
+   */
+  private record VerifiedToken(
+      InternalPolarisToken token, @Nullable PolarisPrincipalSecrets principalSecrets) {}
 }