feat: add MCP OAuth 2.0 support and auth improvements

- Implement OAuth 2.0 Authorization Server Metadata endpoint (RFC 8414)
- Add JWT token validation for MCP endpoints with dynamic clients
- Add `WWW-Authenticate` header support for unauthorized responses
- Add `AuthTokensExt::has_expired()` method for token validation
- Refactor base64 decode to accept generic byte inputs
- Update authentication flow to allow MCP users without DB records
- Clippy delinting

src/common/meta/user.rs

@@ -456,6 +456,13 @@ pub struct AuthTokensExt {
     pub expires_in: i64,
 }
 
+impl AuthTokensExt {
+    /// Checks if the token is still valid or not
+    pub fn has_expired(&self) -> bool {
+        chrono::Utc::now().timestamp() - self.request_time > self.expires_in
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::collections::HashSet;

src/config/src/meta/stream.rs

@@ -986,16 +986,16 @@ impl From<&str> for StreamSettings {
             defined_schema_fields = fields;
         }
 
-        let flatten_level = settings.get("flatten_level").map(|v| v.as_i64().unwrap());
+        let flatten_level = settings.get("flatten_level").and_then(Value::as_i64);
 
         let store_original_data = settings
             .get("store_original_data")
-            .and_then(|v| v.as_bool())
-            .unwrap_or(false);
+            .and_then(Value::as_bool)
+            .unwrap_or_default();
 
         let approx_partition = settings
             .get("approx_partition")
-            .and_then(|v| v.as_bool())
+            .and_then(Value::as_bool)
             .unwrap_or(
                 get_config()
                     .common
@@ -1013,7 +1013,7 @@ impl From<&str> for StreamSettings {
 
         let index_updated_at = settings
             .get("index_updated_at")
-            .and_then(|v| v.as_i64())
+            .and_then(Value::as_i64)
             .unwrap_or_default();
 
         let mut extended_retention_days = vec![];
@@ -1024,26 +1024,26 @@ impl From<&str> for StreamSettings {
             for item in values {
                 let start = item
                     .get("start")
-                    .and_then(|v| v.as_i64())
+                    .and_then(Value::as_i64)
                     .unwrap_or_default();
-                let end = item.get("end").and_then(|v| v.as_i64()).unwrap_or_default();
+                let end = item.get("end").and_then(Value::as_i64).unwrap_or_default();
                 extended_retention_days.push(TimeRange::new(start, end));
             }
         }
 
         let index_original_data = settings
             .get("index_original_data")
-            .and_then(|v| v.as_bool())
-            .unwrap_or(false);
+            .and_then(Value::as_bool)
+            .unwrap_or_default();
 
         let index_all_values = settings
             .get("index_all_values")
-            .and_then(|v| v.as_bool())
-            .unwrap_or(false);
+            .and_then(Value::as_bool)
+            .unwrap_or_default();
         let enable_distinct_fields = settings
             .get("enable_distinct_fields")
-            .and_then(|v| v.as_bool())
-            .unwrap_or(true);
+            .and_then(Value::as_bool)
+            .unwrap_or_default();
         Self {
             partition_time_level,
             partition_keys,

src/config/src/utils/base64.rs

@@ -17,7 +17,7 @@ use std::io::{Error, ErrorKind};
 
 use base64::Engine;
 
-pub fn decode(s: &str) -> Result<String, Error> {
+pub fn decode(s: impl AsRef<[u8]>) -> Result<String, Error> {
     match String::from_utf8(decode_raw(s)?) {
         Ok(v) => Ok(v),
         Err(e) => Err(Error::new(
@@ -27,9 +27,9 @@ pub fn decode(s: &str) -> Result<String, Error> {
     }
 }
 
-pub fn decode_raw(s: &str) -> Result<Vec<u8>, Error> {
+pub fn decode_raw(s: impl AsRef<[u8]>) -> Result<Vec<u8>, Error> {
     base64::engine::general_purpose::STANDARD
-        .decode(s.as_bytes())
+        .decode(s.as_ref())
         .map_err(|e| Error::new(ErrorKind::InvalidData, format!("base64 decode error: {e}")))
 }
 

src/handler/http/auth/token.rs

@@ -50,12 +50,18 @@ pub async fn token_validator(
     };
     let path_columns = path.split('/').collect::<Vec<&str>>();
 
+    // Check if this is an MCP endpoint request
+    let is_mcp_request = path_columns.get(1).map(|s| *s == "mcp").unwrap_or(false);
+
+    // For MCP endpoints with dynamic clients, skip audience validation
+    let login_flow = !is_mcp_request;
+
     match jwt::verify_decode_token(
         auth_info.auth.strip_prefix("Bearer").unwrap().trim(),
         &keys,
         &get_dex_config().client_id,
         false,
-        true,
+        login_flow,
     ) {
         Ok(res) => {
             let user_id = &res.0.user_email;
@@ -75,6 +81,9 @@ pub async fn token_validator(
                 let is_list_invite_call = path_columns.len() <= 2
                     && path_columns.first().is_some_and(|p| p.eq(&"invites"))
                     && (auth_info.method.eq("GET") || auth_info.method.eq("DELETE"));
+
+                // For MCP endpoints, allow any authenticated user from Dex
+                let allow_nonexistent_user = is_mcp_request;
                 let path_suffix = path_columns.last().unwrap_or(&"");
                 if path_suffix.eq(&"organizations")
                     || path_suffix.eq(&"clusters")
@@ -119,7 +128,10 @@ pub async fn token_validator(
                     // nothing else. Similarly for accepting invite, we will not have the
                     // user in db anymore, and the fn checks based on email and token, so ok to
                     // bypass
-                    None if (is_list_invite_call || is_member_subscription) => {
+                    None if (is_list_invite_call
+                        || is_member_subscription
+                        || allow_nonexistent_user) =>
+                    {
                         let mut req = req;
 
                         if req.method().eq(&Method::POST)

src/handler/http/auth/validator.rs

@@ -17,7 +17,7 @@ use core::clone::Clone;
 use std::net::IpAddr;
 
 use actix_web::{
-    Error,
+    Error, HttpResponse,
     dev::ServiceRequest,
     error::{ErrorForbidden, ErrorNotFound, ErrorUnauthorized},
     http::{Method, header},
@@ -672,13 +672,13 @@ async fn oo_validator_internal(
     auth_info: AuthExtractor,
     path_prefix: &str,
 ) -> Result<ServiceRequest, (Error, ServiceRequest)> {
-    if auth_info.auth.starts_with("Basic") {
-        let decoded = match base64::decode(auth_info.auth.strip_prefix("Basic").unwrap().trim()) {
+    if let Some(info) = auth_info.auth.strip_prefix("Basic ").map(str::trim) {
+        let decoded = match base64::decode(info) {
             Ok(val) => val,
             Err(_) => return Err((ErrorUnauthorized("Unauthorized Access"), req)),
         };
 
-        let (username, password) = match get_user_details(decoded) {
+        let (username, password) = match get_user_details(&decoded) {
             Some(value) => value,
             None => return Err((ErrorUnauthorized("Unauthorized Access"), req)),
         };
@@ -686,11 +686,10 @@ async fn oo_validator_internal(
     } else if auth_info.auth.starts_with("Bearer") {
         log::debug!("Bearer token found");
         super::token::token_validator(req, auth_info).await
-    } else if auth_info.auth.starts_with("{\"auth_ext\":") {
+    } else if let Ok(auth_tokens) = config::utils::json::from_str::<AuthTokensExt>(&auth_info.auth)
+    {
         log::debug!("Auth ext token found");
-        let auth_tokens: AuthTokensExt =
-            config::utils::json::from_str(&auth_info.auth).unwrap_or_default();
-        if chrono::Utc::now().timestamp() - auth_tokens.request_time > auth_tokens.expires_in {
+        if auth_tokens.has_expired() {
             Err((ErrorUnauthorized("Unauthorized Access"), req))
         } else {
             log::debug!("Auth ext token found: decoding");
@@ -704,7 +703,7 @@ async fn oo_validator_internal(
                 Ok(val) => val,
                 Err(_) => return Err((ErrorUnauthorized("Unauthorized Access"), req)),
             };
-            let (username, password) = match get_user_details(decoded) {
+            let (username, password) = match get_user_details(&decoded) {
                 Some(value) => value,
                 None => return Err((ErrorUnauthorized("Unauthorized Access"), req)),
             };
@@ -712,7 +711,8 @@ async fn oo_validator_internal(
             validator(req, &username, &password, auth_info, path_prefix).await
         }
     } else {
-        Err((ErrorUnauthorized("Unauthorized Access"), req))
+        // Missing or unrecognized auth - return WWW-Authenticate header
+        Err((create_unauthorized_response("Unauthorized Access"), req))
     }
 }
 
@@ -756,19 +756,40 @@ pub async fn get_user_email_from_auth_str(auth_str: &str) -> Option<String> {
     }
 }
 
-fn get_user_details(decoded: String) -> Option<(String, String)> {
-    let credentials = match String::from_utf8(decoded.into()).map_err(|_| ()) {
-        Ok(val) => val,
-        Err(_) => return None,
+fn get_user_details(decoded: impl AsRef<[u8]>) -> Option<(String, String)> {
+    let credentials = str::from_utf8(decoded.as_ref()).ok()?;
+    credentials
+        .split_once(':')
+        .map(|(u, p)| (u.to_string(), p.to_string()))
+}
+
+/// Creates an Unauthorized error response with WWW-Authenticate header
+fn create_unauthorized_response(message: &str) -> Error {
+    #[cfg(feature = "enterprise")]
+    let auth_server_uri = {
+        let dex_config = get_dex_config();
+        if dex_config.dex_enabled {
+            dex_config.dex_url.clone()
+        } else {
+            String::new()
+        }
     };
-    let parts: Vec<&str> = credentials.splitn(2, ':').collect();
-    if parts.len() != 2 {
-        return None;
-    }
-    let (username, password) = (parts[0], parts[1]);
-    let username = username.to_owned();
-    let password = password.to_owned();
-    Some((username, password))
+    #[cfg(not(feature = "enterprise"))]
+    let auth_server_uri = String::new();
+
+    let www_authenticate = if !auth_server_uri.is_empty() {
+        format!(r#"Bearer as_uri="{auth_server_uri}""#)
+    } else {
+        r#"Bearer realm="openobserve""#.to_string()
+    };
+
+    actix_web::error::InternalError::from_response(
+        message.to_string(),
+        HttpResponse::Unauthorized()
+            .insert_header(("WWW-Authenticate", www_authenticate))
+            .finish(),
+    )
+    .into()
 }
 
 /// Validates the authentication information in the incoming request and returns the request if

src/handler/http/request/mcp/mod.rs

@@ -18,9 +18,19 @@ use actix_web::{HttpResponse, get, post, web};
 use {
     actix_web::HttpRequest,
     futures_util::StreamExt,
-    o2_enterprise::enterprise::mcp::{MCPRequest, handle_mcp_request, handle_mcp_request_stream},
+    o2_enterprise::enterprise::mcp::{
+        MCPRequest, OAuthServerMetadata, handle_mcp_request, handle_mcp_request_stream,
+    },
 };
 
+#[cfg(not(feature = "enterprise"))]
+/// Returns a 404 response for non-enterprise builds
+fn mcp_only_in_enterprise() -> actix_web::Result<HttpResponse> {
+    Ok(HttpResponse::NotFound().json(serde_json::json!({
+        "error": "MCP server is only available in enterprise edition"
+    })))
+}
+
 /// Handler for MCP POST requests (single request/response)
 #[cfg(feature = "enterprise")]
 #[utoipa::path(
@@ -112,6 +122,9 @@ pub async fn handle_mcp_post(
 #[get("/{org_id}/mcp")]
 pub async fn handle_mcp_get(
     _org_id: web::Path<String>,
+    // This is `Bytes` because body in HTTP GET is not defined,
+    // and actix-web tends to drop JSON body.
+    // This is a workaround to prevent that.
     body: web::Bytes,
     req: HttpRequest,
 ) -> actix_web::Result<HttpResponse> {
@@ -176,9 +189,7 @@ pub async fn handle_mcp_post(
     _org_id: web::Path<String>,
     _body: web::Bytes,
 ) -> actix_web::Result<HttpResponse> {
-    Ok(HttpResponse::NotFound().json(serde_json::json!({
-        "error": "MCP server is only available in enterprise edition"
-    })))
+    mcp_only_in_enterprise()
 }
 
 #[cfg(not(feature = "enterprise"))]
@@ -202,7 +213,58 @@ pub async fn handle_mcp_get(
     _org_id: web::Path<String>,
     _body: web::Bytes,
 ) -> actix_web::Result<HttpResponse> {
-    Ok(HttpResponse::NotFound().json(serde_json::json!({
-        "error": "MCP server is only available in enterprise edition"
-    })))
+    mcp_only_in_enterprise()
+}
+
+/// Handler for OAuth 2.0 Authorization Server Metadata (Enterprise)
+/// RFC 8414: https://datatracker.ietf.org/doc/html/rfc8414
+/// This endpoint provides discovery information for OAuth clients
+#[cfg(feature = "enterprise")]
+#[utoipa::path(
+    context_path = "/api",
+    tag = "MCP",
+    operation_id = "OAuthServerMetadata",
+    params(
+        ("org_id" = String, Path, description = "Organization name"),
+    ),
+    responses(
+        (status = 200, description = "Success", content_type = "application/json"),
+    ),
+)]
+#[get("/{org_id}/.well-known/oauth-authorization-server")]
+pub async fn oauth_authorization_server_metadata(
+    _org_id: web::Path<String>,
+) -> actix_web::Result<HttpResponse> {
+    use once_cell::sync::Lazy;
+
+    static METADATA: Lazy<OAuthServerMetadata> = Lazy::new(|| {
+        let dex_config = o2_dex::config::get_config();
+        let base_url = dex_config.dex_url.as_str();
+        OAuthServerMetadata::build(base_url)
+    });
+    Ok(HttpResponse::Ok()
+        .content_type("application/json")
+        .json(&*METADATA))
+}
+
+/// Handler for OAuth 2.0 Authorization Server Metadata (Non-Enterprise)
+/// RFC 8414: https://datatracker.ietf.org/doc/html/rfc8414
+/// This endpoint provides discovery information for OAuth clients
+#[cfg(not(feature = "enterprise"))]
+#[utoipa::path(
+    context_path = "/api",
+    tag = "MCP",
+    operation_id = "OAuthServerMetadata",
+    params(
+        ("org_id" = String, Path, description = "Organization name"),
+    ),
+    responses(
+        (status = 404, description = "Not Found - MCP server is only available in enterprise edition", content_type = "application/json"),
+    ),
+)]
+#[get("/{org_id}/.well-known/oauth-authorization-server")]
+pub async fn oauth_authorization_server_metadata(
+    _org_id: web::Path<String>,
+) -> actix_web::Result<HttpResponse> {
+    mcp_only_in_enterprise()
 }

src/handler/http/request/re_pattern/mod.rs

@@ -436,7 +436,7 @@ pub async fn test(web::Json(req): web::Json<PatternTestRequest>) -> Result<HttpR
         let pattern = req.pattern;
         let inputs = req.test_records;
         // Default to Redact if policy not specified for backward compatibility
-        let policy = req.policy.as_ref().map(|p| p.as_str()).unwrap_or("Redact");
+        let policy = req.policy.as_deref().unwrap_or("Redact");
         let policy = PatternPolicy::from(policy);
 
         let mut ret = Vec::with_capacity(inputs.len());

src/handler/http/request/search/mod.rs

@@ -1238,7 +1238,7 @@ async fn values_v1(
                 .unwrap_or("".to_string());
             let num = row
                 .get("zo_sql_num")
-                .map(|v| v.as_i64().unwrap_or(0))
+                .and_then(json::Value::as_i64)
                 .unwrap_or(0);
             top_hits.push((key, num));
         }

src/handler/http/router/mod.rs

@@ -562,7 +562,10 @@ pub fn get_service_routes(svc: &mut web::ServiceConfig) {
         .service(service_accounts::save)
         .service(service_accounts::delete)
         .service(service_accounts::update)
-        .service(service_accounts::get_api_token);
+        .service(service_accounts::get_api_token)
+        .service(mcp::handle_mcp_post)
+        .service(mcp::handle_mcp_get)
+        .service(mcp::oauth_authorization_server_metadata);
 
     #[cfg(feature = "enterprise")]
     let service = service
@@ -605,9 +608,7 @@ pub fn get_service_routes(svc: &mut web::ServiceConfig) {
         .service(re_pattern::delete)
         .service(re_pattern::test)
         .service(domain_management::get_domain_management_config)
-        .service(domain_management::set_domain_management_config)
-        .service(mcp::handle_mcp_post)
-        .service(mcp::handle_mcp_get);
+        .service(domain_management::set_domain_management_config);
 
     #[cfg(feature = "cloud")]
     let service = service
@@ -626,11 +627,6 @@ pub fn get_service_routes(svc: &mut web::ServiceConfig) {
         .service(organization::org::all_organizations)
         .service(organization::org::extend_trial_period);
 
-    #[cfg(not(feature = "enterprise"))]
-    let service = service
-        .service(mcp::handle_mcp_post)
-        .service(mcp::handle_mcp_get);
-
     svc.service(service);
 }
 

src/infra/src/db/nats.rs

@@ -790,7 +790,7 @@ fn key_encode(key: &str) -> String {
 
 #[inline]
 fn key_decode(key: &str) -> String {
-    base64::decode(&key.replace('-', "+").replace('_', "/")).unwrap()
+    base64::decode(key.replace('-', "+").replace('_', "/")).unwrap()
 }
 
 #[cfg(test)]

src/infra/src/table/migration/m20241119_000002_populate_dashboards_table.rs

@@ -17,6 +17,7 @@
 //! records from the meta table.
 
 use chrono::{DateTime, FixedOffset};
+use config::utils::json;
 use sea_orm::{
     ActiveValue::NotSet, EntityTrait, FromQueryResult, Paginator, PaginatorTrait, SelectModel, Set,
     Statement, TransactionTrait,
@@ -167,11 +168,11 @@ impl TryFrom<MetaDashboard> for dashboards::ActiveModel {
         let role = obj
             .remove("role")
             .and_then(|v| v.as_str().map(|s| s.to_string()))
-            .and_then(|s| if s.is_empty() { None } else { Some(s) });
+            .filter(|s| !s.is_empty());
         let description = obj
             .remove("description")
             .and_then(|v| v.as_str().map(|s| s.to_string()))
-            .and_then(|s| if s.is_empty() { None } else { Some(s) });
+            .filter(|s| !s.is_empty());
 
         // Since the created field includes a timezone that cannot be
         // represented in all databases and since the UI expects to receieve a
@@ -179,9 +180,11 @@ impl TryFrom<MetaDashboard> for dashboards::ActiveModel {
         // JSON while also migrating its value into the created_at Unix timestamp column.
         let created_at_tz: DateTime<FixedOffset> = obj
             .get("created")
-            .and_then(|v| v.as_str().map(|s| s.to_string()))
+            .and_then(json::Value::as_str)
             .ok_or("Dashboard JSON does not have string \"created\" field".to_string())
-            .and_then(|s| DateTime::parse_from_rfc3339(&s).map_err(|e| e.to_string()))?;
+            .map(DateTime::parse_from_rfc3339)?
+            .map_err(|e| e.to_string())?;
+
         let created_at_unix = created_at_tz.timestamp();
 
         Ok(dashboards::ActiveModel {

src/job/query_optimization_recommendation.rs

@@ -13,8 +13,6 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-#![cfg(feature = "enterprise")]
-
 use std::{pin::Pin, sync::Arc};
 
 use config::{META_ORG_ID, get_config, meta::stream::StreamType};
@@ -79,7 +77,7 @@ impl QueryRecommendationEngine for OptimizerContext {
         recommendations: Vec<OptimiserRecommendation>,
     ) -> Pin<Box<dyn Future<Output = Result<IngestionResponse, anyhow::Error>> + Send>> {
         Box::pin(async move {
-            let ingester = Ingester::default();
+            let ingester = Ingester;
             let request = IngestionRequest {
                 org_id: META_ORG_ID.to_string(),
                 stream_type: StreamType::Logs.to_string(),
@@ -95,8 +93,7 @@ impl QueryRecommendationEngine for OptimizerContext {
             Ok(ingester
                 .ingest(tonic::Request::new(request))
                 .await?
-                .into_inner()
-                .into())
+                .into_inner())
         })
     }
 }

src/service/promql/value.rs

@@ -313,7 +313,7 @@ impl<'de> Deserialize<'de> for Exemplar {
 
 impl From<&json::Map<String, json::Value>> for Exemplar {
     fn from(data: &json::Map<String, json::Value>) -> Self {
-        let timestamp = data.get("_timestamp").map(|v| v.as_i64().unwrap_or(0));
+        let timestamp = data.get("_timestamp").and_then(json::Value::as_i64);
         let value = data.get("value").map(|v| v.as_f64().unwrap_or(0.0));
         let mut labels = vec![];
         for (k, v) in data.iter() {

src/service/search/streaming/utils.rs

@@ -69,7 +69,7 @@ pub fn get_top_k_values(
             .unwrap_or_default();
         let num = hit
             .get("zo_sql_num")
-            .and_then(|v| v.as_i64())
+            .and_then(Value::as_i64)
             .unwrap_or_default();
         search_result_hits.push((key, num));
     }

src/service/search/utils.rs

@@ -228,38 +228,38 @@ mod tests {
     #[test]
     fn test_is_cachable_function_error() {
         let error = vec![];
-        assert_eq!(is_permissable_function_error(&error), true);
+        assert!(is_permissable_function_error(&error));
 
         let error = vec!["".to_string()];
-        assert_eq!(is_permissable_function_error(&error), true);
+        assert!(is_permissable_function_error(&error));
 
         let error = vec![
             CAPPED_RESULTS_MSG.to_string(),
             PARTIAL_ERROR_RESPONSE_MESSAGE.to_string(),
         ];
-        assert_eq!(is_permissable_function_error(&error), true); // only this is cachable
+        assert!(is_permissable_function_error(&error)); // only this is cachable
 
         let error = vec![
             CAPPED_RESULTS_MSG.to_string(),
             PARTIAL_ERROR_RESPONSE_MESSAGE.to_string(),
             "parquet not found".to_string(),
         ];
-        assert_eq!(is_permissable_function_error(&error), false);
+        assert!(!is_permissable_function_error(&error));
 
         let error = vec![
             "parquet not found".to_string(),
             PARTIAL_ERROR_RESPONSE_MESSAGE.to_string(),
         ];
-        assert_eq!(is_permissable_function_error(&error), false);
+        assert!(!is_permissable_function_error(&error));
 
         let error = vec!["parquet not found".to_string()];
-        assert_eq!(is_permissable_function_error(&error), false);
+        assert!(!is_permissable_function_error(&error));
 
         let error = vec![
             "parquet not found".to_string(),
             CAPPED_RESULTS_MSG.to_string(),
         ];
-        assert_eq!(is_permissable_function_error(&error), false);
+        assert!(!is_permissable_function_error(&error));
     }
 
     #[test]