feat(auth): support HTTP bearer-token authentication (#8719)

* feat(auth): support HTTP bearer-token authentication (#8718)

Adds an opt-in bearer-token (JWT / OAuth2) authentication path to the HTTP
layer, so clients can authenticate with `Authorization: Bearer <token>`
against any `/v1/` interface. Today such requests are rejected with
`UnsupportedAuthScheme("bearer")` -> 401 before any handler runs.

The token is treated as opaque by the server; validation and identity
derivation stay in the UserProvider, so JWT/JWKS/OIDC policy remains
pluggable and out of core.

Changes:

- `auth::UserProvider` gains `auth_token(token, catalog, schema) ->
  Result<UserInfoRef>` with a default that rejects
  (`Error::UnsupportedAuthMethod`), so password-only providers keep today's
  behavior. A provider that supports token auth overrides it to validate the
  token, resolve it to a user, and authorize the connection.
- `auth::Error::UnsupportedAuthMethod` for the default-reject case.
- `servers::http::authorize::inner_auth` extracts a bearer token
  (`extract_bearer_token`) and, when present, authenticates via
  `UserProvider::auth_token`; otherwise it falls through unchanged to the
  username/password path (Basic / influxdb / splunk). Basic and bearer
  coexist on the same server.

Backward compatible: the default impl preserves existing behavior, and
non-bearer requests take the exact same path as before.

Tests:
- `extract_bearer_token` recognizes `Bearer` (either header) and ignores
  Basic/Token/Splunk/empty.
- `inner_auth` dispatches a bearer token to `auth_token` and populates the
  QueryContext user on success; rejects on failure.
- A password-only provider (default `auth_token`) rejects bearer tokens.

Refs: #8718

* chore: fmt

* refactor(auth): address bearer-auth review feedback (#8719)

Address the review comments on the HTTP bearer-token authentication PR:

- Match the `Bearer` scheme case-insensitively (RFC 9110 §11.1) via
  `eq_ignore_ascii_case`. `extract_bearer_token` previously only accepted
  `Bearer`/`bearer`, so a valid `BEARER <token>` fell through to
  `UnsupportedAuthScheme` and never reached the provider. The opaque token
  itself is deliberately not lowercased.
- Return `Option<&str>` (borrowing the request headers) instead of
  `Option<String>`, avoiding an allocation per bearer request.
- Rename `UserProvider::auth_token` -> `auth_bearer_token` for clarity.
- Route bearer-auth failures on Splunk HEC requests through `splunk_hec_err`
  (FORBIDDEN, code 4) instead of the generic 401 `ErrorResponse`, so HEC
  clients retain their `{"text":"Invalid token","code":4}` endpoint contract.

Adds test coverage for case-insensitive scheme parsing (token preserved
verbatim) and a regression test for the bearer/splunk routing path.

Signed-off-by: Ning Sun <sunning@greptime.com>

---------

Signed-off-by: Ning Sun <sunning@greptime.com>

src/auth/src/error.rs

@@ -88,6 +88,13 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("Unsupported authentication method: {}", method))]
+    UnsupportedAuthMethod {
+        method: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 impl ErrorExt for Error {
@@ -105,6 +112,7 @@ impl ErrorExt for Error {
             Error::UserPasswordMismatch { .. } => StatusCode::UserPasswordMismatch,
             Error::AccessDenied { .. } => StatusCode::AccessDenied,
             Error::PermissionDenied { .. } => StatusCode::PermissionDenied,
+            Error::UnsupportedAuthMethod { .. } => StatusCode::UserPasswordMismatch,
         }
     }
 

src/auth/src/user_provider.rs

@@ -34,8 +34,8 @@ use crate::common::{
     PgScramSha256Verifier, auth_mysql_with_hash_stage_2,
 };
 use crate::error::{
-    IllegalParamSnafu, InvalidConfigSnafu, IoSnafu, Result, UnsupportedPasswordTypeSnafu,
-    UserNotFoundSnafu, UserPasswordMismatchSnafu,
+    IllegalParamSnafu, InvalidConfigSnafu, IoSnafu, Result, UnsupportedAuthMethodSnafu,
+    UnsupportedPasswordTypeSnafu, UserNotFoundSnafu, UserPasswordMismatchSnafu,
 };
 use crate::user_info::{DefaultUserInfo, PermissionMode};
 use crate::{UserInfoRef, auth_mysql};
@@ -66,6 +66,31 @@ pub trait UserProvider: Send + Sync {
         Ok(user_info)
     }
 
+    /// Authenticates and authorizes an opaque bearer token (e.g. a JWT or an
+    /// OAuth2 access token).
+    ///
+    /// Unlike [auth()](Self::auth), the caller has no `Identity`/`Password` —
+    /// the provider validates the token and *derives* the identity from it.
+    /// The token is opaque to the server, so JWT/JWKS/OIDC validation policy
+    /// stays pluggable and out of core.
+    ///
+    /// The default rejects token auth with
+    /// [`Error::UnsupportedAuthMethod`], so password-only providers keep
+    /// today's behavior. Providers that support token auth override this to
+    /// validate the token, resolve it to a user, and
+    /// [`authorize`](Self::authorize) the connection.
+    async fn auth_bearer_token(
+        &self,
+        _token: &str,
+        _catalog: &str,
+        _schema: &str,
+    ) -> Result<UserInfoRef> {
+        UnsupportedAuthMethodSnafu {
+            method: "bearer token",
+        }
+        .fail()
+    }
+
     async fn postgres_auth_info(&self, _id: Identity<'_>) -> Result<PgAuthInfo> {
         Ok(PgAuthInfo::Cleartext)
     }

src/servers/src/http/authorize.rs

@@ -80,7 +80,36 @@ pub async fn inner_auth<B>(
         return Ok(req);
     };
 
-    // 3. get username and pwd
+    // 3. bearer token auth (JWT / OAuth2). When an `Authorization: Bearer
+    //    <token>` header is present, authenticate via
+    //    [`UserProvider::auth_bearer_token`]; otherwise fall through to the
+    //    username/password path (Basic / influxdb / splunk) below.
+    if let Some(token) = extract_bearer_token(&req) {
+        match user_provider
+            .auth_bearer_token(token, &catalog, &schema)
+            .await
+        {
+            Ok(userinfo) => {
+                query_ctx.set_current_user(userinfo);
+                let _ = req.extensions_mut().insert(query_ctx);
+                return Ok(req);
+            }
+            Err(e) => {
+                warn!(e; "bearer token authentication failed");
+                crate::metrics::METRIC_AUTH_FAILURE
+                    .with_label_values(&[e.status_code().as_ref()])
+                    .inc();
+                // Splunk HEC clients expect `{"text":"Invalid token","code":4}`
+                // (FORBIDDEN), not the generic 401 `ErrorResponse`.
+                if is_splunk_request(&req) {
+                    return Err(splunk_hec_err(StatusCode::FORBIDDEN, 4));
+                }
+                return Err(err_response(e));
+            }
+        }
+    }
+
+    // 4. get username and pwd
     let (username, password) = match extract_username_and_password(&req) {
         Ok((username, password)) => (username, password),
         Err(e) => {
@@ -100,7 +129,7 @@ pub async fn inner_auth<B>(
         }
     };
 
-    // 4. auth
+    // 5. auth
     match user_provider
         .auth(
             auth::Identity::UserId(&username, None),
@@ -307,6 +336,29 @@ impl From<AuthScheme> for api::v1::auth_header::AuthScheme {
 
 type Credential<'a> = &'a str;
 
+/// Extracts an opaque bearer token from an `Authorization: Bearer <token>`
+/// header (the standard or `x-greptime-auth` header).
+///
+/// Returns `None` for any other scheme (`Basic`, influxdb `Token`, splunk
+/// `Splunk`, …) so the caller can fall through to the username/password path.
+fn extract_bearer_token<B>(req: &Request<B>) -> Option<&str> {
+    let header = req
+        .headers()
+        .get(AUTHORIZATION_HEADER)
+        .or_else(|| req.headers().get(http::header::AUTHORIZATION))?;
+    let value = header.to_str().ok()?;
+    // HTTP authentication schemes are case-insensitive (RFC 9110 §11.1), so
+    // match the scheme with `eq_ignore_ascii_case` — but never lowercase the
+    // *token* itself, which is opaque and case-sensitive. Returns a borrow
+    // into the request's headers, so there is no allocation.
+    let (scheme, token) = value.split_once(' ')?;
+    if !scheme.eq_ignore_ascii_case("bearer") {
+        return None;
+    }
+    let token = token.trim_start();
+    (!token.is_empty()).then_some(token)
+}
+
 fn auth_header<B>(req: &Request<B>) -> Result<AuthScheme> {
     let auth_header = req
         .headers()
@@ -610,4 +662,174 @@ mod tests {
             (Some("123"), Some("4"))
         );
     }
+
+    #[test]
+    fn test_extract_bearer_token() {
+        let bearer = |scheme: &str, val: &str| {
+            mock_http_request(Some(&format!("{scheme} {val}")), None).unwrap()
+        };
+
+        // Standard bearer scheme, on either header.
+        assert_eq!(
+            extract_bearer_token(&bearer("Bearer", "abc.def.ghi")),
+            Some("abc.def.ghi")
+        );
+        assert_eq!(extract_bearer_token(&bearer("bearer", "tok")), Some("tok"));
+        // HTTP schemes are case-insensitive (RFC 9110 §11.1); the token
+        // itself is opaque and must NOT be lowercased.
+        assert_eq!(
+            extract_bearer_token(&bearer("BEARER", "ABC.DEF.GHI")),
+            Some("ABC.DEF.GHI")
+        );
+        assert_eq!(extract_bearer_token(&bearer("BeArEr", "tok")), Some("tok"));
+        let mut req = mock_http_request(Some("Bearer xyz"), None).unwrap();
+        req.headers_mut().insert(
+            AUTHORIZATION_HEADER,
+            "Bearer from-x-greptime".parse().unwrap(),
+        );
+        assert_eq!(extract_bearer_token(&req), Some("from-x-greptime"));
+
+        // Non-bearer schemes are ignored so the caller falls through to Basic.
+        assert_eq!(extract_bearer_token(&bearer("Basic", "dXNlcjpwYXNz")), None);
+        assert_eq!(extract_bearer_token(&bearer("Token", "u:p")), None);
+        assert_eq!(extract_bearer_token(&bearer("Splunk", "u:p")), None);
+
+        // No header, empty token.
+        assert_eq!(
+            extract_bearer_token(&mock_http_request(None, None).unwrap()),
+            None
+        );
+        assert_eq!(extract_bearer_token(&bearer("Bearer", "")), None);
+    }
+
+    /// A `UserProvider` that resolves exactly one bearer token to a known user
+    /// and rejects everything else (including password auth).
+    struct TokenUserProvider {
+        token: String,
+        user: auth::UserInfoRef,
+    }
+
+    #[async_trait::async_trait]
+    impl auth::UserProvider for TokenUserProvider {
+        fn name(&self) -> &str {
+            "token-test"
+        }
+
+        async fn authenticate(
+            &self,
+            _: auth::Identity<'_>,
+            _: auth::Password<'_>,
+        ) -> auth::error::Result<auth::UserInfoRef> {
+            unreachable!("password auth should not be reached for a bearer request")
+        }
+
+        async fn authorize(
+            &self,
+            _: &str,
+            _: &str,
+            _: &auth::UserInfoRef,
+        ) -> auth::error::Result<()> {
+            Ok(())
+        }
+
+        async fn auth_bearer_token(
+            &self,
+            token: &str,
+            _: &str,
+            _: &str,
+        ) -> auth::error::Result<auth::UserInfoRef> {
+            if token == self.token {
+                Ok(self.user.clone())
+            } else {
+                auth::error::UnsupportedAuthMethodSnafu {
+                    method: "bearer token",
+                }
+                .fail()
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn test_bearer_token_dispatches_to_auth_bearer_token() {
+        let provider = TokenUserProvider {
+            token: "good-token".to_string(),
+            user: auth::userinfo_by_name(Some("alice".into())),
+        };
+        let req = mock_http_request(Some("Bearer good-token"), None).unwrap();
+
+        let req = inner_auth::<()>(
+            Some(std::sync::Arc::new(provider) as auth::UserProviderRef),
+            req,
+        )
+        .await
+        .expect("valid bearer token authenticates");
+
+        let user = req
+            .extensions()
+            .get::<session::context::QueryContext>()
+            .expect("query context is populated")
+            .current_user();
+        assert_eq!(user.username(), "alice");
+    }
+
+    #[tokio::test]
+    async fn test_bearer_token_failure_rejects() {
+        let provider = TokenUserProvider {
+            token: "good-token".to_string(),
+            user: auth::userinfo_by_name(Some("alice".into())),
+        };
+        let req = mock_http_request(Some("Bearer bad-token"), None).unwrap();
+
+        let result = inner_auth::<()>(
+            Some(std::sync::Arc::new(provider) as auth::UserProviderRef),
+            req,
+        )
+        .await;
+        assert!(result.is_err(), "an invalid bearer token is rejected");
+    }
+
+    #[tokio::test]
+    async fn test_default_provider_rejects_bearer() {
+        // A password-only provider uses the default `auth_bearer_token`, which rejects.
+        let provider =
+            auth::user_provider_from_option("static_user_provider:cmd:alice=s3cret").unwrap();
+        let req = mock_http_request(Some("Bearer some-jwt"), None).unwrap();
+        let result = inner_auth::<()>(Some(provider), req).await;
+        assert!(
+            result.is_err(),
+            "password-only providers reject bearer tokens"
+        );
+    }
+
+    /// A bearer-token failure on a Splunk HEC request must keep the HEC
+    /// contract — `{"text":"Invalid token","code":4}` with FORBIDDEN —
+    /// instead of falling through to the generic 401 `ErrorResponse`.
+    /// Regression for the bearer/splunk routing gap.
+    #[tokio::test]
+    async fn test_bearer_failure_on_splunk_keeps_hec_contract() {
+        let provider = TokenUserProvider {
+            token: "good-token".to_string(),
+            user: auth::userinfo_by_name(Some("alice".into())),
+        };
+        let req = mock_http_request(
+            Some("Bearer bad-token"),
+            Some("http://127.0.0.1/v1/splunk/services/collector/event"),
+        )
+        .unwrap();
+
+        let resp = inner_auth::<()>(
+            Some(std::sync::Arc::new(provider) as auth::UserProviderRef),
+            req,
+        )
+        .await
+        .expect_err("an invalid bearer token is rejected");
+
+        assert_eq!(resp.status(), StatusCode::FORBIDDEN);
+        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
+            .await
+            .unwrap();
+        let payload: serde_json::Value = serde_json::from_slice(&body).unwrap();
+        assert_eq!(payload["code"], 4);
+        assert_eq!(payload["text"], "Invalid token");
+    }
 }