liminal-server 0.24.5

//! Installs one identified authority pair before the listener can admit.

use super::{
    Arc, GrantStreamConfig, LysConfig, StreamState, as_config, binding, config_error, engine, http,
};
use crate::auth_revocation::{GrantLogId, RevocationAuthority, RevocationPolicy};

pub(super) struct Installation {
    pub state: Arc<StreamState>,
    pub engine: engine::Engine,
    pub http: http::Http,
    pub root: Arc<crate::auth_lys::root_presence::RootPresence>,
}

pub(super) fn prepare(
    lys: &LysConfig,
    config: &GrantStreamConfig,
    supervisor: &crate::server::ConnectionSupervisor,
    health: &crate::health::HealthServerHandle,
    handle: tokio::runtime::Handle,
) -> Result<Installation, crate::ServerError> {
    config.validate()?;
    let (gate, state) = install_membership(config, supervisor)?;
    let credential =
        http::read_file(&config.credential_path, config.max_credential_bytes).map_err(as_config)?;
    let service_key = service_key(config)?;
    let bindings = binding::BindingSource::new(
        lys,
        config.clone(),
        Arc::clone(&state),
        credential.clone(),
        handle.clone(),
    )
    .map_err(as_config)?;
    let policy = RevocationPolicy {
        log: GrantLogId::new(&config.log_identity, config.log_epoch)
            .map_err(|_| config_error("grant_stream_log_invalid"))?,
        audience: "liminal".into(),
        issuer: bindings.identity(),
    };
    let authority = Arc::new(RevocationAuthority::new(
        policy,
        Arc::new(bindings),
        state.clone(),
    ));
    let enforcing = if config.enforce_revocation {
        supervisor.install_revocation(Arc::clone(&authority))?;
        Some(supervisor.clone())
    } else {
        None
    };
    let api = lys.api_url.as_deref().unwrap_or(&lys.issuer);
    let root = Arc::new(
        crate::auth_lys::root_presence::RootPresence::new(
            &lys.network_id,
            api,
            (*credential).clone(),
            handle,
        )
        .map_err(|_| config_error("lys_root_question_invalid"))?,
    );
    let http = http::Http::new(
        api,
        credential,
        config.max_response_bytes,
        state.cancellation(),
    )
    .map_err(as_config)?;
    let engine = engine::Engine::new(
        Arc::clone(&state),
        authority,
        config.clone(),
        service_key,
        gate,
        enforcing,
    );
    health.install_grant_stream(Arc::clone(&state))?;
    health.install_lys_root(Arc::clone(&root))?;
    Ok(Installation {
        state,
        engine,
        http,
        root,
    })
}

fn install_membership(
    config: &GrantStreamConfig,
    supervisor: &crate::server::ConnectionSupervisor,
) -> Result<
    (
        Option<Arc<crate::auth_membership::MembershipGate>>,
        Arc<StreamState>,
    ),
    crate::ServerError,
> {
    let gate = supervisor.membership_gate();
    if gate.is_none() && !config.enforce_revocation {
        return Err(config_error(
            "auth.lys.grant_stream requires membership or enforce_revocation",
        ));
    }
    let log = lys_pass::membership::GrantLog {
        identity: config.log_identity.clone(),
        epoch: config.log_epoch,
    };
    let state = match &gate {
        Some(gate) => {
            if gate.served_log() != &log {
                return Err(config_error(
                    "auth.lys.grant_stream and membership name different logs",
                ));
            }
            gate.install_stream_authority()?
        }
        None => StreamState::for_log(log),
    };
    if state
        .reader_claim
        .compare_exchange(
            false,
            true,
            std::sync::atomic::Ordering::AcqRel,
            std::sync::atomic::Ordering::Acquire,
        )
        .is_err()
    {
        return Err(config_error("grant_stream_owner_already_installed"));
    }
    Ok((gate, state))
}

fn service_key(config: &GrantStreamConfig) -> Result<[u8; 32], crate::ServerError> {
    let service_key_text =
        http::read_file(&config.service_key_path, config.max_keyset_bytes).map_err(as_config)?;
    if service_key_text.len() != 64
        || service_key_text
            .bytes()
            .any(|byte| !byte.is_ascii_digit() && !(b'a'..=b'f').contains(&byte))
    {
        return Err(config_error("grant_stream_service_key_invalid"));
    }
    let service_key: [u8; 32] = hex::decode(service_key_text.as_str())
        .map_err(|_| config_error("grant_stream_service_key_invalid"))?
        .try_into()
        .map_err(|_| config_error("grant_stream_service_key_invalid"))?;
    let mut y = service_key;
    y[31] &= 0x7f;
    let mut prime = [0xff; 32];
    prime[0] = 0xed;
    prime[31] = 0x7f;
    let canonical = y.iter().rev().cmp(prime.iter().rev()).is_lt();
    if !canonical
        || !ed25519_dalek::VerifyingKey::from_bytes(&service_key).is_ok_and(|key| !key.is_weak())
    {
        return Err(config_error("grant_stream_service_key_invalid"));
    }
    Ok(service_key)
}