diff --git a/test/s3tables/catalog/test_rest_catalog.py b/test/s3tables/catalog/test_rest_catalog.py index 84b6ffb45..36433a885 100644 --- a/test/s3tables/catalog/test_rest_catalog.py +++ b/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 } ) diff --git a/test/s3tables/catalog/test_rest_catalog_auth.py b/test/s3tables/catalog/test_rest_catalog_auth.py index 626d9480e..5b7d2a1ce 100644 --- a/test/s3tables/catalog/test_rest_catalog_auth.py +++ b/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, } diff --git a/test/s3tables/lifecycle/iceberg_lifecycle.py b/test/s3tables/lifecycle/iceberg_lifecycle.py index 8b4545b78..d8965abd7 100644 --- a/test/s3tables/lifecycle/iceberg_lifecycle.py +++ b/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, diff --git a/weed/s3api/iceberg/handlers_oauth.go b/weed/s3api/iceberg/handlers_oauth.go index 9a9b16d92..910ffc279 100644 --- a/weed/s3api/iceberg/handlers_oauth.go +++ b/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) { diff --git a/weed/s3api/iceberg/handlers_oauth_test.go b/weed/s3api/iceberg/handlers_oauth_test.go index 051d67eec..e3888e13e 100644 --- a/weed/s3api/iceberg/handlers_oauth_test.go +++ b/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()) + } +} diff --git a/weed/s3api/iceberg/server.go b/weed/s3api/iceberg/server.go index 3a08320a3..33164df50 100644 --- a/weed/s3api/iceberg/server.go +++ b/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 } diff --git a/weed/s3api/iceberg/server_auth_test.go b/weed/s3api/iceberg/server_auth_test.go new file mode 100644 index 000000000..fcf93067e --- /dev/null +++ b/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") + } +}