liminal-server 0.25.4

//! The link route on one connection (ACCESS-002 R1).
//!
//! A machine admitted on a Lys pass offers to open one end of a link on an
//! application stream (`LinkOpen`). The server proves the offered machine
//! key is the one Lys recorded for the connection's verified holder, asks
//! Lys live for the offered end's grant on the link (never the pass alone),
//! and answers the Noise `KK` handshake keyed from the two machines' Lys
//! identities; the stream is then a keyed link. A message crossing it
//! (`LinkMessage`) is opened by the link's transport and decided from both
//! machines' passes, `link_export` on the exporting machine's and
//! `link_import` on this machine's, before it is published here; a refusal
//! names its end on the link's stream.

use liminal::protocol::{
    Frame, LinkOffer, LinkOpenOutcome, MessageEnvelope, PUBLISH_ACK_POSITION_VERSION,
};
use liminal_protocol::reason_code::PUBLISH_UNRESOLVED_CODE;

use super::apply::unix_seconds_now;
use super::services::{
    ChannelOperation, Reach, access_to_server_error, reserved_namespace_refusal,
};
use super::state::{ConnectionProcessState, FrameAction};
use super::supervisor::runtime::ConnectionRuntime;
use crate::auth_lys::session::LysSession;
use crate::auth_lys::{
    AccessDecisionError, DeliberateDecision, GrantReceipt, RefusalCause, SubjectBinding,
};
use crate::link::{Link, LinkAction, LinkLife, LinkRefusal, LinkRights};
use crate::link_handshake::{LinkHandshake, LinkTransport, NOISE_OVERHEAD};

/// The reason code a refused crossing answers with, on the link's stream.
pub(super) const LINK_CROSSING_REFUSED_CODE: u16 = 0x0203;

#[cfg(test)]
mod open_tests;
#[cfg(test)]
mod refusal_tests;

#[cfg(feature = "test-support")]
impl crate::server::ConnectionSupervisor {
    /// Installs the service adapter's production observations in link wire fixtures.
    ///
    /// # Errors
    /// Returns an error when this supervisor already has an observation owner.
    pub fn install_link_observations_for_tests(
        &self,
        services: &super::services::LiminalConnectionServices,
    ) -> Result<(), crate::ServerError> {
        self.install_signals(services.signals().clone())
    }
}

/// A link open on this connection: its description, the end the remote
/// machine took, and the transport its handshake keyed.
#[derive(Debug)]
pub(super) struct OpenLink {
    link: Link,
    remote_end: LinkAction,
    transport: LinkTransport,
    life: LinkLife,
    calls: calls::IncomingCalls,
    certificate_until: Option<u64>,
}

mod life;
pub(super) use life::ConnectionLinks;

impl OpenLink {
    /// The end the remote machine took.
    pub(super) const fn remote_end(&self) -> LinkAction {
        self.remote_end
    }
}

/// The connection's admitted pass as the remote machine's link rights.
#[derive(Debug)]
struct SessionLinkRights<'a>(&'a dyn LysSession);

impl LinkRights for SessionLinkRights<'_> {
    fn holds(
        &self,
        action: LinkAction,
        subject: &SubjectBinding,
        now: u64,
    ) -> Result<GrantReceipt, AccessDecisionError> {
        self.0.decide_link(subject, action, now)
    }
}

/// Why an open did not key a link.
enum OpenRefusal {
    Held { grant: String, draft: String },
    Refused(String),
    Pass(AccessDecisionError),
}

/// The four-field refusal as the wire carries it, or the cause's words.
fn refusal_text(error: &AccessDecisionError) -> String {
    let envelope =
        serde_json::to_string(&error.refusal).unwrap_or_else(|_| error.cause.to_string());
    if error.cause == RefusalCause::RenewedWithoutGrant {
        format!("{}:{envelope}", error.cause)
    } else {
        envelope
    }
}

fn refused_open(
    state: &ConnectionProcessState,
    offer: &LinkOffer,
    action: &str,
    reason: String,
) -> LinkOpenOutcome {
    let peer = state.lys_session.as_ref().map_or_else(
        || format!("key:{}", blake3::hash(&offer.machine).to_hex()),
        |session| session.holder().to_owned(),
    );
    tracing::warn!(link = %offer.link_id, action, %peer, %reason, "inbound link open refused");
    LinkOpenOutcome::Refused { reason }
}

/// A `LinkOpen` frame: the ack carries the handshake answer, the draft Lys
/// holds the open under, or the refusal by name.
pub(super) fn link_open_frame_response(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    stream_id: u32,
    offer: &LinkOffer,
) -> FrameAction {
    let opened = state
        .links
        .bind(pid, runtime, state.revocation.as_ref())
        .and_then(|()| open_link(runtime, state, stream_id, offer))
        .and_then(|(open, handshake)| {
            state
                .links
                .insert(runtime, stream_id, open)
                .map(|()| handshake)
        });
    let outcome = match opened {
        Ok(handshake) => LinkOpenOutcome::Opened { handshake },
        Err(OpenRefusal::Held { grant, draft }) => LinkOpenOutcome::Held { grant, draft },
        Err(OpenRefusal::Refused(reason)) => refused_open(
            state,
            offer,
            LinkAction::from(offer.end).other().access_action().name(),
            reason,
        ),
        Err(OpenRefusal::Pass(error)) => refused_open(
            state,
            offer,
            &error.refusal.refused.action,
            refusal_text(&error),
        ),
    };
    FrameAction::Respond(Frame::LinkOpenAck {
        flags: 0,
        stream_id,
        outcome,
    })
}

mod open;
use open::open_link;
mod admission;
pub(super) use admission::own_device_connect_refusal;

/// A `LinkMessage` frame: the crossing opened, decided on both ends and
/// published here, answered as a publish on the link's stream; or refused by
/// name under [`LINK_CROSSING_REFUSED_CODE`].
pub(super) fn link_message_frame_response(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    stream_id: u32,
    subject: &str,
    sealed: &[u8],
) -> FrameAction {
    match cross(runtime, state, stream_id, subject, sealed) {
        Ok(Crossed::Publication(crossing)) => {
            publish_crossing(pid, runtime, state, stream_id, subject, &crossing)
        }
        Ok(Crossed::Call(packet)) => {
            calls::dispatch(pid, runtime, state, stream_id, subject, packet)
        }
        Err(reason) => FrameAction::Respond(Frame::PublishError {
            flags: 0,
            stream_id,
            reason_code: LINK_CROSSING_REFUSED_CODE,
            message: Some(reason),
        }),
    }
}

enum Crossed {
    Publication(Crossing),
    Call(crate::link::calls::CallPacket),
}

struct Crossing {
    envelope: MessageEnvelope,
    position: u64,
    link_id: String,
    life: LinkLife,
}

fn publish_crossing(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    stream_id: u32,
    subject: &str,
    crossing: &Crossing,
) -> FrameAction {
    let error_response = |reason_code, message: String| {
        FrameAction::Respond(Frame::PublishError {
            flags: 0,
            stream_id,
            reason_code,
            message: Some(message),
        })
    };
    let services = runtime.services();
    if let Some(error) = reserved_namespace_refusal(subject) {
        return error_response(LINK_CROSSING_REFUSED_CODE, error.to_string());
    }
    if let Err(refusal) = services.admit_channel(ChannelOperation::Publish, subject) {
        // A definite admission refusal settles this subject; a failed roster does not.
        let code = match &refusal {
            super::channel_registry::ChannelAccessError::RosterUnavailable { .. } => {
                refusal.reason_code()
            }
            super::channel_registry::ChannelAccessError::NotRegistered { .. }
            | super::channel_registry::ChannelAccessError::Quiesced { .. }
            | super::channel_registry::ChannelAccessError::ConsoleNamespace { .. }
            | super::channel_registry::ChannelAccessError::ConsoleTapTarget { .. }
            | super::channel_registry::ChannelAccessError::ConsoleTapLimit { .. } => {
                LINK_CROSSING_REFUSED_CODE
            }
        };
        return error_response(code, access_to_server_error(&refusal).to_string());
    }
    let key = format!("link/{}/{subject}/{}", crossing.link_id, crossing.position);
    let outcome = match services.publish(pid, subject, &crossing.envelope, Some(&key)) {
        Ok(outcome) => outcome,
        Err(error) => {
            let code = if matches!(error, crate::ServerError::PublishUnresolved { .. }) {
                PUBLISH_UNRESOLVED_CODE
            } else {
                0xFFFF
            };
            return error_response(code, error.to_string());
        }
    };
    if outcome.reach == Reach::Suppressed {
        if outcome.durable_position.is_none() {
            return error_response(
                PUBLISH_UNRESOLVED_CODE,
                "crossing receipt is still in flight".to_owned(),
            );
        }
        let observed = runtime
            .signals
            .get()
            .ok_or_else(|| "crossing signals are not installed".to_owned())
            .and_then(|signals| signals.already_crossed(&crossing.link_id, &crossing.life));
        if let Err(error) = observed {
            return error_response(0xFFFF, error);
        }
    }
    FrameAction::Respond(Frame::PublishAck {
        // The authenticated source position identifies the crossing even after a restart.
        flags: 0,
        stream_id,
        message_id: crossing.position,
        durable_position: outcome.durable_position.filter(|_| {
            state
                .negotiated_version
                .is_some_and(|version| version >= PUBLISH_ACK_POSITION_VERSION)
        }),
    })
}

fn cross(
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    stream_id: u32,
    subject: &str,
    sealed: &[u8],
) -> Result<Crossed, String> {
    let ConnectionProcessState {
        lys_session, links, ..
    } = state;
    let Some(session) = lys_session.as_ref() else {
        return Err("a crossing is decided on a connection admitted on a Lys pass".to_owned());
    };
    let Some(host) = runtime.link_host() else {
        return Err("this liminal holds no machine identity".to_owned());
    };
    let mut contents = links.lock()?;
    let Some(open) = contents.links.get_mut(&stream_id) else {
        return Err(format!("no link is open on stream {stream_id}"));
    };
    if open
        .certificate_until
        .is_some_and(|until| unix_seconds_now() >= until)
    {
        return Err("certificate_pass_expired".into());
    }
    if open.remote_end() != LinkAction::Export {
        return Err(
            "messages cross from the exporting end; the remote machine imports on this link"
                .to_owned(),
        );
    }
    let mut plain = vec![0_u8; sealed.len()];
    let written = open
        .transport
        .open(sealed, &mut plain)
        .map_err(|error| format!("link frame refused: {error}"))?;
    plain.truncate(written);
    let remote = SessionLinkRights(session.as_ref());
    open.link
        .crossing(&remote, host.rights(), subject, unix_seconds_now())
        .map_err(|refusal| match &refusal {
            LinkRefusal::Export(error) => {
                format!("crossing refused at export: {}", refusal_text(error))
            }
            LinkRefusal::Import(error) => {
                format!("crossing refused at import: {}", refusal_text(error))
            }
            LinkRefusal::Binding(_) => refusal.to_string(),
        })?;
    if crate::link::calls::CallPacket::is_call(&plain) {
        let bound = host
            .peers()
            .ok_or("peer_calls_not_configured")?
            .max_call_bytes
            .get();
        return crate::link::calls::CallPacket::decode(&plain, bound)
            .map(Crossed::Call)
            .map_err(|error| error.to_string());
    }
    let header = plain
        .get(..8)
        .ok_or("crossed message has no position header")?;
    let position = u64::from_be_bytes(
        header
            .try_into()
            .map_err(|_| "crossed position is not eight bytes")?,
    );
    let link_id = open.link.resource().id;
    let life = open.life.clone();
    drop(contents);
    let envelope = MessageEnvelope::from_wire_bytes(&plain[8..])
        .map_err(|error| format!("the crossed message is not an envelope: {error}"))?;
    Ok(Crossed::Publication(Crossing {
        envelope,
        position,
        link_id,
        life,
    }))
}

mod calls;

pub(super) fn settled_calls(
    state: &ConnectionProcessState,
    runtime: &ConnectionRuntime,
) -> Result<Vec<Frame>, String> {
    let mut contents = state.links.lock()?;
    let Some(host) = runtime.link_host() else {
        return Ok(Vec::new());
    };
    let Some(peers) = host.peers() else {
        return Ok(Vec::new());
    };
    let mut frames = Vec::new();
    for (stream, open) in &mut contents.links {
        frames.extend(
            calls::settled(
                open,
                *stream,
                peers.max_call_bytes.get(),
                |open, subject| {
                    if open
                        .certificate_until
                        .is_some_and(|until| unix_seconds_now() >= until)
                    {
                        return Err(crate::link::calls::CallError::PeerGone);
                    }
                    let session = state
                        .lys_session
                        .as_deref()
                        .ok_or(crate::link::calls::CallError::Unavailable)?;
                    open.link
                        .crossing(
                            &SessionLinkRights(session),
                            host.rights(),
                            subject,
                            unix_seconds_now(),
                        )
                        .map(|_| ())
                        .map_err(|_| crate::link::calls::CallError::Unavailable)
                },
            )
            .map_err(|error| error.to_string())?,
        );
    }
    drop(contents);
    Ok(frames)
}