Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,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 JWT token exchange no longer issues an access token that outlives the subject token. Previously each exchange reset expiry to a full `maxTokenGeneration` window, so a stolen access token could be refreshed indefinitely. Exchanged tokens are now capped to the remaining subject lifetime (and still to `maxTokenGeneration`).

## [1.6.0]

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
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;
Expand Down Expand Up @@ -77,18 +78,19 @@ static JWTVerifier buildVerifier(Algorithm algorithm) {

@Override
public PolarisCredential verify(String token) {
return verifyInternal(token);
return verifyInternal(token).token();
}

private InternalPolarisToken verifyInternal(String token) {
private VerifiedSubjectToken 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());

InternalPolarisToken polarisToken =
InternalPolarisToken.of(
decodedJWT.getSubject(),
decodedJWT.getClaim(CLAIM_KEY_PRINCIPAL_ID).asLong(),
decodedJWT.getClaim(CLAIM_KEY_CLIENT_ID).asString(),
decodedJWT.getClaim(CLAIM_KEY_SCOPE).asString());
return new VerifiedSubjectToken(polarisToken, decodedJWT.getExpiresAtAsInstant());
} catch (Exception e) {
throw (NotAuthorizedException)
new NotAuthorizedException("Failed to verify the token").initCause(e);
Expand All @@ -111,26 +113,63 @@ public TokenResponse generateFromToken(
if (subjectToken == null || subjectToken.isBlank()) {
return TokenResponse.of(OAuthError.invalid_request);
}
InternalPolarisToken decodedToken;
VerifiedSubjectToken verified;
try {
decodedToken = verifyInternal(subjectToken);
verified = verifyInternal(subjectToken);
} catch (NotAuthorizedException e) {
LOGGER.error("Failed to verify the token", e.getCause());
return TokenResponse.of(OAuthError.invalid_client);
}
Instant now = Instant.now();
Optional<Integer> expiresInSeconds =
remainingLifetimeSeconds(now, verified.expiresAt(), maxTokenGenerationInSeconds);
if (expiresInSeconds.isEmpty()) {
// Subject is expired, missing exp, or has no remaining lifetime — do not mint a longer
// session.
return TokenResponse.of(OAuthError.invalid_grant);
}
Optional<PrincipalEntity> principalLookup =
metaStoreManager.findPrincipalById(polarisCallContext, decodedToken.getPrincipalId());
metaStoreManager.findPrincipalById(polarisCallContext, verified.token().getPrincipalId());
if (principalLookup.isEmpty()) {
return TokenResponse.of(OAuthError.unauthorized_client);
}
Instant newExpiresAt = now.plus(expiresInSeconds.get(), ChronoUnit.SECONDS);
Comment thread
vigneshio marked this conversation as resolved.
String tokenString =
generateTokenString(
decodedToken.getPrincipalName(),
decodedToken.getPrincipalId(),
decodedToken.getClientId(),
decodedToken.getScope());
return TokenResponse.of(
tokenString, TokenType.ACCESS_TOKEN.getValue(), maxTokenGenerationInSeconds);
verified.token().getPrincipalName(),
verified.token().getPrincipalId(),
verified.token().getClientId(),
verified.token().getScope(),
now,
newExpiresAt);
return TokenResponse.of(tokenString, TokenType.ACCESS_TOKEN.getValue(), expiresInSeconds.get());
}

/**
* Computes the lifetime for a token minted via token exchange.
*
* <p>The exchanged token must not outlive the subject access token. Combined with the configured
* maximum token generation duration, this prevents token-exchange from being used as an infinite
* refresh mechanism that resets expiry to a full new session on every call.
*
* @return remaining lifetime in whole seconds, or empty when the subject must not be exchanged
*/
@VisibleForTesting
static Optional<Integer> remainingLifetimeSeconds(
Instant now, Instant subjectExpiresAt, int maxTokenGenerationInSeconds) {
if (subjectExpiresAt == null || !subjectExpiresAt.isAfter(now)) {
return Optional.empty();
}
if (maxTokenGenerationInSeconds <= 0) {
return Optional.empty();
}
long remainingSeconds = ChronoUnit.SECONDS.between(now, subjectExpiresAt);
if (remainingSeconds < 1) {
return Optional.empty();
}
long capped = Math.min(remainingSeconds, maxTokenGenerationInSeconds);
// TokenResponse / OAuth expires_in is an int; clamp to Integer.MAX_VALUE defensively.
return Optional.of((int) Math.min(capped, Integer.MAX_VALUE));
}

@Override
Expand All @@ -152,20 +191,27 @@ public TokenResponse generateFromClientSecrets(
if (principal.isEmpty()) {
return TokenResponse.of(OAuthError.unauthorized_client);
}
Instant now = Instant.now();
Instant expiresAt = now.plus(maxTokenGenerationInSeconds, ChronoUnit.SECONDS);
String tokenString =
generateTokenString(principal.get().getName(), principal.get().getId(), clientId, scope);
generateTokenString(
principal.get().getName(), principal.get().getId(), clientId, scope, now, expiresAt);
return TokenResponse.of(
tokenString, TokenType.ACCESS_TOKEN.getValue(), maxTokenGenerationInSeconds);
}

private String generateTokenString(
String principalName, long principalId, String clientId, String scope) {
Instant now = Instant.now();
String principalName,
long principalId,
String clientId,
String scope,
Instant issuedAt,
Instant expiresAt) {
return JWT.create()
.withIssuer(ISSUER_KEY)
.withSubject(principalName)
.withIssuedAt(now)
.withExpiresAt(now.plus(maxTokenGenerationInSeconds, ChronoUnit.SECONDS))
.withIssuedAt(issuedAt)
.withExpiresAt(expiresAt)
.withJWTId(UUID.randomUUID().toString())
.withClaim(CLAIM_KEY_ACTIVE, true)
.withClaim(CLAIM_KEY_CLIENT_ID, clientId)
Expand Down Expand Up @@ -201,4 +247,6 @@ private Optional<PrincipalEntity> findPrincipalEntity(String clientId, String cl
return metaStoreManager.findPrincipalById(
polarisCallContext, principalSecrets.getPrincipalSecrets().getPrincipalId());
}

private record VerifiedSubjectToken(InternalPolarisToken token, Instant expiresAt) {}
}
Original file line number Diff line number Diff line change
Expand Up @@ -75,12 +75,8 @@ public Response getToken(
if (!tokenBroker.supportsRequestedTokenType(requestedTokenType)) {
return OAuthUtils.getResponseFromError(OAuthError.invalid_request);
}
if (authHeader == null && clientSecret == null) {
return OAuthUtils.getResponseFromError(OAuthError.invalid_client);
}
// token exchange with client id and client secret in the authorization header means the client
// has previously attempted to refresh an access token, but refreshing was not supported by the
// token broker. Accept the client id and secret and treat it as a new token request
// Parse HTTP Basic credentials first. A non-Basic Authorization header alone is not a
// client credential and must not unlock the token-exchange path.
if (authHeader != null && clientSecret == null && authHeader.startsWith("Basic ")) {
String credentials =
new String(Base64.getDecoder().decode(authHeader.substring(6).getBytes(UTF_8)), UTF_8);
Expand All @@ -99,15 +95,20 @@ public Response getToken(
}
TokenResponse tokenResponse;
if (clientSecret != null) {
// Client presents a secret (body or Basic). Issue/refresh via client authentication.
// Note: when grant_type is token-exchange and a secret is present, historical behavior
// treats this as a fresh client_credentials-style mint (full configured TTL).
tokenResponse =
tokenBroker.generateFromClientSecrets(
clientId, clientSecret, grantType, scope, requestedTokenType);
} else if (subjectToken != null) {
} else if (subjectToken != null && !subjectToken.isBlank()) {
// Token exchange without client secret: mint a new access token that cannot outlive the
// subject token (enforced by JWTBroker). Dummy Authorization headers are not required.
tokenResponse =
tokenBroker.generateFromToken(
subjectTokenType, subjectToken, grantType, scope, requestedTokenType);
} else {
return OAuthUtils.getResponseFromError(OAuthError.invalid_request);
return OAuthUtils.getResponseFromError(OAuthError.invalid_client);
}
if (tokenResponse == null) {
return OAuthUtils.getResponseFromError(OAuthError.unsupported_grant_type);
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,192 @@
/*
* 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.service.auth.internal.broker;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.when;

import com.auth0.jwt.JWT;
import com.auth0.jwt.algorithms.Algorithm;
import com.auth0.jwt.interfaces.DecodedJWT;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.Optional;
import java.util.stream.Stream;
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.internal.service.OAuthError;
import org.apache.polaris.service.types.TokenType;
import org.junit.jupiter.api.BeforeEach;
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.MethodSource;
import org.mockito.Mockito;

/**
* Token-exchange lifetime tests for {@link JWTBroker}: exchanged access tokens must not outlive the
* subject token (no infinite refresh via expiry reset).
*/
public class JWTBrokerTokenExchangeTest {

private static final int MAX_TOKEN_SECONDS = 3600;
private static final long PRINCIPAL_ID = 42L;
private static final String CLIENT_ID = "client-id";
private static final String MAIN_SECRET = "main-secret";
private static final String SCOPE = "PRINCIPAL_ROLE:ANALYST";

private PolarisCallContext polarisCallContext;
private PolarisMetaStoreManager metaStoreManager;
private Algorithm algorithm;
private JWTBroker broker;

@BeforeEach
void setUp() {
polarisCallContext = Mockito.mock(PolarisCallContext.class);
metaStoreManager = Mockito.mock(PolarisMetaStoreManager.class);
algorithm = Algorithm.HMAC256("test-hmac-secret");
broker =
new JWTBroker(
metaStoreManager,
polarisCallContext,
MAX_TOKEN_SECONDS,
algorithm,
JWTBroker.buildVerifier(algorithm));

PolarisPrincipalSecrets secrets =
new PolarisPrincipalSecrets(PRINCIPAL_ID, CLIENT_ID, MAIN_SECRET, "secondary");
when(metaStoreManager.loadPrincipalSecrets(polarisCallContext, CLIENT_ID))
.thenReturn(new PrincipalSecretsResult(secrets));
PrincipalEntity principal =
new PrincipalEntity.Builder().setId(PRINCIPAL_ID).setName("analyst").build();
when(metaStoreManager.findPrincipalById(polarisCallContext, PRINCIPAL_ID))
.thenReturn(Optional.of(principal));
}

@Test
void tokenExchangeDoesNotExtendBeyondSubjectExpiry() {
TokenResponse original =
broker.generateFromClientSecrets(
CLIENT_ID,
MAIN_SECRET,
TokenRequestValidator.CLIENT_CREDENTIALS,
SCOPE,
TokenType.ACCESS_TOKEN);
assertThat(original.getError()).isNull();
DecodedJWT originalJwt = JWT.decode(original.getAccessToken());
Instant originalExpiry = originalJwt.getExpiresAtAsInstant();

TokenResponse exchanged =
broker.generateFromToken(
TokenType.ACCESS_TOKEN,
original.getAccessToken(),
TokenRequestValidator.TOKEN_EXCHANGE,
SCOPE,
TokenType.ACCESS_TOKEN);

assertThat(exchanged.getError()).isNull();
assertThat(exchanged.getExpiresIn()).isPositive();
assertThat(exchanged.getExpiresIn()).isLessThanOrEqualTo(MAX_TOKEN_SECONDS);

DecodedJWT exchangedJwt =
JWT.require(algorithm).withIssuer("polaris").build().verify(exchanged.getAccessToken());
Instant exchangedExpiry = exchangedJwt.getExpiresAtAsInstant();
// Exchanged token must not outlive the subject.
assertThat(exchangedExpiry).isBeforeOrEqualTo(originalExpiry);
assertThat(exchanged.getExpiresIn())
.isLessThanOrEqualTo((int) ChronoUnit.SECONDS.between(Instant.now(), originalExpiry) + 1);
}

@Test
void repeatedTokenExchangeCannotResetToFullSessionLifetime() {
TokenResponse original =
broker.generateFromClientSecrets(
CLIENT_ID,
MAIN_SECRET,
TokenRequestValidator.CLIENT_CREDENTIALS,
SCOPE,
TokenType.ACCESS_TOKEN);
Instant originalExpiry = JWT.decode(original.getAccessToken()).getExpiresAtAsInstant();

String current = original.getAccessToken();
for (int i = 0; i < 5; i++) {
TokenResponse next =
broker.generateFromToken(
TokenType.ACCESS_TOKEN,
current,
TokenRequestValidator.TOKEN_EXCHANGE,
SCOPE,
TokenType.ACCESS_TOKEN);
assertThat(next.getError()).as("exchange iteration %s", i).isNull();
Instant nextExpiry = JWT.decode(next.getAccessToken()).getExpiresAtAsInstant();
assertThat(nextExpiry)
.as("exchange iteration %s must not outlive original subject", i)
.isBeforeOrEqualTo(originalExpiry);
// expires_in reported to client must keep shrinking toward zero, never jump back to max
assertThat(next.getExpiresIn()).isLessThanOrEqualTo(MAX_TOKEN_SECONDS);
current = next.getAccessToken();
}
}

@Test
void tokenExchangeRejectsWhenSubjectMissingPrincipal() {
TokenResponse original =
broker.generateFromClientSecrets(
CLIENT_ID,
MAIN_SECRET,
TokenRequestValidator.CLIENT_CREDENTIALS,
SCOPE,
TokenType.ACCESS_TOKEN);
when(metaStoreManager.findPrincipalById(polarisCallContext, PRINCIPAL_ID))
.thenReturn(Optional.empty());

TokenResponse exchanged =
broker.generateFromToken(
TokenType.ACCESS_TOKEN,
original.getAccessToken(),
TokenRequestValidator.TOKEN_EXCHANGE,
SCOPE,
TokenType.ACCESS_TOKEN);
assertThat(exchanged.getError()).isEqualTo(OAuthError.unauthorized_client);
}

static Stream<Arguments> remainingLifetimeCases() {
Instant now = Instant.parse("2026-01-01T00:00:00Z");
return Stream.of(
Arguments.of(now, null, 3600, Optional.empty()),
Arguments.of(now, now, 3600, Optional.empty()),
Arguments.of(now, now.minusSeconds(1), 3600, Optional.empty()),
Arguments.of(now, now.plusSeconds(10), 3600, Optional.of(10)),
Arguments.of(now, now.plusSeconds(7200), 3600, Optional.of(3600)),
Arguments.of(now, now.plusMillis(500), 3600, Optional.empty()),
Arguments.of(now, now.plusSeconds(100), 0, Optional.empty()),
Arguments.of(now, now.plusSeconds(100), -1, Optional.empty()));
}

@ParameterizedTest
@MethodSource("remainingLifetimeCases")
void remainingLifetimeSeconds(
Instant now, Instant subjectExpiresAt, int maxSeconds, Optional<Integer> expected) {
assertThat(JWTBroker.remainingLifetimeSeconds(now, subjectExpiresAt, maxSeconds))
.isEqualTo(expected);
}
}
Loading