liminal-server 0.23.1

//! 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,
}

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,
    })
}

fn open_link(
    runtime: &ConnectionRuntime,
    state: &ConnectionProcessState,
    stream_id: u32,
    offer: &LinkOffer,
) -> Result<(OpenLink, Vec<u8>), OpenRefusal> {
    let Some(session) = state.lys_session.as_ref() else {
        return Err(OpenRefusal::Refused(
            "a link is opened on a connection admitted on a Lys pass".to_owned(),
        ));
    };
    let Some((_, network_id)) = runtime.lys_admitter() else {
        return Err(OpenRefusal::Refused(
            "no lys admitter is installed".to_owned(),
        ));
    };
    let Some(host) = runtime.link_host() else {
        return Err(OpenRefusal::Refused(
            "this liminal holds no machine identity; [auth.lys.machine] is not configured"
                .to_owned(),
        ));
    };
    if state
        .links
        .contains_key(stream_id)
        .map_err(OpenRefusal::Refused)?
    {
        return Err(OpenRefusal::Refused(format!(
            "a link is already open on stream {stream_id}"
        )));
    }
    let Some(recorded) = host.recorded_machine(session.holder()) else {
        return Err(OpenRefusal::Refused(
            "no machine key is recorded for this holder".to_owned(),
        ));
    };
    if recorded.ed25519_public() != &offer.machine {
        return Err(OpenRefusal::Refused(
            "the offered machine key is not the holder's recorded key".to_owned(),
        ));
    }
    let remote_end = LinkAction::from(offer.end);
    let local = host.local().machine();
    let (exporter, importer) = match remote_end {
        LinkAction::Export => (recorded, local),
        LinkAction::Import => (local, recorded),
    };
    let link = Link::new(
        network_id.to_owned(),
        offer.link_id.clone(),
        exporter,
        importer,
    )
    .map_err(|error| OpenRefusal::Refused(error.to_string()))?;
    match session.open_link(&link.resource(), remote_end, unix_seconds_now()) {
        Ok(DeliberateDecision::Outright(_)) => {}
        Ok(
            DeliberateDecision::ByDraft { grant, draft }
            | DeliberateDecision::ByTwo { grant, draft },
        ) => return Err(OpenRefusal::Held { grant, draft }),
        Err(error) => return Err(OpenRefusal::Pass(error)),
    }
    let handshake_refused = |error: crate::link_handshake::LinkHandshakeError| {
        OpenRefusal::Refused(format!("handshake refused: {error}"))
    };
    let mut handshake =
        LinkHandshake::responder(host.local(), recorded).map_err(handshake_refused)?;
    handshake
        .read(&offer.handshake)
        .map_err(handshake_refused)?;
    let mut answer = vec![0_u8; NOISE_OVERHEAD];
    let written = handshake.write(&mut answer).map_err(handshake_refused)?;
    answer.truncate(written);
    let transport = handshake.finish().map_err(handshake_refused)?;
    let life = LinkLife::new(*transport.peer());
    Ok((
        OpenLink {
            link,
            remote_end,
            transport,
            life,
        },
        answer,
    ))
}

/// 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(crossing) => publish_crossing(pid, runtime, state, stream_id, subject, &crossing),
        Err(reason) => FrameAction::Respond(Frame::PublishError {
            flags: 0,
            stream_id,
            reason_code: LINK_CROSSING_REFUSED_CODE,
            message: Some(reason),
        }),
    }
}

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<Crossing, 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.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(),
        })?;
    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(Crossing {
        envelope,
        position,
        link_id,
        life,
    })
}