diff --git a/test/s3tables/catalog/duckdb_oauth_test.go b/test/s3tables/catalog/duckdb_oauth_test.go
new file mode 100644
index 000000000..6019f107a
--- /dev/null
+++ b/test/s3tables/catalog/duckdb_oauth_test.go
@@ -0,0 +1,465 @@
+package catalog
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/test/testutil"
+)
+
+// oauthTestEnv holds a weed mini instance started with IAM credentials
+// so that the OAuth token endpoint is functional.
+type oauthTestEnv struct {
+	seaweedDir     string
+	weedBinary     string
+	dataDir        string
+	bindIP         string
+	s3Port         int
+	s3GrpcPort     int
+	icebergPort    int
+	masterPort     int
+	masterGrpcPort int
+	filerPort      int
+	filerGrpcPort  int
+	volumePort     int
+	volumeGrpcPort int
+	webdavPort     int
+	weedProcess    *exec.Cmd
+	weedCancel     context.CancelFunc
+	accessKey      string
+	secretKey      string
+}
+
+func newOAuthTestEnv(t *testing.T) *oauthTestEnv {
+	t.Helper()
+
+	wd, err := os.Getwd()
+	if err != nil {
+		t.Fatalf("get working directory: %v", err)
+	}
+
+	seaweedDir := wd
+	for i := 0; i < 6; i++ {
+		if _, err := os.Stat(filepath.Join(seaweedDir, "go.mod")); err == nil {
+			break
+		}
+		seaweedDir = filepath.Dir(seaweedDir)
+	}
+
+	weedBinary := filepath.Join(seaweedDir, "weed", "weed")
+	if info, err := os.Stat(weedBinary); err != nil || info.IsDir() {
+		weedBinary = "weed"
+		if _, err := exec.LookPath(weedBinary); err != nil {
+			t.Skip("weed binary not found, skipping integration test")
+		}
+	}
+
+	dataDir, err := os.MkdirTemp("", "seaweed-oauth-test-*")
+	if err != nil {
+		t.Fatalf("create temp dir: %v", err)
+	}
+
+	bindIP := testutil.FindBindIP()
+	ports := testutil.MustAllocatePorts(t, 10)
+
+	return &oauthTestEnv{
+		seaweedDir:     seaweedDir,
+		weedBinary:     weedBinary,
+		dataDir:        dataDir,
+		bindIP:         bindIP,
+		masterPort:     ports[0],
+		masterGrpcPort: ports[1],
+		volumePort:     ports[2],
+		volumeGrpcPort: ports[3],
+		filerPort:      ports[4],
+		filerGrpcPort:  ports[5],
+		s3Port:         ports[6],
+		s3GrpcPort:     ports[7],
+		icebergPort:    ports[8],
+		webdavPort:     ports[9],
+		accessKey:      "AKIAIOSFODNN7EXAMPLE",
+		secretKey:      "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+	}
+}
+
+func (env *oauthTestEnv) start(t *testing.T) {
+	t.Helper()
+
+	iamConfigPath, err := testutil.WriteIAMConfig(env.dataDir, env.accessKey, env.secretKey)
+	if err != nil {
+		t.Fatalf("write IAM config: %v", err)
+	}
+
+	securityToml := filepath.Join(env.dataDir, "security.toml")
+	if err := os.WriteFile(securityToml, []byte("# Empty security config for testing\n"), 0644); err != nil {
+		t.Fatalf("write security.toml: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	env.weedCancel = cancel
+
+	cmd := exec.CommandContext(ctx, env.weedBinary, "mini",
+		"-master.port", fmt.Sprintf("%d", env.masterPort),
+		"-master.port.grpc", fmt.Sprintf("%d", env.masterGrpcPort),
+		"-volume.port", fmt.Sprintf("%d", env.volumePort),
+		"-volume.port.grpc", fmt.Sprintf("%d", env.volumeGrpcPort),
+		"-filer.port", fmt.Sprintf("%d", env.filerPort),
+		"-filer.port.grpc", fmt.Sprintf("%d", env.filerGrpcPort),
+		"-s3.port", fmt.Sprintf("%d", env.s3Port),
+		"-s3.port.grpc", fmt.Sprintf("%d", env.s3GrpcPort),
+		"-s3.port.iceberg", fmt.Sprintf("%d", env.icebergPort),
+		"-webdav.port", fmt.Sprintf("%d", env.webdavPort),
+		"-s3.config", iamConfigPath,
+		"-ip", env.bindIP,
+		"-ip.bind", "0.0.0.0",
+		"-dir", env.dataDir,
+	)
+	cmd.Dir = env.dataDir
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	cmd.Env = append(os.Environ(),
+		"AWS_ACCESS_KEY_ID="+env.accessKey,
+		"AWS_SECRET_ACCESS_KEY="+env.secretKey,
+	)
+
+	if err := cmd.Start(); err != nil {
+		cancel()
+		t.Fatalf("start weed mini: %v", err)
+	}
+	env.weedProcess = cmd
+
+	icebergURL := fmt.Sprintf("http://%s:%d/v1/config", env.bindIP, env.icebergPort)
+	if !testutil.WaitForService(icebergURL, 30*time.Second) {
+		cancel()
+		cmd.Wait()
+		t.Fatalf("Iceberg REST API did not become ready at %s", icebergURL)
+	}
+}
+
+func (env *oauthTestEnv) cleanup(t *testing.T) {
+	t.Helper()
+	if env.weedCancel != nil {
+		env.weedCancel()
+	}
+	if env.weedProcess != nil {
+		env.weedProcess.Wait()
+	}
+	if env.dataDir != "" {
+		os.RemoveAll(env.dataDir)
+	}
+}
+
+func (env *oauthTestEnv) icebergURL() string {
+	return fmt.Sprintf("http://%s:%d", env.bindIP, env.icebergPort)
+}
+
+// TestOAuthTokenEndpoint tests the /v1/oauth/tokens endpoint directly via HTTP.
+func TestOAuthTokenEndpoint(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	env := newOAuthTestEnv(t)
+	defer env.cleanup(t)
+	env.start(t)
+
+	t.Run("valid credentials", func(t *testing.T) {
+		token := requestOAuthToken(t, env, env.accessKey, env.secretKey)
+		if token == "" {
+			t.Fatal("expected non-empty token")
+		}
+	})
+
+	t.Run("invalid secret", func(t *testing.T) {
+		resp, err := http.PostForm(env.icebergURL()+"/v1/oauth/tokens", url.Values{
+			"grant_type":    {"client_credentials"},
+			"client_id":     {env.accessKey},
+			"client_secret": {"wrong-secret"},
+		})
+		if err != nil {
+			t.Fatalf("POST /v1/oauth/tokens: %v", err)
+		}
+		defer resp.Body.Close()
+
+		if resp.StatusCode != http.StatusUnauthorized {
+			body, _ := io.ReadAll(resp.Body)
+			t.Fatalf("expected 401, got %d: %s", resp.StatusCode, body)
+		}
+	})
+
+	t.Run("missing grant_type", func(t *testing.T) {
+		resp, err := http.PostForm(env.icebergURL()+"/v1/oauth/tokens", url.Values{
+			"client_id":     {env.accessKey},
+			"client_secret": {env.secretKey},
+		})
+		if err != nil {
+			t.Fatalf("POST /v1/oauth/tokens: %v", err)
+		}
+		defer resp.Body.Close()
+
+		if resp.StatusCode != http.StatusBadRequest {
+			body, _ := io.ReadAll(resp.Body)
+			t.Fatalf("expected 400, got %d: %s", resp.StatusCode, body)
+		}
+	})
+
+	t.Run("bearer token auth on catalog endpoint", func(t *testing.T) {
+		token := requestOAuthToken(t, env, env.accessKey, env.secretKey)
+
+		// Use the token to call the catalog
+		req, err := http.NewRequest(http.MethodGet, env.icebergURL()+"/v1/namespaces", nil)
+		if err != nil {
+			t.Fatalf("create request: %v", err)
+		}
+		req.Header.Set("Authorization", "Bearer "+token)
+
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			t.Fatalf("GET /v1/namespaces with Bearer: %v", err)
+		}
+		defer resp.Body.Close()
+
+		// Auth should pass. We accept 200 (success) or 500 (missing warehouse bucket
+		// is an internal error, not an auth error). Reject 401/403/404/405.
+		body, _ := io.ReadAll(resp.Body)
+		switch resp.StatusCode {
+		case http.StatusOK, http.StatusInternalServerError:
+			t.Logf("Bearer token auth succeeded, status=%d", resp.StatusCode)
+		default:
+			t.Fatalf("Bearer auth failed unexpectedly: status=%d body=%s", resp.StatusCode, body)
+		}
+	})
+}
+
+// TestDuckDBOAuthIntegration tests that DuckDB can connect to the Iceberg REST
+// catalog using the OAuth2 client_credentials flow (CREATE SECRET with client_id
+// and client_secret). This is the scenario reported in issue #9015.
+func TestDuckDBOAuthIntegration(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+	if !testutil.HasDocker() {
+		t.Skip("Docker not available, skipping DuckDB OAuth integration test")
+	}
+
+	env := newOAuthTestEnv(t)
+	defer env.cleanup(t)
+	env.start(t)
+
+	// Create a table bucket and namespace so DuckDB has something to query
+	bucketName := "duckdb-oauth-" + randomSuffix()
+	createTableBucketViaShell(t, env, bucketName)
+
+	// Create a namespace via the Iceberg REST API using OAuth token
+	token := requestOAuthToken(t, env, env.accessKey, env.secretKey)
+	createNamespaceWithToken(t, env, token, bucketName, "testns")
+
+	sqlContent := fmt.Sprintf(`
+INSTALL iceberg;
+LOAD iceberg;
+
+CREATE SECRET iceberg_secret (
+    TYPE ICEBERG,
+    ENDPOINT 'http://host.docker.internal:%d',
+    CLIENT_ID '%s',
+    CLIENT_SECRET '%s',
+    SCOPE 's3://%s/'
+);
+
+CREATE SECRET s3_secret (
+    TYPE S3,
+    KEY_ID '%s',
+    SECRET '%s',
+    ENDPOINT 'host.docker.internal:%d',
+    URL_STYLE 'path',
+    USE_SSL false,
+    SCOPE 's3://%s/'
+);
+
+SELECT 'OAuth token obtained successfully' as status;
+
+-- Try listing namespaces via the Iceberg catalog
+SELECT * FROM iceberg_scan('iceberg_secret', ALLOW_MOVED_PATHS => TRUE) LIMIT 0;
+`, env.icebergPort, env.accessKey, env.secretKey, bucketName,
+		env.accessKey, env.secretKey, env.s3Port, bucketName)
+
+	sqlFile := filepath.Join(env.dataDir, "duckdb_oauth_test.sql")
+	if err := os.WriteFile(sqlFile, []byte(sqlContent), 0644); err != nil {
+		t.Fatalf("write SQL file: %v", err)
+	}
+
+	// Run DuckDB in Docker.
+	// We use a simple test: get a token, create a secret, and verify the
+	// Iceberg extension can communicate with the catalog.
+	// The simpler fallback test just verifies CREATE SECRET succeeds (no 404).
+	fallbackSQL := fmt.Sprintf(`
+INSTALL iceberg;
+LOAD iceberg;
+
+CREATE SECRET (
+    TYPE ICEBERG,
+    ENDPOINT 'http://host.docker.internal:%d',
+    CLIENT_ID '%s',
+    CLIENT_SECRET '%s'
+);
+
+SELECT 'DuckDB Iceberg OAuth secret created successfully' as result;
+`, env.icebergPort, env.accessKey, env.secretKey)
+
+	fallbackFile := filepath.Join(env.dataDir, "duckdb_oauth_fallback.sql")
+	if err := os.WriteFile(fallbackFile, []byte(fallbackSQL), 0644); err != nil {
+		t.Fatalf("write fallback SQL file: %v", err)
+	}
+
+	// Try the full SQL script first (CREATE SECRET + catalog access).
+	// Fall back to the simple CREATE SECRET test if iceberg_scan isn't supported.
+	cmd := exec.Command("docker", "run", "--rm",
+		"-v", fmt.Sprintf("%s:/test", env.dataDir),
+		"--add-host", "host.docker.internal:host-gateway",
+		"--entrypoint", "duckdb",
+		"duckdb/duckdb:latest",
+		"-init", "/test/duckdb_oauth_test.sql",
+		"-c", "SELECT 1",
+	)
+
+	output, err := cmd.CombinedOutput()
+	outputStr := string(output)
+	t.Logf("DuckDB output:\n%s", outputStr)
+
+	if err != nil {
+		if strings.Contains(outputStr, "iceberg extension is not available") ||
+			strings.Contains(outputStr, "Failed to load") {
+			t.Skip("Skipping: Iceberg extension not available in DuckDB Docker image")
+		}
+		// The key check: the old error was "HTTP NotFound_404" on /v1/oauth/tokens.
+		// With our fix, this should no longer happen.
+		if strings.Contains(outputStr, "NotFound_404") && strings.Contains(outputStr, "/v1/oauth/tokens") {
+			t.Fatal("OAuth token endpoint returned 404 - the fix is not working")
+		}
+		// If iceberg_scan failed but CREATE SECRET worked, fall back to simpler test
+		if strings.Contains(outputStr, "OAuth token obtained successfully") {
+			t.Logf("Full SQL had partial success (token obtained), iceberg_scan may not be supported. Continuing.")
+		} else {
+			// Try the fallback script that only tests CREATE SECRET
+			t.Logf("Full SQL failed, trying fallback CREATE SECRET test...")
+			fallbackCmd := exec.Command("docker", "run", "--rm",
+				"-v", fmt.Sprintf("%s:/test", env.dataDir),
+				"--add-host", "host.docker.internal:host-gateway",
+				"--entrypoint", "duckdb",
+				"duckdb/duckdb:latest",
+				"-init", "/test/duckdb_oauth_fallback.sql",
+				"-c", "SELECT 1",
+			)
+			fallbackOutput, fallbackErr := fallbackCmd.CombinedOutput()
+			fallbackStr := string(fallbackOutput)
+			t.Logf("DuckDB fallback output:\n%s", fallbackStr)
+
+			if fallbackErr != nil {
+				if strings.Contains(fallbackStr, "NotFound_404") && strings.Contains(fallbackStr, "/v1/oauth/tokens") {
+					t.Fatal("OAuth token endpoint returned 404 - the fix is not working")
+				}
+				t.Fatalf("DuckDB fallback also failed: %v\nOutput: %s", fallbackErr, fallbackStr)
+			}
+			if !strings.Contains(fallbackStr, "DuckDB Iceberg OAuth secret created successfully") {
+				t.Errorf("expected success message in fallback output, got:\n%s", fallbackStr)
+			}
+			return
+		}
+	}
+
+	if strings.Contains(outputStr, "OAuth token obtained successfully") {
+		t.Logf("DuckDB OAuth CREATE SECRET succeeded")
+	} else {
+		t.Errorf("expected OAuth success message in output, got:\n%s", outputStr)
+	}
+}
+
+// requestOAuthToken obtains a bearer token from the OAuth endpoint.
+func requestOAuthToken(t *testing.T, env *oauthTestEnv, accessKey, secretKey string) string {
+	t.Helper()
+
+	resp, err := http.PostForm(env.icebergURL()+"/v1/oauth/tokens", url.Values{
+		"grant_type":    {"client_credentials"},
+		"client_id":     {accessKey},
+		"client_secret": {secretKey},
+	})
+	if err != nil {
+		t.Fatalf("POST /v1/oauth/tokens: %v", err)
+	}
+	defer resp.Body.Close()
+
+	body, _ := io.ReadAll(resp.Body)
+
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("OAuth token request failed: status=%d body=%s", resp.StatusCode, body)
+	}
+
+	var tokenResp struct {
+		AccessToken string `json:"access_token"`
+		TokenType   string `json:"token_type"`
+		ExpiresIn   int    `json:"expires_in"`
+	}
+	if err := json.Unmarshal(body, &tokenResp); err != nil {
+		t.Fatalf("decode token response: %v", err)
+	}
+
+	if tokenResp.AccessToken == "" {
+		t.Fatal("got empty access_token")
+	}
+	if tokenResp.TokenType != "bearer" {
+		t.Errorf("expected token_type=bearer, got %s", tokenResp.TokenType)
+	}
+	return tokenResp.AccessToken
+}
+
+// createTableBucketViaShell creates a table bucket using weed shell,
+// which bypasses S3 auth. This is the same approach used by the Trino tests.
+func createTableBucketViaShell(t *testing.T, env *oauthTestEnv, bucketName string) {
+	t.Helper()
+
+	cmd := exec.Command(env.weedBinary, "shell",
+		fmt.Sprintf("-master=%s:%d.%d", env.bindIP, env.masterPort, env.masterGrpcPort),
+	)
+	cmd.Stdin = strings.NewReader(fmt.Sprintf("s3tables.bucket -create -name %s -account 000000000000\nexit\n", bucketName))
+	output, err := cmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("create table bucket %s via weed shell: %v\nOutput: %s", bucketName, err, output)
+	}
+	t.Logf("Created table bucket %s", bucketName)
+}
+
+// createNamespaceWithToken creates a namespace using a Bearer token.
+func createNamespaceWithToken(t *testing.T, env *oauthTestEnv, token, bucketName, namespace string) {
+	t.Helper()
+
+	path := fmt.Sprintf("/v1/%s/namespaces", bucketName)
+	body := fmt.Sprintf(`{"namespace":["%s"]}`, namespace)
+	req, err := http.NewRequest(http.MethodPost, env.icebergURL()+path, strings.NewReader(body))
+	if err != nil {
+		t.Fatalf("create request: %v", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Bearer "+token)
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("create namespace: %v", err)
+	}
+	defer resp.Body.Close()
+
+	respBody, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusConflict {
+		t.Fatalf("create namespace failed: status=%d body=%s", resp.StatusCode, respBody)
+	}
+	t.Logf("Created namespace %s in bucket %s", namespace, bucketName)
+}
diff --git a/weed/command/s3.go b/weed/command/s3.go
index dc2191bcf..7c8b72b0e 100644
--- a/weed/command/s3.go
+++ b/weed/command/s3.go
@@ -514,6 +514,7 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) {
 
 	// Create Iceberg server using the S3ApiServer as filer client
 	icebergServer := iceberg.NewServer(s3ApiServer, s3ApiServer)
+	icebergServer.SetCredentialValidator(s3ApiServer)
 	icebergServer.RegisterRoutes(icebergRouter)
 
 	listenAddress := fmt.Sprintf("%s:%d", *s3opt.bindIp, *s3opt.portIceberg)
diff --git a/weed/s3api/iceberg/handlers_oauth.go b/weed/s3api/iceberg/handlers_oauth.go
new file mode 100644
index 000000000..58ebf5435
--- /dev/null
+++ b/weed/s3api/iceberg/handlers_oauth.go
@@ -0,0 +1,196 @@
+package iceberg
+
+import (
+	"crypto/hmac"
+	"crypto/sha256"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	jwt "github.com/golang-jwt/jwt/v5"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+)
+
+// OAuthTokenResponse is the response for POST /v1/oauth/tokens.
+type OAuthTokenResponse struct {
+	AccessToken string `json:"access_token"`
+	TokenType   string `json:"token_type"`
+	ExpiresIn   int    `json:"expires_in"`
+	Scope       string `json:"scope,omitempty"`
+}
+
+// OAuthErrorResponse is the error response for the OAuth endpoint.
+type OAuthErrorResponse struct {
+	Error       string `json:"error"`
+	Description string `json:"error_description,omitempty"`
+}
+
+// IcebergClaims are JWT claims for Iceberg catalog OAuth tokens.
+type IcebergClaims struct {
+	IdentityName string `json:"identity_name"`
+	AccessKey    string `json:"access_key"`
+	jwt.RegisteredClaims
+}
+
+const oauthTokenExpiry = 3600 // 1 hour in seconds
+
+// handleOAuthTokens implements the OAuth2 client_credentials flow.
+// POST /v1/oauth/tokens
+func (s *Server) handleOAuthTokens(w http.ResponseWriter, r *http.Request) {
+	if err := r.ParseForm(); err != nil {
+		writeOAuthError(w, http.StatusBadRequest, "invalid_request", "Could not parse form body")
+		return
+	}
+
+	// Reject credentials in query string to prevent leaking secrets into logs and caches.
+	if r.URL.Query().Get("client_secret") != "" {
+		writeOAuthError(w, http.StatusBadRequest, "invalid_request", "client_secret must not be sent in the URL")
+		return
+	}
+
+	grantType := r.PostFormValue("grant_type")
+	if grantType != "client_credentials" {
+		writeOAuthError(w, http.StatusBadRequest, "unsupported_grant_type",
+			fmt.Sprintf("Unsupported grant_type: %s", grantType))
+		return
+	}
+
+	clientID := r.PostFormValue("client_id")
+	clientSecret := r.PostFormValue("client_secret")
+
+	// Also support HTTP Basic auth per OAuth2 spec
+	if clientID == "" && clientSecret == "" {
+		var ok bool
+		clientID, clientSecret, ok = r.BasicAuth()
+		if !ok {
+			writeOAuthError(w, http.StatusUnauthorized, "invalid_client", "Missing client credentials")
+			return
+		}
+	}
+
+	if clientID == "" || clientSecret == "" {
+		writeOAuthError(w, http.StatusUnauthorized, "invalid_client", "Missing client_id or client_secret")
+		return
+	}
+
+	if s.credentialValidator == nil {
+		writeOAuthError(w, http.StatusInternalServerError, "server_error", "Credential validation not configured")
+		return
+	}
+
+	identityName, _, err := s.credentialValidator.ValidateS3Credential(clientID, clientSecret)
+	if err != nil {
+		glog.V(2).Infof("Iceberg OAuth: credential validation failed for client_id=%s: %v", clientID, err)
+		writeOAuthError(w, http.StatusUnauthorized, "invalid_client", "Invalid client credentials")
+		return
+	}
+
+	// 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)
+	if err != nil {
+		glog.Errorf("Iceberg OAuth: failed to sign token: %v", err)
+		writeOAuthError(w, http.StatusInternalServerError, "server_error", "Failed to generate token")
+		return
+	}
+
+	scope := r.PostFormValue("scope")
+	resp := OAuthTokenResponse{
+		AccessToken: tokenString,
+		TokenType:   "bearer",
+		ExpiresIn:   oauthTokenExpiry,
+		Scope:       scope,
+	}
+	w.Header().Set("Cache-Control", "no-store")
+	writeJSON(w, http.StatusOK, resp)
+}
+
+// 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) {
+	auth := r.Header.Get("Authorization")
+	if auth == "" {
+		return "", nil, false
+	}
+	if !strings.HasPrefix(strings.ToLower(auth), "bearer ") {
+		return "", nil, false
+	}
+	tokenString := strings.TrimSpace(auth[7:])
+	if tokenString == "" {
+		return "", nil, false
+	}
+
+	if s.credentialValidator == nil {
+		return "", nil, false
+	}
+
+	// Parse the token without verification first to get the access key,
+	// then look up the exact credential to verify the signature.
+	parser := jwt.NewParser(jwt.WithoutClaimsValidation())
+	unverified := &IcebergClaims{}
+	_, _, err := parser.ParseUnverified(tokenString, unverified)
+	if err != nil {
+		glog.V(2).Infof("Iceberg OAuth: failed to parse token: %v", err)
+		return "", nil, false
+	}
+
+	if unverified.AccessKey == "" {
+		return "", nil, false
+	}
+
+	// Look up the credential by access key to get the signing key for verification
+	identityName, identity, secretKey, err := s.credentialValidator.GetCredentialByAccessKey(unverified.AccessKey)
+	if err != nil {
+		glog.V(2).Infof("Iceberg OAuth: failed to get credential for access key: %v", err)
+		return "", nil, false
+	}
+
+	signingKey := deriveSigningKey(unverified.AccessKey, secretKey)
+	claims := &IcebergClaims{}
+	verified, err := jwt.ParseWithClaims(tokenString, 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
+	})
+	if err != nil || !verified.Valid {
+		glog.V(2).Infof("Iceberg OAuth: token verification failed: %v", err)
+		return "", nil, false
+	}
+
+	return identityName, identity, true
+}
+
+// deriveSigningKey derives a signing key from the access key and secret using HMAC-SHA256.
+// Including the access key prevents cross-credential token forgery when two
+// credentials happen to share the same secret.
+func deriveSigningKey(accessKey, secret string) []byte {
+	h := hmac.New(sha256.New, []byte("seaweedfs-iceberg-oauth"))
+	h.Write([]byte(accessKey))
+	h.Write([]byte{0}) // null separator
+	h.Write([]byte(secret))
+	return h.Sum(nil)
+}
+
+func writeOAuthError(w http.ResponseWriter, status int, errCode, description string) {
+	resp := OAuthErrorResponse{
+		Error:       errCode,
+		Description: description,
+	}
+	writeJSON(w, status, resp)
+}
+
diff --git a/weed/s3api/iceberg/handlers_oauth_test.go b/weed/s3api/iceberg/handlers_oauth_test.go
new file mode 100644
index 000000000..051d67eec
--- /dev/null
+++ b/weed/s3api/iceberg/handlers_oauth_test.go
@@ -0,0 +1,155 @@
+package iceberg
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+type mockCredentialValidator struct {
+	credentials map[string]string // accessKey -> secretKey
+	identities  map[string]string // accessKey -> identityName
+}
+
+func (m *mockCredentialValidator) ValidateS3Credential(accessKey, secretKey string) (string, interface{}, error) {
+	expected, ok := m.credentials[accessKey]
+	if !ok {
+		return "", nil, fmt.Errorf("access key not found")
+	}
+	if expected != secretKey {
+		return "", nil, fmt.Errorf("invalid secret key")
+	}
+	return m.identities[accessKey], nil, nil
+}
+
+func (m *mockCredentialValidator) GetCredentialByAccessKey(accessKey string) (string, interface{}, string, error) {
+	secret, ok := m.credentials[accessKey]
+	if !ok {
+		return "", nil, "", fmt.Errorf("access key not found")
+	}
+	return m.identities[accessKey], nil, secret, nil
+}
+
+func newTestServerWithOAuth() *Server {
+	cv := &mockCredentialValidator{
+		credentials: map[string]string{"AKID123": "secret456"},
+		identities:  map[string]string{"AKID123": "testuser"},
+	}
+	s := &Server{
+		credentialValidator: cv,
+	}
+	return s
+}
+
+func TestHandleOAuthTokens_Success(t *testing.T) {
+	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: %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" {
+		t.Errorf("expected token_type=bearer, got %s", resp.TokenType)
+	}
+	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)
+	}
+}
+
+func TestHandleOAuthTokens_InvalidCredentials(t *testing.T) {
+	s := newTestServerWithOAuth()
+
+	body := "grant_type=client_credentials&client_id=AKID123&client_secret=wrongsecret"
+	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.StatusUnauthorized {
+		t.Fatalf("expected 401, got %d: %s", w.Code, w.Body.String())
+	}
+}
+
+func TestHandleOAuthTokens_UnsupportedGrantType(t *testing.T) {
+	s := newTestServerWithOAuth()
+
+	body := "grant_type=authorization_code&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.StatusBadRequest {
+		t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String())
+	}
+}
+
+func TestBearerTokenRoundTrip(t *testing.T) {
+	s := newTestServerWithOAuth()
+
+	// Get a token
+	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)
+
+	var resp OAuthTokenResponse
+	if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
+		t.Fatal(err)
+	}
+
+	// Use the token for Bearer auth
+	authReq := httptest.NewRequest(http.MethodGet, "/v1/namespaces", nil)
+	authReq.Header.Set("Authorization", "Bearer "+resp.AccessToken)
+
+	identityName, _, ok := s.authenticateBearer(authReq)
+	if !ok {
+		t.Fatal("expected Bearer auth to succeed")
+	}
+	if identityName != "testuser" {
+		t.Errorf("expected identity 'testuser', got '%s'", identityName)
+	}
+}
+
+func TestBearerTokenInvalid(t *testing.T) {
+	s := newTestServerWithOAuth()
+
+	req := httptest.NewRequest(http.MethodGet, "/v1/namespaces", nil)
+	req.Header.Set("Authorization", "Bearer invalid-token")
+
+	_, _, ok := s.authenticateBearer(req)
+	if ok {
+		t.Error("expected Bearer auth to fail with invalid token")
+	}
+}
+
+func TestBearerTokenNone(t *testing.T) {
+	s := newTestServerWithOAuth()
+
+	req := httptest.NewRequest(http.MethodGet, "/v1/namespaces", nil)
+
+	_, _, ok := s.authenticateBearer(req)
+	if ok {
+		t.Error("expected Bearer auth to fail with no token")
+	}
+}
diff --git a/weed/s3api/iceberg/server.go b/weed/s3api/iceberg/server.go
index a4023b115..db8bee826 100644
--- a/weed/s3api/iceberg/server.go
+++ b/weed/s3api/iceberg/server.go
@@ -21,12 +21,25 @@ type S3Authenticator interface {
 	DefaultAllow() bool
 }
 
+// CredentialValidator validates S3 access key / secret key pairs
+// and provides credential lookup for OAuth token verification.
+type CredentialValidator interface {
+	// ValidateS3Credential checks if the access key and secret key are valid.
+	// Returns the identity name and identity object on success.
+	ValidateS3Credential(accessKey, secretKey string) (identityName string, identity interface{}, err error)
+	// GetCredentialByAccessKey looks up a credential by access key.
+	// Returns the identity name, identity object, and secret key.
+	// Used for verifying Bearer tokens signed with a specific credential.
+	GetCredentialByAccessKey(accessKey string) (identityName string, identity interface{}, secretKey string, err error)
+}
+
 // Server implements the Iceberg REST Catalog API.
 type Server struct {
-	filerClient   FilerClient
-	tablesManager *s3tables.Manager
-	prefix        string // optional prefix for routes
-	authenticator S3Authenticator
+	filerClient         FilerClient
+	tablesManager       *s3tables.Manager
+	prefix              string // optional prefix for routes
+	authenticator       S3Authenticator
+	credentialValidator CredentialValidator
 }
 
 // NewServer creates a new Iceberg REST Catalog server.
@@ -40,6 +53,11 @@ func NewServer(filerClient FilerClient, authenticator S3Authenticator) *Server {
 	}
 }
 
+// SetCredentialValidator sets the credential validator for OAuth token support.
+func (s *Server) SetCredentialValidator(cv CredentialValidator) {
+	s.credentialValidator = cv
+}
+
 // RegisterRoutes registers Iceberg REST API routes on the provided router.
 func (s *Server) RegisterRoutes(router *mux.Router) {
 	// Add middleware to log all requests/responses
@@ -48,6 +66,9 @@ func (s *Server) RegisterRoutes(router *mux.Router) {
 	// Configuration endpoint - no auth needed for config
 	router.HandleFunc("/v1/config", s.handleConfig).Methods(http.MethodGet)
 
+	// OAuth2 token endpoint - no auth needed (this IS the auth endpoint)
+	router.HandleFunc("/v1/oauth/tokens", s.handleOAuthTokens).Methods(http.MethodPost)
+
 	// Namespace endpoints - wrapped with Auth middleware
 	router.HandleFunc("/v1/namespaces", s.Auth(s.handleListNamespaces)).Methods(http.MethodGet)
 	router.HandleFunc("/v1/namespaces", s.Auth(s.handleCreateNamespace)).Methods(http.MethodPost)
@@ -122,6 +143,18 @@ 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)
+			}
+			r = r.WithContext(ctx)
+			handler(w, r)
+			return
+		}
+
 		if s.authenticator == nil {
 			writeError(w, http.StatusUnauthorized, "NotAuthorizedException", "Authentication required")
 			return
diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go
index 1596b247f..c48aa3072 100644
--- a/weed/s3api/s3api_server.go
+++ b/weed/s3api/s3api_server.go
@@ -1055,3 +1055,52 @@ func (s3a *S3ApiServer) DefaultAllow() bool {
 	}
 	return s3a.iam.iamIntegration.DefaultAllow()
 }
+
+// ValidateS3Credential validates an S3 access key / secret key pair.
+// Returns the identity name and identity object on success.
+func (s3a *S3ApiServer) ValidateS3Credential(accessKey, secretKey string) (string, interface{}, error) {
+	if s3a.iam == nil {
+		return "", nil, fmt.Errorf("IAM not configured")
+	}
+	identity, cred, found := s3a.iam.LookupByAccessKey(accessKey)
+	if !found {
+		return "", nil, fmt.Errorf("access key not found")
+	}
+	if cred.SecretKey != secretKey {
+		return "", nil, fmt.Errorf("invalid secret key")
+	}
+	if identity.Disabled {
+		return "", nil, fmt.Errorf("identity is disabled")
+	}
+	if cred.isCredentialExpired() {
+		return "", nil, fmt.Errorf("credential expired")
+	}
+	if cred.Status == "Inactive" {
+		return "", nil, fmt.Errorf("credential is inactive")
+	}
+	return identity.Name, identity, nil
+}
+
+// GetCredentialByAccessKey looks up a credential by access key.
+// Returns the identity name, identity object, and secret key.
+// Used for verifying Iceberg OAuth Bearer tokens with the exact credential
+// that was used to sign the token.
+func (s3a *S3ApiServer) GetCredentialByAccessKey(accessKey string) (string, interface{}, string, error) {
+	if s3a.iam == nil {
+		return "", nil, "", fmt.Errorf("IAM not configured")
+	}
+	identity, cred, found := s3a.iam.LookupByAccessKey(accessKey)
+	if !found {
+		return "", nil, "", fmt.Errorf("access key not found")
+	}
+	if identity.Disabled {
+		return "", nil, "", fmt.Errorf("identity is disabled")
+	}
+	if cred.isCredentialExpired() {
+		return "", nil, "", fmt.Errorf("credential expired")
+	}
+	if cred.Status == "Inactive" {
+		return "", nil, "", fmt.Errorf("credential is inactive")
+	}
+	return identity.Name, identity, cred.SecretKey, nil
+}