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    agent_id: String,
70    network: Option<&'a crate::link_open::network::Network>,
71    pass: Result<crate::link_pass::MachinePass, crate::link_pass::PassRefusal>,
72    topology:
73        Option<std::sync::Arc<crate::server::connection::services::LiminalConnectionServices>>,
74}
75
76impl std::fmt::Debug for LinkDialer<'_> {
77    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
78        formatter
79            .debug_struct("LinkDialer")
80            .field("host", &self.host)
81            .field(
82                "pass",
83                &self.pass.as_ref().map(crate::link_pass::debug::PassLength),
84            )
85            .field("topology", &self.topology)
86            .finish()
87    }
88}
89
90impl<'a> LinkDialer<'a> {
91    /// Uses the host already built from this machine's verified configuration.
92    #[must_use]
93    pub fn new(host: &'a LinkHost, machine: &MachineConfig) -> Self {
94        Self {
95            host,
96            agent_id: machine.agent_id.clone(),
97            pass: crate::link_pass::MachinePass::configured(machine),
98            topology: None,
99            network: None,
100        }
101    }
102
103    pub(crate) const fn with_network(
104        mut self,
105        network: &'a crate::link_open::network::Network,
106    ) -> Self {
107        self.network = Some(network);
108        self
109    }
110
111    /// Publishes the keyed open before returning it to its carrier owner.
112    #[must_use]
113    pub fn with_topology(
114        mut self,
115        services: std::sync::Arc<crate::server::connection::services::LiminalConnectionServices>,
116    ) -> Self {
117        self.topology = Some(services);
118        self
119    }
120
121    /// Opens the exporting end with exactly one TCP connection attempt.
122    ///
123    /// DNS is resolved once and only its first address is attempted. A peer's
124    /// Held or Refused answer is returned without reconnecting.
125    ///
126    /// # Errors
127    /// Missing address/key, unreadable pass, admission refusal, malformed wire
128    /// answer, I/O failure, or a handshake against an unexpected machine.
129    pub fn open(&self, entry: &OutboundLinkConfig) -> Result<LinkDialOutcome, LinkDialError> {
130        let address = entry.peer_address.as_deref().ok_or_else(|| {
131            LinkDialError::Configuration(format!("link '{}' has no peer_address", entry.link_id))
132        })?;
133        let mut handshake = self.handshake(entry)?;
134        let mut first = vec![0; NOISE_OVERHEAD];
135        let length = handshake.write(&mut first)?;
136        first.truncate(length);
137        let first = self.certificate_offer(&first)?;
138        let (mut socket, interruption) = self.connect(address)?;
139        let stream_id = 1;
140        write_frame(
141            &mut socket,
142            &Frame::LinkOpen {
143                flags: 0,
144                stream_id,
145                offer: LinkOffer {
146                    link_id: entry.link_id.clone(),
147                    end: LinkEnd::Export,
148                    machine: *self.host.local().machine().ed25519_public(),
149                    handshake: first,
150                },
151            },
152        )?;
153        let outcome = match read_frame(&mut socket)? {
154            Some(Frame::LinkOpenAck {
155                stream_id: answered_stream,
156                outcome,
157                ..
158            }) if answered_stream == stream_id => outcome,
159            None => return Err(LinkDialError::Ended("LinkOpenAck")),
160            Some(_) => return Err(CarrierError::UnexpectedFrame.into()),
161        };
162        socket.set_read_timeout(None)?;
163        socket.set_write_timeout(None)?;
164        drop(interruption);
165        match outcome {
166            LinkOpenOutcome::Opened { handshake: answer } => {
167                handshake.read(&answer)?;
168                let transport = handshake.finish()?;
169                let mut carrier = WireCarrier::new(socket, stream_id)?;
170                let life = crate::link::LinkLife::new(*transport.peer());
171                if let Some(services) = &self.topology {
172                    let local = services.console().device().to_owned();
173                    let services = std::sync::Arc::clone(services);
174                    life.observe(
175                        local,
176                        std::sync::Arc::new(move |change| {
177                            if let Err(error) = services.console().emit(change) {
178                                tracing::error!(%error, "dialed link topology failed");
179                            }
180                        }),
181                    )
182                    .map_err(LinkDialError::Configuration)?;
183                }
184                carrier.install_life(life)?;
185                Ok(LinkDialOutcome::Opened(Box::new(DialedLink {
186                    link_id: entry.link_id.clone(),
187                    exports: entry.exports.clone(),
188                    transport,
189                    carrier,
190                })))
191            }
192            LinkOpenOutcome::Held { grant, draft } => Ok(LinkDialOutcome::Held { grant, draft }),
193            LinkOpenOutcome::Refused { reason } => Ok(LinkDialOutcome::Refused { reason }),
194        }
195    }
196
197    fn handshake(&self, entry: &OutboundLinkConfig) -> Result<LinkHandshake, LinkDialError> {
198        if entry.link_id.is_empty()
199            || entry
200                .link_id
201                .chars()
202                .any(|c| c.is_whitespace() || c.is_control())
203        {
204            return Err(LinkDialError::Configuration(
205                "link_id is not an identifier".to_owned(),
206            ));
207        }
208        let peer = self
209            .host
210            .recorded_machine(&entry.peer_holder)
211            .ok_or_else(|| {
212                LinkDialError::Configuration(format!(
213                    "peer_holder '{}' has no recorded machine key",
214                    entry.peer_holder
215                ))
216            })?;
217        for subject in &entry.exports {
218            crate::auth_lys::SubjectBinding::new(
219                "wire".to_owned(),
220                subject.clone(),
221                subject.clone(),
222            )
223            .map_err(|error| LinkDialError::Configuration(error.to_string()))?;
224        }
225        Ok(LinkHandshake::initiator(self.host.local(), peer)?)
226    }
227
228    fn connect(
229        &self,
230        address: &str,
231    ) -> Result<(TcpStream, Option<std::sync::Arc<TcpStream>>), LinkDialError> {
232        let pass = self
233            .pass
234            .as_ref()
235            .map_err(|error| LinkDialError::Configuration(error.to_string()))?;
236        let token = pass
237            .for_open()
238            .map_err(|error| LinkDialError::Configuration(error.to_string()))?;
239        let token = token.trim();
240        if token.is_empty() {
241            return Err(LinkDialError::Configuration(
242                "machine pass is empty".to_owned(),
243            ));
244        }
245        let credential = WireCredential::lys_record(token.as_bytes());
246        if credential.len() > crate::server::connection::pre_admission::MAX_CONNECT_AUTH_TOKEN_BYTES
247        {
248            return Err(LinkDialError::Configuration(
249                "machine pass exceeds the Connect bound".to_owned(),
250            ));
251        }
252        let (mut socket, interruption) = if let Some(network) = self.network {
253            let (socket, interruption) = network
254                .wire_socket(address)
255                .map_err(|error| CarrierError::Lifecycle(error.0))?;
256            (socket, Some(interruption))
257        } else {
258            let target = address.to_socket_addrs()?.next().ok_or_else(|| {
259                LinkDialError::Configuration(
260                    "peer_address resolved to no socket address".to_owned(),
261                )
262            })?;
263            (TcpStream::connect(target)?, None)
264        };
265        socket.set_nodelay(true)?;
266        write_frame(
267            &mut socket,
268            &Frame::Connect {
269                flags: 0,
270                min_version: ProtocolVersion::new(1, 0),
271                max_version: REVOCATION_CLOSE_VERSION,
272                auth_token: credential,
273            },
274        )?;
275        match read_frame(&mut socket)? {
276            Some(Frame::ConnectAck {
277                selected_version, ..
278            }) => {
279                if selected_version < ProtocolVersion::new(1, 0)
280                    || selected_version > REVOCATION_CLOSE_VERSION
281                {
282                    return Err(LinkDialError::Version);
283                }
284            }
285            Some(Frame::ConnectError {
286                reason_code,
287                message,
288                ..
289            }) => {
290                return Err(LinkDialError::ConnectRefused {
291                    reason_code,
292                    reason: message,
293                });
294            }
295            None => return Err(LinkDialError::Ended("ConnectAck")),
296            Some(_) => return Err(CarrierError::UnexpectedFrame.into()),
297        }
298        Ok((socket, interruption))
299    }
300}
301
302#[path = "link_dial/certificate.rs"]
303mod certificate;
liminal_server/link_dial.rs

liminal_server/
link_dial.rs
