liminal_server/
link_relay.rs

1//! Only the endpoints key and open a relayed Noise link.
2
3use crate::link_carrier::{CarrierError, CarrierReceive, LinkCarrier, RelayCarrier};
4use crate::link_handshake::{
5    LinkHandshake, LinkHandshakeError, LinkTransport, LocalMachine, MachineIdentity, NOISE_OVERHEAD,
6};
7
8const HANDSHAKE_SUBJECT: &str = "net.link.handshake";
9
10/// The proved endpoint transport and its opaque relay path.
11#[derive(Debug)]
12pub struct RelayedLink {
13    /// Noise keys known only to the two linked machines.
14    pub transport: LinkTransport,
15    /// The admitted relay connection and marked application stream.
16    pub carrier: RelayCarrier,
17}
18
19/// A relayed open retains the endpoint or relay's own refusal.
20#[derive(Debug, thiserror::Error)]
21pub enum RelayOpenError {
22    /// The ordinary wire connection failed or answered an unexpected frame.
23    #[error(transparent)]
24    Carrier(#[from] CarrierError),
25    /// The endpoint's Noise peer proof failed.
26    #[error(transparent)]
27    Handshake(#[from] LinkHandshakeError),
28    /// The relay or the far endpoint refused or ended the open.
29    #[error("{0}")]
30    Refused(String),
31}
32
33/// Sends the opening KK message over the same relay stream as later frames.
34///
35/// # Errors
36/// Wire failure, relay refusal, unexpected handshake traffic or failed peer proof.
37pub fn open(
38    mut carrier: RelayCarrier,
39    local: &LocalMachine,
40    peer: MachineIdentity,
41) -> Result<RelayedLink, RelayOpenError> {
42    let mut handshake = LinkHandshake::initiator(local, peer)?;
43    send_handshake(&mut carrier, &mut handshake)?;
44    let mut acknowledged = false;
45    let mut received = false;
46    while !acknowledged || !received {
47        match next(&mut carrier)? {
48            CarrierReceive::Acknowledged { .. } if !acknowledged => acknowledged = true,
49            CarrierReceive::Sealed { subject, sealed }
50                if !received && subject == HANDSHAKE_SUBJECT =>
51            {
52                read_handshake(&mut carrier, &mut handshake, &sealed)?;
53                received = true;
54            }
55            _ => return Err(CarrierError::UnexpectedFrame.into()),
56        }
57    }
58    Ok(RelayedLink {
59        transport: handshake.finish()?,
60        carrier,
61    })
62}
63
64/// Answers the opening KK message without granting the relay any key material.
65///
66/// # Errors
67/// Wire failure, relay refusal, unexpected handshake traffic or failed peer proof.
68pub fn accept(
69    mut carrier: RelayCarrier,
70    local: &LocalMachine,
71    peer: MachineIdentity,
72) -> Result<RelayedLink, RelayOpenError> {
73    let mut handshake = LinkHandshake::responder(local, peer)?;
74    match next(&mut carrier)? {
75        CarrierReceive::Sealed { subject, sealed } if subject == HANDSHAKE_SUBJECT => {
76            read_handshake(&mut carrier, &mut handshake, &sealed)?;
77        }
78        _ => return Err(CarrierError::UnexpectedFrame.into()),
79    }
80    send_handshake(&mut carrier, &mut handshake)?;
81    if !matches!(next(&mut carrier)?, CarrierReceive::Acknowledged { .. }) {
82        return Err(CarrierError::UnexpectedFrame.into());
83    }
84    Ok(RelayedLink {
85        transport: handshake.finish()?,
86        carrier,
87    })
88}
89
90fn send_handshake(
91    carrier: &mut RelayCarrier,
92    handshake: &mut LinkHandshake,
93) -> Result<(), RelayOpenError> {
94    let mut bytes = [0; NOISE_OVERHEAD];
95    let length = handshake.write(&mut bytes)?;
96    carrier.send(HANDSHAKE_SUBJECT, &bytes[..length])?;
97    Ok(())
98}
99
100fn read_handshake(
101    carrier: &mut RelayCarrier,
102    handshake: &mut LinkHandshake,
103    bytes: &[u8],
104) -> Result<(), RelayOpenError> {
105    if let Err(error) = handshake.read(bytes) {
106        carrier.refuse(0x0302, "relay_handshake_refused")?;
107        return Err(error.into());
108    }
109    carrier.acknowledge(0, None)?;
110    Ok(())
111}
112
113fn next(carrier: &mut RelayCarrier) -> Result<CarrierReceive, RelayOpenError> {
114    match carrier.receive()? {
115        CarrierReceive::Refused {
116            reason_code,
117            reason,
118        } => {
119            Err(RelayOpenError::Refused(reason.unwrap_or_else(|| {
120                format!("relay_publish_error_{reason_code}")
121            })))
122        }
123        CarrierReceive::End { reason } => Err(RelayOpenError::Refused(reason)),
124        received => Ok(received),
125    }
126}
127
128#[cfg(feature = "test-support")]
129impl crate::server::connection::ConnectionSupervisor {
130    /// Boots a relay fixture with its configured machine and service bindings.
131    ///
132    /// # Errors
133    /// Configuration, machine admission, relay binding or scheduler startup failed.
134    pub fn with_relay_for_tests(
135        config: &crate::config::ServerConfig,
136        services: std::sync::Arc<crate::server::connection::LiminalConnectionServices>,
137        handle: tokio::runtime::Handle,
138    ) -> Result<Self, crate::ServerError> {
139        let refused = |message: &str| crate::ServerError::ConfigValidation {
140            message: message.to_owned(),
141        };
142        let auth = config.auth.as_ref().ok_or_else(|| refused("auth absent"))?;
143        let lys = auth
144            .lys
145            .as_ref()
146            .ok_or_else(|| refused("auth.lys absent"))?;
147        let mut install = crate::server::connection::configured_lys(auth, Some(handle.clone()))?
148            .ok_or_else(|| refused("Lys install absent"))?;
149        if install.link_host.is_none() {
150            return Err(refused("relay fixture has no installed machine identity"));
151        }
152        let bindings = crate::services::production::ServiceBindings::configure(
153            lys,
154            std::sync::Arc::clone(&services),
155            handle,
156            1,
157        )
158        .map_err(|error| crate::ServerError::ConfigValidation {
159            message: error.to_string(),
160        })?
161        .ok_or_else(|| refused("relay bindings absent"))?;
162        if bindings.relay().is_none() {
163            return Err(refused("relay not configured"));
164        }
165        install.services = Some(std::sync::Arc::new(bindings));
166        let observations = std::sync::Arc::clone(&services);
167        let supervisor = Self::with_lys_and_fatal_shutdown(
168            services,
169            None,
170            None,
171            Some(install),
172            config.limits,
173            crate::server::shutdown::ShutdownHandle::new(),
174        )?;
175        supervisor.install_link_observations_for_tests(&observations)?;
176        Ok(supervisor)
177    }
178}
liminal_server/link_relay.rs

liminal_server/
link_relay.rs
