mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-15 11:00:51 +02:00
* 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
232 lines
7.1 KiB
Go
232 lines
7.1 KiB
Go
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")
|
|
}
|
|
}
|