ntex-basicauth 0.4.1

//! Core implementation of basic authentication

use crate::{
    error::{AuthError, AuthResult},
    is_valid_username,
    limiter::{ConcurrencyLimiter, RateLimiter},
};
use base64::{Engine, engine::general_purpose::STANDARD};
use ntex::time::timeout;
use ntex::{Middleware, Service, ServiceCtx, web};
use std::collections::HashMap;
use std::fmt::Debug;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

#[cfg(feature = "timing-safe")]
use subtle::ConstantTimeEq;

#[cfg(feature = "secure-memory")]
use zeroize::{Zeroize, ZeroizeOnDrop};

#[cfg(feature = "cache")]
use {
    crate::cache::{AuthCache, CacheConfig},
    sha2::{Digest, Sha256},
};

/// User credentials
#[derive(Debug, Clone, PartialEq, Eq)]
#[cfg_attr(feature = "secure-memory", derive(Zeroize, ZeroizeOnDrop))]
pub struct Credentials {
    /// Username
    #[cfg_attr(feature = "secure-memory", zeroize(skip))]
    pub username: String,
    /// Password
    pub password: String,
}

impl Credentials {
    /// Create new credentials instance
    pub fn new(username: String, password: String) -> Self {
        Self { username, password }
    }

    /// Generate secure cache key (using SHA256 hash with application-specific salt)
    #[cfg(feature = "cache")]
    pub fn cache_key(&self) -> [u8; 32] {
        let mut hasher = Sha256::new();
        // Add application-specific salt to prevent rainbow table attacks
        hasher.update(b"ntex-basicauth-v1:");
        hasher.update(self.username.as_bytes());
        hasher.update(b":");
        hasher.update(self.password.as_bytes());
        hasher.finalize().into()
    }

    /// Timing-safe password verification.
    ///
    /// Both passwords are reduced to fixed-length SHA-256 digests before the
    /// constant-time comparison, so the comparison always runs over equal-length
    /// buffers and does not leak whether the plaintext lengths match (a raw
    /// `ct_eq` skips comparison on length mismatch, leaking that information).
    #[cfg(feature = "timing-safe")]
    pub fn verify_password(&self, expected: &str) -> bool {
        use sha2::{Digest, Sha256};

        let mut ha = Sha256::new();
        ha.update(self.password.as_bytes());
        let mut hb = Sha256::new();
        hb.update(expected.as_bytes());

        let a = ha.finalize();
        let b = hb.finalize();
        a.as_slice().ct_eq(b.as_slice()).into()
    }

    /// Non-timing-safe password verification (fallback if timing-safe feature is off)
    #[cfg(not(feature = "timing-safe"))]
    pub fn verify_password(&self, expected: &str) -> bool {
        self.password == expected
    }

    /// Get username reference (avoid clone)
    pub fn username_ref(&self) -> &str {
        &self.username
    }

    /// Validate credentials format
    pub fn is_valid_format(&self) -> bool {
        is_valid_username(&self.username) && !self.password.chars().any(|c| c.is_control())
    }
}

/// User validator trait for custom authentication logic
pub trait UserValidator: Send + Sync + Debug {
    /// Validate user credentials
    fn validate<'a>(
        &'a self,
        credentials: &'a Credentials,
    ) -> Pin<Box<dyn Future<Output = AuthResult<bool>> + Send + 'a>>;

    /// Get validator name (for logging)
    fn name(&self) -> &'static str {
        "UserValidator"
    }

    /// Pre-validation check (optional)
    fn pre_validate(&self, credentials: &Credentials) -> AuthResult<()> {
        if !credentials.is_valid_format() {
            return Err(AuthError::InvalidCredentials);
        }
        Ok(())
    }

    /// Get user count
    fn user_count(&self) -> usize {
        0
    }
}

/// Static user list validator
#[derive(Debug)]
pub struct StaticUserValidator {
    users: HashMap<String, String>,
    case_sensitive: bool,
}

impl StaticUserValidator {
    /// Create a new static user validator
    pub fn new() -> Self {
        Self {
            users: HashMap::new(),
            case_sensitive: true,
        }
    }

    /// Set to case insensitive
    pub fn case_insensitive(mut self) -> Self {
        self.case_sensitive = false;
        self
    }

    /// Add a user
    pub fn add_user(&mut self, username: String, password: String) -> &mut Self {
        let key = if self.case_sensitive {
            username
        } else {
            username.to_lowercase()
        };
        self.users.insert(key, password);
        self
    }

    /// Create validator from HashMap
    pub fn from_map(users: HashMap<String, String>) -> Self {
        Self {
            users,
            case_sensitive: true,
        }
    }

    /// Create validator from HashMap (case insensitive)
    pub fn from_map_case_insensitive(users: HashMap<String, String>) -> Self {
        let normalized_users: HashMap<String, String> = users
            .into_iter()
            .map(|(k, v)| (k.to_lowercase(), v))
            .collect();

        Self {
            users: normalized_users,
            case_sensitive: false,
        }
    }

    /// Check if user exists
    pub fn contains_user(&self, username: &str) -> bool {
        let key = if self.case_sensitive {
            username
        } else {
            &username.to_lowercase()
        };
        self.users.contains_key(key)
    }
}

impl Default for StaticUserValidator {
    fn default() -> Self {
        Self::new()
    }
}

impl UserValidator for StaticUserValidator {
    fn validate<'a>(
        &'a self,
        credentials: &'a Credentials,
    ) -> Pin<Box<dyn Future<Output = AuthResult<bool>> + Send + 'a>> {
        Box::pin(async move {
            let username = if self.case_sensitive {
                &credentials.username
            } else {
                &credentials.username.to_lowercase()
            };

            match self.users.get(username) {
                Some(stored_password) => Ok(credentials.verify_password(stored_password)),
                None => Ok(false),
            }
        })
    }

    fn name(&self) -> &'static str {
        "StaticUserValidator"
    }

    /// Get user count
    fn user_count(&self) -> usize {
        self.users.len()
    }
}

/// BCrypt password validator (requires bcrypt feature)
#[cfg(feature = "bcrypt")]
#[derive(Debug)]
pub struct BcryptUserValidator {
    users: HashMap<String, String>, // username -> bcrypt hash
    cost: u32,
}

#[cfg(feature = "bcrypt")]
impl BcryptUserValidator {
    /// Create a new BCrypt user validator
    pub fn new() -> Self {
        Self {
            users: HashMap::new(),
            cost: bcrypt::DEFAULT_COST,
        }
    }

    /// Set BCrypt cost factor
    pub fn with_cost(mut self, cost: u32) -> Self {
        self.cost = cost;
        self
    }

    /// Add a user with precomputed BCrypt hash
    pub fn add_user(&mut self, username: String, bcrypt_hash: String) -> &mut Self {
        self.users.insert(username, bcrypt_hash);
        self
    }

    /// Add a user with password, automatically hashing it with BCrypt
    pub fn add_user_with_password(
        &mut self,
        username: String,
        password: &str,
    ) -> AuthResult<&mut Self> {
        let hash = bcrypt::hash(password, self.cost)
            .map_err(|e| AuthError::ValidationFailed(format!("BCrypt hash failed: {}", e)))?;
        self.users.insert(username, hash);
        Ok(self)
    }

    /// Create validator from HashMap of usernames and BCrypt hashes
    pub fn from_hashes(users: HashMap<String, String>) -> Self {
        Self {
            users,
            cost: bcrypt::DEFAULT_COST,
        }
    }
}

#[cfg(feature = "bcrypt")]
impl Default for BcryptUserValidator {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(feature = "bcrypt")]
impl UserValidator for BcryptUserValidator {
    fn validate<'a>(
        &'a self,
        credentials: &'a Credentials,
    ) -> Pin<Box<dyn Future<Output = AuthResult<bool>> + Send + 'a>> {
        Box::pin(async move {
            match self.users.get(&credentials.username) {
                Some(stored_hash) => {
                    // Bcrypt verification is blocking, so we use spawn_blocking
                    let password = credentials.password.clone();
                    let hash = stored_hash.clone();

                    let result =
                        ntex::rt::spawn_blocking(move || bcrypt::verify(&password, &hash)).await;

                    match result {
                        Ok(Ok(is_valid)) => Ok(is_valid),
                        Ok(Err(e)) => Err(AuthError::ValidationFailed(format!(
                            "BCrypt verify failed: {}",
                            e
                        ))),
                        Err(e) => Err(AuthError::InternalError(format!("Task join failed: {}", e))),
                    }
                }
                None => Ok(false),
            }
        })
    }

    fn name(&self) -> &'static str {
        "BcryptUserValidator"
    }

    fn user_count(&self) -> usize {
        self.users.len()
    }
}

/// Authentication metrics for monitoring
#[derive(Debug, Default)]
pub struct AuthMetrics {
    /// Total authentication requests
    pub total_requests: AtomicU64,
    /// Successful authentications
    pub successful_auths: AtomicU64,
    /// Failed authentications
    pub failed_auths: AtomicU64,
    /// Cached authentication hits
    pub cached_hits: AtomicU64,
    /// Total validation time in milliseconds
    pub validation_time_ms: AtomicU64,
}

impl AuthMetrics {
    /// Create new metrics instance
    pub fn new() -> Self {
        Self::default()
    }

    /// Get total requests
    pub fn total_requests(&self) -> u64 {
        self.total_requests.load(Ordering::Relaxed)
    }

    /// Get successful authentications
    pub fn successful_auths(&self) -> u64 {
        self.successful_auths.load(Ordering::Relaxed)
    }

    /// Get failed authentications
    pub fn failed_auths(&self) -> u64 {
        self.failed_auths.load(Ordering::Relaxed)
    }

    /// Get cache hits
    pub fn cached_hits(&self) -> u64 {
        self.cached_hits.load(Ordering::Relaxed)
    }

    /// Get average validation time in milliseconds
    pub fn avg_validation_time_ms(&self) -> f64 {
        let total_time = self.validation_time_ms.load(Ordering::Relaxed);
        let total_requests = self.total_requests.load(Ordering::Relaxed);
        if total_requests > 0 {
            total_time as f64 / total_requests as f64
        } else {
            0.0
        }
    }

    /// Get success rate as percentage
    pub fn success_rate(&self) -> f64 {
        let successful = self.successful_auths.load(Ordering::Relaxed);
        let total = self.total_requests.load(Ordering::Relaxed);
        if total > 0 {
            (successful as f64 / total as f64) * 100.0
        } else {
            0.0
        }
    }

    /// Get cache hit rate as percentage
    pub fn cache_hit_rate(&self) -> f64 {
        let hits = self.cached_hits.load(Ordering::Relaxed);
        let total = self.total_requests.load(Ordering::Relaxed);
        if total > 0 {
            (hits as f64 / total as f64) * 100.0
        } else {
            0.0
        }
    }

    /// Reset all metrics
    pub fn reset(&self) {
        self.total_requests.store(0, Ordering::Relaxed);
        self.successful_auths.store(0, Ordering::Relaxed);
        self.failed_auths.store(0, Ordering::Relaxed);
        self.cached_hits.store(0, Ordering::Relaxed);
        self.validation_time_ms.store(0, Ordering::Relaxed);
    }

    /// Increment total authentication request counter
    pub fn incr_total_requests(&self) {
        self.total_requests.fetch_add(1, Ordering::Relaxed);
    }

    /// Increment successful authentication counter
    pub fn incr_successful_auths(&self) {
        self.successful_auths.fetch_add(1, Ordering::Relaxed);
    }

    /// Increment failed authentication counter
    pub fn incr_failed_auths(&self) {
        self.failed_auths.fetch_add(1, Ordering::Relaxed);
    }

    /// Increment cached authentication hit counter
    pub fn incr_cached_hits(&self) {
        self.cached_hits.fetch_add(1, Ordering::Relaxed);
    }

    /// Add validation time in milliseconds
    pub fn add_validation_time(&self, duration: Duration) {
        let ms = duration.as_millis() as u64;
        self.validation_time_ms.fetch_add(ms, Ordering::Relaxed);
    }
}

/// Custom error handler type for authentication failures
pub type CustomErrorHandler = Arc<dyn Fn(&AuthError, &str) -> web::HttpResponse + Send + Sync>;

/// Basic authentication config
pub struct BasicAuthConfig {
    /// Authentication realm (for WWW-Authenticate header)
    pub realm: String,
    /// User validator
    pub validator: Arc<dyn UserValidator>,
    #[cfg(feature = "cache")]
    /// Auth result cache (optional)
    pub cache: Option<Arc<AuthCache>>,
    /// Path filter (optional)
    pub path_filter: Option<Arc<crate::utils::PathFilter>>,
    /// Request header size limit (bytes)
    pub max_header_size: usize,
    /// Log details on authentication failure
    pub log_failures: bool,
    /// Custom error handler
    pub custom_error_handler: Option<CustomErrorHandler>,
    /// Maximum concurrent authentication validations
    pub max_concurrent_validations: Option<usize>,
    /// Validation timeout
    pub validation_timeout: Option<Duration>,
    /// Rate limiting: (max_requests, time_window)
    pub rate_limit_per_ip: Option<(usize, Duration)>,
    /// Header to read the client IP from for rate limiting (e.g.
    /// `"x-forwarded-for"`). When set, the first address in the header is used
    /// instead of the transport peer address. Only enable behind a trusted
    /// proxy, since clients can spoof this header.
    pub client_ip_header: Option<String>,
    /// Enable metrics collection
    pub enable_metrics: bool,
    /// Log usernames in production (security risk)
    pub log_usernames_in_production: bool,
}

impl BasicAuthConfig {
    /// Create new basic auth config
    pub fn new(validator: Arc<dyn UserValidator>) -> Self {
        Self {
            realm: "Restricted Area".to_string(),
            validator,
            #[cfg(feature = "cache")]
            cache: None,
            path_filter: None,
            max_header_size: 8192, // 8KB
            log_failures: false,
            custom_error_handler: None,
            max_concurrent_validations: None,
            validation_timeout: Some(Duration::from_secs(30)),
            rate_limit_per_ip: None,
            client_ip_header: None,
            enable_metrics: true,
            log_usernames_in_production: false,
        }
    }

    /// Set authentication realm
    pub fn realm(mut self, realm: String) -> Self {
        self.realm = realm;
        self
    }

    #[cfg(feature = "cache")]
    /// Create auth config with cache config
    pub fn with_cache(mut self, cache_config: CacheConfig) -> AuthResult<Self> {
        self.cache = Some(Arc::new(AuthCache::new(cache_config)?));
        Ok(self)
    }

    #[cfg(feature = "cache")]
    /// Disable cache
    pub fn disable_cache(mut self) -> Self {
        self.cache = None;
        self
    }

    /// Set path filter
    pub fn path_filter(mut self, filter: crate::utils::PathFilter) -> Self {
        self.path_filter = Some(Arc::new(filter));
        self
    }

    /// Set max request header size
    pub fn max_header_size(mut self, size: usize) -> Self {
        self.max_header_size = size;
        self
    }

    /// Enable or disable logging on authentication failure
    pub fn log_failures(mut self, enabled: bool) -> Self {
        self.log_failures = enabled;
        self
    }

    /// Set custom error handler function
    pub fn custom_error_handler<F>(mut self, handler: F) -> Self
    where
        F: Fn(&AuthError, &str) -> web::HttpResponse + Send + Sync + 'static,
    {
        self.custom_error_handler = Some(Arc::new(handler));
        self
    }

    /// Set maximum concurrent validations
    pub fn max_concurrent_validations(mut self, max: usize) -> Self {
        self.max_concurrent_validations = Some(max);
        self
    }

    /// Set validation timeout
    pub fn validation_timeout(mut self, timeout: Duration) -> Self {
        self.validation_timeout = Some(timeout);
        self
    }

    /// Set rate limiting per IP
    pub fn rate_limit_per_ip(mut self, max_requests: usize, window: Duration) -> Self {
        self.rate_limit_per_ip = Some((max_requests, window));
        self
    }

    /// Set the header to read the client IP from for rate limiting (e.g.
    /// `"x-forwarded-for"`). Only enable this behind a trusted proxy, since
    /// clients can otherwise spoof the header to evade or forge rate limits.
    pub fn client_ip_header(mut self, header: impl Into<String>) -> Self {
        self.client_ip_header = Some(header.into());
        self
    }

    /// Enable or disable metrics collection
    pub fn enable_metrics(mut self, enabled: bool) -> Self {
        self.enable_metrics = enabled;
        self
    }

    /// Enable or disable logging usernames in production (security risk)
    pub fn log_usernames_in_production(mut self, enabled: bool) -> Self {
        self.log_usernames_in_production = enabled;
        self
    }

    /// Enhanced config validation
    pub fn validate(&self) -> AuthResult<()> {
        if self.realm.is_empty() {
            return Err(AuthError::ConfigError("realm cannot be empty".to_string()));
        }
        if self.max_header_size == 0 {
            return Err(AuthError::ConfigError(
                "max_header_size must be greater than 0".to_string(),
            ));
        }
        if self.max_header_size > 1024 * 1024 {
            // 1MB limit
            return Err(AuthError::ConfigError(
                "max_header_size too large (max 1MB)".to_string(),
            ));
        }

        if let Some(max_concurrent) = self.max_concurrent_validations {
            if max_concurrent == 0 {
                return Err(AuthError::ConfigError(
                    "max_concurrent_validations must be greater than 0".to_string(),
                ));
            }
            if max_concurrent > 10000 {
                return Err(AuthError::ConfigError(
                    "max_concurrent_validations too large (max 10000)".to_string(),
                ));
            }
        }

        if let Some(timeout) = self.validation_timeout {
            if timeout.is_zero() {
                return Err(AuthError::ConfigError(
                    "validation_timeout must be greater than 0".to_string(),
                ));
            }
            if timeout > Duration::from_secs(300) {
                // 5 minutes
                return Err(AuthError::ConfigError(
                    "validation_timeout too large (max 5 minutes)".to_string(),
                ));
            }
        }

        if let Some((max_requests, window)) = self.rate_limit_per_ip {
            if max_requests == 0 {
                return Err(AuthError::ConfigError(
                    "rate_limit max_requests must be greater than 0".to_string(),
                ));
            }
            if window.is_zero() {
                return Err(AuthError::ConfigError(
                    "rate_limit window must be greater than 0".to_string(),
                ));
            }
        }

        #[cfg(feature = "cache")]
        if let Some(cache) = &self.cache {
            let stats = cache.stats();
            if stats.total_entries > 100000 {
                // Reasonable cache size limit
                eprintln!(
                    "Warning: Cache has {} entries, consider reducing TTL",
                    stats.total_entries
                );
            }
        }

        Ok(())
    }
}

/// Basic authentication middleware
pub struct BasicAuth {
    pub(crate) config: BasicAuthConfig,
    pub(crate) metrics: Arc<AuthMetrics>,
    pub(crate) concurrency_limiter: Option<Arc<ConcurrencyLimiter>>,
    pub(crate) rate_limiter: Option<Arc<RateLimiter>>,
}

impl BasicAuth {
    /// Create new BasicAuth instance
    pub fn new(config: BasicAuthConfig) -> AuthResult<Self> {
        config.validate()?;
        let concurrency_limiter = config
            .max_concurrent_validations
            .map(|max| Arc::new(ConcurrencyLimiter::new(max)));
        let rate_limiter = config
            .rate_limit_per_ip
            .map(|(max_requests, window)| Arc::new(RateLimiter::new(max_requests, window)));
        Ok(Self {
            config,
            metrics: Arc::new(AuthMetrics::new()),
            concurrency_limiter,
            rate_limiter,
        })
    }

    /// Get metrics reference
    pub fn metrics(&self) -> &AuthMetrics {
        &self.metrics
    }

    /// Create BasicAuth with static user list
    pub fn with_users(users: HashMap<String, String>) -> AuthResult<Self> {
        let validator = Arc::new(StaticUserValidator::from_map(users));
        let config = BasicAuthConfig::new(validator);
        Self::new(config)
    }

    /// Create BasicAuth with a single user
    pub fn with_user(username: String, password: String) -> AuthResult<Self> {
        let mut users = HashMap::new();
        users.insert(username, password);
        Self::with_users(users)
    }

    /// Parse Authorization header and extract credentials
    /// Supports colons in password
    fn parse_credentials(auth_header: &str, max_size: usize) -> AuthResult<Credentials> {
        if auth_header.len() > max_size {
            return Err(AuthError::InvalidFormat);
        }

        // Safe slicing: `get(..6)` returns None when the cut is not on a UTF-8
        // boundary, which prevents a panic on malformed non-ASCII headers.
        let scheme = auth_header.get(..6).ok_or(AuthError::InvalidFormat)?;
        if !scheme.eq_ignore_ascii_case("Basic ") {
            return Err(AuthError::InvalidFormat);
        }

        let encoded = &auth_header[6..]; // Remove "Basic " prefix

        // Check Base64 string length
        if encoded.len() > (max_size * 3 / 4) {
            return Err(AuthError::InvalidFormat);
        }

        let decoded = STANDARD
            .decode(encoded)
            .map_err(|_| AuthError::InvalidBase64)?;

        // Validate UTF-8 without an intermediate String allocation
        let decoded_str = std::str::from_utf8(&decoded).map_err(|_| AuthError::InvalidBase64)?;

        // Split only at the first colon, support colons in password
        let (username, password) = decoded_str
            .split_once(':')
            .ok_or(AuthError::InvalidFormat)?;

        let credentials = Credentials::new(username.to_string(), password.to_string());

        // Validate credentials format
        if !credentials.is_valid_format() {
            return Err(AuthError::InvalidCredentials);
        }

        Ok(credentials)
    }

    /// Authenticate user credentials
    async fn authenticate(&self, credentials: &Credentials) -> AuthResult<bool> {
        // Pre-validation check
        self.config.validator.pre_validate(credentials)?;

        // Check cache and cache result (compute key only once)
        #[cfg(feature = "cache")]
        {
            if let Some(cache) = &self.config.cache {
                let cache_key = credentials.cache_key();
                if let Some(cached_result) = cache.get(&cache_key) {
                    if self.config.enable_metrics {
                        self.metrics.incr_cached_hits();
                    }
                    return Ok(cached_result);
                }

                let start = Instant::now();
                let result = self.run_validation(credentials).await?;

                if self.config.enable_metrics {
                    self.metrics.add_validation_time(start.elapsed());
                }

                // Cache result using the same key
                if let Err(e) = cache.insert(cache_key, result) {
                    // Cache failure should not affect authentication result, just log error
                    eprintln!("Failed to cache authentication result: {}", e);
                }

                return Ok(result);
            }
        }

        // Validate using configured validator (when cache is disabled)
        let start = Instant::now();
        let result = self.run_validation(credentials).await?;

        if self.config.enable_metrics {
            self.metrics.add_validation_time(start.elapsed());
        }
        Ok(result)
    }

    /// Run the configured validator, applying the optional concurrency limit
    /// and validation timeout.
    ///
    /// When a concurrency limiter is configured, validation runs inside a
    /// spawned task so the permit is held until the validator actually finishes
    /// — even if *this* call times out. That keeps `max_concurrent_validations`
    /// effective for validators that cannot be cancelled (e.g. bcrypt on a
    /// blocking thread): a timed-out request no longer releases its permit early
    /// and lets work pile up past the limit. Permit acquisition is also covered
    /// by the timeout, so a saturated limiter cannot make a request wait longer
    /// than `validation_timeout`.
    async fn run_validation(&self, credentials: &Credentials) -> AuthResult<bool> {
        // Without a concurrency limiter there is no permit to protect, so we can
        // validate directly (optionally under a timeout).
        let Some(limiter) = &self.concurrency_limiter else {
            let validate = self.config.validator.validate(credentials);
            return match self.config.validation_timeout {
                Some(timeout_dur) => timeout(timeout_dur, validate)
                    .await
                    .map_err(|_| AuthError::InternalError("Validation timed out".to_string()))?,
                None => validate.await,
            };
        };

        let validator = Arc::clone(&self.config.validator);
        let limiter = Arc::clone(limiter);
        let credentials = credentials.clone();

        let task = ntex::rt::spawn(async move {
            // Held until validation completes, even if the caller times out below.
            let _permit = limiter.acquire().await;
            validator.validate(&credentials).await
        });

        match self.config.validation_timeout {
            Some(timeout_dur) => match timeout(timeout_dur, task).await {
                Ok(join_result) => join_result
                    .map_err(|_| AuthError::InternalError("Validation task failed".to_string()))?,
                Err(_) => Err(AuthError::InternalError("Validation timed out".to_string())),
            },
            None => task
                .await
                .map_err(|_| AuthError::InternalError("Validation task failed".to_string()))?,
        }
    }

    /// Resolve the client IP used for rate limiting.
    ///
    /// When `client_ip_header` is configured (trusted-proxy deployments), the
    /// first address in that header wins; otherwise the transport peer address
    /// is used. Only enable the header when running behind a trusted proxy,
    /// since clients can otherwise spoof it to evade or forge rate limits.
    fn client_ip<Err>(&self, req: &web::WebRequest<Err>) -> String
    where
        Err: web::ErrorRenderer,
    {
        if let Some(header) = &self.config.client_ip_header
            && let Some(value) = req.headers().get(header).and_then(|v| v.to_str().ok())
            && let Some(first) = value.split(',').next()
        {
            let ip = first.trim();
            if !ip.is_empty() {
                return ip.to_string();
            }
        }
        req.peer_addr()
            .map(|addr| addr.ip().to_string())
            .unwrap_or_default()
    }

    /// Handle authentication error
    fn handle_auth_error(&self, error: &AuthError) -> web::HttpResponse {
        if let Some(handler) = &self.config.custom_error_handler {
            handler(error, &self.config.realm)
        } else {
            error.to_response(&self.config.realm)
        }
    }

    /// Log authentication failure (if enabled) with enhanced security
    fn log_auth_failure(&self, error: &AuthError, username: Option<&str>) {
        if self.config.log_failures {
            // In production, avoid logging usernames unless explicitly configured
            let safe_username = if self.config.log_usernames_in_production || cfg!(debug_assertions)
            {
                username
            } else {
                None // Don't log usernames in production for security
            };

            match safe_username {
                Some(user) => eprintln!("Authentication failed - user: {}, error: {}", user, error),
                None => eprintln!("Authentication failed - error: {}", error),
            }
        }
    }
}

impl<S, Cfg> Middleware<S, Cfg> for BasicAuth {
    type Service = BasicAuthMiddlewareService<S>;

    fn create(&self, service: S, _cfg: Cfg) -> Self::Service {
        BasicAuthMiddlewareService {
            service,
            auth: BasicAuth {
                config: BasicAuthConfig {
                    realm: self.config.realm.clone(),
                    validator: Arc::clone(&self.config.validator),
                    #[cfg(feature = "cache")]
                    cache: self.config.cache.clone(),
                    path_filter: self.config.path_filter.clone(),
                    max_header_size: self.config.max_header_size,
                    log_failures: self.config.log_failures,
                    custom_error_handler: self.config.custom_error_handler.clone(),
                    max_concurrent_validations: self.config.max_concurrent_validations,
                    validation_timeout: self.config.validation_timeout,
                    rate_limit_per_ip: self.config.rate_limit_per_ip,
                    client_ip_header: self.config.client_ip_header.clone(),
                    enable_metrics: self.config.enable_metrics,
                    log_usernames_in_production: self.config.log_usernames_in_production,
                },
                metrics: Arc::clone(&self.metrics),
                concurrency_limiter: self.concurrency_limiter.clone(),
                rate_limiter: self.rate_limiter.clone(),
            },
        }
    }
}

pub struct BasicAuthMiddlewareService<S> {
    service: S,
    auth: BasicAuth,
}

impl<S, Err> Service<web::WebRequest<Err>> for BasicAuthMiddlewareService<S>
where
    S: Service<web::WebRequest<Err>, Response = web::WebResponse, Error = web::Error> + 'static,
    Err: web::ErrorRenderer,
{
    type Response = web::WebResponse;
    type Error = web::Error;

    async fn call(
        &self,
        req: web::WebRequest<Err>,
        ctx: ServiceCtx<'_, Self>,
    ) -> Result<Self::Response, Self::Error> {
        let metrics_enabled = self.auth.config.enable_metrics;

        // Check if path filter is configured and should skip authentication
        if let Some(filter) = &self.auth.config.path_filter
            && filter.should_skip(req.path())
        {
            return ctx.call(&self.service, req).await;
        }

        if metrics_enabled {
            self.auth.metrics.incr_total_requests();
        }

        // Per-IP rate limiting (checked before credential parsing)
        if let Some(rate_limiter) = &self.auth.rate_limiter {
            let ip = self.auth.client_ip(&req);
            if let Err(err) = rate_limiter.check(&ip) {
                self.auth.log_auth_failure(&err, None);
                let response = self.auth.handle_auth_error(&err);
                if metrics_enabled {
                    self.auth.metrics.incr_failed_auths();
                }
                return Ok(req.into_response(response));
            }
        }

        // Extract authorization header
        let auth_header = req
            .headers()
            .get("authorization")
            .and_then(|h| h.to_str().ok());

        // Handle missing or malformed Authorization header
        let auth_header = match auth_header {
            Some(header) => header,
            None => {
                let error = AuthError::MissingHeader;
                self.auth.log_auth_failure(&error, None);
                let response = self.auth.handle_auth_error(&error);
                if metrics_enabled {
                    self.auth.metrics.incr_failed_auths();
                }
                return Ok(req.into_response(response));
            }
        };

        // Parse credentials from Authorization header
        let credentials =
            match BasicAuth::parse_credentials(auth_header, self.auth.config.max_header_size) {
                Ok(creds) => creds,
                Err(err) => {
                    self.auth.log_auth_failure(&err, None);
                    let response = self.auth.handle_auth_error(&err);
                    if metrics_enabled {
                        self.auth.metrics.incr_failed_auths();
                    }
                    return Ok(req.into_response(response));
                }
            };

        // Authenticate user credentials
        let is_authenticated = match self.auth.authenticate(&credentials).await {
            Ok(result) => result,
            Err(err) => {
                self.auth
                    .log_auth_failure(&err, Some(&credentials.username));
                let response = self.auth.handle_auth_error(&err);
                if metrics_enabled {
                    self.auth.metrics.incr_failed_auths();
                }
                return Ok(req.into_response(response));
            }
        };

        if !is_authenticated {
            let error = AuthError::InvalidCredentials;
            self.auth
                .log_auth_failure(&error, Some(&credentials.username));
            let response = self.auth.handle_auth_error(&error);
            if metrics_enabled {
                self.auth.metrics.incr_failed_auths();
            }
            return Ok(req.into_response(response));
        }

        if metrics_enabled {
            self.auth.metrics.incr_successful_auths();
        }

        // Add credentials to request extensions for further processing
        req.extensions_mut().insert(credentials);
        ctx.call(&self.service, req).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;

    #[tokio::test]
    async fn test_static_validator() {
        let mut users = HashMap::new();
        users.insert("admin".to_string(), "secret".to_string());
        users.insert("user".to_string(), "password:with:colons".to_string());

        let validator = StaticUserValidator::from_map(users);

        let valid_creds = Credentials::new("admin".to_string(), "secret".to_string());
        let colon_password_creds =
            Credentials::new("user".to_string(), "password:with:colons".to_string());
        let invalid_creds = Credentials::new("admin".to_string(), "wrong".to_string());

        assert!(validator.validate(&valid_creds).await.unwrap());
        assert!(validator.validate(&colon_password_creds).await.unwrap());
        assert!(!validator.validate(&invalid_creds).await.unwrap());
    }

    #[test]
    fn test_parse_credentials_with_colons() {
        use base64::Engine;
        let credentials = "admin:pass:word:with:colons";
        let encoded = STANDARD.encode(credentials.as_bytes());
        let auth_header = format!("Basic {}", encoded);

        let creds = BasicAuth::parse_credentials(&auth_header, 8192).unwrap();
        assert_eq!(creds.username, "admin");
        assert_eq!(creds.password, "pass:word:with:colons");
    }

    #[test]
    fn test_parse_credentials_multibyte_no_panic() {
        // "abc" + a 4-byte emoji: byte index 6 falls inside the emoji, which
        // previously caused `&auth_header[..6]` to panic. Must be rejected
        // gracefully instead of crashing the worker (DoS).
        let malicious = "abc😀garbage";
        let result = BasicAuth::parse_credentials(malicious, 8192);
        assert!(matches!(result, Err(AuthError::InvalidFormat)));
    }

    #[test]
    fn test_credentials_validation() {
        let valid_creds = Credentials::new("user".to_string(), "pass".to_string());
        let valid_empty_user = Credentials::new("".to_string(), "pass".to_string()); // Now valid per RFC 7617
        let invalid_creds1 = Credentials::new("user:name".to_string(), "pass".to_string());
        let invalid_creds2 = Credentials::new("user".to_string(), "pass\nword".to_string());
        let invalid_creds3 = Credentials::new("user".to_string(), "pass\tword".to_string()); // Tab is control character

        assert!(valid_creds.is_valid_format());
        assert!(valid_empty_user.is_valid_format());
        assert!(!invalid_creds1.is_valid_format());
        assert!(!invalid_creds2.is_valid_format());
        assert!(!invalid_creds3.is_valid_format());
    }

    #[cfg(feature = "cache")]
    #[test]
    fn test_secure_cache_key() {
        let creds = Credentials::new("admin".to_string(), "secret".to_string());

        let key1 = creds.cache_key();
        let key2 = creds.cache_key();

        // The same credentials should produce the same cache key
        assert_eq!(key1, key2);

        // Cache key should be a 32-byte array (does not contain sensitive information)
        assert_eq!(key1.len(), 32);
    }

    #[test]
    fn test_case_insensitive_validator() {
        let mut users = HashMap::new();
        users.insert("admin".to_string(), "secret".to_string());

        let validator = StaticUserValidator::from_map_case_insensitive(users);

        assert!(validator.contains_user("admin"));
        assert!(validator.contains_user("ADMIN"));
        assert!(validator.contains_user("Admin"));
    }

    #[tokio::test]
    async fn test_validator_pre_validation() {
        let validator = StaticUserValidator::new();
        let invalid_creds = Credentials::new("user:name".to_string(), "pass".to_string());

        assert!(validator.pre_validate(&invalid_creds).is_err());
    }

    #[test]
    fn test_config_validation() {
        let validator = Arc::new(StaticUserValidator::new());

        let valid_config = BasicAuthConfig::new(validator.clone());
        assert!(valid_config.validate().is_ok());

        let invalid_config = BasicAuthConfig::new(validator).realm("".to_string());
        assert!(invalid_config.validate().is_err());
    }

    #[cfg(feature = "bcrypt")]
    #[tokio::test]
    async fn test_bcrypt_validator() {
        let mut validator = BcryptUserValidator::new();
        validator
            .add_user_with_password("admin".to_string(), "secret")
            .unwrap();

        let valid_creds = Credentials::new("admin".to_string(), "secret".to_string());
        let invalid_creds = Credentials::new("admin".to_string(), "wrong".to_string());

        assert!(validator.validate(&valid_creds).await.unwrap());
        assert!(!validator.validate(&invalid_creds).await.unwrap());
    }

    #[ntex::test]
    async fn test_rate_limit_per_ip() {
        use base64::Engine;
        use std::time::Duration;

        let auth = crate::BasicAuthBuilder::new()
            .user("admin", "secret")
            .rate_limit_per_ip(1, Duration::from_secs(60))
            .build()
            .unwrap();
        let app = ntex::web::test::init_service(
            ntex::web::App::new()
                .middleware(auth)
                .route("/", ntex::web::get().to(|| async { "ok" })),
        )
        .await;

        let auth_hdr = format!(
            "Basic {}",
            base64::engine::general_purpose::STANDARD.encode("admin:secret")
        );
        let ip: std::net::SocketAddr = "1.2.3.4:80".parse().unwrap();

        // First request from this IP succeeds.
        let req = ntex::web::test::TestRequest::get()
            .peer_addr(ip)
            .header("authorization", auth_hdr.as_str())
            .to_request();
        let resp = ntex::web::test::call_service(&app, req).await;
        assert!(resp.status().is_success());

        // Second request from the same IP exceeds the limit (1 req / 60s).
        let req = ntex::web::test::TestRequest::get()
            .peer_addr(ip)
            .header("authorization", auth_hdr.as_str())
            .to_request();
        let resp = ntex::web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), ntex::http::StatusCode::TOO_MANY_REQUESTS);
    }

    #[ntex::test]
    async fn test_validation_timeout() {
        use base64::Engine;
        use std::future::Future;
        use std::pin::Pin;
        use std::sync::Arc;
        use std::time::Duration;

        #[derive(Debug)]
        struct SlowValidator;

        impl UserValidator for SlowValidator {
            fn validate<'a>(
                &'a self,
                _: &'a Credentials,
            ) -> Pin<Box<dyn Future<Output = AuthResult<bool>> + Send + 'a>> {
                Box::pin(async {
                    ntex::time::sleep(Duration::from_secs(5)).await;
                    Ok(true)
                })
            }
        }

        let validator = Arc::new(SlowValidator);
        let config = BasicAuthConfig::new(validator).validation_timeout(Duration::from_millis(100));
        let auth = BasicAuth::new(config).unwrap();
        let app = ntex::web::test::init_service(
            ntex::web::App::new()
                .middleware(auth)
                .route("/", ntex::web::get().to(|| async { "ok" })),
        )
        .await;

        let auth_hdr = format!(
            "Basic {}",
            base64::engine::general_purpose::STANDARD.encode("admin:secret")
        );
        let req = ntex::web::test::TestRequest::get()
            .header("authorization", auth_hdr.as_str())
            .to_request();
        let resp = ntex::web::test::call_service(&app, req).await;
        // Validation timed out -> InternalError (500).
        assert_eq!(resp.status(), ntex::http::StatusCode::INTERNAL_SERVER_ERROR);
    }

    #[ntex::test]
    async fn test_concurrency_limited_auth_succeeds() {
        use base64::Engine;

        // Exercises the spawned-task validation path (concurrency limiter set +
        // default timeout): a normal request must still authenticate.
        let auth = crate::BasicAuthBuilder::new()
            .user("admin", "secret")
            .max_concurrent_validations(2)
            .build()
            .unwrap();
        let app = ntex::web::test::init_service(
            ntex::web::App::new()
                .middleware(auth)
                .route("/", ntex::web::get().to(|| async { "ok" })),
        )
        .await;

        let auth_hdr = format!(
            "Basic {}",
            base64::engine::general_purpose::STANDARD.encode("admin:secret")
        );
        let req = ntex::web::test::TestRequest::get()
            .header("authorization", auth_hdr.as_str())
            .to_request();
        let resp = ntex::web::test::call_service(&app, req).await;
        assert!(resp.status().is_success());
    }

    #[ntex::test]
    async fn test_rate_limit_uses_forwarded_header() {
        use base64::Engine;
        use std::time::Duration;

        // With client_ip_header configured, rate limiting keys on the forwarded
        // client IP rather than the (shared) proxy peer address.
        let auth = crate::BasicAuthBuilder::new()
            .user("admin", "secret")
            .rate_limit_per_ip(1, Duration::from_secs(60))
            .client_ip_header("x-forwarded-for")
            .build()
            .unwrap();
        let app = ntex::web::test::init_service(
            ntex::web::App::new()
                .middleware(auth)
                .route("/", ntex::web::get().to(|| async { "ok" })),
        )
        .await;
        let auth_hdr = format!(
            "Basic {}",
            base64::engine::general_purpose::STANDARD.encode("admin:secret")
        );

        // Two requests from the same forwarded IP: second exceeds the 1/60s limit.
        for (i, expect_ok) in [(0u8, true), (1u8, false)] {
            let req = ntex::web::test::TestRequest::get()
                .header("authorization", auth_hdr.as_str())
                .header("x-forwarded-for", "9.9.9.9, 10.0.0.1")
                .to_request();
            let resp = ntex::web::test::call_service(&app, req).await;
            assert_eq!(
                resp.status().is_success(),
                expect_ok,
                "request {i} unexpected status {}",
                resp.status()
            );
        }

        // A different forwarded IP has its own budget even though the peer
        // address (absent here) is identical.
        let req = ntex::web::test::TestRequest::get()
            .header("authorization", auth_hdr.as_str())
            .header("x-forwarded-for", "8.8.8.8")
            .to_request();
        let resp = ntex::web::test::call_service(&app, req).await;
        assert!(resp.status().is_success());
    }
}