praxis-proxy 0.4.0

// SPDX-License-Identifier: MIT
// Copyright (c) 2024 Praxis Contributors

//! Server bootstrap: protocol registration and startup.

use std::{
    path::PathBuf,
    sync::{Arc, Mutex},
};

use praxis_core::{
    PingoraServerRuntime,
    config::{Config, ProtocolKind},
    health::{HealthRegistry, build_health_registry},
};
use praxis_filter::FilterRegistry;
use praxis_protocol::{CertWatcherShutdowns, ListenerPipelines, Protocol as _, http::PingoraHttp, tcp::PingoraTcp};
use tokio_util::sync::CancellationToken;
use tracing::info;

use crate::pipelines::resolve_pipelines;

// -----------------------------------------------------------------------------
// Config Path Resolution
// -----------------------------------------------------------------------------

/// Resolve the config file path without loading the config.
///
/// Returns `Some` if an explicit path was given or `praxis.yaml`
/// exists in the working directory. Returns `None` when using the
/// built-in default (no file to watch).
///
/// ```
/// let path = praxis::resolve_config_path(None);
/// // Returns None if ./praxis.yaml doesn't exist.
/// ```
pub fn resolve_config_path(explicit: Option<&str>) -> Option<PathBuf> {
    if let Some(path) = explicit {
        return Some(PathBuf::from(path));
    }
    let default_path = PathBuf::from("praxis.yaml");
    default_path.exists().then_some(default_path)
}

// -----------------------------------------------------------------------------
// Server
// -----------------------------------------------------------------------------

/// Build filter pipelines using built-in and auto-discovered external filters, register
/// protocols and run the server.
///
/// # Security: Root Check
///
/// On Unix, this function refuses to start if the effective UID is 0 (root). Set
/// `insecure_options.allow_root: true` in the configuration to override. Prefer
/// `CAP_NET_BIND_SERVICE` or a reverse proxy for low-port binding.
///
/// Config is owned for the server's lifetime (never returns).
#[expect(clippy::allow_attributes, reason = "lint is platform/config-dependent")]
#[allow(clippy::needless_pass_by_value, reason = "server owns config")]
pub fn run_server(config: Config, config_path: Option<PathBuf>) -> ! {
    run_server_with_registry(config, crate::build_full_registry(), config_path)
}

/// Build filter pipelines from the given registry, register protocols and run the server.
///
/// Use this variant when you need custom filters beyond the built-ins (e.g. via [`register_filters!`]).
///
/// Assumes tracing is already initialized. Blocks until the process is terminated; never returns.
///
/// Config is owned for the server's lifetime (never returns).
///
/// [`register_filters!`]: praxis_filter::register_filters
#[expect(clippy::allow_attributes, reason = "lint is platform/config-dependent")]
#[allow(clippy::needless_pass_by_value, reason = "server owns config")]
pub fn run_server_with_registry(config: Config, registry: FilterRegistry, config_path: Option<PathBuf>) -> ! {
    enforce_root_check(&config);
    warn_insecure_options(&config);
    init_runtime_limits(&config.runtime);
    warn_insecure_key_permissions(&config);

    let health_registry = build_health_registry(&config.clusters);
    let state = build_server_state(&config, &registry, &health_registry);

    info!("initializing server");
    let mut server = PingoraServerRuntime::new(&config);
    let _cert_shutdowns = register_protocols(&mut server, &config, &state.pipelines);
    register_admin_endpoints(&mut server, &config, health_registry, &state.kv_stores);

    let _watcher = spawn_watcher(config_path, config, registry, state);

    info!("starting server");
    server.run()
}

// -----------------------------------------------------------------------------
// Server State
// -----------------------------------------------------------------------------

/// State built during server initialization and shared with the
/// file watcher for hot reload.
struct ServerState {
    /// Resolved filter pipelines per listener.
    pipelines: Arc<ListenerPipelines>,
    /// KV store registry.
    kv_stores: praxis_core::kv::KvStoreRegistry,
    /// Health check cancellation token.
    health_shutdown: Arc<Mutex<CancellationToken>>,
}

/// Build filter pipelines, health checks, and registries.
fn build_server_state(config: &Config, registry: &FilterRegistry, health_registry: &HealthRegistry) -> ServerState {
    info!("building filter pipelines");
    let kv_stores = praxis_core::kv::KvStoreRegistry::new();

    let pipelines = resolve_pipelines(config, registry, health_registry, &kv_stores).unwrap_or_else(|e| fatal(&e));

    let health_shutdown = Arc::new(Mutex::new(CancellationToken::new()));
    spawn_health_check_tasks(config, Arc::clone(health_registry), &health_shutdown);

    ServerState {
        pipelines: Arc::new(pipelines),
        kv_stores,
        health_shutdown,
    }
}

// -----------------------------------------------------------------------------
// Protocol Registration
// -----------------------------------------------------------------------------

/// Register HTTP and TCP protocol handlers with the Pingora server.
fn register_protocols(
    server: &mut PingoraServerRuntime,
    config: &Config,
    pipelines: &ListenerPipelines,
) -> CertWatcherShutdowns {
    let mut all_shutdowns = Vec::new();

    if config.listeners.iter().any(|l| l.protocol == ProtocolKind::Http) {
        let shutdowns = Box::new(PingoraHttp)
            .register(server, config, pipelines)
            .unwrap_or_else(|e| fatal(&e));
        all_shutdowns.extend(shutdowns);
    }

    if config.listeners.iter().any(|l| l.protocol == ProtocolKind::Tcp) {
        let shutdowns = Box::new(PingoraTcp)
            .register(server, config, pipelines)
            .unwrap_or_else(|e| fatal(&e));
        all_shutdowns.extend(shutdowns);
    }

    CertWatcherShutdowns::new(all_shutdowns)
}

/// Spawn the config file watcher if a config path is available.
fn spawn_watcher(
    config_path: Option<PathBuf>,
    config: Config,
    registry: FilterRegistry,
    state: ServerState,
) -> Option<std::thread::JoinHandle<()>> {
    let path = config_path?;
    let handle = crate::watcher::spawn_config_watcher(crate::watcher::WatcherParams {
        config_path: path,
        health_shutdown: state.health_shutdown,
        initial_config: config,
        kv_stores: state.kv_stores,
        pipelines: state.pipelines,
        registry: Arc::new(registry),
        shutdown: CancellationToken::new(),
    });
    Some(handle)
}

// -----------------------------------------------------------------------------
// Admin
// -----------------------------------------------------------------------------

/// Register admin/health endpoints with the Pingora server.
fn register_admin_endpoints(
    server: &mut PingoraServerRuntime,
    config: &Config,
    health_registry: HealthRegistry,
    kv_stores: &praxis_core::kv::KvStoreRegistry,
) {
    if let Some(admin_addr) = &config.admin.address {
        praxis_protocol::http::pingora::health::add_admin_endpoints_to_pingora_server(
            server.server_mut(),
            admin_addr,
            Some(health_registry),
            Some(kv_stores.clone()),
            config.admin.verbose,
        );
    }
}

// -----------------------------------------------------------------------------
// Runtime Limits
// -----------------------------------------------------------------------------

/// Initialize global connection and memory limits from runtime config.
fn init_runtime_limits(runtime: &praxis_core::config::RuntimeConfig) {
    if let Some(max) = runtime.max_connections {
        praxis_protocol::connections::init_global_limit(max as usize);
        info!(max_connections = max, "global connection limit enabled");
    }
    if let Some(threshold) = runtime.max_memory_bytes {
        praxis_core::memory::init(threshold);
        info!(
            threshold_mib = threshold / 1_048_576,
            "memory pressure monitoring enabled"
        );
    }
}

// -----------------------------------------------------------------------------
// Insecure Options Warnings
// -----------------------------------------------------------------------------

/// Emit startup warnings for every active insecure option.
fn warn_insecure_options(config: &Config) {
    let o = &config.insecure_options;
    insecure_warn(
        o.allow_unbounded_body,
        "allow_unbounded_body: body size ceiling relaxed",
    );
    insecure_warn(
        o.allow_open_security_filters,
        "allow_open_security_filters: open failure_mode allowed",
    );
    insecure_warn(
        o.allow_public_admin,
        "allow_public_admin: admin may bind all interfaces",
    );
    insecure_warn(
        o.allow_tls_without_sni,
        "allow_tls_without_sni: TLS hostname verification weakened",
    );
    insecure_warn(
        o.allow_private_health_checks,
        "allow_private_health_checks: loopback health checks allowed",
    );
    insecure_warn(o.csrf_log_only, "csrf_log_only: CSRF violations logged, not rejected");
    insecure_warn(
        o.skip_pipeline_validation,
        "skip_pipeline_validation: pipeline errors demoted to warnings",
    );
}

/// Log a warning if an insecure option is active.
fn insecure_warn(active: bool, msg: &str) {
    if active {
        tracing::warn!("insecure_options.{msg}");
    }
}

// -----------------------------------------------------------------------------
// Root Privilege Check
// -----------------------------------------------------------------------------

/// Refuse to start when running as root (UID 0) unless `allow_root` is set.
///
/// # Errors
///
/// Returns an error message when the effective UID is 0 and `allow_root` is `false`.
///
/// ```
/// let msg = praxis::check_root_privilege(false, 0);
/// assert!(msg.is_some());
///
/// let msg = praxis::check_root_privilege(true, 0);
/// assert!(msg.is_none());
///
/// let msg = praxis::check_root_privilege(false, 1000);
/// assert!(msg.is_none());
/// ```
pub fn check_root_privilege(allow_root: bool, euid: u32) -> Option<String> {
    if euid != 0 {
        return None;
    }

    if allow_root {
        tracing::warn!("running as root (UID 0) with insecure_options.allow_root override; this is not recommended");
        return None;
    }

    Some(
        "Praxis refuses to run as root (UID 0). Running a proxy as root is a security risk.\n\
         Use one of these alternatives:\n  \
         - Run as a non-root user with CAP_NET_BIND_SERVICE for low ports\n  \
         - Use a reverse proxy or socket activation\n  \
         - Set insecure_options.allow_root: true in config to override (not recommended)"
            .to_owned(),
    )
}

/// Enforce the root privilege check on Unix, using the real effective UID.
#[cfg(unix)]
fn enforce_root_check(config: &Config) {
    let euid = nix::unistd::geteuid().as_raw();
    if let Some(msg) = check_root_privilege(config.insecure_options.allow_root, euid) {
        fatal(&msg);
    }
}

/// No-op on non-Unix platforms.
#[cfg(not(unix))]
fn enforce_root_check(_config: &Config) {}

// -----------------------------------------------------------------------------
// TLS Key Permission Checks
// -----------------------------------------------------------------------------

/// Warn if any TLS private key file has group or world read/write permissions.
///
/// This check is intentionally advisory-only (warning, not error) because
/// Kubernetes secret volume mounts often use permissions that would fail a
/// strict check (e.g. `0644`). The warning gives operators visibility without
/// blocking legitimate deployments.
#[cfg(unix)]
fn warn_insecure_key_permissions(config: &Config) {
    use std::os::unix::fs::PermissionsExt as _;

    for listener in &config.listeners {
        if let Some(tls) = &listener.tls {
            for cert in &tls.certificates {
                let key_path = &cert.key_path;
                if let Ok(meta) = std::fs::metadata(key_path) {
                    let mode = meta.permissions().mode();
                    if mode & 0o077 != 0 {
                        tracing::warn!(
                            listener = %listener.name,
                            path = %key_path,
                            mode = format!("{:04o}", mode & 0o7777),
                            "TLS private key file has overly permissive \
                             permissions; recommend chmod 0600"
                        );
                    }
                } else {
                    tracing::trace!(
                        listener = %listener.name,
                        path = %key_path,
                        "skipped permission check: could not read file metadata"
                    );
                }
            }
        }
    }
}

/// No-op on non-Unix platforms.
#[cfg(not(unix))]
fn warn_insecure_key_permissions(_config: &Config) {}

// -----------------------------------------------------------------------------
// Health Check Tasks
// -----------------------------------------------------------------------------

/// Spawn background health check tasks on a dedicated tokio runtime.
///
/// The spawned thread listens for `ctrl_c` and cancels the
/// [`CancellationToken`] so that every health check loop exits
/// cleanly via `shutdown.cancelled()` before the thread returns.
///
/// Pingora's `server.run()` installs its own signal handlers and may
/// terminate the process before this thread receives `ctrl_c`. This is
/// acceptable: the OS reaps the thread on process exit, so the graceful
/// shutdown path here is best-effort.
///
/// [`CancellationToken`]: tokio_util::sync::CancellationToken
#[expect(clippy::expect_used, reason = "fatal")]
fn spawn_health_check_tasks(
    config: &Config,
    registry: HealthRegistry,
    health_shutdown: &Arc<Mutex<CancellationToken>>,
) {
    if registry.is_empty() {
        return;
    }

    let shutdown = health_shutdown.lock().expect("health shutdown lock").clone();
    let clusters = config.clusters.clone();

    std::thread::spawn(move || {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("health check runtime");
        rt.block_on(async {
            praxis_protocol::http::pingora::health::runner::spawn_health_checks(&clusters, &registry, &shutdown);
            shutdown.cancelled().await;
        });
    });
}

// -----------------------------------------------------------------------------
// Utility Functions
// -----------------------------------------------------------------------------

/// Print a fatal error to stderr and exit the process.
#[expect(
    clippy::print_stderr,
    clippy::exit,
    reason = "fatal error output before runtime is available"
)]
pub fn fatal(err: &dyn std::fmt::Display) -> ! {
    eprintln!("fatal: {err}");
    std::process::exit(1)
}

// -----------------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------------

#[cfg(test)]
#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::too_many_lines,
    reason = "tests"
)]
mod tests {
    use super::*;

    #[test]
    fn root_uid_without_override_returns_error() {
        let result = check_root_privilege(false, 0);
        assert!(result.is_some(), "UID 0 without allow_root should return an error");
        let msg = result.unwrap();
        assert!(
            msg.contains("refuses to run as root"),
            "error message should explain the refusal"
        );
    }

    #[test]
    fn root_uid_with_override_returns_none() {
        let result = check_root_privilege(true, 0);
        assert!(result.is_none(), "UID 0 with allow_root should be allowed");
    }

    #[test]
    fn non_root_uid_returns_none() {
        let result = check_root_privilege(false, 1000);
        assert!(result.is_none(), "non-root UID should always be allowed");
    }

    #[test]
    fn non_root_uid_with_override_returns_none() {
        let result = check_root_privilege(true, 1000);
        assert!(result.is_none(), "non-root UID with allow_root should be allowed");
    }

    #[test]
    fn error_message_suggests_alternatives() {
        let msg = check_root_privilege(false, 0).unwrap();
        assert!(
            msg.contains("CAP_NET_BIND_SERVICE"),
            "should suggest CAP_NET_BIND_SERVICE"
        );
        assert!(
            msg.contains("insecure_options.allow_root: true"),
            "should mention the config override"
        );
    }

    #[test]
    fn resolve_config_path_explicit() {
        let path = resolve_config_path(Some("/tmp/test.yaml"));
        assert_eq!(
            path,
            Some(PathBuf::from("/tmp/test.yaml")),
            "explicit path should be returned as-is"
        );
    }

    #[test]
    fn resolve_config_path_none_no_file() {
        let path = resolve_config_path(None);
        if !std::path::Path::new("praxis.yaml").exists() {
            assert!(path.is_none(), "should return None when praxis.yaml does not exist");
        }
    }

    // -----------------------------------------------------------------------
    // insecure_warn
    // -----------------------------------------------------------------------

    #[test]
    fn insecure_warn_inactive_does_not_panic() {
        insecure_warn(false, "test_option: this should not panic");
    }

    #[test]
    fn insecure_warn_active_does_not_panic() {
        insecure_warn(true, "test_option: active warning");
    }

    // -----------------------------------------------------------------------
    // init_runtime_limits
    // -----------------------------------------------------------------------

    #[test]
    fn init_runtime_limits_no_limits_does_not_panic() {
        let runtime = praxis_core::config::RuntimeConfig::default();
        init_runtime_limits(&runtime);
    }

    #[test]
    fn init_runtime_limits_with_memory_does_not_panic() {
        let runtime = praxis_core::config::RuntimeConfig {
            max_memory_bytes: Some(1_073_741_824),
            ..Default::default()
        };
        init_runtime_limits(&runtime);
    }

    // -----------------------------------------------------------------------
    // warn_insecure_key_permissions (Unix)
    // -----------------------------------------------------------------------

    #[cfg(unix)]
    #[test]
    fn key_permissions_restrictive_no_warning() {
        use std::os::unix::fs::PermissionsExt as _;

        let dir = tempfile::TempDir::new().expect("tempdir");
        let key_path = dir.path().join("key.pem");
        let cert_path = dir.path().join("cert.pem");
        std::fs::write(&key_path, "fake-key").expect("write key");
        std::fs::write(&cert_path, "fake-cert").expect("write cert");
        std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600)).expect("chmod");

        let config = config_with_tls(cert_path.to_str().expect("cert"), key_path.to_str().expect("key"));
        warn_insecure_key_permissions(&config);
    }

    #[cfg(unix)]
    #[test]
    fn key_permissions_permissive_does_not_panic() {
        use std::os::unix::fs::PermissionsExt as _;

        let dir = tempfile::TempDir::new().expect("tempdir");
        let key_path = dir.path().join("key.pem");
        let cert_path = dir.path().join("cert.pem");
        std::fs::write(&key_path, "fake-key").expect("write key");
        std::fs::write(&cert_path, "fake-cert").expect("write cert");
        std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o644)).expect("chmod");

        let config = config_with_tls(cert_path.to_str().expect("cert"), key_path.to_str().expect("key"));
        warn_insecure_key_permissions(&config);
    }

    #[cfg(unix)]
    #[test]
    fn key_permissions_missing_file_does_not_panic() {
        let config = config_with_tls("/nonexistent/cert.pem", "/nonexistent/key.pem");
        warn_insecure_key_permissions(&config);
    }

    // -----------------------------------------------------------------------
    // Test Utilities
    // -----------------------------------------------------------------------

    #[cfg(unix)]
    fn config_with_tls(cert_path: &str, key_path: &str) -> Config {
        let yaml = format!(
            r#"
listeners:
  - name: tls
    address: "127.0.0.1:8443"
    filter_chains: [main]
    tls:
      certificates:
        - cert_path: "{cert_path}"
          key_path: "{key_path}"
          server_names: ["localhost"]
filter_chains:
  - name: main
    filters:
      - filter: router
        routes:
          - path_prefix: "/"
            cluster: backend
      - filter: load_balancer
        clusters:
          - name: backend
            endpoints:
              - "127.0.0.1:3000"
"#
        );
        Config::from_yaml(&yaml).expect("test config should parse")
    }
}