liminal_server/server/
link_udp_endpoint.rs

1//! An authenticated direct carrier reaches the ordinary machine and crossing admission route.
2use crate::link_handshake::MachineIdentity;
3use crate::link_open::network::Network;
4use crate::link_open::{OpenError, Stop};
5use crate::link_udp::{UdpCarrier, streams::Inbox};
6use crate::server::connection::{
7    ConnectionSupervisor, LiminalConnectionServices, LoopbackClientEnd, LoopbackDuplex,
8};
9use liminal::protocol::{Frame, FrameHeader, LinkEnd, codec};
10use std::collections::BTreeMap;
11use std::io::Write;
12use std::sync::Arc;
13use std::thread::JoinHandle;
14use zeroize::Zeroizing;
15
16type Worker = JoinHandle<Result<(), OpenError>>;
17/// Owns the incoming broker instruction reader and the keyed application workers.
18pub struct UdpEndpoint {
19    stop: Stop,
20    worker: Option<JoinHandle<Result<Vec<Worker>, OpenError>>>,
21}
22impl std::fmt::Debug for UdpEndpoint {
23    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24        f.debug_struct("UdpEndpoint").finish_non_exhaustive()
25    }
26}
27impl UdpEndpoint {
28    pub(crate) fn start(
29        network: Arc<Network>,
30        supervisor: ConnectionSupervisor,
31        services: Arc<LiminalConnectionServices>,
32    ) -> Result<Self, OpenError> {
33        let (mut puncher, connections) = network.incoming()?;
34        let stop = network.stop_handle();
35        let worker = std::thread::Builder::new()
36            .name("link-udp-accept".to_owned())
37            .spawn(move || {
38                let mut peers = BTreeMap::new();
39                let mut workers = Vec::new();
40                loop {
41                    let instruction = match puncher.next_instruction() {
42                        Ok(word) => word,
43                        Err(error) => {
44                            if !network.is_stopped() {
45                                tracing::warn!(%error, "broker instruction stream ended; keyed UDP imports remain served");
46                            }
47                            break;
48                        }
49                    };
50                    let peer = instruction.peer;
51                    let key = hex::encode(peer.ed25519_public());
52                    let existing = peers.get(&key).copied();
53                    let (index,refresh) = match network.accept_instruction(&mut puncher, &instruction, existing) {
54                        Ok(Some(index)) => index,
55                        Ok(None) => continue,
56                        Err(error) => {
57                            tracing::warn!(%error, "incoming punch refused");
58                            continue;
59                        }
60                    };
61                    if refresh { continue; }
62                    let carrier = puncher.shared_carrier().map_err(named)?;
63                    carrier.watch_acknowledgements(index).map_err(named)?;
64                    let inbox = carrier.subscribe(index).map_err(named)?;
65                    let supervisor = supervisor.clone();
66                    let services = services.clone();
67                    workers.push(std::thread::Builder::new()
68                        .name("link-udp-import".to_owned())
69                        .spawn(move || serve(&carrier, index, peer, &inbox, &supervisor, &services))
70                        .map_err(named)?);
71                    peers.insert(key, index);
72                }
73                drop(connections);
74                Ok(workers)
75            }).map_err(named)?;
76        Ok(Self {
77            stop,
78            worker: Some(worker),
79        })
80    }
81    pub(crate) fn finish(&mut self) -> Result<(), OpenError> {
82        let stopped = (self.stop)();
83        let Some(worker) = self.worker.take() else {
84            return stopped;
85        };
86        let workers = worker
87            .join()
88            .map_err(|_| OpenError("udp_accept_worker_panicked".to_owned()))??;
89        let mut failure = stopped.err();
90        for worker in workers {
91            match worker.join() {
92                Ok(Ok(())) => {}
93                Ok(Err(error)) => {
94                    tracing::error!(%error,"UDP importer ended by refusal");
95                    if failure.is_none() {
96                        failure = Some(error);
97                    }
98                }
99                Err(_) => {
100                    failure = Some(OpenError("udp_import_worker_panicked".to_owned()));
101                }
102            }
103        }
104        failure.map_or(Ok(()), Err)
105    }
106}
107impl Drop for UdpEndpoint {
108    fn drop(&mut self) {
109        if self.worker.is_some()
110            && let Err(error) = self.finish()
111        {
112            tracing::error!(%error,"UDP endpoint shutdown failed");
113        }
114    }
115}
116fn named(error: impl std::fmt::Display) -> OpenError {
117    OpenError(error.to_string())
118}
119
120fn serve(
121    carrier: &UdpCarrier,
122    index: u32,
123    peer: MachineIdentity,
124    inbox: &Inbox,
125    supervisor: &ConnectionSupervisor,
126    services: &LiminalConnectionServices,
127) -> Result<(), OpenError> {
128    let (mut connection, server) = LoopbackDuplex::bounded(crate::link_carrier::WIRE_FRAME_BOUND);
129    supervisor
130        .spawn_loopback_connection(server)
131        .map_err(named)?;
132    let mut assembled = Zeroizing::new(Vec::new());
133    let mut total = 0;
134    let mut connected = false;
135    let mut opened = false;
136    loop {
137        let datagram = match inbox.receive() {
138            Ok(datagram) => datagram,
139            Err(crate::link_udp::UdpError::Stopped) => return Ok(()),
140            Err(error) => return Err(named(error)),
141        };
142        let payload = &datagram.payload;
143        if payload.get(..4) != Some(b"LMUC") {
144            return Err(OpenError("invalid_udp_request".to_owned()));
145        }
146        let length = read_u32(payload.get(4..8))?;
147        let offset = read_u32(payload.get(8..12))?;
148        let body = payload
149            .get(12..)
150            .ok_or_else(|| OpenError("short_udp_request".to_owned()))?;
151        if length == 0
152            || length > crate::link_carrier::WIRE_FRAME_BOUND
153            || offset != assembled.len()
154            || body.is_empty()
155            || body.len() > length.saturating_sub(offset)
156        {
157            return Err(OpenError("invalid_udp_request_extent".to_owned()));
158        }
159        if offset == 0 {
160            total = length;
161        } else if total != length {
162            return Err(OpenError("udp_request_length_changed".to_owned()));
163        }
164        assembled.extend_from_slice(body);
165        if assembled.len() != total {
166            continue;
167        }
168        let (frame, read) = codec::decode(&assembled)
169            .map_err(|_| OpenError("invalid_udp_application_frame".to_owned()))?;
170        if read != assembled.len() {
171            return Err(OpenError("udp_request_has_trailing_bytes".to_owned()));
172        }
173        match &frame {
174            Frame::Connect { .. } if !connected => connected = true,
175            Frame::LinkOpen {
176                stream_id: 1,
177                offer,
178                ..
179            } if connected
180                && !opened
181                && offer.machine == *peer.ed25519_public()
182                && offer.end == LinkEnd::Export =>
183            {
184                opened = true;
185            }
186            Frame::LinkMessage { stream_id: 1, .. } if opened => {}
187            _ => return Err(OpenError("udp_frame_not_admitted_on_link".to_owned())),
188        }
189        connection.write_all(&assembled).map_err(named)?;
190        assembled.clear();
191        total = 0;
192        // The route emits this answer after the importing publish has applied.
193        let response = read_frame(&mut connection)?;
194        let mut reply = b"LMUR".to_vec();
195        reply.extend_from_slice(&datagram.counter.to_be_bytes());
196        reply.extend(response);
197        carrier
198            .reply(index, datagram.counter, &reply)
199            .map_err(named)?;
200        services
201            .console()
202            .emit(crate::console::TopologyChange::Link {
203                link_id: crate::console::records::Link::to_peer(
204                    services.console().device(),
205                    &hex::encode(peer.ed25519_public()),
206                )
207                .id,
208                peer: hex::encode(peer.ed25519_public()),
209                carrier: Some(crate::console::records::Carrier::Direct),
210                state: crate::console::records::LinkState::Up,
211                reason: None,
212                counters: crate::console::records::LinkCounters::default(),
213            })
214            .map_err(named)?;
215    }
216}
217fn read_u32(bytes: Option<&[u8]>) -> Result<usize, OpenError> {
218    usize::try_from(u32::from_be_bytes(
219        bytes
220            .ok_or_else(|| OpenError("short_udp_request".to_owned()))?
221            .try_into()
222            .map_err(|_| OpenError("short_udp_request".to_owned()))?,
223    ))
224    .map_err(named)
225}
226fn read_frame(connection: &mut LoopbackClientEnd) -> Result<Vec<u8>, OpenError> {
227    let mut header = [0; FrameHeader::WIRE_LEN];
228    read_exact(connection, &mut header)?;
229    let length = usize::try_from(u32::from_be_bytes(header[6..10].try_into().map_err(named)?))
230        .map_err(named)?;
231    if length > crate::link_carrier::WIRE_FRAME_BOUND {
232        return Err(OpenError("udp_route_response_too_large".to_owned()));
233    }
234    let mut bytes = header.to_vec();
235    bytes.resize(header.len() + length, 0);
236    read_exact(connection, &mut bytes[header.len()..])?;
237    Ok(bytes)
238}
239fn read_exact(connection: &mut LoopbackClientEnd, mut bytes: &mut [u8]) -> Result<(), OpenError> {
240    while !bytes.is_empty() {
241        let read = connection.read_timeout(bytes, None).map_err(named)?;
242        if read == 0 {
243            return Err(OpenError("udp_route_ended_before_answer".to_owned()));
244        }
245        bytes = &mut bytes[read..];
246    }
247    Ok(())
248}
liminal_server/server/link_udp_endpoint.rs

liminal_server/server/
link_udp_endpoint.rs
