From bb725e94ee154ae0baecb3f256d9f0bc9c85f06e Mon Sep 17 00:00:00 2001 From: ByteBaker <42913098+ByteBaker@users.noreply.github.com> Date: Mon, 20 Oct 2025 13:33:56 +0530 Subject: [PATCH] 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 | 7 ++ src/config/src/meta/stream.rs | 26 +++---- src/config/src/utils/base64.rs | 6 +- src/handler/http/auth/token.rs | 16 +++- src/handler/http/auth/validator.rs | 65 ++++++++++------ src/handler/http/request/mcp/mod.rs | 76 +++++++++++++++++-- src/handler/http/request/re_pattern/mod.rs | 2 +- src/handler/http/request/search/mod.rs | 2 +- src/handler/http/router/mod.rs | 14 ++-- src/infra/src/db/nats.rs | 2 +- ...241119_000002_populate_dashboards_table.rs | 11 ++- src/job/query_optimization_recommendation.rs | 7 +- src/service/promql/value.rs | 2 +- src/service/search/streaming/utils.rs | 2 +- src/service/search/utils.rs | 14 ++-- 15 files changed, 175 insertions(+), 77 deletions(-) diff --git a/src/common/meta/user.rs b/src/common/meta/user.rs index 89de47805a..23156389d8 100644 --- a/src/common/meta/user.rs +++ b/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; diff --git a/src/config/src/meta/stream.rs b/src/config/src/meta/stream.rs index c01fcdf9fa..f2e92a4772 100644 --- a/src/config/src/meta/stream.rs +++ b/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, diff --git a/src/config/src/utils/base64.rs b/src/config/src/utils/base64.rs index c113bb85b6..b6d94bfdbb 100644 --- a/src/config/src/utils/base64.rs +++ b/src/config/src/utils/base64.rs @@ -17,7 +17,7 @@ use std::io::{Error, ErrorKind}; use base64::Engine; -pub fn decode(s: &str) -> Result { +pub fn decode(s: impl AsRef<[u8]>) -> Result { 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 { } } -pub fn decode_raw(s: &str) -> Result, Error> { +pub fn decode_raw(s: impl AsRef<[u8]>) -> Result, 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}"))) } diff --git a/src/handler/http/auth/token.rs b/src/handler/http/auth/token.rs index c1ffd261ee..8cbb50d608 100644 --- a/src/handler/http/auth/token.rs +++ b/src/handler/http/auth/token.rs @@ -50,12 +50,18 @@ pub async fn token_validator( }; let path_columns = path.split('/').collect::>(); + // 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) diff --git a/src/handler/http/auth/validator.rs b/src/handler/http/auth/validator.rs index 739260b1e7..2780033f8c 100644 --- a/src/handler/http/auth/validator.rs +++ b/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 { - 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::(&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 { } } -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 diff --git a/src/handler/http/request/mcp/mod.rs b/src/handler/http/request/mcp/mod.rs index 528a18cf41..c6bdd32951 100644 --- a/src/handler/http/request/mcp/mod.rs +++ b/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 { + 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, + // 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 { @@ -176,9 +189,7 @@ pub async fn handle_mcp_post( _org_id: web::Path, _body: web::Bytes, ) -> actix_web::Result { - 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, _body: web::Bytes, ) -> actix_web::Result { - 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, +) -> actix_web::Result { + use once_cell::sync::Lazy; + + static METADATA: Lazy = 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, +) -> actix_web::Result { + mcp_only_in_enterprise() } diff --git a/src/handler/http/request/re_pattern/mod.rs b/src/handler/http/request/re_pattern/mod.rs index 1581003dd3..2b8335e37c 100644 --- a/src/handler/http/request/re_pattern/mod.rs +++ b/src/handler/http/request/re_pattern/mod.rs @@ -436,7 +436,7 @@ pub async fn test(web::Json(req): web::Json) -> Result String { #[inline] fn key_decode(key: &str) -> String { - base64::decode(&key.replace('-', "+").replace('_', "/")).unwrap() + base64::decode(key.replace('-', "+").replace('_', "/")).unwrap() } #[cfg(test)] diff --git a/src/infra/src/table/migration/m20241119_000002_populate_dashboards_table.rs b/src/infra/src/table/migration/m20241119_000002_populate_dashboards_table.rs index 76a43d99b6..6db83db3ad 100644 --- a/src/infra/src/table/migration/m20241119_000002_populate_dashboards_table.rs +++ b/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 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 for dashboards::ActiveModel { // JSON while also migrating its value into the created_at Unix timestamp column. let created_at_tz: DateTime = 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 { diff --git a/src/job/query_optimization_recommendation.rs b/src/job/query_optimization_recommendation.rs index baf7efb4bd..3de682453f 100644 --- a/src/job/query_optimization_recommendation.rs +++ b/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 . -#![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, ) -> Pin> + 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()) }) } } diff --git a/src/service/promql/value.rs b/src/service/promql/value.rs index 04557b4936..bfdeb45cf2 100644 --- a/src/service/promql/value.rs +++ b/src/service/promql/value.rs @@ -313,7 +313,7 @@ impl<'de> Deserialize<'de> for Exemplar { impl From<&json::Map> for Exemplar { fn from(data: &json::Map) -> 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() { diff --git a/src/service/search/streaming/utils.rs b/src/service/search/streaming/utils.rs index e9b9632237..bd20ad6c76 100644 --- a/src/service/search/streaming/utils.rs +++ b/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)); } diff --git a/src/service/search/utils.rs b/src/service/search/utils.rs index 7eeab42877..f9d45806e7 100644 --- a/src/service/search/utils.rs +++ b/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]