liminal_server/config/

validation.rs

use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};

use crate::ServerError;

use super::types::{LoadedSchema, ServerConfig, ServiceProfile};

mod links;
mod profiles;
mod websocket;
use links::validate_outbound_links;
use profiles::validate_services;
pub(crate) use profiles::worker_front_door_field_errors;
use websocket::validate_websocket;

#[cfg(test)]
mod link_export_tests;

/// Validates a fully loaded server configuration before startup.
///
/// Validation is intentionally limited to deterministic semantic checks and
/// filesystem inspection. It does not bind sockets, connect to peers, or perform
/// any other network I/O. Beyond the semantic checks it also resolves and loads
/// each channel's `schema_ref` from disk (relative to `base_dir`, or verbatim for
/// absolute paths), parses the JSON Schema document, and stores it on the channel
/// so the channel can later be built with a real schema. A missing, unreadable, or
/// non-JSON schema file is an accumulated validation error like any other.
/// Link exports refuse missing placed subjects, non-durable channels, and the reserved `liminal.` namespace by link and prefix before startup.
///
/// `base_dir` is the directory the config file was loaded from; relative
/// `schema_ref` paths resolve against it. When it is `None` (e.g. a config
/// assembled in memory), relative paths resolve against the process working
/// directory, so callers that construct a config directly should use absolute
/// `schema_ref` paths.
///
/// # Errors
///
/// Returns [`ServerError::ConfigValidation`] containing all discovered validation
/// errors when the configuration is not safe to use for startup.
pub fn validate(config: &mut ServerConfig, base_dir: Option<&Path>) -> Result<(), ServerError> {
    let mut errors = Vec::new();

    validate_listen_address(config, &mut errors);
    validate_health_listen_address(config, &mut errors);
    // NO `validate_drain_timeout`. `drain_timeout_ms` has been ignored since
    // 0.14.3 (see `ServerConfig::drain_timeout_ms`), and a refusal is an effect:
    // a field that changes nothing must not be able to stop a startup.
    validate_channels(config, &mut errors);
    validate_routing_rules(config, &mut errors);
    validate_persistence_path(config, &mut errors);
    validate_cluster(config, &mut errors);
    validate_auth(config, &mut errors);
    if let Some(lys) = config.auth.as_ref().and_then(|auth| auth.lys.as_ref()) {
        crate::services::broker::validate_configuration(lys, &mut errors);
        crate::link_pass::validate_configuration(lys, &mut errors);
    }
    validate_outbound_links(config, &mut errors);
    super::templates::validate_templates(config, &mut errors);
    validate_services(config, &mut errors);
    validate_websocket(config, &mut errors);
    config.limits.collect_errors(&mut errors);
    validate_participant(config, &mut errors);
    load_channel_schemas(config, base_dir, &mut errors);

    if errors.is_empty() {
        Ok(())
    } else {
        Err(ServerError::ConfigValidation {
            message: errors.join("\n"),
        })
    }
}

/// Resolves, reads, and parses each channel's `schema_ref`, storing the loaded
/// document on the channel. Follows the same deterministic-local-FS discipline as
/// [`validate_persistence_path`]: every failure is accumulated rather than
/// short-circuiting, so an operator sees all schema problems at once.
fn load_channel_schemas(
    config: &mut ServerConfig,
    base_dir: Option<&Path>,
    errors: &mut Vec<String>,
) {
    for channel in &mut config.channels {
        let Some(schema_ref) = channel.schema_ref.as_ref() else {
            continue;
        };
        let path = resolve_schema_path(schema_ref, base_dir);
        match load_schema_document(&path) {
            Ok(loaded) => channel.loaded_schema = Some(loaded),
            Err(reason) => errors.push(format!(
                "channels.schema_ref '{}': {reason}",
                schema_ref.display()
            )),
        }
    }
}

/// Resolves a `schema_ref` to a concrete path: absolute refs are used verbatim,
/// relative refs are joined onto `base_dir` (or the working directory when there
/// is no base directory).
fn resolve_schema_path(schema_ref: &Path, base_dir: Option<&Path>) -> PathBuf {
    // `Path::join` returns `schema_ref` unchanged when it is absolute, so the
    // base-dir arm covers both the absolute and relative cases.
    base_dir.map_or_else(|| schema_ref.to_path_buf(), |dir| dir.join(schema_ref))
}

/// Reads, JSON-parses, and schema-compiles a schema file, returning the loaded
/// document or a human-readable reason on failure (missing/unreadable file,
/// invalid JSON, or valid JSON that is not a compilable JSON Schema). The
/// compile check runs here so every schema problem surfaces in the accumulated
/// validation pass instead of deferring to a different error class at channel
/// construction.
fn load_schema_document(path: &Path) -> Result<LoadedSchema, String> {
    let bytes = std::fs::read(path)
        .map_err(|error| format!("schema file '{}' is unreadable: {error}", path.display()))?;
    let document: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| {
        format!(
            "schema file '{}' is not valid JSON: {error}",
            path.display()
        )
    })?;
    liminal::channel::Schema::new(document.clone()).map_err(|error| {
        format!(
            "schema file '{}' is not a valid JSON Schema: {error}",
            path.display()
        )
    })?;
    Ok(LoadedSchema { bytes, document })
}

fn validate_listen_address(config: &ServerConfig, errors: &mut Vec<String>) {
    if config.listen_address.port() == 0 {
        errors.push("listen_address: port must be non-zero".to_owned());
    }
}

fn validate_health_listen_address(config: &ServerConfig, errors: &mut Vec<String>) {
    if config.health_listen_address.port() == 0 {
        errors.push("health_listen_address: port must be non-zero".to_owned());
    }

    if config.health_listen_address == config.listen_address {
        errors.push(
            "health_listen_address: must differ from listen_address for probe isolation".to_owned(),
        );
    } else if config.health_listen_address.port() == config.listen_address.port() {
        errors.push(
            "health_listen_address: port must differ from listen_address port for probe isolation"
                .to_owned(),
        );
    }
}

fn validate_channels(config: &ServerConfig, errors: &mut Vec<String>) {
    let mut seen = BTreeSet::new();
    let mut duplicates = BTreeSet::new();

    for channel in &config.channels {
        let name = channel.name.trim();
        if name.is_empty() {
            errors.push("channels.name: channel name must not be empty".to_owned());
            continue;
        }

        if !seen.insert(name.to_owned()) {
            duplicates.insert(name.to_owned());
        }

        // SRV-012 R4: the console's channels are ephemeral and server-owned;
        // configured durable they would write the store.
        if channel.durable && crate::server::connection::is_console_subject(name) {
            errors.push(format!(
                "channels.durable: console channel '{name}' must not be durable"
            ));
        }
    }

    if !duplicates.is_empty() {
        let names = duplicates.into_iter().collect::<Vec<_>>().join(", ");
        errors.push(format!("channels.name: duplicate channel names: {names}"));
    }
}

fn validate_routing_rules(config: &ServerConfig, errors: &mut Vec<String>) {
    let channel_names = config
        .channels
        .iter()
        .map(|channel| channel.name.as_str())
        .collect::<BTreeSet<_>>();

    for (index, rule) in config.routing_rules.iter().enumerate() {
        let source = rule.source_channel.trim();
        if source.is_empty() {
            errors.push(format!(
                "routing_rules[{index}].source_channel: source channel must not be empty"
            ));
        } else if !channel_names.contains(source) {
            errors.push(format!(
                "routing_rules[{index}].source_channel: unknown channel '{source}'"
            ));
        }

        let target = rule.target_channel.trim();
        if target.is_empty() {
            errors.push(format!(
                "routing_rules[{index}].target_channel: target channel must not be empty"
            ));
        } else if !channel_names.contains(target) {
            errors.push(format!(
                "routing_rules[{index}].target_channel: unknown channel '{target}'"
            ));
        }
    }
}

pub(crate) const MISSING_PERSISTENCE_PATH: &str = "persistence_path: full services require an existing directory; set persistence_path or LIMINAL_PERSISTENCE_PATH; no store is opened";

pub(crate) fn required_persistence_path(config: &ServerConfig) -> Result<&Path, ServerError> {
    config
        .persistence_path
        .as_deref()
        .ok_or_else(|| ServerError::ConfigValidation {
            message: MISSING_PERSISTENCE_PATH.to_owned(),
        })
}

fn validate_persistence_path(config: &ServerConfig, errors: &mut Vec<String>) {
    let Some(path) = config.persistence_path.as_deref() else {
        if matches!(config.services.profile(), Ok(ServiceProfile::Full)) {
            errors.push(MISSING_PERSISTENCE_PATH.to_owned());
        }
        return;
    };

    match std::fs::metadata(path) {
        Ok(metadata) => {
            if !metadata.is_dir() {
                errors.push(format!(
                    "persistence_path '{}': path must be an existing directory",
                    path.display()
                ));
            } else if metadata.permissions().readonly() {
                errors.push(format!(
                    "persistence_path '{}': path is not writable",
                    path.display()
                ));
            }
        }
        Err(error) => {
            errors.push(format!(
                "persistence_path '{}': path is unreachable: {error}",
                path.display()
            ));
        }
    }
}

fn validate_cluster(config: &ServerConfig, errors: &mut Vec<String>) {
    let Some(cluster) = config.cluster.as_ref() else {
        return;
    };

    if cluster.node_name.trim().is_empty() {
        errors.push("cluster.node_name: node name must not be empty".to_owned());
    }

    if cluster.cookie.is_empty() {
        errors.push("cluster.cookie: distribution cookie must not be empty".to_owned());
    } else if cluster.cookie == super::types::DEFAULT_COOKIE {
        errors.push(
            "cluster.cookie: secret cookie must be set; the public default is refused".to_owned(),
        );
    }

    if cluster.listen_address.port() == 0 {
        errors.push("cluster.listen_address: distribution port must be non-zero".to_owned());
    }

    if cluster.listen_address == config.listen_address {
        errors.push(
            "cluster.listen_address: distribution port must differ from the client listen_address"
                .to_owned(),
        );
    }

    let mut seed_node_counts = BTreeMap::new();
    for (index, seed_node) in cluster.seed_nodes.iter().enumerate() {
        if seed_node.port() == 0 {
            errors.push(format!(
                "cluster.seed_nodes[{index}]: seed node port must be non-zero"
            ));
        }
        seed_node_counts
            .entry(seed_node.to_string())
            .and_modify(|count| *count += 1)
            .or_insert(1_usize);
    }

    let duplicates = seed_node_counts
        .into_iter()
        .filter_map(|(seed_node, count)| (count > 1).then_some(seed_node))
        .collect::<Vec<_>>();

    if !duplicates.is_empty() {
        errors.push(format!(
            "cluster.seed_nodes: duplicate seed nodes: {}",
            duplicates.join(", ")
        ));
    }
}

/// Validates the optional `[auth]` section. When present its token must be
/// non-empty: an empty token would gate nothing (every client's empty `auth_token`
/// would match), so it is rejected rather than silently leaving the server open.
/// The token is not trimmed — a shared secret may legitimately contain leading or
/// trailing whitespace.
fn validate_auth(config: &ServerConfig, errors: &mut Vec<String>) {
    let Some(auth) = config.auth.as_ref() else {
        return;
    };

    if auth.token.is_empty() {
        errors.push("auth.token: authentication token must not be empty".to_owned());
    }
    // A token longer than a pre-admission `Connect` can carry could never be
    // presented: every client would be refused before its token was compared.
    let max = crate::server::connection::pre_admission::MAX_CONNECT_AUTH_TOKEN_BYTES;
    if auth.token.len() > max {
        errors.push(format!(
            "auth.token: authentication token is {} bytes; a Connect can carry at most {max}",
            auth.token.len()
        ));
    }
    // ACCESS-004 R3: only a pass proves a holder, so requiring one without a
    // pass verifier would refuse every worker.
    if auth.require_verified_worker && auth.pass.is_none() {
        errors.push(
            "auth.require_verified_worker: requires [auth.pass], or no worker could register"
                .to_owned(),
        );
    }
}

/// Semantic checks for the optional `[participant]` section: the shared
/// nonzero/ordering rules plus the protocol codec's own minimum-frame check on
/// `wire_frame_limit`, so an impossible limit fails at validation rather than
/// at service construction.
fn validate_participant(config: &ServerConfig, errors: &mut Vec<String>) {
    let Some(participant) = config.participant.as_ref() else {
        return;
    };
    participant.collect_errors(errors);
    if participant.wire_frame_limit != 0
        && let Err(error) = crate::server::participant::normalize_configured_frame_limit(
            participant.wire_frame_limit,
        )
    {
        errors.push(format!(
            "participant.wire_frame_limit: {} is below the protocol's minimum complete \
             participant frame ({error:?})",
            participant.wire_frame_limit
        ));
    }
}

#[cfg(test)]
mod self_stream_tests;
#[cfg(test)]
mod tests;