liminal_server/

link_pass.rs

//! The agent's own pass, signed for the issuer's agent route at boot.
//!
//! Crossings read the verified rights offline. An open renews a due agent
//! pass and judges the renewed rights and lifetime on that same open.
use crate::ServerError;
use crate::auth_lys::lys_pass_adapter::{LysPassAdapter, MachineLinkRights};
use crate::auth_lys::session::{ConsumerAdmitter, LysAdmissionRefusal, LysAdmitter, LysSession};
use crate::auth_lys::{
    AccessAction, AccessDecisionError, AccessRefusal, AccessResource, DeliberateDecision,
    GrantReceipt, HotAction, LysPassConsumer, RefusalCause, SubjectBinding,
};
use crate::config::types::{LysConfig, MachineConfig};
use crate::link::{LinkAction, LinkHost, LinkRights};
use crate::link_handshake::{LocalMachine, MachineIdentity};
use lys_core::keys::Ed25519Identity;
use std::net::IpAddr;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use zeroize::Zeroizing;
mod agent_dial;
pub(crate) mod cose;
pub(crate) mod debug;
mod holder;
mod peer_security;
mod start_issue;
use agent_dial::{Dial, Renewal};
pub(crate) use holder::MachinePass;
pub(crate) use start_issue::StartIssue;

type GrantAnswer = Result<GrantReceipt, AccessDecisionError>;
type LinkAnswer = Result<DeliberateDecision, AccessDecisionError>;
static REFUSALS: AtomicU64 = AtomicU64::new(0);
#[derive(Debug, thiserror::Error)]
#[error("{0}")]
pub(crate) struct PassRefusal(&'static str);
impl PassRefusal {
    pub(crate) fn source_absent(&self) -> bool {
        self.0 == "machine_pass_not_configured"
    }
}
fn named(name: &str) -> PassRefusal {
    // Only contract names enter logs and metric labels, never response words.
    const NAMES: &str = "agent_dial_refused agent_signature_refused HolderRetired ProviderUnavailable \
        ClientUnknown ScopeUnknown ScopeNotGranted PassRefused RequestMalformed signature_refused \
        wrong_audience wrong_issuer pass_expired pass_not_yet_valid unpublished_key \
        pass_encoding_refused contract_refused rights_outside_audience rights_truncated endpoint_refused";
    let name = match name {
        "AgentSignatureRefused" => "agent_signature_refused",
        "agent_pass_unproven" | "agent_not_holder" | "audience_not_approved" => {
            "agent_dial_refused"
        }
        "agent_retired" => "HolderRetired",
        "agent_pass_unavailable" => "ProviderUnavailable",
        name => name,
    };
    let known = NAMES.split_ascii_whitespace().find(|known| *known == name);
    PassRefusal(known.unwrap_or("machine_pass_refused"))
}
fn record_refusal(error: &PassRefusal, phase: &'static str) {
    let count = REFUSALS.fetch_add(1, Ordering::Relaxed) + 1;
    // Open refusals are named once by their connection or retry owner.
    if phase == "boot" {
        tracing::warn!(
            reason = error.0,
            phase,
            count,
            "machine pass refused; asked again at the next link open"
        );
    }
    if let Some(registry) = liminal::metrics::global_registry() {
        match registry
            .register_counter("liminal_machine_pass_refusals_total", [("reason", error.0)])
        {
            Ok(counter) => counter.increment(),
            Err(error) => {
                tracing::error!(%error, "machine pass refusal counter registration failed");
            }
        }
    }
}
fn identifier(value: &str) -> bool {
    !value.is_empty() && !value.chars().any(|c| c.is_whitespace() || c.is_control())
}
fn api_url(value: &str) -> Result<reqwest::Url, PassRefusal> {
    let mut url = reqwest::Url::parse(value).map_err(|_| PassRefusal("endpoint_refused"))?;
    let host = url.host_str().ok_or(PassRefusal("endpoint_refused"))?;
    let loopback = host == "localhost"
        || host
            .trim_matches(['[', ']'])
            .parse::<IpAddr>()
            .is_ok_and(|ip| ip.is_loopback());
    if !matches!(url.scheme(), "http" | "https")
        || !url.username().is_empty()
        || url.password().is_some()
        || url.query().is_some()
        || url.fragment().is_some()
        || (url.scheme() == "http" && !loopback)
    {
        return Err(PassRefusal("endpoint_refused"));
    }
    if !url.path().ends_with('/') {
        url.set_path(&format!("{}/", url.path()));
    }
    Ok(url)
}
/// Collects errors for the two mutually exclusive agent pass sources.
pub(crate) fn validate_configuration(lys: &LysConfig, errors: &mut Vec<String>) {
    let Some(machine) = &lys.machine else {
        return;
    };
    let file = !machine.pass_path.as_os_str().is_empty();
    let dial = !machine.agent_id.is_empty() || !machine.audience.is_empty();
    if file && dial {
        errors.push(
            "auth.lys.machine: pass_path and agent_id/audience are mutually exclusive".into(),
        );
    } else if !file && !dial {
        errors.push("auth.lys.machine: configure pass_path or agent_id and audience".into());
    }
    if dial {
        if !identifier(&machine.agent_id)
            || !machine
                .agent_id
                .bytes()
                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
        {
            errors.push("auth.lys.machine.agent_id: expected one nonempty route segment".into());
        }
        if !identifier(&machine.audience) {
            errors.push("auth.lys.machine.audience: expected a nonempty audience".into());
        }
        if lys
            .api_url
            .as_deref()
            .is_none_or(|url| api_url(url).is_err())
        {
            errors.push("auth.lys.api_url: agent_id requires HTTPS or loopback HTTP".into());
        }
    }
}

#[derive(Debug)]
struct Current {
    rights: MachineLinkRights,
    expiry: u64,
    renew_at: u64,
    renewed: bool,
}
struct Authority {
    /// The pass held and its rights; `None` after a boot whose ask Lys
    /// refused, until the next link open asks (ACCESS-009 R1).
    current: Mutex<Option<Current>>,
    pass: MachinePass,
    dial: Option<Box<dyn Renewal>>,
    adapter: Arc<LysPassAdapter>,
    /// The boot left its ask for after the bind: whichever ask comes first
    /// is the boot's (SRV-041 R4).
    boot_pending: AtomicBool,
}
fn verified(adapter: &LysPassAdapter, token: &[u8], now: u64) -> Result<Current, PassRefusal> {
    let observed = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_err(|_| PassRefusal("machine_pass_clock_refused"))?
        .as_secs();
    let now = now.max(observed);
    let pass = adapter
        .verify_offline(token, crate::auth_lys::LIMINAL_PASS_AUDIENCE, now)
        .map_err(|error| named(error.name()))?;
    if pass.verified().claims().holder.kind != "agent" {
        return Err(PassRefusal("agent_holder_refused"));
    }
    let expiry = pass.verified().claims().exp;
    Ok(Current {
        rights: MachineLinkRights::new(adapter, pass),
        expiry,
        renew_at: renewal_at(expiry, now),
        renewed: false,
    })
}
impl Authority {
    fn refusal(
        &self,
        resource: &AccessResource,
        action: AccessAction,
        cause: RefusalCause,
    ) -> AccessDecisionError {
        AccessDecisionError {
            cause,
            refusal: AccessRefusal::new(resource.clone(), action, self.adapter.request_url()),
        }
    }
    fn open(
        &self,
        link: &AccessResource,
        end: LinkAction,
        now: u64,
    ) -> Result<(), AccessDecisionError> {
        let action = end.other();
        self.refresh_expired(now).map_err(|error| {
            let cause = if error.0 == "pass_expired" {
                RefusalCause::Expired
            } else {
                RefusalCause::DeliberateUnavailable
            };
            self.refusal(link, action.access_action(), cause)
        })?;
        let unavailable = || {
            self.refusal(
                link,
                action.access_action(),
                RefusalCause::DeliberateUnavailable,
            )
        };
        self.current
            .lock()
            .map_err(|_| unavailable())?
            .as_ref()
            .map_or_else(
                || Err(unavailable()),
                |current| {
                    current
                        .rights
                        .holds_resource(action, link, now)
                        .map(|_| ())
                        .map_err(|mut error| {
                            if current.renewed && error.cause == RefusalCause::MissingGrant {
                                error.cause = RefusalCause::RenewedWithoutGrant;
                            }
                            error
                        })
                },
            )
    }
    fn refresh_expired(&self, now: u64) -> Result<(), PassRefusal> {
        self.refresh(now, "link_open")
    }
    /// Asks for the pass when none is held or it is due; `phase` names a
    /// refusal as a link open's or as the boot's (SRV-041 R4).
    fn refresh(&self, now: u64, phase: &'static str) -> Result<(), PassRefusal> {
        let mut current = self
            .current
            .lock()
            .map_err(|_| PassRefusal("pass_holder_refused"))?;
        let due = current
            .as_ref()
            .is_none_or(|held| held.expiry <= now || (self.dial.is_some() && held.renew_at <= now));
        if !due {
            return Ok(());
        }
        let dial = self.dial.as_ref().ok_or(PassRefusal("pass_expired"))?;
        let boot = self.boot_pending.swap(false, Ordering::AcqRel);
        let phase = if boot { "boot" } else { phase };
        // Serialize renewals so rights and the presented token change together.
        let (token, mut renewed) = dial.renew(&self.adapter, now).inspect_err(|error| {
            record_refusal(error, phase);
        })?;
        if renewed.expiry <= now {
            return Err(PassRefusal("pass_expired"));
        }
        let mut held = self
            .pass
            .0
            .lock()
            .map_err(|_| PassRefusal("pass_holder_refused"))?;
        // A first issue after a refused boot is not a renewal.
        renewed.renewed = current.is_some();
        held.token = token;
        *current = Some(renewed);
        drop(held);
        drop(current);
        Ok(())
    }
}
#[derive(Debug)]
struct Rights(Arc<Authority>);
impl LinkRights for Rights {
    fn holds(&self, action: LinkAction, subject: &SubjectBinding, now: u64) -> GrantAnswer {
        let unavailable = || {
            self.0.refusal(
                subject.resource(),
                action.access_action(),
                RefusalCause::DeliberateUnavailable,
            )
        };
        self.0
            .current
            .lock()
            .map_err(|_| unavailable())?
            .as_ref()
            .map_or_else(
                || Err(unavailable()),
                |current| current.rights.holds(action, subject, now),
            )
    }
}
#[derive(Debug)]
struct Admitter {
    inner: Arc<dyn LysAdmitter>,
    authority: Arc<Authority>,
}
impl LysAdmitter for Admitter {
    fn admit(&self, pass: &[u8], now: u64) -> Result<Box<dyn LysSession>, LysAdmissionRefusal> {
        Ok(Box::new(Session {
            inner: self.inner.admit(pass, now)?,
            authority: Arc::clone(&self.authority),
        }))
    }
}
#[derive(Debug)]
struct Session {
    inner: Box<dyn LysSession>,
    authority: Arc<Authority>,
}
impl LysSession for Session {
    // The link authority wraps a session only to guard link opens; every prepared hold is the
    // inner session's, or permissions and construction would fall to the trait's refusals.
    fn hold_permission(
        &self,
        request: crate::auth_lys::prepared::PermissionRequest<'_>,
    ) -> Result<serde_json::Value, lys_pass::Error> {
        self.inner.hold_permission(request)
    }
    fn hold_site(
        &self,
        request: crate::auth_lys::prepared::SiteRequest<'_>,
    ) -> Result<serde_json::Value, lys_pass::Error> {
        self.inner.hold_site(request)
    }
    fn holder(&self) -> &str {
        self.inner.holder()
    }
    fn holder_kind(&self) -> Option<&str> {
        self.inner.holder_kind()
    }
    fn issued_at(&self) -> Option<u64> {
        self.inner.issued_at()
    }
    fn decide(&self, subject: &SubjectBinding, action: HotAction, now: u64) -> GrantAnswer {
        self.inner.decide(subject, action, now)
    }
    fn decide_link(&self, subject: &SubjectBinding, action: LinkAction, now: u64) -> GrantAnswer {
        self.inner.decide_link(subject, action, now)
    }
    fn open_link(&self, link: &AccessResource, end: LinkAction, now: u64) -> LinkAnswer {
        self.authority.open(link, end, now)?;
        self.inner.open_link(link, end, now)
    }
    fn decide_live(&self, target: &AccessResource, action: AccessAction, now: u64) -> LinkAnswer {
        self.inner.decide_live(target, action, now)
    }
}
pub(crate) struct PassInstall {
    pub host: Option<Arc<LinkHost>>,
    pub admitter: Arc<dyn LysAdmitter>,
    /// The first pass issue, left for after the listener binds (SRV-041 R4).
    pub issue: Option<StartIssue>,
}
/// Verifies the configured pass before creating any link host. With
/// `after_bind`, a dialled pass that nothing built before the bind reads is
/// not asked for here: [`PassInstall::issue`] asks once the listener binds.
pub(crate) fn configured(
    lys: &LysConfig,
    adapter: Arc<LysPassAdapter>,
    handle: Option<tokio::runtime::Handle>,
    now: u64,
    after_bind: bool,
) -> Result<PassInstall, ServerError> {
    let mut install = PassInstall {
        host: None,
        admitter: Arc::new(ConsumerAdmitter::new(Arc::clone(&adapter))),
        issue: None,
    };
    let Some(machine) = &lys.machine else {
        return Ok(install);
    };
    let mut errors = Vec::new();
    validate_configuration(lys, &mut errors);
    if !errors.is_empty() {
        return Err(ServerError::ConfigValidation {
            message: errors.join("; "),
        });
    }
    let defer = after_bind && start_issue::defers(lys, machine);
    match build(lys, machine, adapter, handle, now, defer) {
        Ok((host, authority, refused)) => {
            if let Some(error) = &refused {
                record_refusal(error, "boot");
            }
            install.issue = defer.then(|| StartIssue::new(Arc::clone(&authority), now));
            install.host = Some(Arc::new(host));
            install.admitter = Arc::new(Admitter {
                inner: install.admitter,
                authority,
            });
        }
        Err(error) if !machine.agent_id.is_empty() && machine.carrier_choice.peers.is_none() => {
            record_refusal(&error, "boot");
        }
        Err(error) => {
            return Err(ServerError::ConfigValidation {
                message: format!("auth.lys.machine: {error}"),
            });
        }
    }
    Ok(install)
}
fn hex_key(digits: &str) -> Result<Zeroizing<[u8; 32]>, PassRefusal> {
    let mut key = Zeroizing::new([0; 32]);
    hex::decode_to_slice(digits.trim(), &mut *key)
        .map_err(|_| PassRefusal("machine_key_refused"))?;
    Ok(key)
}
fn build(
    lys: &LysConfig,
    machine: &MachineConfig,
    adapter: Arc<LysPassAdapter>,
    handle: Option<tokio::runtime::Handle>,
    now: u64,
    defer: bool,
) -> Result<(LinkHost, Arc<Authority>, Option<PassRefusal>), PassRefusal> {
    let digits = Zeroizing::new(
        std::fs::read_to_string(&machine.identity_path)
            .map_err(|_| PassRefusal("machine_identity_refused"))?,
    );
    let seed = hex_key(&digits)?;
    let local = LocalMachine::new(Ed25519Identity::from_seed(&seed));
    let peers = peer_security::prepare(lys, machine, &adapter, handle.as_ref())?;
    let dial = if machine.agent_id.is_empty() {
        None
    } else {
        Some(Dial::new(
            lys.api_url
                .as_deref()
                .ok_or(PassRefusal("endpoint_refused"))?,
            machine,
            Ed25519Identity::from_seed(&seed),
            handle.ok_or(PassRefusal("lys_ask_runtime_absent"))?,
        )?)
    };
    let (token, current, refused) = if defer {
        (None, None, None)
    } else if let Some(dial) = &dial {
        let asked = dial.ask().and_then(|issued| {
            let current = dial.verified(&adapter, &issued, now)?;
            Ok((issued.token, current))
        });
        match asked {
            Ok((token, current)) => (Some(token), Some(current), None),
            // The host is built without a pass and the next link open asks
            // again (ACCESS-009 R1); a machine with a peers block cannot
            // serve without its pass, so its refusal stays a refusal.
            Err(error) if machine.carrier_choice.peers.is_none() => (None, None, Some(error)),
            Err(error) => return Err(error),
        }
    } else {
        let token = Zeroizing::new(
            std::fs::read_to_string(&machine.pass_path)
                .map_err(|_| PassRefusal("machine_pass_path_refused"))?,
        );
        let current = verified(&adapter, token.as_bytes(), now)?;
        (Some(token), Some(current), None)
    };
    let pass = MachinePass::hold(machine, token)?;
    let authority = Arc::new(Authority {
        current: Mutex::new(current),
        pass,
        dial: dial.map(|dial| Box::new(dial) as Box<dyn Renewal>),
        adapter,
        boot_pending: AtomicBool::new(defer),
    });
    authority.pass.attach(&authority)?;
    let recorded = machine
        .machines
        .iter()
        .map(|(holder, key)| Ok((holder.clone(), MachineIdentity::new(*hex_key(key)?))))
        .collect::<Result<_, PassRefusal>>()?;
    let host = LinkHost::new(local, Box::new(Rights(Arc::clone(&authority))), recorded);
    let host = if let Some(peers) = peers {
        peers.install(host)?
    } else {
        host
    };
    Ok((host, authority, refused))
}

fn renewal_at(expiry: u64, now: u64) -> u64 {
    expiry.saturating_sub((expiry.saturating_sub(now) / 10).clamp(1, 30))
}

#[cfg(test)]
pub(crate) mod fixture;
#[cfg(test)]
mod open_tests;
#[cfg(test)]
mod renewal_tests;