openobserve/src/core/src/auth.rs

// Copyright 2026 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 <http://www.gnu.org/licenses/>.

use std::{collections::HashMap, fmt::Debug, num::NonZeroUsize, sync::LazyLock};

use axum::{
    Json,
    extract::FromRequestParts,
    http::{HeaderMap, StatusCode, request::Parts},
    response::{IntoResponse, Response},
};
use base64::Engine;
#[cfg(feature = "enterprise")]
use common::meta::user::AuthTokensExt;
use common::meta::user::{AuthTokens, UserOrgRole};
use config::{
    meta::user::UserRole,
    utils::{hash::try_get_passcode_hash, json},
};
use lru::LruCache;
use parking_lot::Mutex;
use sha2::{Digest, Sha256};
#[cfg(feature = "enterprise")]
use {
    crate::users::get_user, db::user::is_root_user, jsonwebtoken::TokenData,
    o2_dex::service::auth::get_dex_jwks, o2_openfga::meta::mapping::OFGA_MODELS, serde_json::Value,
    std::str::FromStr,
};

pub const V2_API_PREFIX: &str = "v2";
/// Modules routed as `/v2/{org_id}/{module}`; any other path names its org in the first segment.
pub const V2_ORG_MODULES: [&str; 9] = [
    "alerts",
    "folders",
    "reports",
    "synthetics",
    "incidents",
    "downtimes",
    "workflows",
    "alert_calibration",
    "fleets",
];
pub const SESSION_AUTH_MARKER: &str = "Session::";
const PASSWORD_HASH_CACHE_CAPACITY: NonZeroUsize = NonZeroUsize::new(10_000).unwrap();

// Every login attempt inserts, so it must stay bounded, and it must never key by the plaintext.
static PASSWORD_HASH: LazyLock<Mutex<LruCache<String, String>>> =
    LazyLock::new(|| Mutex::new(LruCache::new(PASSWORD_HASH_CACHE_CAPACITY)));

#[cfg(feature = "enterprise")]
pub async fn get_user_email_from_auth_str(auth_str: &str) -> Option<String> {
    if auth_str.starts_with("Basic") {
        let decoded = config::utils::base64::decode(auth_str.strip_prefix("Basic")?.trim()).ok()?;
        get_user_details(decoded).map(|value| value.0)
    } else if auth_str.starts_with("Bearer") {
        common::utils::jwt::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 auth_tokens.has_expired() {
            return None;
        }
        let decoded =
            config::utils::base64::decode(auth_tokens.auth_ext.strip_prefix("auth_ext")?.trim())
                .ok()?;
        get_user_details(decoded).map(|value| value.0)
    } else {
        None
    }
}

pub fn get_user_details(decoded: impl AsRef<[u8]>) -> Option<(String, String)> {
    let credentials = std::str::from_utf8(decoded.as_ref()).ok()?;
    credentials
        .split_once(':')
        .map(|(user, password)| (user.to_string(), password.to_string()))
}

/// The org the router serves for a path below the `/api` mount, which a v2 route carries second.
pub fn org_from_path<'a>(path_columns: &[&'a str]) -> &'a str {
    match *path_columns {
        [V2_API_PREFIX, org, module, ..] if V2_ORG_MODULES.contains(&module) => org,
        [org, ..] => org,
        [] => "",
    }
}

/// The license routes name no org, so a user lookup cannot take one from their path.
pub fn is_license_path_without_org(path_columns: &[&str]) -> bool {
    matches!(path_columns, ["license"] | ["license", "refresh"])
}

/// The path as the router matched it, below the mount prefixes axum has already stripped.
pub fn routed_path(uri_path: &str) -> &str {
    uri_path.strip_prefix('/').unwrap_or(uri_path)
}

/// Resolves the effective permission method for write requests (PUT/DELETE/PATCH).
///
/// Certain endpoints use HTTP methods that don't map 1:1 to their required
/// permission level. This function handles those overrides:
/// - `delete_fields` endpoints: PATCH/PUT → DELETE (needs delete permission)
/// - General PATCH normalization: PATCH → PUT (treat as update)
#[cfg(any(feature = "enterprise", test))]
#[allow(dead_code)]
pub fn resolve_write_method(method: &str, path_columns: &[&str]) -> String {
    let url_len = path_columns.len();
    let mut resolved = method.to_string();

    if url_len > 0 && path_columns[url_len - 1] == "delete_fields" {
        resolved = "DELETE".to_string();
    }

    if resolved == "PATCH" {
        resolved = "PUT".to_string();
    }

    resolved
}

// Email validation and OpenFGA name sanitization live in the shared `config` crate so the OSS
// auth layer, the enterprise domain-management blocklist, and the enterprise OpenFGA route checks
// all use identical logic. Re-exported here to preserve the existing
// `crate::auth::{EMAIL_REGEX, is_valid_email, into_ofga_supported_format, is_ofga_unsupported,
// RE_OFGA_UNSUPPORTED_NAME}` API.
pub use config::utils::str::{
    EMAIL_REGEX, RE_OFGA_UNSUPPORTED_NAME, into_ofga_supported_format, is_ofga_unsupported,
    is_valid_email,
};

#[cfg(feature = "enterprise")]
pub fn is_ofga_object_visible(
    org_id: &str,
    object_type: &str,
    object_id: &str,
    permitted_objects: Option<&[String]>,
) -> bool {
    permitted_objects.is_none_or(|permitted_objects| {
        permitted_objects.contains(&format!("{object_type}:{object_id}"))
            || permitted_objects.contains(&format!("{object_type}:_all_{org_id}"))
    })
}

pub fn get_hash(pass: &str, salt: &str) -> String {
    try_get_hash(pass, salt).expect("salt length outside what argon2 accepts")
}

/// `None` when the salt cannot be hashed with, as for external users stored with an empty salt.
pub fn try_get_hash(pass: &str, salt: &str) -> Option<String> {
    let key = password_hash_cache_key(pass, salt);
    if let Some(hash) = PASSWORD_HASH.lock().get(&key) {
        return Some(hash.clone());
    }
    let password_hash = try_get_passcode_hash(pass, salt)?;
    PASSWORD_HASH.lock().put(key, password_hash.clone());
    Some(password_hash)
}

#[cfg(feature = "enterprise")]
pub async fn save_org_tuples(org_id: &str) {
    use o2_openfga::config::get_config as get_openfga_config;

    if get_openfga_config().enabled {
        o2_openfga::authorizer::authz::save_org_tuples(org_id).await
    }
}

#[cfg(not(feature = "enterprise"))]
pub async fn save_org_tuples(_org_id: &str) {}

#[cfg(feature = "enterprise")]
pub async fn delete_org_tuples(org_id: &str) -> Result<(), anyhow::Error> {
    use o2_openfga::config::get_config as get_openfga_config;

    if get_openfga_config().enabled {
        o2_openfga::authorizer::authz::delete_org_tuples(org_id)
            .await
            .map_err(|e| anyhow::anyhow!("failed to delete org tuples for {org_id}: {e}"))?;
    }
    Ok(())
}

#[cfg(not(feature = "enterprise"))]
pub async fn delete_org_tuples(_org_id: &str) -> Result<(), anyhow::Error> {
    Ok(())
}

#[cfg(feature = "enterprise")]
pub fn get_role(role: &UserOrgRole) -> UserRole {
    use std::str::FromStr;

    let role = o2_openfga::authorizer::roles::get_role(format!("{}", role.base_role));
    UserRole::from_str(&role).unwrap()
}

#[cfg(not(feature = "enterprise"))]
pub fn get_role(_role: &UserOrgRole) -> UserRole {
    UserRole::Admin
}

fn deserialize_trimmed<'de, D>(deserializer: D) -> Result<String, D::Error>
where
    D: serde::Deserializer<'de>,
{
    let s: String = serde::Deserialize::deserialize(deserializer)?;
    Ok(s.trim().to_string())
}

#[derive(Debug, serde::Deserialize)]
pub struct UserEmail {
    #[serde(deserialize_with = "deserialize_trimmed")]
    pub user_id: String,
}

/// A per-object GET check, taken as a trait so tests can stand in for OpenFGA.
#[cfg(feature = "enterprise")]
pub trait ReadCheck {
    fn can_read(&self, resource: &str, object_id: &str) -> impl Future<Output = bool> + Send;
}

/// One user's reads in one org, checked as the `GET /{org}/{resource}/{id}` routes check them.
#[cfg(feature = "enterprise")]
pub struct UserReads<'a> {
    pub org_id: &'a str,
    pub user_id: &'a str,
}

#[cfg(feature = "enterprise")]
impl ReadCheck for UserReads<'_> {
    async fn can_read(&self, resource: &str, object_id: &str) -> bool {
        check_permissions(
            object_id,
            self.org_id,
            self.user_id,
            resource,
            "GET",
            None,
            false,
            false,
            true,
        )
        .await
    }
}

#[derive(Debug, PartialEq, Eq, Clone)]
pub struct AuthExtractor {
    pub auth: String,
    pub method: String,
    pub o2_type: String,
    pub org_id: String,
    pub bypass_check: bool,
    pub parent_id: String,
    pub use_all_org: bool,
    pub use_self_context: bool,
    pub use_self_parent: bool,
}

impl AuthExtractor {
    /// Create an AuthExtractor that bypasses the FGA check.
    /// Used for routes where auth is handler-managed or not needed.
    pub fn bypass(auth: String, parent_id: String) -> Self {
        Self {
            auth,
            method: String::new(),
            o2_type: String::new(),
            org_id: String::new(),
            bypass_check: true,
            parent_id,
            use_all_org: false,
            use_self_context: false,
            use_self_parent: false,
        }
    }
}

/// Rejection type for AuthExtractor
pub struct AuthExtractorRejection {
    message: String,
    status_code: StatusCode,
}

impl AuthExtractorRejection {
    fn unauthorized(message: impl Into<String>) -> Self {
        Self {
            message: message.into(),
            status_code: StatusCode::UNAUTHORIZED,
        }
    }
}

impl IntoResponse for AuthExtractorRejection {
    fn into_response(self) -> Response {
        (
            self.status_code,
            Json(serde_json::json!({
                "code": self.status_code.as_u16(),
                "message": self.message
            })),
        )
            .into_response()
    }
}

impl<S> FromRequestParts<S> for AuthExtractor
where
    S: Send + Sync,
{
    type Rejection = AuthExtractorRejection;

    #[cfg(feature = "enterprise")]
    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
        use common::utils::http::{get_folder, get_stream_type_from_request};
        use config::meta::stream::StreamType;
        use hashbrown::HashMap;
        use o2_openfga::meta::route_permissions::{self as rp, StreamType as RpStreamType};

        let start = std::time::Instant::now();

        // Parse query string
        let query_string = parts.uri.query().unwrap_or("");
        let query: HashMap<String, String> = url::form_urlencoded::parse(query_string.as_bytes())
            .into_owned()
            .collect();

        let stream_type = get_stream_type_from_request(&query);
        let folder = get_folder(&query);

        let method = parts.method.to_string();
        let local_path = parts.uri.path().to_string();
        let path = routed_path(&local_path);

        let mut path_columns = path.split('/').collect::<Vec<&str>>();
        // Drop a trailing empty segment produced by a trailing slash (e.g.
        // "default/" -> ["default", ""]). Without this, the segment count is off
        // by one and exact-length route matching in `resolve_permission` fails —
        // e.g. the ES-compat org-root route `["{org}"]` never matches the
        // slash-mandatory `/{org_id}/` handler. Mirrors `validate_credentials`.
        if path_columns.len() > 1
            && let Some(v) = path_columns.last()
            && v.is_empty()
        {
            path_columns.pop();
        }
        let org_id = org_from_path(&path_columns).to_string();

        // Synthetics probe job + agent APIs — `/{org_id}/synthetics/{jobs,agent}/*`.
        // Authenticated via the o2syn_ probe token (not a user), so OpenFGA is
        // bypassed; the handler asserts the token's org == the path org.
        if method.eq("POST")
            && path_columns.get(1) == Some(&"synthetics")
            && matches!(path_columns.get(2), Some(&"jobs") | Some(&"agent"))
        {
            if let Some(auth_header) = parts.headers.get("Authorization")
                && let Ok(auth_str) = auth_header.to_str()
            {
                return Ok(AuthExtractor {
                    auth: reject_forged_session_marker(auth_str.to_owned()),
                    method,
                    o2_type: String::new(),
                    org_id: org_id.clone(),
                    bypass_check: true,
                    parent_id: folder,
                    use_all_org: false,
                    use_self_context: false,
                    use_self_parent: false,
                });
            }
            return Err(AuthExtractorRejection::unauthorized("Unauthorized Access"));
        }

        if common::meta::ingestion_routes::is_opamp(path) {
            return opamp_extractor(&parts.headers, &method, &org_id, &folder);
        }

        // This is case for ingestion endpoints where we need to check
        // permissions on the stream. Classified against the authoritative
        // ingestion-route table (method + exact path shape) rather than
        // substring-matching an ingestion word, so only real writes take the
        // stream-level bypass.
        if common::meta::ingestion_routes::is_ingestion_write(&parts.method, path) {
            if let Some(auth_header) = parts.headers.get("Authorization")
                && let Ok(auth_str) = auth_header.to_str()
            {
                return Ok(AuthExtractor {
                    auth: reject_forged_session_marker(auth_str.to_owned()),
                    method,
                    o2_type: format!("stream:{org_id}"),
                    org_id,
                    bypass_check: true,
                    parent_id: folder,
                    use_all_org: false,
                    use_self_context: false,
                    use_self_parent: false,
                });
            }
            return Err(AuthExtractorRejection::unauthorized("Unauthorized Access"));
        }

        // Resolve permission via the declarative route table.
        // Falls back to the legacy if-else for routes not yet in the table.
        let resolved = match rp::resolve_permission(
            &path_columns,
            &method,
            &org_id,
            &folder,
            stream_type.map(|st| match st {
                StreamType::Logs => RpStreamType::Logs,
                StreamType::Metrics => RpStreamType::Metrics,
                StreamType::Traces => RpStreamType::Traces,
                StreamType::Profiles => RpStreamType::Profiles,
                StreamType::EnrichmentTables => RpStreamType::EnrichmentTables,
                _ => RpStreamType::Logs,
            }),
        ) {
            Some(r) => r,
            None => {
                // Route not yet in ROUTE_PERMISSIONS table.
                // Root users bypass in check_permissions; non-root users are denied
                // via is_allowed with empty method/o2_type/org_id.
                log::warn!(
                    "route missing from ROUTE_PERMISSIONS: {method} {path} — non-root users will be denied"
                );
                let auth_str = extract_auth_str_from_headers(&parts.headers).await;
                if auth_str.is_empty() {
                    return Err(AuthExtractorRejection::unauthorized("Unauthorized Access"));
                }
                return Ok(AuthExtractor {
                    auth: auth_str,
                    method: String::new(),
                    o2_type: String::new(),
                    org_id: String::new(),
                    bypass_check: false,
                    parent_id: String::new(),
                    use_all_org: false,
                    use_self_context: false,
                    use_self_parent: false,
                });
            }
        };

        let auth_str = extract_auth_str_from_headers(&parts.headers).await;
        if auth_str.is_empty() {
            return Err(AuthExtractorRejection::unauthorized("Unauthorized Access"));
        }

        if resolved.bypass_check {
            return Ok(AuthExtractor {
                auth: auth_str,
                method: resolved.method,
                o2_type: String::new(),
                org_id: String::new(),
                bypass_check: true,
                parent_id: resolved.parent_id,
                use_all_org: false,
                use_self_context: false,
                use_self_parent: false,
            });
        }

        log::debug!(
            "AuthExtractor::from_request took {} ms",
            start.elapsed().as_millis()
        );

        Ok(AuthExtractor {
            auth: auth_str,
            method: resolved.method,
            o2_type: resolved.o2_type,
            org_id: resolved.org_id,
            bypass_check: false,
            parent_id: resolved.parent_id,
            use_all_org: resolved.use_all_org,
            use_self_context: resolved.use_self_context,
            use_self_parent: resolved.use_self_parent,
        })
    }

    #[cfg(not(feature = "enterprise"))]
    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
        let cookies = extract_cookies_from_headers(&parts.headers);
        let auth_str = if let Some(cookie) = cookies.get("auth_tokens") {
            let val = config::utils::base64::decode_raw(cookie).unwrap_or_default();
            let auth_tokens: AuthTokens =
                json::from_str(std::str::from_utf8(&val).unwrap_or_default()).unwrap_or_default();
            let access_token = auth_tokens.access_token;
            if access_token.starts_with("Basic") || access_token.starts_with("Bearer") {
                access_token
            } else {
                format!("Bearer {access_token}")
            }
        } else if let Some(auth_header) = parts.headers.get("Authorization") {
            if let Ok(auth_str) = auth_header.to_str() {
                reject_forged_session_marker(auth_str.to_owned())
            } else {
                "".to_string()
            }
        } else {
            "".to_string()
        };

        // if let Some(auth_header) = parts.headers.get("Authorization") {
        if !auth_str.is_empty() {
            return Ok(AuthExtractor {
                auth: auth_str.to_owned(),
                method: "".to_string(),
                o2_type: "".to_string(),
                org_id: "".to_string(),
                bypass_check: true, // bypass check permissions
                parent_id: "".to_string(),
                use_all_org: false,
                use_self_context: false,
                use_self_parent: false,
            });
        }

        Err(AuthExtractorRejection::unauthorized("Unauthorized Access"))
    }
}

// The separator keeps `("ab", "c")` and `("a", "bc")` from sharing an entry.
fn password_hash_cache_key(pass: &str, salt: &str) -> String {
    let mut hasher = Sha256::new();
    hasher.update(salt.as_bytes());
    hasher.update([0u8]);
    hasher.update(pass.as_bytes());
    format!("{:x}", hasher.finalize())
}

// Only server-side session resolution may emit the marker; from a client it is a forged trust
// claim.
fn reject_forged_session_marker(auth: String) -> String {
    if auth.starts_with(SESSION_AUTH_MARKER) {
        log::warn!("rejected client-supplied session trust marker in credentials");
        String::new()
    } else {
        auth
    }
}

/// §4.2: a fleet token in the header only; a cookie would let any page the user visits drive OpAMP.
#[cfg(feature = "enterprise")]
fn opamp_extractor(
    headers: &HeaderMap,
    method: &str,
    org_id: &str,
    folder: &str,
) -> Result<AuthExtractor, AuthExtractorRejection> {
    let Some(auth) = headers
        .get("Authorization")
        .and_then(|v| v.to_str().ok())
        .map(|v| reject_forged_session_marker(v.to_owned()))
        .filter(|v| {
            config::utils::base64::basic_password(v)
                .is_some_and(|p| p.starts_with(config::meta::fleet::FLEET_TOKEN_PREFIX))
        })
    else {
        return Err(AuthExtractorRejection::unauthorized("Unauthorized Access"));
    };
    // The token's fixed scope is its authorization; the validator confines it to its org's writes.
    Ok(AuthExtractor {
        auth,
        method: method.to_string(),
        o2_type: String::new(),
        org_id: org_id.to_string(),
        bypass_check: true,
        parent_id: folder.to_string(),
        use_all_org: false,
        use_self_context: false,
        use_self_parent: false,
    })
}

fn extract_cookies_from_headers(headers: &HeaderMap) -> HashMap<&str, &str> {
    // Check cookies
    let cookies = headers
        .get(http::header::COOKIE)
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");

    // Parse cookies
    let cookie_map: HashMap<&str, &str> = cookies
        .split(';')
        .filter_map(|c| {
            let mut parts = c.trim().splitn(2, '=');
            Some((parts.next()?, parts.next()?))
        })
        .collect();

    cookie_map
}

#[cfg(feature = "enterprise")]
pub async fn extract_auth_str_from_headers(headers: &HeaderMap) -> String {
    let cookies = extract_cookies_from_headers(headers);
    if let Some(cookie) = cookies.get("auth_tokens") {
        let val = config::utils::base64::decode_raw(cookie).unwrap_or_default();
        let auth_tokens: AuthTokens =
            json::from_str(std::str::from_utf8(&val).unwrap_or_default()).unwrap_or_default();
        let access_token = auth_tokens.access_token;
        if access_token.is_empty() {
            // If cookie was set but access token is still empty
            // we check auth_ext cookie to get the token.
            cookies
                .get("auth_ext")
                .map(|cookie| {
                    let val = config::utils::base64::decode_raw(cookie).unwrap_or_default();
                    reject_forged_session_marker(
                        std::str::from_utf8(&val).unwrap_or_default().to_string(),
                    )
                })
                .unwrap_or_default()
        } else if access_token.starts_with("Basic") || access_token.starts_with("Bearer") {
            access_token
        } else if let Some(session_key) = access_token.strip_prefix("session ") {
            let session_key = session_key.to_string();
            match crate::db::session::get(&session_key).await {
                Ok(token) => {
                    log::debug!("Session '{}' resolved to token", session_key);
                    // Check if token already has auth prefix
                    if token.starts_with("Basic ") || token.starts_with("Bearer ") {
                        // Add session marker prefix to bypass allow_static_token check
                        // Format: "Session::<session_id>::<actual_token>"
                        format!("Session::{}::{}", session_key, token)
                    } else {
                        // Plain JWT token from Dex/OAuth, needs Bearer prefix
                        format!("Bearer {}", token)
                    }
                }
                Err(e) => {
                    log::error!("Failed to resolve session '{}': {}", session_key, e);
                    access_token
                }
            }
        } else {
            format!("Bearer {access_token}")
        }
    } else if let Some(cookie) = cookies.get("auth_ext") {
        let val = config::utils::base64::decode_raw(cookie).unwrap_or_default();
        reject_forged_session_marker(std::str::from_utf8(&val).unwrap_or_default().to_string())
    } else if let Some(auth_header) = headers.get("Authorization") {
        if let Ok(auth_str) = auth_header.to_str() {
            // Log auth type without exposing sensitive tokens
            let auth_type = if auth_str.starts_with("Basic ") {
                "Basic"
            } else if auth_str.starts_with("Bearer ") {
                "Bearer"
            } else if auth_str.starts_with("session ") {
                "session"
            } else {
                "Other"
            };
            log::debug!(
                "Authorization header (extract_auth_str_from_parts): type='{}', len={}",
                auth_type,
                auth_str.len()
            );
            // Handle session tokens from Authorization header
            if auth_str.starts_with("session ") {
                let session_key = auth_str.strip_prefix("session ").unwrap().to_string();
                match crate::db::session::get(&session_key).await {
                    Ok(token) => {
                        log::debug!("Session '{}' resolved to token from header", session_key);
                        // Check if token already has auth prefix
                        if token.starts_with("Basic ") || token.starts_with("Bearer ") {
                            // Add session marker prefix to bypass allow_static_token check
                            // Format: "Session::<session_id>::<actual_token>"
                            format!("Session::{}::{}", session_key, token)
                        } else {
                            // Plain JWT token from Dex/OAuth, needs Bearer prefix
                            format!("Bearer {}", token)
                        }
                    }
                    Err(e) => {
                        log::error!(
                            "Failed to resolve session '{}' from header: {}",
                            session_key,
                            e
                        );
                        auth_str.to_owned()
                    }
                }
            } else {
                reject_forged_session_marker(auth_str.to_owned())
            }
        } else {
            "".to_string()
        }
    } else {
        "".to_string()
    }
}

/// Constructs the login URL with the provided parameters.
///
/// # Arguments
///
/// * `base_url` - The base URL of the authentication service.
/// * `time` - The request time.
/// * `exp_in` - The expiration time.
/// * `auth` - The authentication token.
///
/// # Returns
///
/// The constructed login URL.
pub fn generate_presigned_url(
    username: &str,
    password: &str,
    salt: &str,
    base_url: &str,
    exp_in: i64,
    time: i64,
) -> String {
    // let time = chrono::Utc::now().timestamp();
    let stage1 = get_hash(password, salt);
    let stage2 = get_hash(&format!("{}{}", stage1, time), salt);
    let stage3 = get_hash(&format!("{}{}", stage2, exp_in), salt);

    let user_pass = format!("{username}:{stage3}");
    let auth = base64::engine::general_purpose::STANDARD.encode(user_pass);

    format!("{base_url}/auth/login?request_time={time}&exp_in={exp_in}&auth={auth}")
}

#[cfg(not(feature = "enterprise"))]
#[allow(clippy::too_many_arguments)]
pub async fn check_permissions(
    _object_id: &str,
    _org_id: &str,
    _user_id: &str,
    _object_type: &str,
    _method: &str,
    _parent_id: Option<&str>,
    _use_all_org: bool,
    _use_self_context: bool,
    _use_self_parent: bool,
) -> bool {
    false
}

/// Returns false if Auth fails
#[cfg(feature = "enterprise")]
#[allow(clippy::too_many_arguments)]
pub async fn check_permissions(
    object_id: &str,
    org_id: &str,
    user_id: &str,
    object_type: &str,
    method: &str,
    parent_id: Option<&str>,
    use_all_org: bool,
    use_self_context: bool,
    use_self_parent: bool,
) -> bool {
    if !is_root_user(user_id) {
        let user: config::meta::user::User = match get_user(Some(org_id), user_id).await {
            Some(user) => user.clone(),
            None => return false,
        }
        .clone();

        // user_id here is generally safe because it comes from the `user_id` request header,
        // which the auth middleware sets to the DB-resolved email. However, we use user.email
        // directly to avoid any inconsistency between the input identifier and the canonical
        // DB email (e.g. casing differences or aliased identifiers).
        return crate::authz::check_permissions(
            &user.email,
            AuthExtractor {
                auth: "".to_string(),
                method: method.to_string(),
                o2_type: format!(
                    "{}:{}",
                    OFGA_MODELS
                        .get(object_type)
                        .map_or(object_type, |model| model.key),
                    object_id
                ),
                org_id: org_id.to_string(),
                bypass_check: false,
                parent_id: parent_id.unwrap_or("").to_string(),
                use_all_org,
                use_self_context,
                use_self_parent,
            },
            user.role,
            user.is_external,
        )
        .await;
    }
    true
}

/// The first object `reads` refuses, asking once per distinct trimmed id.
#[cfg(feature = "enterprise")]
pub async fn first_unreadable<'a>(
    objects: &[(&'static str, &'a str)],
    reads: &impl ReadCheck,
) -> Option<(&'static str, &'a str)> {
    let mut asked = std::collections::HashSet::new();
    for &(resource, id) in objects {
        let id = id.trim();
        if asked.insert((resource, id)) && !reads.can_read(resource, id).await {
            return Some((resource, id));
        }
    }
    None
}

// OSS has no folder boundary to cross, so this allows where the `check_permissions`
// stub above denies: there is no folder ACL to consult, not a failed lookup.
#[cfg(not(feature = "enterprise"))]
pub async fn check_folder_write_permissions(
    _org_id: &str,
    _user_id: &str,
    _folder_type: &str,
    _folder_id: &str,
) -> bool {
    true
}

/// A source-folder check proves the caller may take the object out, never that
/// they may put it into a destination folder they cannot otherwise reach.
///
/// Checks POST, the verb the create routes require on a folder: putting an
/// object into a folder is authorized the same way whether it arrives by
/// create, move or clone.
#[cfg(feature = "enterprise")]
pub async fn check_folder_write_permissions(
    org_id: &str,
    user_id: &str,
    folder_type: &str,
    folder_id: &str,
) -> bool {
    check_permissions(
        folder_id,
        org_id,
        user_id,
        folder_type,
        "POST",
        None,
        false,
        false,
        true,
    )
    .await
}

#[cfg(feature = "enterprise")]
pub async fn extract_auth_expiry_and_user_id(
    parts: &Parts,
) -> (Option<chrono::DateTime<chrono::Utc>>, Option<String>) {
    let decode = async |token: &str| match decode_expiry(token).await {
        Ok(token_data) => token_data
            .claims
            .get("exp")
            .and_then(|exp| exp.as_i64())
            .and_then(|exp_ts| chrono::DateTime::from_timestamp(exp_ts, 0)),
        Err(e) => {
            log::error!("Error verifying token: {e}");
            None
        }
    };

    let auth_str = extract_auth_str_from_headers(&parts.headers).await;
    if auth_str.is_empty() {
        return (None, None);
    } else if auth_str.starts_with("Basic") {
        let user_id = get_user_email_from_auth_str(&auth_str).await;
        return (None, user_id);
    } else if auth_str.starts_with("Bearer") {
        let user_id = get_user_email_from_auth_str(&auth_str).await;
        let stripped_bearer_token = auth_str.strip_prefix("Bearer ").unwrap();
        let exp = decode(stripped_bearer_token).await;
        return (exp, user_id);
    } else if auth_str.starts_with("session ") {
        let session_key = auth_str.strip_prefix("session ").unwrap();
        let stripped_bearer_token = match crate::db::session::get(session_key).await {
            Ok(bearer_token) => bearer_token,
            Err(e) => {
                log::error!("Error getting session: {e}");
                return (None, None);
            }
        };
        let exp = decode(&stripped_bearer_token).await;
        let bearer_full_token = format!("Bearer {stripped_bearer_token}");
        let user_id = get_user_email_from_auth_str(&bearer_full_token).await;
        return (exp, user_id);
    }
    (None, None)
}

#[cfg(feature = "enterprise")]
async fn decode_expiry(token: &str) -> Result<TokenData<HashMap<String, Value>>, anyhow::Error> {
    use infra::errors::JwtError;
    use jsonwebtoken::{
        Algorithm, DecodingKey, Validation, decode, decode_header,
        jwk::{self, AlgorithmParameters},
    };

    let header = decode_header(token)?;
    let kid = match header.kid {
        Some(k) => k,
        None => return Err(JwtError::MissingAttribute("`kid` header".to_owned()).into()),
    };
    let dex_jwks = get_dex_jwks().await;
    let jwks: jwk::JwkSet = serde_json::from_str(&dex_jwks).unwrap();

    if let Some(j) = jwks.find(&kid) {
        match &j.algorithm {
            AlgorithmParameters::RSA(rsa) => {
                let decoding_key = DecodingKey::from_rsa_components(&rsa.n, &rsa.e).unwrap();

                let mut validation = Validation::new(
                    Algorithm::from_str(j.common.key_algorithm.unwrap().to_string().as_str())
                        .unwrap(),
                );
                validation.validate_exp = true;
                let aud = &o2_dex::config::get_config().client_id;
                validation.set_audience(&[aud]);
                Ok(decode::<HashMap<String, serde_json::Value>>(
                    token,
                    &decoding_key,
                    &validation,
                )?)
            }
            _ => Err(JwtError::ValidationFailed().into()),
        }
    } else {
        Err(JwtError::KeyNotExists().into())
    }
}

pub fn build_basic_auth_header(email: &str, token: &str) -> String {
    let credentials = format!("{email}:{token}");
    let encoded = base64::engine::general_purpose::STANDARD.encode(credentials.as_bytes());
    format!("Basic {encoded}")
}

#[cfg(test)]
mod tests {
    use config::{DEFAULT_ORG, utils::hash::get_passcode_hash};
    use db::user::is_root_user;
    use infra::{db as infra_db, table as infra_table};

    use super::*;
    use crate::{common::meta::user::UserRequest, organization, users};

    #[test]
    fn test_valid_emails() {
        assert!(is_valid_email("user@example.com"));
        assert!(is_valid_email("john.doe+123@mail.co.in"));
        assert!(is_valid_email("a_b-c.d+e@domain.org"));
        // Regression for #12961: local part ending in a single-char dotted segment.
        assert!(is_valid_email("first.x@example.com"));
        assert!(is_valid_email("s.a@example.com"));
        assert!(is_valid_email("first.x.y@example.com"));
        assert!(is_valid_email("first.xy@example.com"));
        // TLDs longer than 6 letters, incl. the RFC 2606 reserved `.invalid`
        // used across the API test suite.
        assert!(is_valid_email("sa_88dddbd4@test.invalid"));
        assert!(is_valid_email("user@example.software"));
        assert!(!is_valid_email("no-at-symbol.com"));
        assert!(!is_valid_email("@missing-user.com"));
        assert!(!is_valid_email("user@.com"));
        assert!(!is_valid_email("user@com"));
        assert!(!is_valid_email("user@domain..com"));
    }

    #[test]
    fn test_generate_presigned_url() {
        let password = "password";
        let salt = "saltsalt";
        let username = "user";
        let base_url = "https://example.com";
        let exp_in = 3600;
        let time = 1634567890;

        let expected_url = format!(
            "{}/auth/login?request_time={}&exp_in={}&auth={}",
            base_url,
            time,
            exp_in,
            "dXNlcjokYXJnb24yZCR2PTE2JG09MjA0OCx0PTQscD0yJGMyRnNkSE5oYkhRJGNwTElHZzdEaFl1Vi9nSWxMaCtRZksrS29Vd2ZFaGVpdHkwc3Z0c243Y1E="
        );

        let generated_url =
            generate_presigned_url(username, password, salt, base_url, exp_in, time);

        assert_eq!(generated_url, expected_url);
    }

    #[tokio::test]
    async fn test_is_root_user() {
        assert!(!is_root_user("dummy"));
    }

    #[tokio::test]
    #[ignore]
    async fn test_is_root_user2() {
        infra_db::create_table().await.unwrap();
        infra_table::create_user_tables().await.unwrap();
        organization::check_and_create_org_without_ofga(DEFAULT_ORG)
            .await
            .unwrap();
        let _ = users::create_root_user_if_not_exists(
            DEFAULT_ORG,
            UserRequest {
                email: "root@example.com".to_string(),
                password: "Complexpass#123".to_string(),
                role: UserOrgRole {
                    base_role: config::meta::user::UserRole::Root,
                    custom_role: None,
                },
                first_name: "root".to_owned(),
                last_name: "".to_owned(),
                is_external: false,
                token: None,
            },
        )
        .await;
        db::user::cache().await.unwrap();
        db::organization::cache().await.unwrap();
        db::org_users::cache().await.unwrap();
        assert!(is_root_user("root@example.com"));
        assert!(!is_root_user("root2@example.com"));
    }

    #[tokio::test]
    async fn test_get_hash() {
        let hash =
            "$argon2d$v=16$m=2048,t=4,p=2$VGVzdFNhbHQ$CZzrFPtqjY4mIPYwoDztCJ3OGD5M0P37GH4QddwrbZk";
        assert_eq!(get_hash("Pass#123", "TestSalt"), hash);
    }

    #[tokio::test]
    async fn test_get_hash_for_pass() {
        let pass1 = get_hash("Pass#123", "openobserve");
        let time = chrono::Utc::now().timestamp();
        let pass2 = get_hash(&format!("{}{}", pass1, time), "openobserve");
        let exp_in = 600;
        let pass3 = get_hash(&format!("{}{}", pass2, exp_in), "openobserve");
        println!("time: {time}");
        println!("pass3: {pass3}");

        let user_pass = format!("{}:{}", "b@b.com", pass3);
        let auth = base64::engine::general_purpose::STANDARD.encode(user_pass);
        println!(
            "http://localhost:5080/auth/login?request_time={time}&exp_in={exp_in}&auth={auth}"
        );
    }

    #[test]
    fn password_hash_cache_never_holds_the_plaintext() {
        let (pass, salt) = ("plaintext-marker-pass", "plaintext-marker-salt");
        let hash = get_hash(pass, salt);
        let key = password_hash_cache_key(pass, salt);
        assert!(!key.contains(pass) && !key.contains(salt));
        assert_eq!(PASSWORD_HASH.lock().peek(&key), Some(&hash));
        assert!(
            PASSWORD_HASH
                .lock()
                .iter()
                .all(|(k, _)| !k.contains(pass) && !k.contains(salt))
        );
    }

    #[test]
    fn password_hash_cache_stays_bounded() {
        let cap = PASSWORD_HASH_CACHE_CAPACITY.get();
        let hash = get_passcode_hash("bounded", "bounded-salt");
        for i in 0..cap + 50 {
            PASSWORD_HASH.lock().put(
                password_hash_cache_key(&format!("guess-{i}"), "bounded-salt"),
                hash.clone(),
            );
        }
        get_hash("one-more-guess", "bounded-salt");
        assert!(PASSWORD_HASH.lock().len() <= cap);
    }

    #[test]
    fn unusable_salts_hash_to_none_instead_of_panicking() {
        assert_eq!(try_get_hash("anything", ""), None);
        assert_eq!(try_get_hash("anything", "ab"), None);
        assert_eq!(try_get_hash("anything", &"s".repeat(49)), None);
        assert_eq!(
            try_get_hash("Pass#123", "TestSalt").as_deref(),
            Some(
                "$argon2d$v=16$m=2048,t=4,p=2$VGVzdFNhbHQ$CZzrFPtqjY4mIPYwoDztCJ3OGD5M0P37GH4QddwrbZk"
            )
        );
    }

    #[test]
    fn test_get_hash_caching() {
        let pass = "testpass";
        let salt = "testsalt";

        // First call should create hash and cache it
        let hash1 = get_hash(pass, salt);

        // Second call should return cached value
        let hash2 = get_hash(pass, salt);

        assert_eq!(hash1, hash2);
        assert!(!hash1.is_empty());
    }

    #[test]
    fn test_email_regex_edge_cases() {
        // Valid emails
        assert!(is_valid_email("test@example.com"));
        assert!(is_valid_email("user.name@example.com"));
        assert!(is_valid_email("user+tag@example.com"));
        assert!(is_valid_email("user_name@example.co.uk"));
        assert!(is_valid_email("123@example.com"));

        // Invalid emails
        assert!(!is_valid_email(""));
        assert!(!is_valid_email("no-at-sign"));
        assert!(!is_valid_email("@no-user.com"));
        assert!(!is_valid_email("user@"));
        assert!(!is_valid_email("user..double@example.com"));
        assert!(!is_valid_email("user@no-domain"));
        assert!(!is_valid_email("user@domain.c")); // TLD too short
        assert!(!is_valid_email("user with space@example.com"));
        assert!(!is_valid_email("user@domain with space.com"));
        assert!(!is_valid_email(".user@example.com"));
        assert!(!is_valid_email("user@.example.com"));
        assert!(!is_valid_email("user.@example.com")); // trailing dot in local part
        assert!(!is_valid_email("user@example.com trailing")); // trailing garbage
    }

    #[test]
    fn test_generate_presigned_url_variations() {
        let username = "testuser";
        let password = "testpass";
        let salt = "testsalt";
        let base_url = "https://auth.example.com";
        let exp_in = 7200;
        let time = 1600000000;

        let url = generate_presigned_url(username, password, salt, base_url, exp_in, time);

        assert!(url.starts_with(base_url));
        assert!(url.contains("/auth/login"));
        assert!(url.contains(&format!("request_time={time}")));
        assert!(url.contains(&format!("exp_in={exp_in}")));
        assert!(url.contains("auth="));

        // Test with different parameters
        let url2 = generate_presigned_url(username, password, salt, base_url, exp_in, time + 1);
        assert_ne!(url, url2); // Different time should generate different URL

        let url3 = generate_presigned_url("different", password, salt, base_url, exp_in, time);
        assert_ne!(url, url3); // Different username should generate different URL
    }

    #[tokio::test]
    async fn test_save_org_tuples_non_enterprise() {
        // In non-enterprise mode, this should not panic and return immediately
        save_org_tuples("test_org").await;
        // If we reach here, the function completed successfully
    }

    #[tokio::test]
    async fn test_delete_org_tuples_non_enterprise() {
        // In non-enterprise mode, this should not panic and return immediately
        assert!(delete_org_tuples("test_org").await.is_ok());
        // If we reach here, the function completed successfully
    }

    // Gated to match what it asserts. Without this the test runs under
    // `--features enterprise` too, where `get_role` returns the mapped role
    // rather than collapsing to Admin — so it failed the entire enterprise
    // test build on a claim it never made about enterprise.
    #[cfg(not(feature = "enterprise"))]
    #[test]
    fn test_get_role_non_enterprise() {
        let user_role = UserOrgRole {
            base_role: UserRole::User,
            custom_role: None,
        };

        // In non-enterprise mode, should always return Admin
        assert_eq!(get_role(&user_role), UserRole::Admin);
    }

    #[cfg(not(feature = "enterprise"))]
    #[tokio::test]
    async fn test_check_permissions_non_enterprise() {
        // In non-enterprise mode, should always return false
        let result = check_permissions(
            "test_object",
            "test_org",
            "test_user",
            "dashboard",
            "GET",
            None,
            false,
            false,
            false,
        )
        .await;

        assert!(!result);
    }

    // OSS has no folder boundary to cross, so the destination check must never
    // block a move there — only enterprise+OpenFGA gates folders.
    #[cfg(not(feature = "enterprise"))]
    #[tokio::test]
    async fn test_check_folder_write_permissions_non_enterprise_allows() {
        let result =
            check_folder_write_permissions("test_org", "test_user", "alert_folders", "dst_folder")
                .await;

        assert!(result);
    }

    #[test]
    fn test_regex_compilation() {
        // Smoke-test the re-exported regexes compile and behave. Exhaustive
        // coverage lives in `config::utils::str` where they are defined.
        assert!(RE_OFGA_UNSUPPORTED_NAME.is_match("test:name"));
        assert!(!RE_OFGA_UNSUPPORTED_NAME.is_match("valid_name"));
        assert!(EMAIL_REGEX.is_match("test@example.com"));
        assert!(!EMAIL_REGEX.is_match("invalid-email"));
    }

    #[test]
    fn test_resolve_write_method_alert_crud_put() {
        let path: Vec<&str> = "v2/default/alerts/abc123".split('/').collect();
        assert_eq!(resolve_write_method("PUT", &path), "PUT");
    }

    #[test]
    fn test_resolve_write_method_patch_normalizes_to_put() {
        let path: Vec<&str> = "v2/default/alerts/abc123".split('/').collect();
        assert_eq!(resolve_write_method("PATCH", &path), "PUT");
    }

    #[test]
    fn test_resolve_write_method_delete_fields_uses_delete() {
        let path: Vec<&str> = "default/streams/mystream/delete_fields"
            .split('/')
            .collect();
        assert_eq!(resolve_write_method("PUT", &path), "DELETE");
        assert_eq!(resolve_write_method("PATCH", &path), "DELETE");
    }

    #[test]
    fn test_build_basic_auth_header_format() {
        use base64::Engine as _;
        let header = build_basic_auth_header("user@example.com", "mytoken");
        assert!(header.starts_with("Basic "));
        let encoded = header.strip_prefix("Basic ").unwrap();
        let decoded = String::from_utf8(
            base64::engine::general_purpose::STANDARD
                .decode(encoded)
                .unwrap(),
        )
        .unwrap();
        assert_eq!(decoded, "user@example.com:mytoken");
    }

    #[test]
    fn test_reject_forged_session_marker() {
        assert_eq!(
            reject_forged_session_marker("Session::forged::Basic dXNlcjpwYXNz".to_string()),
            "",
            "a client-supplied session marker must not survive extraction"
        );
        assert_eq!(
            reject_forged_session_marker("Session::".to_string()),
            "",
            "a bare marker must not survive extraction either"
        );
        for untouched in ["Basic dXNlcjpwYXNz", "Bearer jwt", "session abc", ""] {
            assert_eq!(
                reject_forged_session_marker(untouched.to_string()),
                untouched,
                "ordinary credentials must pass through unchanged"
            );
        }
    }

    #[cfg(feature = "enterprise")]
    #[tokio::test]
    async fn auth_ext_with_an_overflowing_request_time_is_expired() {
        let auth_ext = format!(
            "auth_ext {}",
            config::utils::base64::encode("u@example.com:x")
        );
        let auth = format!(
            r#"{{"auth_ext":"{auth_ext}","refresh_token":"","request_time":{},"expires_in":300}}"#,
            i64::MIN
        );
        assert_eq!(get_user_email_from_auth_str(&auth).await, None);
    }

    // Each URI is what the extractor sees once axum has stripped the `/api` mount.
    #[cfg(feature = "enterprise")]
    #[tokio::test]
    async fn the_extractor_reads_the_org_the_route_serves() {
        let mut misread = vec![];
        for (method, uri, served) in [
            ("POST", "/kv/kv/_search", "kv"),
            ("POST", "/api/kv/_search", "api"),
            ("POST", "/v2/kv/_search", "v2"),
            ("GET", "/v2/acme/alerts", "acme"),
        ] {
            let (mut parts, ()) = http::Request::builder()
                .method(method)
                .uri(uri)
                .header(http::header::AUTHORIZATION, "Basic eDp5")
                .body(())
                .unwrap()
                .into_parts();
            let auth = AuthExtractor::from_request_parts(&mut parts, &())
                .await
                .unwrap_or_else(|e| panic!("{method} {uri}: {}", e.message));
            if auth.org_id != served {
                misread.push(format!(
                    "{method} /api{uri}: resolved org {:?}",
                    auth.org_id
                ));
            }
        }
        assert!(misread.is_empty(), "{misread:#?}");
    }

    #[cfg(feature = "enterprise")]
    #[tokio::test]
    async fn session_cookie_without_a_space_does_not_panic() {
        for access_token in ["sessionX", "session"] {
            let tokens = format!(r#"{{"access_token":"{access_token}","refresh_token":""}}"#);
            let cookie = format!("auth_tokens={}", config::utils::base64::encode(&tokens));
            let mut headers = HeaderMap::new();
            headers.insert(http::header::COOKIE, cookie.parse().unwrap());
            assert_eq!(
                extract_auth_str_from_headers(&headers).await,
                format!("Bearer {access_token}")
            );
        }
    }

    #[cfg(feature = "enterprise")]
    fn basic(credentials: &str) -> String {
        format!("Basic {}", config::utils::base64::encode(credentials))
    }

    #[cfg(feature = "enterprise")]
    #[test]
    fn opamp_extractor_takes_fleet_tokens_only() {
        let headers = |name: http::header::HeaderName, value: &str| {
            let mut headers = HeaderMap::new();
            headers.insert(name, value.parse().unwrap());
            headers
        };

        // The validator trims the password, so a padded token is still a fleet token.
        for credentials in [
            "agent:o2ft_x",
            "admin@example.com: o2ft_x",
            "agent:o2ft_x\n",
        ] {
            let fleet_token = headers(http::header::AUTHORIZATION, &basic(credentials));
            for method in ["GET", "POST"] {
                let ext = opamp_extractor(&fleet_token, method, "default", "")
                    .unwrap_or_else(|_| panic!("{method} rejected"));
                assert!(ext.bypass_check, "{method} {credentials:?}");
            }
        }

        for login in [
            basic("user@example.com:pw"),
            "Bearer o2ft_x".to_string(),
            basic("o2ft_x:pw"),
        ] {
            let login = headers(http::header::AUTHORIZATION, &login);
            for method in ["GET", "POST"] {
                let Err(rejection) = opamp_extractor(&login, method, "default", "") else {
                    panic!("only a fleet token may connect, {method}");
                };
                assert_eq!(rejection.status_code, StatusCode::UNAUTHORIZED);
            }
        }

        let cookie = headers(
            http::header::COOKIE,
            "auth_tokens=eyJhY2Nlc3NfdG9rZW4iOiJ4In0=",
        );
        let Err(rejection) = opamp_extractor(&cookie, "GET", "default", "") else {
            panic!("a cookie must never authenticate OpAMP");
        };
        assert_eq!(rejection.status_code, StatusCode::UNAUTHORIZED);
    }
}