// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see .
use std::net::IpAddr;
use actix_web::{
Error,
dev::ServiceRequest,
error::{ErrorForbidden, ErrorUnauthorized},
http::{Method, header},
web,
};
use config::{get_config, utils::base64};
#[cfg(feature = "enterprise")]
use o2_dex::config::get_config as get_dex_config;
#[cfg(feature = "enterprise")]
use o2_openfga::config::get_config as get_openfga_config;
use url::Url;
use crate::{
common::{
infra::cluster,
meta::{
ingestion::INGESTION_EP,
user::{
AuthTokensExt, DBUser, TokenValidationResponse, TokenValidationResponseBuilder,
UserRole,
},
},
utils::{
auth::{AuthExtractor, V2_API_PREFIX, get_hash, is_root_user},
redirect_response::RedirectResponseBuilder,
},
},
service::{db, users},
};
pub const PKCE_STATE_ORG: &str = "o2_pkce_state";
pub const ACCESS_TOKEN: &str = "access_token";
pub const REFRESH_TOKEN: &str = "refresh_token";
pub const ID_TOKEN_HEADER: &str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9";
pub async fn validator(
req: ServiceRequest,
user_id: &str,
password: &str,
auth_info: AuthExtractor,
path_prefix: &str,
) -> Result {
let cfg = get_config();
let path = match req
.request()
.path()
.strip_prefix(format!("{}{}", cfg.common.base_uri, path_prefix).as_str())
{
Some(path) => path,
None => req.request().path(),
};
match if auth_info.auth.starts_with("{\"auth_ext\":") {
let auth_token: AuthTokensExt =
config::utils::json::from_str(&auth_info.auth).unwrap_or_default();
validate_credentials_ext(user_id, password, path, auth_token).await
} else {
validate_credentials(user_id, password.trim(), path).await
} {
Ok(res) => {
if res.is_valid {
// / Hack for prometheus, need support POST and check the header
let mut req = req;
if req.method().eq(&Method::POST) && !req.headers().contains_key("content-type") {
req.headers_mut().insert(
header::CONTENT_TYPE,
header::HeaderValue::from_static("application/x-www-form-urlencoded"),
);
}
req.headers_mut().insert(
header::HeaderName::from_static("user_id"),
header::HeaderValue::from_str(&res.user_email).unwrap(),
);
if auth_info.bypass_check
|| check_permissions(
user_id,
auth_info,
res.user_role.unwrap_or_default(),
!res.is_internal_user,
)
.await
{
Ok(req)
} else {
Err((ErrorForbidden("Unauthorized Access"), req))
}
} else {
Err((ErrorUnauthorized("Unauthorized Access"), req))
}
}
Err(err) => Err((err, req)),
}
}
/// `validate_token` validates the endpoints which are token only.
/// This includes endpoints like `rum` etc.
///
/// ### Args:
/// - token: The token to validate
///
pub async fn validate_token(token: &str, org_id: &str) -> Result<(), Error> {
match users::get_user_by_token(org_id, token).await {
Some(_user) => Ok(()),
None => Err(ErrorForbidden("User associated with this token not found")),
}
}
pub async fn validate_credentials(
user_id: &str,
user_password: &str,
path: &str,
) -> Result {
let user;
let mut path_columns = path.split('/').collect::>();
if let Some(v) = path_columns.last() {
if v.is_empty() {
path_columns.pop();
}
}
// this is only applicable for super admin user
if is_root_user(user_id) {
user = users::get_user(None, user_id).await;
if user.is_none() {
return Ok(TokenValidationResponse {
is_valid: false,
user_email: "".to_string(),
is_internal_user: false,
user_role: None,
user_name: "".to_string(),
family_name: "".to_string(),
given_name: "".to_string(),
});
}
} else if path_columns.last().unwrap_or(&"").eq(&"organizations") {
let db_user = db::user::get_db_user(user_id).await;
user = match db_user {
Ok(user) => {
let all_users = user.get_all_users();
if all_users.is_empty() {
None
} else {
all_users.first().cloned()
}
}
Err(_) => None,
}
} else {
user = match path.find('/') {
Some(index) => {
let org_id = if path_columns.len() > 1 && path_columns[0].eq(V2_API_PREFIX) {
path_columns[1]
} else {
&path[0..index]
};
users::get_user(Some(org_id), user_id).await
}
None => users::get_user(None, user_id).await,
}
};
if user.is_none() {
return Ok(TokenValidationResponse {
is_valid: false,
user_email: "".to_string(),
is_internal_user: false,
user_role: None,
user_name: "".to_string(),
family_name: "".to_string(),
given_name: "".to_string(),
});
}
let user = user.unwrap();
#[cfg(feature = "enterprise")]
{
if !get_dex_config().native_login_enabled && !user.is_external {
return Ok(TokenValidationResponse {
is_valid: false,
user_email: "".to_string(),
is_internal_user: false,
user_role: None,
user_name: "".to_string(),
family_name: "".to_string(),
given_name: "".to_string(),
});
}
if get_dex_config().root_only_login && !is_root_user(user_id) {
return Ok(TokenValidationResponse {
is_valid: false,
user_email: "".to_string(),
is_internal_user: false,
user_role: None,
user_name: "".to_string(),
family_name: "".to_string(),
given_name: "".to_string(),
});
}
}
if user.role.eq(&UserRole::ServiceAccount) && user.token.eq(&user_password) {
return Ok(TokenValidationResponse {
is_valid: true,
user_email: user.email,
is_internal_user: !user.is_external,
user_role: Some(user.role),
user_name: user.first_name.to_owned(),
family_name: user.last_name,
given_name: user.first_name,
});
}
if (path_columns.len() == 1 || INGESTION_EP.iter().any(|s| path_columns.contains(s)))
&& user.token.eq(&user_password)
{
return Ok(TokenValidationResponse {
is_valid: true,
user_email: user.email,
is_internal_user: !user.is_external,
user_role: Some(user.role),
user_name: user.first_name.to_owned(),
family_name: user.last_name,
given_name: user.first_name,
});
}
let in_pass = get_hash(user_password, &user.salt);
if !user.password.eq(&in_pass)
&& !user
.password_ext
.unwrap_or("".to_string())
.eq(&user_password)
{
return Ok(TokenValidationResponse {
is_valid: false,
user_email: "".to_string(),
is_internal_user: false,
user_role: None,
user_name: "".to_string(),
family_name: "".to_string(),
given_name: "".to_string(),
});
}
if !path.contains("/user")
|| (path.contains("/user")
&& (user.role.eq(&UserRole::Admin)
|| user.role.eq(&UserRole::Root)
|| user.email.eq(user_id)))
{
Ok(TokenValidationResponse {
is_valid: true,
user_email: user.email,
is_internal_user: !user.is_external,
user_role: Some(user.role),
user_name: user.first_name.to_owned(),
family_name: user.last_name,
given_name: user.first_name,
})
} else {
Err(ErrorForbidden("Not allowed"))
}
}
#[cfg(feature = "enterprise")]
pub async fn validate_credentials_ext(
user_id: &str,
in_password: &str,
path: &str,
auth_token: AuthTokensExt,
) -> Result {
let user;
let config = get_config();
let password_ext_salt = config.auth.ext_auth_salt.as_str();
let mut path_columns = path.split('/').collect::>();
if let Some(v) = path_columns.last() {
if v.is_empty() {
path_columns.pop();
}
}
// this is only applicable for super admin user
if is_root_user(user_id) {
user = users::get_user(None, user_id).await;
if user.is_none() {
return Ok(TokenValidationResponse {
is_valid: false,
user_email: "".to_string(),
is_internal_user: false,
user_role: None,
user_name: "".to_string(),
family_name: "".to_string(),
given_name: "".to_string(),
});
}
} else if path_columns.last().unwrap_or(&"").eq(&"organizations") {
let db_user = db::user::get_db_user(user_id).await;
user = match db_user {
Ok(user) => {
let all_users = user.get_all_users();
if all_users.is_empty() {
None
} else {
all_users.first().cloned()
}
}
Err(_) => None,
}
} else {
user = match path.find('/') {
Some(index) => {
let org_id = if path_columns.len() > 1 && path_columns[0].eq(V2_API_PREFIX) {
path_columns[1]
} else {
&path[0..index]
};
users::get_user(Some(org_id), user_id).await
}
None => users::get_user(None, user_id).await,
}
};
if user.is_none() {
return Ok(TokenValidationResponse::default());
}
let user = user.unwrap();
let hashed_pass = get_hash(
&format!(
"{}{}",
get_hash(
&format!("{}{}", user.password_ext.unwrap(), auth_token.request_time),
password_ext_salt
),
auth_token.expires_in
),
password_ext_salt,
);
if !hashed_pass.eq(&in_password) {
return Ok(TokenValidationResponse::default());
}
if !path.contains("/user")
|| (path.contains("/user")
&& (user.role.eq(&UserRole::Admin)
|| user.role.eq(&UserRole::Root)
|| user.email.eq(user_id)))
{
Ok(TokenValidationResponse {
is_valid: true,
user_email: user.email,
is_internal_user: !user.is_external,
user_role: Some(user.role),
user_name: user.first_name.to_owned(),
family_name: user.last_name,
given_name: user.first_name,
})
} else {
Err(ErrorForbidden("Not allowed"))
}
}
#[cfg(not(feature = "enterprise"))]
pub async fn validate_credentials_ext(
_user_id: &str,
_in_password: &str,
_path: &str,
_auth_token: AuthTokensExt,
) -> Result {
Err(ErrorForbidden("Not allowed"))
}
async fn validate_user_from_db(
db_user: Result,
user_password: &str,
req_time: Option<&String>,
exp_in: i64,
password_ext_salt: &str,
) -> Result {
// let db_user = db::user::get_db_user(user_id).await;
match db_user {
Ok(mut user) => {
let in_pass = get_hash(user_password, &user.salt);
if req_time.is_none() && user.password.eq(&in_pass) {
log::debug!("Validating internal user");
if user.password_ext.is_none() {
let password_ext = get_hash(user_password, password_ext_salt);
user.password_ext = Some(password_ext);
let _ = db::user::set(&user).await;
}
let resp = TokenValidationResponseBuilder::from_db_user(&user).build();
Ok(resp)
} else if user.password_ext.is_some() && req_time.is_some() {
log::debug!("Validating user for query params");
let hashed_pass = get_hash(
&format!(
"{}{}",
get_hash(
&format!(
"{}{}",
user.password_ext.as_ref().unwrap(),
req_time.unwrap()
),
password_ext_salt
),
exp_in
),
password_ext_salt,
);
if hashed_pass.eq(&user_password) {
let resp = TokenValidationResponseBuilder::from_db_user(&user).build();
return Ok(resp);
} else {
Err(ErrorForbidden("Not allowed"))
}
} else {
Err(ErrorForbidden("Not allowed"))
}
}
Err(_) => Err(ErrorForbidden("Not allowed")),
}
}
pub async fn validate_user(
user_id: &str,
user_password: &str,
) -> Result {
let db_user = db::user::get_db_user(user_id).await;
let config = get_config();
validate_user_from_db(db_user, user_password, None, 0, &config.auth.ext_auth_salt).await
}
pub async fn validate_user_for_query_params(
user_id: &str,
user_password: &str,
req_time: Option<&String>,
exp_in: i64,
) -> Result {
let db_user = db::user::get_db_user(user_id).await;
let config = get_config();
validate_user_from_db(
db_user,
user_password,
req_time,
exp_in,
&config.auth.ext_auth_salt,
)
.await
}
pub async fn validator_aws(
req: ServiceRequest,
_thread_id: web::Data,
) -> Result {
let cfg = get_config();
let path = req
.request()
.path()
.strip_prefix(format!("{}/aws/", cfg.common.base_uri).as_str())
.unwrap_or(req.request().path());
match req.headers().get("X-Amz-Firehose-Access-Key") {
Some(val) => match val.to_str() {
Ok(val) => {
let amz_creds = match base64::decode(val) {
Ok(val) => val,
Err(_) => return Err((ErrorUnauthorized("Unauthorized Access"), req)),
};
let creds = amz_creds
.split(':')
.map(|s| s.to_string())
.collect::>();
match validate_credentials(&creds[0], &creds[1], path).await {
Ok(res) => {
if res.is_valid {
let mut req = req;
req.headers_mut().insert(
header::HeaderName::from_static("user_id"),
header::HeaderValue::from_str(&res.user_email).unwrap(),
);
Ok(req)
} else {
Err((ErrorUnauthorized("Unauthorized Access"), req))
}
}
Err(err) => Err((err, req)),
}
}
Err(_) => Err((ErrorUnauthorized("Unauthorized Access"), req)),
},
None => Err((ErrorUnauthorized("Unauthorized Access"), req)),
}
}
pub async fn validator_gcp(
req: ServiceRequest,
_thread_id: web::Data,
) -> Result {
let cfg = get_config();
let path = req
.request()
.path()
.strip_prefix(format!("{}/gcp/", cfg.common.base_uri).as_str())
.unwrap_or(req.request().path());
let query =
web::Query::>::from_query(req.query_string())
.unwrap();
match query.get("API-Key") {
Some(val) => {
let gcp_creds = base64::decode(val).unwrap();
let creds = gcp_creds
.split(':')
.map(|s| s.to_string())
.collect::>();
match validate_credentials(&creds[0], &creds[1], path).await {
Ok(res) => {
if res.is_valid {
let mut req = req;
req.headers_mut().insert(
header::HeaderName::from_static("user_id"),
header::HeaderValue::from_str(&res.user_email).unwrap(),
);
Ok(req)
} else {
Err((ErrorUnauthorized("Unauthorized Access"), req))
}
}
Err(err) => Err((err, req)),
}
}
None => Err((ErrorUnauthorized("Unauthorized Access"), req)),
}
}
pub async fn validator_rum(
req: ServiceRequest,
_thread_id: web::Data,
) -> Result {
let path = req
.request()
.path()
.strip_prefix(format!("{}/rum/v1/", get_config().common.base_uri).as_str())
.unwrap_or(req.request().path());
// After this previous path clean we should get only the
// remaining `org_id/rum` or `org_id/replay` or `org_id/logs`
let org_id_end_point: Vec<&str> = path.split('/').collect();
if org_id_end_point.len() != 2 {
return Err((
ErrorUnauthorized("Unauthorized Access. Please pass a valid org_id."),
req,
));
}
let query =
web::Query::>::from_query(req.query_string())
.unwrap();
let token = query.get("oo-api-key").or_else(|| query.get("o2-api-key"));
match token {
Some(token) => match validate_token(token, org_id_end_point[0]).await {
Ok(_res) => Ok(req),
Err(err) => {
log::error!(
"validate_token: Token not found for org_id: {}",
org_id_end_point[0]
);
Err((err, req))
}
},
None => {
log::error!(
"validate_token: Missing api key for rum endpoint org_id: {}",
org_id_end_point[0]
);
Err((ErrorUnauthorized("Unauthorized Access"), req))
}
}
}
async fn oo_validator_internal(
req: ServiceRequest,
auth_info: AuthExtractor,
path_prefix: &str,
) -> Result {
// Check if the ws request is using internal grpc token
if get_config().websocket.enabled && auth_info.auth.eq(&get_config().grpc.internal_grpc_token) {
return validate_http_internal(req).await;
}
if auth_info.auth.starts_with("Basic") {
let decoded = match base64::decode(auth_info.auth.strip_prefix("Basic").unwrap().trim()) {
Ok(val) => val,
Err(_) => return Err((ErrorUnauthorized("Unauthorized Access"), req)),
};
let (username, password) = match get_user_details(decoded) {
Some(value) => value,
None => return Err((ErrorUnauthorized("Unauthorized Access"), req)),
};
validator(req, &username, &password, auth_info, path_prefix).await
} else if auth_info.auth.starts_with("Bearer") {
super::token::token_validator(req, auth_info).await
} else if auth_info.auth.starts_with("{\"auth_ext\":") {
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 {
Err((ErrorUnauthorized("Unauthorized Access"), req))
} else {
let decoded = match base64::decode(
auth_tokens
.auth_ext
.strip_prefix("auth_ext")
.unwrap()
.trim(),
) {
Ok(val) => val,
Err(_) => return Err((ErrorUnauthorized("Unauthorized Access"), req)),
};
let (username, password) = match get_user_details(decoded) {
Some(value) => value,
None => return Err((ErrorUnauthorized("Unauthorized Access"), req)),
};
validator(req, &username, &password, auth_info, path_prefix).await
}
} else {
Err((ErrorUnauthorized("Unauthorized Access"), req))
}
}
#[cfg(feature = "enterprise")]
pub async fn get_user_email_from_auth_str(auth_str: &str) -> Option {
if auth_str.starts_with("Basic") {
let decoded = match base64::decode(auth_str.strip_prefix("Basic").unwrap().trim()) {
Ok(val) => val,
Err(_) => return None,
};
match get_user_details(decoded) {
Some(value) => Some(value.0),
None => None,
}
} else if auth_str.starts_with("Bearer") {
super::token::get_user_name_from_token(auth_str).await
} else if auth_str.starts_with("{\"auth_ext\":") {
let auth_tokens: AuthTokensExt =
config::utils::json::from_str(auth_str).unwrap_or_default();
if chrono::Utc::now().timestamp() - auth_tokens.request_time > auth_tokens.expires_in {
None
} else {
let decoded = match base64::decode(
auth_tokens
.auth_ext
.strip_prefix("auth_ext")
.unwrap()
.trim(),
) {
Ok(val) => val,
Err(_) => return None,
};
match get_user_details(decoded) {
Some(value) => Some(value.0),
None => None,
}
}
} else {
None
}
}
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,
};
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))
}
/// Validates the authentication information in the incoming request and returns the request if
/// valid, or an error if invalid.
///
/// This function is responsible for validating the authentication information in the incoming
/// request. It supports both Basic and Bearer authentication (in enterprise).
/// Works exclusively on `/api` prefix
///
/// For Basic authentication, it decodes the base64-encoded credentials, splits them into username
/// and password, and calls the `validator` function to validate the credentials.
///
/// For Bearer authentication, it calls the `token_validator` function to validate the token.
///
/// If the authentication is invalid, it returns an `ErrorUnauthorized` error.
pub async fn oo_validator(
req: ServiceRequest,
auth_result: Result,
) -> Result {
let path_prefix = "/api/";
let path = extract_relative_path(req.request().path(), path_prefix);
let path_columns = path.split('/').collect::>();
let is_short_url = is_short_url_path(&path_columns);
let auth_info = match auth_result {
Ok(info) => info,
Err(e) => {
return if is_short_url {
Err(handle_auth_failure_for_redirect(req, &e))
} else {
Err((e, req))
};
}
};
match oo_validator_internal(req, auth_info, path_prefix).await {
Ok(service_req) => Ok(service_req),
Err((err, err_req)) => {
if is_short_url {
Err(handle_auth_failure_for_redirect(err_req, &err))
} else {
Err((err, err_req))
}
}
}
}
/// Validates the authentication information in the request and returns the request if valid, or an
/// error if invalid.
///
/// This function is a proxy for the `oo_validator_internal` function, setting the `path_prefix` to
/// "/proxy/".
///
/// # Arguments
/// * `req` - The `ServiceRequest` to validate.
/// * `auth_info` - The authentication information extracted from the request.
///
/// # Returns
/// * `Result` - The validated request, or an error if the
/// authentication is invalid.
pub async fn validator_proxy_url(
req: ServiceRequest,
auth_info: AuthExtractor,
) -> Result {
let path_prefix = "/proxy/";
oo_validator_internal(req, auth_info, path_prefix).await
}
pub async fn validate_http_internal(
req: ServiceRequest,
) -> Result {
let router_nodes = cluster::get_cached_online_router_nodes()
.await
.unwrap_or_default();
// Get the peer address early and own the string
let peer = req
.connection_info()
.peer_addr()
.unwrap_or("unknown")
.to_string();
let router_node = router_nodes.iter().find(|node| {
let node_url = match Url::parse(&node.http_addr) {
Ok(node_url) => node_url,
Err(e) => {
log::error!("Failed to parse node URL: {}", e);
return false;
}
};
// Get IP from peer address (strips port if present)
let peer_ip = match peer
.split(':')
.next()
.and_then(|addr| addr.parse::().ok())
{
Some(ip) => ip,
None => {
log::debug!("Failed to parse peer IP from: {}", peer);
return false;
}
};
let node_ip = match node_url
.host()
.and_then(|h| h.to_string().parse::().ok())
{
Some(ip) => ip,
None => {
log::debug!("Failed to parse node IP");
return false;
}
};
peer_ip == node_ip
});
if router_node.is_none() {
return Err((ErrorUnauthorized("Unauthorized Access"), req));
}
Ok(req)
}
#[cfg(feature = "enterprise")]
pub(crate) async fn check_permissions(
user_id: &str,
auth_info: AuthExtractor,
role: UserRole,
_is_external: bool,
) -> bool {
if !get_openfga_config().enabled || role.eq(&UserRole::Root) {
return true;
}
let object_str = auth_info.o2_type;
let obj_str = if object_str.contains("##user_id##") {
object_str.replace("##user_id##", user_id)
} else {
object_str
};
let org_id = if auth_info.org_id.eq("organizations") {
user_id
} else {
&auth_info.org_id
};
o2_openfga::authorizer::authz::is_allowed(
org_id,
user_id,
&auth_info.method,
&obj_str,
&auth_info.parent_id,
&role.to_string(),
)
.await
}
#[cfg(not(feature = "enterprise"))]
pub(crate) async fn check_permissions(
_user_id: &str,
_auth_info: AuthExtractor,
_role: UserRole,
_is_external: bool,
) -> bool {
true
}
#[cfg(feature = "enterprise")]
async fn list_objects(
user_id: &str,
permission: &str,
object_type: &str,
org_id: &str,
) -> Result, anyhow::Error> {
o2_openfga::authorizer::authz::list_objects(user_id, permission, object_type, org_id).await
}
#[cfg(feature = "enterprise")]
pub(crate) async fn list_objects_for_user(
org_id: &str,
user_id: &str,
permission: &str,
object_type: &str,
) -> Result