liminal_server/link_open/
network.rs

1//! Configured dial, broker, punch and relay operations share one device context.
2
3use super::carriers::named;
4use super::{ChoiceConfig, OpenError};
5use crate::config::types::{MachineConfig, OutboundLinkConfig};
6use crate::link::LinkHost;
7use crate::link_carrier::{read_frame, write_frame};
8use crate::link_handshake::{LocalMachine, MachineIdentity};
9use crate::link_punch::{Puncher, WireBroker};
10use crate::server::connection::services::LiminalConnectionServices;
11use liminal::protocol::{Frame, ProtocolVersion, REVOCATION_CLOSE_VERSION};
12use liminal_protocol::credential::WireCredential;
13use std::net::{Shutdown, TcpStream, UdpSocket};
14use std::sync::atomic::{AtomicBool, Ordering};
15use std::sync::{Arc, Mutex, Weak};
16use std::time::Duration;
17use zeroize::Zeroizing;
18
19#[derive(Default)]
20struct DirectState {
21    puncher: Option<Puncher>,
22    next: u32,
23    connections: Vec<Arc<TcpStream>>,
24}
25
26/// The device owns one UDP socket and one parked receiver across its links.
27pub struct Network {
28    host: Arc<LinkHost>,
29    machine: MachineConfig,
30    choice: ChoiceConfig,
31    direct: Mutex<DirectState>,
32    sockets: Mutex<Vec<Weak<TcpStream>>>,
33    stopped: AtomicBool,
34    cancellation: tokio::sync::Notify,
35    udp: Mutex<Option<Weak<crate::link_udp::UdpCarrier>>>,
36    pending: Mutex<std::collections::BTreeSet<String>>,
37    handshakes: Mutex<()>,
38}
39
40impl std::fmt::Debug for Network {
41    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
42        formatter.debug_struct("Network").finish_non_exhaustive()
43    }
44}
45
46impl Network {
47    #[must_use]
48    pub fn new(host: Arc<LinkHost>, machine: MachineConfig, choice: ChoiceConfig) -> Arc<Self> {
49        Arc::new(Self {
50            host,
51            machine,
52            choice,
53            direct: Mutex::new(DirectState {
54                next: 1,
55                ..DirectState::default()
56            }),
57            sockets: Mutex::new(Vec::new()),
58            stopped: AtomicBool::new(false),
59            cancellation: tokio::sync::Notify::new(),
60            udp: Mutex::default(),
61            pending: Mutex::default(),
62            handshakes: Mutex::new(()),
63        })
64    }
65    fn retain(&self, socket: TcpStream) -> Result<Arc<TcpStream>, OpenError> {
66        self.require_running()?;
67        let socket = Arc::new(socket);
68        let mut sockets = self.sockets.lock().map_err(named)?;
69        self.require_running()?;
70        sockets.retain(|socket| socket.strong_count() != 0);
71        sockets.push(Arc::downgrade(&socket));
72        drop(sockets);
73        Ok(socket)
74    }
75    fn require_running(&self) -> Result<(), OpenError> {
76        if self.stopped.load(Ordering::Acquire) {
77            return Err(OpenError("network_stopped".to_owned()));
78        }
79        Ok(())
80    }
81    fn with_puncher<T>(
82        &self,
83        action: impl FnOnce(&mut Puncher) -> Result<T, OpenError>,
84    ) -> Result<T, OpenError> {
85        self.require_running()?;
86        let mut direct = self.direct.lock().map_err(named)?;
87        if direct.puncher.is_none() {
88            let (puncher, connections) = self.boot_puncher()?;
89            let carrier = puncher.shared_carrier().map_err(named)?;
90            *self.udp.lock().map_err(named)? = Some(Arc::downgrade(&carrier));
91            if self.is_stopped() {
92                carrier.cancel_waits().map_err(named)?;
93                return Err(OpenError("network_stopped".to_owned()));
94            }
95            direct.puncher = Some(puncher);
96            direct.connections = connections;
97        } else {
98            let (broker, connections) = self.broker()?;
99            direct
100                .puncher
101                .as_mut()
102                .ok_or_else(|| OpenError("device_not_observed".to_owned()))?
103                .rebind_broker(broker)
104                .map_err(named)?;
105            direct.connections = connections;
106        }
107        action(
108            direct
109                .puncher
110                .as_mut()
111                .ok_or_else(|| OpenError("device_not_observed".to_owned()))?,
112        )
113    }
114    fn connect(&self, address: &str) -> Result<(TcpStream, Arc<TcpStream>), OpenError> {
115        self.require_running()?;
116        let pass = crate::link_pass::MachinePass::configured(&self.machine).map_err(named)?;
117        let token = pass.for_open().map_err(named)?;
118        let credential = WireCredential::lys_record(token.trim().as_bytes());
119        if token.trim().is_empty()
120            || credential.len()
121                > crate::server::connection::pre_admission::MAX_CONNECT_AUTH_TOKEN_BYTES
122        {
123            return Err(OpenError("machine_pass_refused".to_owned()));
124        }
125        let (mut socket, interrupt) = self.wire_socket(address)?;
126        let bound = Some(Duration::from_millis(self.choice.crossing_deadline_ms));
127        socket.set_read_timeout(bound).map_err(named)?;
128        socket.set_write_timeout(bound).map_err(named)?;
129        write_frame(
130            &mut socket,
131            &Frame::Connect {
132                flags: 0,
133                min_version: ProtocolVersion::new(1, 0),
134                max_version: REVOCATION_CLOSE_VERSION,
135                auth_token: credential,
136            },
137        )
138        .map_err(named)?;
139        match read_frame(&mut socket).map_err(named)? {
140            Some(Frame::ConnectAck {
141                selected_version, ..
142            }) if (ProtocolVersion::new(1, 0)..=REVOCATION_CLOSE_VERSION)
143                .contains(&selected_version) =>
144            {
145                Ok((socket, interrupt))
146            }
147            Some(Frame::ConnectError { reason_code, .. }) => {
148                Err(OpenError(format!("service_connect_refused:{reason_code}")))
149            }
150            _ => Err(OpenError("unexpected_service_connect_answer".to_owned())),
151        }
152    }
153    fn broker(&self) -> Result<(WireBroker, Vec<Arc<TcpStream>>), OpenError> {
154        let address = self
155            .choice
156            .broker_address
157            .as_deref()
158            .ok_or_else(|| OpenError("broker_not_configured".to_owned()))?;
159        let (control, control_interrupt) = self.connect(address)?;
160        let (words, words_interrupt) = self.connect(address)?;
161        Ok((
162            WireBroker::new(control, words),
163            vec![control_interrupt, words_interrupt],
164        ))
165    }
166    fn boot_puncher(&self) -> Result<(Puncher, Vec<Arc<TcpStream>>), OpenError> {
167        let echo = self
168            .choice
169            .broker_echo
170            .ok_or_else(|| OpenError("broker_echo_not_configured".to_owned()))?;
171        let digits = Zeroizing::new(
172            std::fs::read_to_string(&self.machine.identity_path)
173                .map_err(|_| OpenError("machine_identity_refused".to_owned()))?,
174        );
175        let mut seed = Zeroizing::new([0; 32]);
176        hex::decode_to_slice(digits.trim(), &mut *seed)
177            .map_err(|_| OpenError("machine_identity_refused".to_owned()))?;
178        let local = LocalMachine::new(lys_core::keys::Ed25519Identity::from_seed(&seed));
179        if local.machine() != self.host.local().machine() {
180            return Err(OpenError("machine_identity_changed".to_owned()));
181        }
182        let socket = Arc::new(UdpSocket::bind(self.choice.udp_listen).map_err(named)?);
183        let (broker, connections) = self.broker()?;
184        let mut puncher = Puncher::boot(
185            local,
186            socket,
187            broker,
188            echo,
189            Duration::from_millis(self.choice.crossing_deadline_ms),
190        )
191        .map_err(named)?;
192        puncher.observe().map_err(named)?;
193        puncher.register().map_err(named)?;
194        Ok((puncher, connections))
195    }
196    fn stop(&self) -> Result<(), OpenError> {
197        self.stopped.store(true, Ordering::Release);
198        self.cancellation.notify_waiters();
199        let mut failure = None;
200        let sockets = self.sockets.lock().map_err(named)?;
201        for socket in sockets.iter().filter_map(Weak::upgrade) {
202            if let Err(error) = socket.shutdown(Shutdown::Both)
203                && error.kind() != std::io::ErrorKind::NotConnected
204            {
205                tracing::error!(%error, "owned carrier shutdown failed");
206                if failure.is_none() {
207                    failure = Some(named(error));
208                }
209            }
210        }
211        drop(sockets);
212        let carrier = self
213            .udp
214            .lock()
215            .map_err(named)?
216            .as_ref()
217            .and_then(Weak::upgrade);
218        if let Some(carrier) = carrier {
219            carrier.cancel_waits().map_err(named)?;
220        }
221        let cancelled = self.with_puncher_if_present(|puncher| {
222            puncher
223                .carrier()
224                .map_err(named)?
225                .cancel_waits()
226                .map_err(named)
227        });
228        if let Err(error) = cancelled {
229            tracing::error!(%error, "direct crossing cancellation failed");
230            if failure.is_none() {
231                failure = Some(error);
232            }
233        }
234        failure.map_or(Ok(()), Err)
235    }
236    fn with_puncher_if_present(
237        &self,
238        action: impl FnOnce(&Puncher) -> Result<(), OpenError>,
239    ) -> Result<(), OpenError> {
240        let mut direct = self.direct.lock().map_err(named)?;
241        let result = direct.puncher.as_ref().map_or(Ok(()), action);
242        direct.connections.clear();
243        drop(direct);
244        result
245    }
246}
247
248/// One configured link's observed carrier operations.
249#[derive(Debug)]
250pub struct NetworkBackend {
251    network: Arc<Network>,
252    peer: MachineIdentity,
253    services: Arc<LiminalConnectionServices>,
254    refused: u64,
255    link_id: String,
256    peer_holder: String,
257}
258
259impl NetworkBackend {
260    /// # Errors
261    /// The peer holder has no recorded machine key.
262    pub fn new(
263        network: Arc<Network>,
264        services: Arc<LiminalConnectionServices>,
265        config: &OutboundLinkConfig,
266    ) -> Result<Self, OpenError> {
267        let peer = network
268            .host
269            .recorded_machine(&config.peer_holder)
270            .ok_or_else(|| OpenError("peer_has_no_recorded_machine".to_owned()))?;
271        Ok(Self {
272            network,
273            peer,
274            services,
275            refused: 0,
276            link_id: config.link_id.clone(),
277            peer_holder: config.peer_holder.clone(),
278        })
279    }
280    fn require_config(&self, config: &OutboundLinkConfig) -> Result<(), OpenError> {
281        if config.link_id != self.link_id || config.peer_holder != self.peer_holder {
282            return Err(OpenError(
283                "carrier_backend_bound_to_another_link".to_owned(),
284            ));
285        }
286        Ok(())
287    }
288}
289
290mod backend;
291mod dial;
292mod incoming;
liminal_server/link_open/network.rs

liminal_server/link_open/
network.rs
