liminal_server/
link_dial.rs

1//! One outbound exporting link, admitted with this machine's own Lys pass.
2
3use crate::config::types::{MachineConfig, OutboundLinkConfig};
4use crate::link::LinkHost;
5use crate::link_carrier::{CarrierError, WireCarrier, read_frame, write_frame};
6use crate::link_handshake::{LinkHandshake, LinkHandshakeError, LinkTransport, NOISE_OVERHEAD};
7use liminal::protocol::{
8    Frame, LinkEnd, LinkOffer, LinkOpenOutcome, ProtocolVersion, REVOCATION_CLOSE_VERSION,
9};
10use liminal_protocol::credential::WireCredential;
11use std::net::{TcpStream, ToSocketAddrs};
12
13/// A keyed link and the carrier retained for its subsequent traffic.
14#[derive(Debug)]
15pub struct DialedLink {
16    /// The configured link under the network.
17    pub link_id: String,
18    /// The placed prefixes this link exports.
19    pub exports: Vec<String>,
20    /// The finished handshake, bound to the importing peer's machine.
21    pub transport: LinkTransport,
22    /// The connection and application stream the handshake opened.
23    pub carrier: WireCarrier,
24}
25
26/// The peer's answer to one exporting open; Held and Refused are final answers.
27#[derive(Debug)]
28pub enum LinkDialOutcome {
29    /// The handshake finished and its keyed link is held by the caller.
30    Opened(Box<DialedLink>),
31    /// Lys retained the open under an approval draft.
32    Held { grant: String, draft: String },
33    /// The peer refused the open by name.
34    Refused { reason: String },
35}
36
37/// Why an outbound open could not reach a peer's link answer.
38#[derive(Debug, thiserror::Error)]
39pub enum LinkDialError {
40    /// The entry cannot be dialed as configured.
41    #[error("outbound link configuration refused: {0}")]
42    Configuration(String),
43    /// A resolver or connection failed without exposing credentials.
44    #[error("outbound link I/O failed: {0}")]
45    Io(#[from] std::io::Error),
46    /// Framing or the carrier's I/O failed.
47    #[error(transparent)]
48    Carrier(#[from] CarrierError),
49    /// The KK peer proof did not finish.
50    #[error(transparent)]
51    Handshake(#[from] LinkHandshakeError),
52    /// The peer refused machine-pass admission at Connect.
53    #[error("peer refused Connect ({reason_code}): {reason:?}")]
54    ConnectRefused {
55        reason_code: u16,
56        reason: Option<String>,
57    },
58    /// A Connect ack selected a version outside the offered range.
59    #[error("peer selected an unoffered protocol version")]
60    Version,
61    /// The peer ended before answering `Connect` or `LinkOpen`.
62    #[error("wire ended before {0}")]
63    Ended(&'static str),
64}
65
66/// Dials one configured entry with this machine's recorded identity and pass.
67pub struct LinkDialer<'a> {
68    host: &'a LinkHost,
69    network: Option<&'a crate::link_open::network::Network>,
70    pass: Result<crate::link_pass::MachinePass, crate::link_pass::PassRefusal>,
71    topology:
72        Option<std::sync::Arc<crate::server::connection::services::LiminalConnectionServices>>,
73}
74
75impl std::fmt::Debug for LinkDialer<'_> {
76    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        formatter
78            .debug_struct("LinkDialer")
79            .field("host", &self.host)
80            .field(
81                "pass",
82                &self.pass.as_ref().map(crate::link_pass::debug::PassLength),
83            )
84            .field("topology", &self.topology)
85            .finish()
86    }
87}
88
89impl<'a> LinkDialer<'a> {
90    /// Uses the host already built from this machine's verified configuration.
91    #[must_use]
92    pub fn new(host: &'a LinkHost, machine: &MachineConfig) -> Self {
93        Self {
94            host,
95            pass: crate::link_pass::MachinePass::configured(machine),
96            topology: None,
97            network: None,
98        }
99    }
100
101    pub(crate) const fn with_network(
102        mut self,
103        network: &'a crate::link_open::network::Network,
104    ) -> Self {
105        self.network = Some(network);
106        self
107    }
108
109    /// Publishes the keyed open before returning it to its carrier owner.
110    #[must_use]
111    pub fn with_topology(
112        mut self,
113        services: std::sync::Arc<crate::server::connection::services::LiminalConnectionServices>,
114    ) -> Self {
115        self.topology = Some(services);
116        self
117    }
118
119    /// Opens the exporting end with exactly one TCP connection attempt.
120    ///
121    /// DNS is resolved once and only its first address is attempted. A peer's
122    /// Held or Refused answer is returned without reconnecting.
123    ///
124    /// # Errors
125    /// Missing address/key, unreadable pass, admission refusal, malformed wire
126    /// answer, I/O failure, or a handshake against an unexpected machine.
127    pub fn open(&self, entry: &OutboundLinkConfig) -> Result<LinkDialOutcome, LinkDialError> {
128        let address = entry.peer_address.as_deref().ok_or_else(|| {
129            LinkDialError::Configuration(format!("link '{}' has no peer_address", entry.link_id))
130        })?;
131        let mut handshake = self.handshake(entry)?;
132        let mut first = vec![0; NOISE_OVERHEAD];
133        let length = handshake.write(&mut first)?;
134        first.truncate(length);
135        let (mut socket, interruption) = self.connect(address)?;
136        let stream_id = 1;
137        write_frame(
138            &mut socket,
139            &Frame::LinkOpen {
140                flags: 0,
141                stream_id,
142                offer: LinkOffer {
143                    link_id: entry.link_id.clone(),
144                    end: LinkEnd::Export,
145                    machine: *self.host.local().machine().ed25519_public(),
146                    handshake: first,
147                },
148            },
149        )?;
150        let outcome = match read_frame(&mut socket)? {
151            Some(Frame::LinkOpenAck {
152                stream_id: answered_stream,
153                outcome,
154                ..
155            }) if answered_stream == stream_id => outcome,
156            None => return Err(LinkDialError::Ended("LinkOpenAck")),
157            Some(_) => return Err(CarrierError::UnexpectedFrame.into()),
158        };
159        socket.set_read_timeout(None)?;
160        socket.set_write_timeout(None)?;
161        drop(interruption);
162        match outcome {
163            LinkOpenOutcome::Opened { handshake: answer } => {
164                handshake.read(&answer)?;
165                let transport = handshake.finish()?;
166                let mut carrier = WireCarrier::new(socket, stream_id)?;
167                let life = crate::link::LinkLife::new(*transport.peer());
168                if let Some(services) = &self.topology {
169                    let local = services.console().device().to_owned();
170                    let services = std::sync::Arc::clone(services);
171                    life.observe(
172                        local,
173                        std::sync::Arc::new(move |change| {
174                            if let Err(error) = services.console().emit(change) {
175                                tracing::error!(%error, "dialed link topology failed");
176                            }
177                        }),
178                    )
179                    .map_err(LinkDialError::Configuration)?;
180                }
181                carrier.install_life(life)?;
182                Ok(LinkDialOutcome::Opened(Box::new(DialedLink {
183                    link_id: entry.link_id.clone(),
184                    exports: entry.exports.clone(),
185                    transport,
186                    carrier,
187                })))
188            }
189            LinkOpenOutcome::Held { grant, draft } => Ok(LinkDialOutcome::Held { grant, draft }),
190            LinkOpenOutcome::Refused { reason } => Ok(LinkDialOutcome::Refused { reason }),
191        }
192    }
193
194    fn handshake(&self, entry: &OutboundLinkConfig) -> Result<LinkHandshake, LinkDialError> {
195        if entry.link_id.is_empty()
196            || entry
197                .link_id
198                .chars()
199                .any(|c| c.is_whitespace() || c.is_control())
200        {
201            return Err(LinkDialError::Configuration(
202                "link_id is not an identifier".to_owned(),
203            ));
204        }
205        let peer = self
206            .host
207            .recorded_machine(&entry.peer_holder)
208            .ok_or_else(|| {
209                LinkDialError::Configuration(format!(
210                    "peer_holder '{}' has no recorded machine key",
211                    entry.peer_holder
212                ))
213            })?;
214        for subject in &entry.exports {
215            crate::auth_lys::SubjectBinding::new(
216                "wire".to_owned(),
217                subject.clone(),
218                subject.clone(),
219            )
220            .map_err(|error| LinkDialError::Configuration(error.to_string()))?;
221        }
222        Ok(LinkHandshake::initiator(self.host.local(), peer)?)
223    }
224
225    fn connect(
226        &self,
227        address: &str,
228    ) -> Result<(TcpStream, Option<std::sync::Arc<TcpStream>>), LinkDialError> {
229        let pass = self
230            .pass
231            .as_ref()
232            .map_err(|error| LinkDialError::Configuration(error.to_string()))?;
233        let token = pass
234            .for_open()
235            .map_err(|error| LinkDialError::Configuration(error.to_string()))?;
236        let token = token.trim();
237        if token.is_empty() {
238            return Err(LinkDialError::Configuration(
239                "machine pass is empty".to_owned(),
240            ));
241        }
242        let credential = WireCredential::lys_record(token.as_bytes());
243        if credential.len() > crate::server::connection::pre_admission::MAX_CONNECT_AUTH_TOKEN_BYTES
244        {
245            return Err(LinkDialError::Configuration(
246                "machine pass exceeds the Connect bound".to_owned(),
247            ));
248        }
249        let (mut socket, interruption) = if let Some(network) = self.network {
250            let (socket, interruption) = network
251                .wire_socket(address)
252                .map_err(|error| CarrierError::Lifecycle(error.0))?;
253            (socket, Some(interruption))
254        } else {
255            let target = address.to_socket_addrs()?.next().ok_or_else(|| {
256                LinkDialError::Configuration(
257                    "peer_address resolved to no socket address".to_owned(),
258                )
259            })?;
260            (TcpStream::connect(target)?, None)
261        };
262        socket.set_nodelay(true)?;
263        write_frame(
264            &mut socket,
265            &Frame::Connect {
266                flags: 0,
267                min_version: ProtocolVersion::new(1, 0),
268                max_version: REVOCATION_CLOSE_VERSION,
269                auth_token: credential,
270            },
271        )?;
272        match read_frame(&mut socket)? {
273            Some(Frame::ConnectAck {
274                selected_version, ..
275            }) => {
276                if selected_version < ProtocolVersion::new(1, 0)
277                    || selected_version > REVOCATION_CLOSE_VERSION
278                {
279                    return Err(LinkDialError::Version);
280                }
281            }
282            Some(Frame::ConnectError {
283                reason_code,
284                message,
285                ..
286            }) => {
287                return Err(LinkDialError::ConnectRefused {
288                    reason_code,
289                    reason: message,
290                });
291            }
292            None => return Err(LinkDialError::Ended("ConnectAck")),
293            Some(_) => return Err(CarrierError::UnexpectedFrame.into()),
294        }
295        Ok((socket, interruption))
296    }
297}
liminal_server/link_dial.rs

liminal_server/
link_dial.rs
