liminal-server 0.25.4

use super::{
    Arc, AuthConfig, ConnectionNotifier, ConnectionRuntime, ConnectionServices,
    ConnectionSupervisor, LimitsConfig, PassVerifier, ServerError, SupervisorInner,
};

/// The configured `[auth]` section as the supervisor carries it: the bearer
/// token as opaque bytes for the handshake's constant-time comparison, and the
/// registry pass verifier built from `auth.pass` when one is configured. The
/// single derivation behind [`ConnectionSupervisor::from_config`] and
/// [`ConnectionSupervisorBuilder::auth`], so the two cannot drift.
pub(super) fn configured_authentication(
    auth: &AuthConfig,
) -> Result<(Vec<u8>, Option<PassVerifier>), ServerError> {
    let pass_verifier = PassVerifier::from_auth(auth)?;
    Ok((auth.token.clone().into_bytes(), pass_verifier))
}

/// Composes a [`ConnectionSupervisor`] over caller-owned services.
///
/// Any of the optional parts — `[auth]` (bearer + pass verifier), a connection
/// notifier, operational limits — can be set together. Minted by
/// [`ConnectionSupervisor::builder`].
pub struct ConnectionSupervisorBuilder {
    pub(super) services: Arc<dyn ConnectionServices>,
    pub(super) notifier: Option<Arc<dyn ConnectionNotifier>>,
    pub(super) auth_token: Option<Vec<u8>>,
    pub(super) pass_verifier: Option<PassVerifier>,
    pub(super) lys: Option<LysInstall>,
    pub(super) limits: LimitsConfig,
}

// The bearer token prints as its length, never its bytes (PROTO-007 R2).
impl std::fmt::Debug for ConnectionSupervisorBuilder {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ConnectionSupervisorBuilder")
            .field("services", &self.services)
            .field("notifier", &self.notifier)
            .field("auth_token_len", &self.auth_token.as_ref().map(Vec::len))
            .field("pass_verifier", &self.pass_verifier)
            .field("lys", &self.lys)
            .field("limits", &self.limits)
            .finish()
    }
}

/// What `[auth.lys]` installs on the connection runtime: the admitter with
/// the network id its subjects are placed under (ACCESS-001 R2), the
/// machine's link host (ACCESS-002 R1), the service bindings (ACCESS-003 R1)
/// and the channel membership gate (ACCESS-006 R1).
#[derive(Debug)]
pub struct LysInstall {
    /// Admits a Lys pass at `Connect` and decides its routes.
    pub(crate) admitter: Arc<dyn crate::auth_lys::session::LysAdmitter>,
    /// The network every subject is placed under.
    pub(crate) network_id: String,
    /// This machine's link host, `[auth.lys.machine]`.
    pub(crate) link_host: Option<Arc<crate::link::LinkHost>>,
    /// The service effect seams, `[auth.lys.services]`.
    pub(crate) services: Option<Arc<crate::services::production::ServiceBindings>>,
    /// The channel membership gate `[auth.lys.membership]` configures, over
    /// the same issuer.
    pub(crate) membership: Option<Arc<crate::auth_membership::MembershipGate>>,
}

/// The tokio runtime Lys is asked on, live deliberate decisions and the
/// service bindings' HTTP alike, owned by the server for its life; `None`
/// without `[auth.lys]`. One worker: every ask is one short request made
/// from a connection's slice through `Handle::block_on` on a scoped thread,
/// never a runtime nested in the connection scheduler.
///
/// # Errors
/// The runtime could not start, by name.
pub fn lys_ask_runtime(
    auth: Option<&AuthConfig>,
) -> Result<Option<tokio::runtime::Runtime>, ServerError> {
    if auth.is_none_or(|auth| auth.lys.is_none()) {
        return Ok(None);
    }
    tokio::runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .thread_name("liminal-lys-ask")
        .enable_all()
        .build()
        .map(Some)
        .map_err(|error| ServerError::ConfigValidation {
            message: format!("the Lys ask runtime could not start: {error}"),
        })
}

/// Builds the Lys admitter `[auth.lys]` configures, reading the issuer's key
/// set once; `None` without the section. `ask_handle` is the runtime live
/// asks run on ([`lys_ask_runtime`], or an embedder's own); `auth.lys.live`
/// without one is refused at start, by name.
///
/// # Errors
/// A key set, pass or identity that cannot be read or is refused, by name.
pub fn configured_lys(
    auth: &AuthConfig,
    ask_handle: Option<tokio::runtime::Handle>,
) -> Result<Option<LysInstall>, ServerError> {
    Ok(lys_install(auth, ask_handle, false)?.map(|(install, _)| install))
}

/// [`configured_lys`] for a server that binds before it asks: the machine
/// pass's first issue, when nothing built before the bind reads the pass,
/// is returned to be begun once the listener binds (SRV-041 R4).
///
/// # Errors
/// As [`configured_lys`].
pub fn configured_lys_after_bind(
    auth: &AuthConfig,
    ask_handle: Option<tokio::runtime::Handle>,
) -> Result<(Option<LysInstall>, Option<crate::link_pass::StartIssue>), ServerError> {
    Ok(lys_install(auth, ask_handle, true)?
        .map_or((None, None), |(install, issue)| (Some(install), issue)))
}

fn lys_install(
    auth: &AuthConfig,
    ask_handle: Option<tokio::runtime::Handle>,
    after_bind: bool,
) -> Result<Option<(LysInstall, Option<crate::link_pass::StartIssue>)>, ServerError> {
    let Some(lys) = auth.lys.as_ref() else {
        return Ok(None);
    };
    let keys_json =
        std::fs::read_to_string(&lys.keys_path).map_err(|error| ServerError::ConfigValidation {
            message: format!("auth.lys.keys_path {}: {error}", lys.keys_path.display()),
        })?;
    let mut adapter = crate::auth_lys::lys_pass_adapter::LysPassAdapter::from_keys_json(
        &keys_json,
        lys.issuer.clone(),
    )
    .map_err(|error| ServerError::ConfigValidation {
        message: format!("auth.lys key set refused: {error}"),
    })?;
    if lys.live {
        // Deliberate decisions ask the issuer live, with the holder's pass,
        // on the server's runtime; the ask never nests a runtime.
        let Some(handle) = ask_handle.as_ref() else {
            return Err(ServerError::ConfigValidation {
                message:
                    "auth.lys.live needs the server's Lys ask runtime at start; none was given"
                        .to_owned(),
            });
        };
        let mut live = crate::auth_lys::live::HttpLysAsk::new(
            lys.api_url.as_deref().unwrap_or(&lys.issuer),
            handle.clone(),
        )
        .map_err(|error| ServerError::ConfigValidation {
            message: format!("auth.lys.issuer refused for live asks: {error}"),
        })?;
        // Lys answers its batch check for an administrator or an app, never
        // a seat's pass: with an app credential configured the question is
        // asked as this liminal, about the holder.
        if let Some(path) = lys.app_credential_path.as_ref() {
            let credential =
                std::fs::read_to_string(path).map_err(|error| ServerError::ConfigValidation {
                    message: format!(
                        "auth.lys.app_credential_path '{}' cannot be read: {}",
                        path.display(),
                        error.kind()
                    ),
                })?;
            live = live.asking_as(credential.trim().to_owned());
        }
        adapter = adapter.with_live(Arc::new(live));
    }
    let adapter = Arc::new(adapter);
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_err(|_| ServerError::ConfigValidation {
            message: "machine pass clock is before the Unix epoch".to_owned(),
        })?
        .as_secs();
    let machine_pass = crate::link_pass::configured(lys, adapter, ask_handle, now, after_bind)?;
    let membership = lys
        .membership
        .as_ref()
        .map(|membership| crate::auth_membership::configured_gate(membership, &lys.issuer))
        .transpose()?;
    Ok(Some((
        LysInstall {
            admitter: machine_pass.admitter,
            network_id: lys.network_id.clone(),
            link_host: machine_pass.host,
            services: None,
            membership,
        },
        machine_pass.issue,
    )))
}

impl ConnectionSupervisorBuilder {
    /// Installs the configured `[auth]` section: the bearer token, and the
    /// registry pass verifier when `auth.pass` is present — derived exactly as
    /// [`ConnectionSupervisor::from_config`] derives them.
    ///
    /// # Errors
    /// Returns [`ServerError::ConfigValidation`] when `auth.pass` names a key
    /// that is not exactly one valid Ed25519 verifying key as 64 hexadecimal
    /// digits — the same error, with the same text, that `from_config` returns
    /// for that configuration.
    pub fn auth(mut self, auth: &AuthConfig) -> Result<Self, ServerError> {
        let (auth_token, pass_verifier) = configured_authentication(auth)?;
        self.auth_token = Some(auth_token);
        self.pass_verifier = pass_verifier;
        self.lys = configured_lys(auth, tokio::runtime::Handle::try_current().ok())?;
        Ok(self)
    }

    /// Test seam: installs `lys` exactly as `[auth.lys]` would, with no
    /// configuration file and no issuer key behind it.
    #[cfg(test)]
    #[must_use]
    pub(in crate::server::connection) fn lys_for_tests(mut self, lys: LysInstall) -> Self {
        self.lys = Some(lys);
        self
    }

    /// Installs the connection-keyed notifier, invoked on worker registration
    /// lifecycle and on pass-stamped attach/detach.
    #[must_use]
    pub fn notifier(mut self, notifier: Arc<dyn ConnectionNotifier>) -> Self {
        self.notifier = Some(notifier);
        self
    }

    /// Installs the operational limits (§5); unset, the signed defaults apply.
    #[must_use]
    pub const fn limits(mut self, limits: LimitsConfig) -> Self {
        self.limits = limits;
        self
    }

    /// Starts the supervisor.
    ///
    /// # Errors
    /// Returns [`ServerError`] when incarnation startup or scheduler startup
    /// fails.
    pub fn build(self) -> Result<ConnectionSupervisor, ServerError> {
        SupervisorInner::new(
            self.services,
            self.notifier,
            self.auth_token,
            self.pass_verifier,
            self.lys,
            self.limits,
            None,
        )
        .map(|inner| ConnectionSupervisor {
            inner: Arc::new(inner),
        })
    }
}

/// Installs the `[auth.lys]` parts on the runtime, each once: the admitter,
/// the link host, the membership gate and the service bindings.
pub(super) fn install_lys(runtime: &ConnectionRuntime, lys: LysInstall) -> Result<(), ServerError> {
    let LysInstall {
        admitter,
        network_id,
        link_host,
        services: bindings,
        membership,
    } = lys;
    runtime
        .install_lys_admitter(admitter, network_id)
        .map_err(|_| ServerError::ConfigValidation {
            message: "auth.lys admitter was already installed".to_owned(),
        })?;
    if let Some(host) = link_host {
        runtime
            .install_link_host(host)
            .map_err(|_| ServerError::ConfigValidation {
                message: "auth.lys.machine was already installed".to_owned(),
            })?;
    }
    if let Some(gate) = membership {
        runtime
            .install_membership_gate(gate)
            .map_err(|_| ServerError::ConfigValidation {
                message: "auth.lys.membership was already installed".to_owned(),
            })?;
    }
    if let Some(bindings) = bindings {
        runtime
            .install_service_bindings(bindings)
            .map_err(|_| ServerError::ConfigValidation {
                message: "auth.lys.services was already installed".to_owned(),
            })?;
    }
    Ok(())
}