iceberg: fix OAuth token expiry handling (401 + token-exchange + configurable TTL) (#11242)

* iceberg: return 401 for invalid or expired Bearer tokens

BUG-0001: when the OAuth JWT expired, Server.Auth fell through to the S3
SigV4 authenticator, which rejects the "Authorization: Bearer" scheme
with NotImplemented — a 501. Iceberg clients (Java OAuth2Manager,
pyiceberg) only refresh tokens on 401, so they retried the dead token
forever: RisingWave sinks stalled and Doris catalog queries failed every
token TTL (1h) until the client process was restarted.

A request carrying a Bearer header is an Iceberg REST client: answer 401
(+ WWW-Authenticate: Bearer, RFC 6750) when the token fails, and only
fall through to the S3 authenticator when no Bearer header is present.

* iceberg: make OAuth token TTL configurable via ICEBERG_OAUTH_TOKEN_EXPIRY

BUG-0001 follow-up: production evidence shows Iceberg Java 1.10.x
clients (RisingWave connector node, Doris FE) never re-fetch tokens on
401 — the sink stalled again on token expiry even with the 501→401 fix,
and no POST /v1/oauth/tokens appeared in server logs across dozens of
retries. 401 is necessary but not sufficient for these clients.

The TTL was hardcoded to 3600 with no knob. Read the expiry (seconds)
from ICEBERG_OAUTH_TOKEN_EXPIRY, defaulting to 3600, so deployments can
issue longer-lived tokens (e.g. 86400) to survive client restart cycles.

* iceberg: support OAuth token exchange (RFC 8693) for client refresh

Decompiling the Iceberg Java 1.10.1 client bundled with Doris FE showed
the missing half of BUG-0001: OAuth2Manager refreshes via token-exchange
(AuthConfig.exchangeEnabled defaults to true — the client_credentials
re-fetch branch only runs with exchange disabled), so a server that only
accepts client_credentials leaves Iceberg clients unable to ever refresh
their token, regardless of 401 correctness.

Accept grant_type=urn:ietf:params:oauth:grant-type:token-exchange on
POST /v1/oauth/tokens: verify the subject_token signature against the
issuing credential, allow exchange within a recovery grace window
(max(2*TTL, 1h), capped 24h) so clients holding tokens that expired
while the grant was unsupported recover without a restart, and mint a
fresh access token with the configured TTL.

* iceberg: harden OAuth token exchange and Bearer matching per review

- match the Bearer scheme case-insensitively (RFC 7235), like
  authenticateBearer already does
- accept optional client authentication on the token-exchange grant
  (Basic or form credentials, bound to the subject token's client);
  expired subject tokens now require it. Iceberg Java's proactive
  refresh sends Bearer-only headers, so the grant cannot require it
- reject subject tokens without an exp claim, and re-check the issuer
  on the verified claims
- unauthenticated exchange cannot extend the lifetime past the
  subject token's own expiry (no chain-refresh from a leaked token)
- return 400 invalid_grant per RFC 6749 §5.2 (was 401)
- include issued_token_type on exchange responses (RFC 8693)
- clamp ICEBERG_OAUTH_TOKEN_EXPIRY to 365d so Duration math cannot
  overflow into already-expired tokens

* iceberg: give authenticated token exchanges a fresh full TTL

The remaining-lifetime cap only guards unauthenticated (Bearer-only)
exchanges; an authenticated client renewing a live token must get the
full configured TTL, matching client_credentials.

* iceberg: reject token exchange when no lifetime remains

A Bearer-only exchange with under a second of subject lifetime would
mint a token with expires_in: 0. Reject with invalid_grant instead.

* iceberg: pin near-expiry test token to the next second boundary

jwt/v5 serializes exp at one-second precision, so a 300 ms offset can
round into the current second and route the test through the expired
branch instead of the ttlSeconds<=0 guard. Mint the subject with the
next whole-second expiry: live at exchange time, deterministically
under a second of remaining lifetime.

* iceberg: drop internal ticket reference from comments

* iceberg: clamp oversized OAuth TTLs on 32-bit platforms

strconv.Atoi on an int-sized value fails with ErrRange on 386, so an
oversized ICEBERG_OAUTH_TOKEN_EXPIRY silently fell back to the default
instead of clamping. Parse in 64-bit space and clamp, then narrow.

* iceberg: make OAuth TTL narrowing explicit

* iceberg: disable legacy OAuth in PyIceberg integration tests

test/s3tables/catalog/test_rest_catalog.py

@@ -201,6 +201,7 @@ def main():
                     "uri": args.catalog_url,
                     "warehouse": args.warehouse,
                     "prefix": args.prefix,
+                    "auth": {"type": "noop"},
                     "s3.anonymous": "true",  # Disable AWS request signing for unauthenticated access
                 }
             )

test/s3tables/catalog/test_rest_catalog_auth.py

@@ -173,6 +173,7 @@ def main():
                     "uri": args.catalog_url,
                     "warehouse": args.warehouse,
                     "prefix": args.prefix,
+                    "auth": {"type": "noop"},
                     "s3.access-key-id": args.access_key,
                     "s3.secret-access-key": args.secret_key,
                 }

test/s3tables/lifecycle/iceberg_lifecycle.py

@@ -99,6 +99,7 @@ def connect(args):
         "uri": args.catalog_url,
         "warehouse": f"s3://{args.bucket}/",
         "prefix": args.bucket,
+        "auth": {"type": "noop"},
         "s3.endpoint": args.s3_endpoint,
         "s3.access-key-id": args.access_key,
         "s3.secret-access-key": args.secret_key,

weed/s3api/iceberg/handlers_oauth.go

@@ -5,6 +5,8 @@ import (
 	"crypto/sha256"
 	"fmt"
 	"net/http"
+	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -18,6 +20,9 @@ type OAuthTokenResponse struct {
 	TokenType   string `json:"token_type"`
 	ExpiresIn   int    `json:"expires_in"`
 	Scope       string `json:"scope,omitempty"`
+	// IssuedTokenType is set to the access-token URN on token-exchange
+	// responses, where RFC 8693 requires it.
+	IssuedTokenType string `json:"issued_token_type,omitempty"`
 }
 
 // OAuthErrorResponse is the error response for the OAuth endpoint.
@@ -33,7 +38,58 @@ type IcebergClaims struct {
 	jwt.RegisteredClaims
 }
 
-const oauthTokenExpiry = 3600 // 1 hour in seconds
+const defaultOauthTokenExpiry = 3600
+
+// maxOauthTokenExpiry bounds the configured TTL so seconds-to-Duration
+// conversions cannot overflow into already-expired tokens.
+const maxOauthTokenExpiry = 365 * 24 * 3600
+
+// accessTokenTokenType is the RFC 8693 token-type URN for access tokens,
+// required in issued_token_type on token-exchange responses.
+const accessTokenTokenType = "urn:ietf:params:oauth:token-type:access_token"
+
+// grant types accepted by POST /v1/oauth/tokens. Iceberg Java 1.10.x
+// refreshes tokens via token-exchange (exchangeEnabled defaults true), so a
+// server that only accepts client_credentials leaves those clients unable
+// to refresh.
+const (
+	grantTypeClientCredentials = "client_credentials"
+	grantTypeTokenExchange     = "urn:ietf:params:oauth:grant-type:token-exchange"
+)
+
+// oauthExpirySeconds returns the OAuth token TTL. Deployments whose
+// clients cannot refresh tokens on 401 can raise this to survive client
+// restart cycles instead of dying every hour.
+func oauthExpirySeconds() int {
+	if v := os.Getenv("ICEBERG_OAUTH_TOKEN_EXPIRY"); v != "" {
+		// Parse in 64-bit space: on 32-bit platforms Atoi overflows and
+		// errors on oversized values, which would silently fall back to
+		// the default instead of clamping.
+		if n, err := strconv.ParseInt(v, 10, 64); err == nil && n > 0 {
+			if n > maxOauthTokenExpiry {
+				return maxOauthTokenExpiry
+			}
+			return int(n)
+		}
+	}
+	return defaultOauthTokenExpiry
+}
+
+// tokenExchangeGrace is how long an expired subject token may still be
+// exchanged for a fresh access token. Signature verification is the real
+// gate; the grace exists so a client holding a token that expired while the
+// exchange grant was unsupported (or during a server outage) recovers
+// without a process restart.
+func tokenExchangeGrace() time.Duration {
+	grace := 2 * oauthExpirySeconds()
+	if grace < 3600 {
+		grace = 3600
+	}
+	if grace > 86400 {
+		grace = 86400
+	}
+	return time.Duration(grace) * time.Second
+}
 
 // handleOAuthTokens implements the OAuth2 client_credentials flow.
 // POST /v1/oauth/tokens
@@ -50,7 +106,13 @@ func (s *Server) handleOAuthTokens(w http.ResponseWriter, r *http.Request) {
 	}
 
 	grantType := r.PostFormValue("grant_type")
-	if grantType != "client_credentials" {
+	switch grantType {
+	case grantTypeClientCredentials:
+		// handled below
+	case grantTypeTokenExchange, "token_exchange":
+		s.handleTokenExchange(w, r)
+		return
+	default:
 		writeOAuthError(w, http.StatusBadRequest, "unsupported_grant_type",
 			fmt.Sprintf("Unsupported grant_type: %s", grantType))
 		return
@@ -88,20 +150,7 @@ func (s *Server) handleOAuthTokens(w http.ResponseWriter, r *http.Request) {
 
 	// Generate a JWT signed with a key derived from the client secret.
 	// Include the access key in claims so we can look up the exact credential for verification.
-	signingKey := deriveSigningKey(clientID, clientSecret)
-	now := time.Now()
-	claims := IcebergClaims{
-		IdentityName: identityName,
-		AccessKey:    clientID,
-		RegisteredClaims: jwt.RegisteredClaims{
-			IssuedAt:  jwt.NewNumericDate(now),
-			ExpiresAt: jwt.NewNumericDate(now.Add(time.Duration(oauthTokenExpiry) * time.Second)),
-			Issuer:    "seaweedfs-iceberg",
-		},
-	}
-
-	token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
-	tokenString, err := token.SignedString(signingKey)
+	tokenString, err := mintIcebergToken(identityName, clientID, clientSecret, oauthExpirySeconds())
 	if err != nil {
 		glog.Errorf("Iceberg OAuth: failed to sign token: %v", err)
 		writeOAuthError(w, http.StatusInternalServerError, "server_error", "Failed to generate token")
@@ -112,13 +161,166 @@ func (s *Server) handleOAuthTokens(w http.ResponseWriter, r *http.Request) {
 	resp := OAuthTokenResponse{
 		AccessToken: tokenString,
 		TokenType:   "bearer",
-		ExpiresIn:   oauthTokenExpiry,
+		ExpiresIn:   oauthExpirySeconds(),
 		Scope:       scope,
 	}
 	w.Header().Set("Cache-Control", "no-store")
 	writeJSON(w, http.StatusOK, resp)
 }
 
+// handleTokenExchange implements RFC 8693 token exchange for Iceberg REST
+// clients: a previously issued access token (subject_token) is exchanged for
+// a fresh one. Iceberg Java's OAuth2Manager refreshes via this grant
+// (exchangeEnabled defaults true), so supporting it lets those clients
+// self-heal before their token expires — no client restart needed.
+//
+// Client authentication (RFC 8693 §2.1) is accepted but not required:
+// Iceberg Java's refreshExpiredToken sends Basic credentials with the
+// exchange, while its proactive scheduled refresh sends only the Bearer
+// session headers. An expired subject_token is only exchangeable with valid
+// client credentials, and an unauthenticated exchange never extends the
+// token's lifetime beyond the subject_token's own expiry.
+func (s *Server) handleTokenExchange(w http.ResponseWriter, r *http.Request) {
+	if s.credentialValidator == nil {
+		writeOAuthError(w, http.StatusInternalServerError, "server_error", "Credential validation not configured")
+		return
+	}
+
+	subjectToken := r.PostFormValue("subject_token")
+	if subjectToken == "" {
+		writeOAuthError(w, http.StatusBadRequest, "invalid_request", "Missing subject_token")
+		return
+	}
+
+	// Verify the subject token by signature (exp checked separately against
+	// the recovery grace).
+	unverified := &IcebergClaims{}
+	parser := jwt.NewParser(jwt.WithoutClaimsValidation())
+	if _, _, err := parser.ParseUnverified(subjectToken, unverified); err != nil {
+		writeOAuthError(w, http.StatusBadRequest, "invalid_grant", "Invalid subject_token")
+		return
+	}
+	if unverified.AccessKey == "" || unverified.Issuer != "seaweedfs-iceberg" {
+		writeOAuthError(w, http.StatusBadRequest, "invalid_grant", "Invalid subject_token")
+		return
+	}
+	identityName, _, secretKey, err := s.credentialValidator.GetCredentialByAccessKey(unverified.AccessKey)
+	if err != nil {
+		glog.V(2).Infof("Iceberg OAuth: token exchange failed to get credential for access key: %v", err)
+		writeOAuthError(w, http.StatusBadRequest, "invalid_grant", "Invalid subject_token")
+		return
+	}
+
+	// Optional client authentication. When present, the client must
+	// authenticate with the same credential that issued the subject_token.
+	clientID, clientSecret, hasBasic := r.BasicAuth()
+	if !hasBasic {
+		clientID = r.PostFormValue("client_id")
+		clientSecret = r.PostFormValue("client_secret")
+		hasBasic = clientID != "" && clientSecret != ""
+	}
+	clientAuthenticated := false
+	if hasBasic {
+		if _, _, err := s.credentialValidator.ValidateS3Credential(clientID, clientSecret); err != nil {
+			glog.V(2).Infof("Iceberg OAuth: token exchange client authentication failed for client_id=%s: %v", clientID, err)
+			writeOAuthError(w, http.StatusUnauthorized, "invalid_client", "Invalid client credentials")
+			return
+		}
+		if clientID != unverified.AccessKey {
+			writeOAuthError(w, http.StatusBadRequest, "invalid_grant", "subject_token was issued to a different client")
+			return
+		}
+		clientAuthenticated = true
+	}
+
+	signingKey := deriveSigningKey(unverified.AccessKey, secretKey)
+	claims := &IcebergClaims{}
+	parsed, err := jwt.ParseWithClaims(subjectToken, claims, func(token *jwt.Token) (interface{}, error) {
+		if _, ok := token.Method.(*jwt.SigningMethodHMAC); !ok {
+			return nil, fmt.Errorf("unexpected signing method: %v", token.Header["alg"])
+		}
+		return signingKey, nil
+	}, jwt.WithoutClaimsValidation())
+	if err != nil || !parsed.Valid {
+		glog.V(2).Infof("Iceberg OAuth: token exchange signature verification failed: %v", err)
+		writeOAuthError(w, http.StatusBadRequest, "invalid_grant", "Invalid subject_token")
+		return
+	}
+	if claims.Issuer != "seaweedfs-iceberg" {
+		writeOAuthError(w, http.StatusBadRequest, "invalid_grant", "Invalid subject_token")
+		return
+	}
+	if claims.ExpiresAt == nil {
+		// Tokens minted by this server always carry an expiry; a subject
+		// token without one must not be exchangeable forever.
+		writeOAuthError(w, http.StatusBadRequest, "invalid_grant", "subject_token has no expiry")
+		return
+	}
+
+	ttlSeconds := oauthExpirySeconds()
+	if time.Now().After(claims.ExpiresAt.Time) {
+		// Expired subject tokens are only exchangeable by an authenticated
+		// client, within the recovery grace.
+		if !clientAuthenticated {
+			writeOAuthError(w, http.StatusBadRequest, "invalid_grant",
+				"subject_token is expired; client authentication is required to exchange it")
+			return
+		}
+		if time.Since(claims.ExpiresAt.Time) > tokenExchangeGrace() {
+			writeOAuthError(w, http.StatusBadRequest, "invalid_grant", "subject_token expired beyond the exchange grace window")
+			return
+		}
+	} else if remaining := int(time.Until(claims.ExpiresAt.Time).Seconds()); !clientAuthenticated && remaining < ttlSeconds {
+		// Unauthenticated exchange (proactive refresh carrying only the
+		// Bearer) must not extend the lifetime past the subject's expiry;
+		// otherwise a leaked token could chain-refresh forever. An
+		// authenticated client gets a fresh full TTL — it could mint one
+		// via client_credentials anyway.
+		ttlSeconds = remaining
+	}
+	if ttlSeconds <= 0 {
+		// A subject token with under a second left would otherwise be
+		// exchanged for a token that is born expired.
+		writeOAuthError(w, http.StatusBadRequest, "invalid_grant", "subject_token has no remaining lifetime")
+		return
+	}
+
+	tokenString, err := mintIcebergToken(identityName, unverified.AccessKey, secretKey, ttlSeconds)
+	if err != nil {
+		glog.Errorf("Iceberg OAuth: failed to sign exchanged token: %v", err)
+		writeOAuthError(w, http.StatusInternalServerError, "server_error", "Failed to generate token")
+		return
+	}
+
+	resp := OAuthTokenResponse{
+		AccessToken:     tokenString,
+		TokenType:       "bearer",
+		ExpiresIn:       ttlSeconds,
+		Scope:           r.PostFormValue("scope"),
+		IssuedTokenType: accessTokenTokenType,
+	}
+	w.Header().Set("Cache-Control", "no-store")
+	writeJSON(w, http.StatusOK, resp)
+}
+
+// mintIcebergToken issues a signed access token for the given identity and
+// credential, valid for ttlSeconds.
+func mintIcebergToken(identityName, accessKey, secret string, ttlSeconds int) (string, error) {
+	signingKey := deriveSigningKey(accessKey, secret)
+	now := time.Now()
+	claims := IcebergClaims{
+		IdentityName: identityName,
+		AccessKey:    accessKey,
+		RegisteredClaims: jwt.RegisteredClaims{
+			IssuedAt:  jwt.NewNumericDate(now),
+			ExpiresAt: jwt.NewNumericDate(now.Add(time.Duration(ttlSeconds) * time.Second)),
+			Issuer:    "seaweedfs-iceberg",
+		},
+	}
+	token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
+	return token.SignedString(signingKey)
+}
+
 // authenticateBearer validates a Bearer token from the Authorization header.
 // Returns the identity name, identity object, and whether auth succeeded.
 func (s *Server) authenticateBearer(r *http.Request) (string, interface{}, bool) {

weed/s3api/iceberg/handlers_oauth_test.go

@@ -5,8 +5,12 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
 	"strings"
 	"testing"
+	"time"
+
+	jwt "github.com/golang-jwt/jwt/v5"
 )
 
 type mockCredentialValidator struct {
@@ -68,8 +72,8 @@ func TestHandleOAuthTokens_Success(t *testing.T) {
 	if resp.AccessToken == "" {
 		t.Error("expected non-empty access_token")
 	}
-	if resp.ExpiresIn != oauthTokenExpiry {
-		t.Errorf("expected expires_in=%d, got %d", oauthTokenExpiry, resp.ExpiresIn)
+	if resp.ExpiresIn != oauthExpirySeconds() {
+		t.Errorf("expected expires_in=%d, got %d", oauthExpirySeconds(), resp.ExpiresIn)
 	}
 }
 
@@ -153,3 +157,254 @@ func TestBearerTokenNone(t *testing.T) {
 		t.Error("expected Bearer auth to fail with no token")
 	}
 }
+
+// TestOauthExpirySecondsEnvOverride pins the TTL knob: the token TTL must
+// be configurable so clients that cannot refresh on 401 can be given
+// longer-lived tokens instead of dying every hour.
+func TestOauthExpirySecondsEnvOverride(t *testing.T) {
+	if got := oauthExpirySeconds(); got != defaultOauthTokenExpiry {
+		t.Fatalf("default TTL = %d, want %d", got, defaultOauthTokenExpiry)
+	}
+	t.Setenv("ICEBERG_OAUTH_TOKEN_EXPIRY", "86400")
+	if got := oauthExpirySeconds(); got != 86400 {
+		t.Fatalf("env TTL = %d, want 86400", got)
+	}
+	t.Setenv("ICEBERG_OAUTH_TOKEN_EXPIRY", "-5")
+	if got := oauthExpirySeconds(); got != defaultOauthTokenExpiry {
+		t.Fatalf("negative TTL must fall back to default, got %d", got)
+	}
+	t.Setenv("ICEBERG_OAUTH_TOKEN_EXPIRY", "garbage")
+	if got := oauthExpirySeconds(); got != defaultOauthTokenExpiry {
+		t.Fatalf("invalid TTL must fall back to default, got %d", got)
+	}
+	// oversized values are clamped so Duration math cannot overflow
+	t.Setenv("ICEBERG_OAUTH_TOKEN_EXPIRY", "99999999999999999")
+	if got := oauthExpirySeconds(); got != maxOauthTokenExpiry {
+		t.Fatalf("oversized TTL must clamp to %d, got %d", maxOauthTokenExpiry, got)
+	}
+
+	// a token minted under an override carries the override's expiry
+	t.Setenv("ICEBERG_OAUTH_TOKEN_EXPIRY", "7200")
+	s := newTestServerWithOAuth()
+	body := "grant_type=client_credentials&client_id=AKID123&client_secret=secret456"
+	req := httptest.NewRequest(http.MethodPost, "/v1/oauth/tokens", strings.NewReader(body))
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	w := httptest.NewRecorder()
+	s.handleOAuthTokens(w, req)
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200, got %d", w.Code)
+	}
+	var resp OAuthTokenResponse
+	if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
+		t.Fatal(err)
+	}
+	if resp.ExpiresIn != 7200 {
+		t.Fatalf("minted token expires_in = %d, want 7200", resp.ExpiresIn)
+	}
+}
+
+func exchangeRequest(t *testing.T, subjectToken string) *httptest.ResponseRecorder {
+	return exchangeRequestWithBasic(t, subjectToken, "", "")
+}
+
+// exchangeRequestWithBasic mirrors Iceberg Java: refreshExpiredToken sends
+// Basic client credentials with the token-exchange grant, the proactive
+// scheduled refresh sends only the form body.
+func exchangeRequestWithBasic(t *testing.T, subjectToken, basicUser, basicPass string) *httptest.ResponseRecorder {
+	t.Helper()
+	s := newTestServerWithOAuth()
+	body := "grant_type=urn:ietf:params:oauth:grant-type:token-exchange&subject_token=" + url.QueryEscape(subjectToken)
+	req := httptest.NewRequest(http.MethodPost, "/v1/oauth/tokens", strings.NewReader(body))
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	if basicUser != "" || basicPass != "" {
+		req.SetBasicAuth(basicUser, basicPass)
+	}
+	w := httptest.NewRecorder()
+	s.handleOAuthTokens(w, req)
+	return w
+}
+
+// TestTokenExchangeLiveSubject: the proactive refresh path Iceberg Java
+// 1.10.x uses (exchangeEnabled defaults true, Bearer-only headers) must
+// mint a fresh working token.
+func TestTokenExchangeLiveSubject(t *testing.T) {
+	s := newTestServerWithOAuth()
+	now := time.Now()
+	live := mintTestToken(t, "AKID123", "secret456", now, now.Add(30*time.Minute))
+
+	w := exchangeRequest(t, live)
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
+	}
+	var resp OAuthTokenResponse
+	if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
+		t.Fatal(err)
+	}
+	if resp.TokenType != "bearer" || resp.AccessToken == "" {
+		t.Fatalf("bad token response: %+v", resp)
+	}
+	if resp.IssuedTokenType != accessTokenTokenType {
+		t.Fatalf("issued_token_type = %q, want %s", resp.IssuedTokenType, accessTokenTokenType)
+	}
+	// Unauthenticated exchange must not outlive the subject token.
+	if resp.ExpiresIn > 30*60 {
+		t.Fatalf("expires_in = %d, must be bounded by the subject's remaining lifetime (<= 1800)", resp.ExpiresIn)
+	}
+	// the exchanged token must authenticate like a normal Bearer
+	req := httptest.NewRequest(http.MethodGet, "/v1/namespaces", nil)
+	req.Header.Set("Authorization", "Bearer "+resp.AccessToken)
+	if _, _, ok := s.authenticateBearer(req); !ok {
+		t.Fatalf("exchanged token must pass authenticateBearer")
+	}
+}
+
+// TestTokenExchangeExpiredWithinGrace: a client whose token expired while
+// exchange was unsupported must recover without a restart. This is the
+// refreshExpiredToken path, which authenticates with Basic credentials.
+func TestTokenExchangeExpiredWithinGrace(t *testing.T) {
+	now := time.Now()
+	expiredRecently := mintTestToken(t, "AKID123", "secret456", now.Add(-20*time.Minute), now.Add(-10*time.Minute))
+
+	w := exchangeRequestWithBasic(t, expiredRecently, "AKID123", "secret456")
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200 within grace, got %d: %s", w.Code, w.Body.String())
+	}
+	var resp OAuthTokenResponse
+	if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
+		t.Fatal(err)
+	}
+	// Authenticated refresh gets the full configured TTL.
+	if resp.ExpiresIn != oauthExpirySeconds() {
+		t.Fatalf("expires_in = %d, want %d", resp.ExpiresIn, oauthExpirySeconds())
+	}
+}
+
+// TestTokenExchangeAuthenticatedLiveSubjectFullTTL: an authenticated client
+// exchanging a live token renews the session — the TTL cap only applies to
+// unauthenticated (Bearer-only) exchanges.
+func TestTokenExchangeAuthenticatedLiveSubjectFullTTL(t *testing.T) {
+	now := time.Now()
+	live := mintTestToken(t, "AKID123", "secret456", now, now.Add(30*time.Minute))
+
+	w := exchangeRequestWithBasic(t, live, "AKID123", "secret456")
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
+	}
+	var resp OAuthTokenResponse
+	if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
+		t.Fatal(err)
+	}
+	if resp.ExpiresIn != oauthExpirySeconds() {
+		t.Fatalf("authenticated live exchange expires_in = %d, want full TTL %d", resp.ExpiresIn, oauthExpirySeconds())
+	}
+}
+
+// TestTokenExchangeSubjectNearlyExpired: a Bearer-only exchange when the
+// subject token has under a second left must be rejected, not minted into
+// an already-expired token (expires_in: 0).
+func TestTokenExchangeSubjectNearlyExpired(t *testing.T) {
+	// jwt/v5 serializes exp at one-second precision, so pin the expiry to
+	// the next whole-second boundary: the token is always live when minted
+	// and exchanged, yet has under a second of remaining lifetime.
+	exp := time.Now().Truncate(time.Second).Add(time.Second)
+	nearlyDead := mintTestToken(t, "AKID123", "secret456", exp.Add(-time.Hour), exp)
+
+	w := exchangeRequest(t, nearlyDead)
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String())
+	}
+	if !strings.Contains(w.Body.String(), "invalid_grant") {
+		t.Fatalf("expected invalid_grant, got: %s", w.Body.String())
+	}
+}
+
+// TestTokenExchangeExpiredWithoutClientAuth: a leaked expired token must
+// not be exchangeable without client credentials.
+func TestTokenExchangeExpiredWithoutClientAuth(t *testing.T) {
+	now := time.Now()
+	expiredRecently := mintTestToken(t, "AKID123", "secret456", now.Add(-20*time.Minute), now.Add(-10*time.Minute))
+
+	w := exchangeRequest(t, expiredRecently)
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String())
+	}
+	if !strings.Contains(w.Body.String(), "invalid_grant") {
+		t.Fatalf("expected invalid_grant, got: %s", w.Body.String())
+	}
+}
+
+// TestTokenExchangeExpiredBeyondGrace: stale tokens far past expiry must not
+// act as eternal credentials, even with client auth.
+func TestTokenExchangeExpiredBeyondGrace(t *testing.T) {
+	now := time.Now()
+	longDead := mintTestToken(t, "AKID123", "secret456", now.Add(-48*time.Hour), now.Add(-47*time.Hour))
+
+	w := exchangeRequestWithBasic(t, longDead, "AKID123", "secret456")
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("expected 400 beyond grace, got %d: %s", w.Code, w.Body.String())
+	}
+	if !strings.Contains(w.Body.String(), "invalid_grant") {
+		t.Fatalf("expected invalid_grant, got: %s", w.Body.String())
+	}
+}
+
+// TestTokenExchangeGarbageSubject: malformed subject tokens are rejected
+// with 400 invalid_grant per RFC 6749 §5.2.
+func TestTokenExchangeGarbageSubject(t *testing.T) {
+	w := exchangeRequest(t, "not-a-jwt")
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("expected 400, got %d", w.Code)
+	}
+}
+
+// TestTokenExchangeWrongClientCredentials: valid subject token, but Basic
+// auth with wrong credentials or a different client is rejected.
+func TestTokenExchangeWrongClientCredentials(t *testing.T) {
+	now := time.Now()
+	live := mintTestToken(t, "AKID123", "secret456", now, now.Add(30*time.Minute))
+
+	// wrong secret → invalid_client 401
+	w := exchangeRequestWithBasic(t, live, "AKID123", "wrongsecret")
+	if w.Code != http.StatusUnauthorized {
+		t.Fatalf("wrong secret: expected 401, got %d", w.Code)
+	}
+
+	// a different valid client must not exchange someone else's token
+	s := &Server{credentialValidator: &mockCredentialValidator{
+		credentials: map[string]string{"AKID123": "secret456", "AKID999": "other-secret"},
+		identities:  map[string]string{"AKID123": "testuser", "AKID999": "otheruser"},
+	}}
+	body := "grant_type=urn:ietf:params:oauth:grant-type:token-exchange&subject_token=" + url.QueryEscape(live)
+	req := httptest.NewRequest(http.MethodPost, "/v1/oauth/tokens", strings.NewReader(body))
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	req.SetBasicAuth("AKID999", "other-secret")
+	rec := httptest.NewRecorder()
+	s.handleOAuthTokens(rec, req)
+	if rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "invalid_grant") {
+		t.Fatalf("cross-client exchange: expected 400 invalid_grant, got %d: %s", rec.Code, rec.Body.String())
+	}
+}
+
+// TestTokenExchangeSubjectWithoutExpiry: a correctly signed subject token
+// without an exp claim must be rejected — it would otherwise refresh forever.
+func TestTokenExchangeSubjectWithoutExpiry(t *testing.T) {
+	claims := IcebergClaims{
+		IdentityName: "testuser",
+		AccessKey:    "AKID123",
+		RegisteredClaims: jwt.RegisteredClaims{
+			IssuedAt: jwt.NewNumericDate(time.Now()),
+			Issuer:   "seaweedfs-iceberg",
+			// no ExpiresAt
+		},
+	}
+	token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
+	signed, err := token.SignedString(deriveSigningKey("AKID123", "secret456"))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	w := exchangeRequestWithBasic(t, signed, "AKID123", "secret456")
+	if w.Code != http.StatusBadRequest || !strings.Contains(w.Body.String(), "invalid_grant") {
+		t.Fatalf("expected 400 invalid_grant for nil-exp subject, got %d: %s", w.Code, w.Body.String())
+	}
+}

weed/s3api/iceberg/server.go

@@ -3,6 +3,7 @@ package iceberg
 import (
 	"context"
 	"net/http"
+	"strings"
 	"time"
 
 	"github.com/gorilla/mux"
@@ -217,15 +218,28 @@ func (w *responseWriter) WriteHeader(code int) {
 
 func (s *Server) Auth(handler http.HandlerFunc) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
-		// Try Bearer token authentication first (from OAuth2 flow)
-		if identityName, identity, ok := s.authenticateBearer(r); ok {
-			ctx := r.Context()
-			ctx = s3_constants.SetIdentityNameInContext(ctx, identityName)
-			if identity != nil {
-				ctx = s3_constants.SetIdentityInContext(ctx, identity)
+		// A request carrying a Bearer token is an Iceberg REST client. When
+		// the token is invalid or expired, answer 401 immediately so clients
+		// (Iceberg Java OAuth2Manager, pyiceberg) refresh their token and
+		// retry. Falling through to the S3 authenticator instead would parse
+		// the Authorization header as SigV4, fail with NotImplemented (501),
+		// and clients would retry the dead token forever.
+		// The auth scheme is case-insensitive (RFC 7235), matching
+		// authenticateBearer below.
+		if strings.HasPrefix(strings.ToLower(r.Header.Get("Authorization")), "bearer ") {
+			if identityName, identity, ok := s.authenticateBearer(r); ok {
+				ctx := r.Context()
+				ctx = s3_constants.SetIdentityNameInContext(ctx, identityName)
+				if identity != nil {
+					ctx = s3_constants.SetIdentityInContext(ctx, identity)
+				}
+				r = r.WithContext(ctx)
+				handler(w, r)
+				return
 			}
-			r = r.WithContext(ctx)
-			handler(w, r)
+			// RFC 6750 / Iceberg REST spec: 401 is the refresh signal.
+			w.Header().Set("WWW-Authenticate", "Bearer")
+			writeError(w, http.StatusUnauthorized, "NotAuthorizedException", "Bearer token is invalid or expired")
 			return
 		}
 

weed/s3api/iceberg/server_auth_test.go

@@ -0,0 +1,231 @@
+package iceberg
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	jwt "github.com/golang-jwt/jwt/v5"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+)
+
+type mockS3Authenticator struct {
+	called  bool
+	errCode s3err.ErrorCode
+}
+
+func (m *mockS3Authenticator) AuthenticateRequest(r *http.Request) (string, interface{}, s3err.ErrorCode) {
+	m.called = true
+	if m.errCode != s3err.ErrNone {
+		return "", nil, m.errCode
+	}
+	return "s3user", nil, s3err.ErrNone
+}
+
+func (m *mockS3Authenticator) DefaultAllow() bool { return false }
+
+func mintTestToken(t *testing.T, accessKey, secret string, issuedAt, expiresAt time.Time) string {
+	t.Helper()
+	key := deriveSigningKey(accessKey, secret)
+	claims := IcebergClaims{
+		IdentityName: "testuser",
+		AccessKey:    accessKey,
+		RegisteredClaims: jwt.RegisteredClaims{
+			IssuedAt:  jwt.NewNumericDate(issuedAt),
+			ExpiresAt: jwt.NewNumericDate(expiresAt),
+			Issuer:    "seaweedfs-iceberg",
+		},
+	}
+	token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
+	signed, err := token.SignedString(key)
+	if err != nil {
+		t.Fatalf("sign token: %v", err)
+	}
+	return signed
+}
+
+func newAuthTestServer() (*Server, *mockS3Authenticator) {
+	s := newTestServerWithOAuth()
+	auth := &mockS3Authenticator{errCode: s3err.ErrNotImplemented}
+	s.authenticator = auth
+	return s, auth
+}
+
+// TestAuthExpiredBearerReturns401: an expired Bearer token must get 401
+// (the refresh signal for Iceberg clients), never fall through to the S3
+// authenticator and surface as 501 NotImplemented.
+func TestAuthExpiredBearerReturns401(t *testing.T) {
+	s, auth := newAuthTestServer()
+	now := time.Now()
+	expired := mintTestToken(t, "AKID123", "secret456", now.Add(-2*time.Hour), now.Add(-1*time.Hour))
+
+	var handlerCalled bool
+	handler := s.Auth(func(w http.ResponseWriter, r *http.Request) {
+		handlerCalled = true
+		w.WriteHeader(http.StatusOK)
+	})
+
+	req := httptest.NewRequest(http.MethodGet, "/v1/namespaces", nil)
+	req.Header.Set("Authorization", "Bearer "+expired)
+	rec := httptest.NewRecorder()
+	handler(rec, req)
+
+	if rec.Code != http.StatusUnauthorized {
+		t.Fatalf("expired Bearer: status = %d, want 401 (previously fell through to 501)", rec.Code)
+	}
+	if auth.called {
+		t.Fatalf("expired Bearer must not fall through to the S3 authenticator")
+	}
+	if handlerCalled {
+		t.Fatalf("handler must not run for an expired token")
+	}
+	if wa := rec.Header().Get("WWW-Authenticate"); wa != "Bearer" {
+		t.Fatalf("WWW-Authenticate = %q, want Bearer", wa)
+	}
+	if !strings.Contains(rec.Body.String(), "NotAuthorizedException") {
+		t.Fatalf("error type must be NotAuthorizedException, got: %s", rec.Body.String())
+	}
+}
+
+// TestAuthCaseInsensitiveBearerScheme: RFC 7235 auth schemes are
+// case-insensitive; a lowercase or mixed-case "bearer" must take the same
+// Bearer path (401 refresh signal) instead of falling through to S3 auth.
+func TestAuthCaseInsensitiveBearerScheme(t *testing.T) {
+	s, auth := newAuthTestServer()
+	now := time.Now()
+	expired := mintTestToken(t, "AKID123", "secret456", now.Add(-2*time.Hour), now.Add(-1*time.Hour))
+
+	for _, scheme := range []string{"bearer", "BEARER", "BeArEr"} {
+		var handlerCalled bool
+		handler := s.Auth(func(w http.ResponseWriter, r *http.Request) {
+			handlerCalled = true
+			w.WriteHeader(http.StatusOK)
+		})
+
+		req := httptest.NewRequest(http.MethodGet, "/v1/namespaces", nil)
+		req.Header.Set("Authorization", scheme+" "+expired)
+		rec := httptest.NewRecorder()
+		handler(rec, req)
+
+		if rec.Code != http.StatusUnauthorized {
+			t.Fatalf("scheme %q: status = %d, want 401", scheme, rec.Code)
+		}
+		if auth.called {
+			t.Fatalf("scheme %q: must not fall through to the S3 authenticator", scheme)
+		}
+		if handlerCalled {
+			t.Fatalf("scheme %q: handler must not run for an expired token", scheme)
+		}
+		if wa := rec.Header().Get("WWW-Authenticate"); wa != "Bearer" {
+			t.Fatalf("scheme %q: WWW-Authenticate = %q, want Bearer", scheme, wa)
+		}
+	}
+}
+
+// TestAuthGarbageBearerReturns401 pins the same contract for malformed tokens.
+func TestAuthGarbageBearerReturns401(t *testing.T) {
+	s, auth := newAuthTestServer()
+
+	handler := s.Auth(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})
+
+	req := httptest.NewRequest(http.MethodGet, "/v1/namespaces", nil)
+	req.Header.Set("Authorization", "Bearer not-a-jwt")
+	rec := httptest.NewRecorder()
+	handler(rec, req)
+
+	if rec.Code != http.StatusUnauthorized {
+		t.Fatalf("garbage Bearer: status = %d, want 401", rec.Code)
+	}
+	if auth.called {
+		t.Fatalf("garbage Bearer must not fall through to the S3 authenticator")
+	}
+}
+
+// TestAuthFreshBearerRunsHandler verifies the happy path still authenticates
+// and hands the identity to the handler.
+func TestAuthFreshBearerRunsHandler(t *testing.T) {
+	s, auth := newAuthTestServer()
+	now := time.Now()
+	fresh := mintTestToken(t, "AKID123", "secret456", now, now.Add(time.Hour))
+
+	var gotIdentity string
+	handler := s.Auth(func(w http.ResponseWriter, r *http.Request) {
+		gotIdentity = s3_constants.GetIdentityNameFromContext(r)
+		w.WriteHeader(http.StatusOK)
+	})
+
+	req := httptest.NewRequest(http.MethodGet, "/v1/namespaces", nil)
+	req.Header.Set("Authorization", "Bearer "+fresh)
+	rec := httptest.NewRecorder()
+	handler(rec, req)
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("fresh Bearer: status = %d, want 200", rec.Code)
+	}
+	if auth.called {
+		t.Fatalf("fresh Bearer should not need the S3 authenticator")
+	}
+	if gotIdentity != "testuser" {
+		t.Fatalf("identity in context = %q, want testuser", gotIdentity)
+	}
+}
+
+// TestAuthNoBearerStillUsesS3Authenticator keeps the non-Bearer path intact:
+// SigV4 requests keep flowing through the S3 authenticator as before.
+func TestAuthNoBearerStillUsesS3Authenticator(t *testing.T) {
+	s := newTestServerWithOAuth()
+	auth := &mockS3Authenticator{errCode: s3err.ErrNone}
+	s.authenticator = auth
+
+	var handlerCalled bool
+	handler := s.Auth(func(w http.ResponseWriter, r *http.Request) {
+		handlerCalled = true
+		w.WriteHeader(http.StatusOK)
+	})
+
+	req := httptest.NewRequest(http.MethodGet, "/v1/namespaces", nil)
+	rec := httptest.NewRecorder()
+	handler(rec, req)
+
+	if !auth.called {
+		t.Fatalf("request without Bearer header must use the S3 authenticator")
+	}
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200 after successful S3 auth", rec.Code)
+	}
+	if !handlerCalled {
+		t.Fatalf("handler must run after S3 auth succeeds")
+	}
+}
+
+// TestAuthNonBearerAuthorizationHeaderUsesS3Authenticator pins that other
+// Authorization schemes (e.g. SigV4) never take the Bearer fast path.
+func TestAuthNonBearerAuthorizationHeaderUsesS3Authenticator(t *testing.T) {
+	s := newTestServerWithOAuth()
+	auth := &mockS3Authenticator{errCode: s3err.ErrNone}
+	s.authenticator = auth
+
+	var handlerCalled bool
+	handler := s.Auth(func(w http.ResponseWriter, r *http.Request) {
+		handlerCalled = true
+		w.WriteHeader(http.StatusOK)
+	})
+
+	req := httptest.NewRequest(http.MethodGet, "/v1/namespaces", nil)
+	req.Header.Set("Authorization", "AWS4-HMAC-SHA256 Credential=AKID123/...")
+	rec := httptest.NewRecorder()
+	handler(rec, req)
+
+	if !auth.called {
+		t.Fatalf("SigV4 Authorization header must use the S3 authenticator")
+	}
+	if !handlerCalled {
+		t.Fatalf("handler must run after S3 auth succeeds")
+	}
+}