use crate::link_handshake::MachineIdentity;
use crate::link_open::network::Network;
use crate::link_open::{OpenError, Stop};
use crate::link_udp::{UdpCarrier, streams::Inbox};
use crate::server::connection::{
ConnectionSupervisor, LiminalConnectionServices, LoopbackClientEnd, LoopbackDuplex,
};
use liminal::protocol::{Frame, FrameHeader, LinkEnd, codec};
use std::collections::BTreeMap;
use std::io::Write;
use std::sync::Arc;
use std::thread::JoinHandle;
use zeroize::Zeroizing;
type Worker = JoinHandle<Result<(), OpenError>>;
pub struct UdpEndpoint {
stop: Stop,
worker: Option<JoinHandle<Result<Vec<Worker>, OpenError>>>,
}
impl std::fmt::Debug for UdpEndpoint {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("UdpEndpoint").finish_non_exhaustive()
}
}
impl UdpEndpoint {
pub(crate) fn start(
network: Arc<Network>,
supervisor: ConnectionSupervisor,
services: Arc<LiminalConnectionServices>,
) -> Result<Self, OpenError> {
let (mut puncher, connections) = network.incoming()?;
let stop = network.stop_handle();
let worker = std::thread::Builder::new()
.name("link-udp-accept".to_owned())
.spawn(move || {
let mut peers = BTreeMap::new();
let mut workers = Vec::new();
loop {
let instruction = match puncher.next_instruction() {
Ok(word) => word,
Err(error) => {
if !network.is_stopped() {
tracing::warn!(%error, "broker instruction stream ended; keyed UDP imports remain served");
}
break;
}
};
let peer = instruction.peer;
let key = hex::encode(peer.ed25519_public());
let existing = peers.get(&key).copied();
let (index,refresh) = match network.accept_instruction(&mut puncher, &instruction, existing) {
Ok(Some(index)) => index,
Ok(None) => continue,
Err(error) => {
tracing::warn!(%error, "incoming punch refused");
continue;
}
};
if refresh { continue; }
let carrier = puncher.shared_carrier().map_err(named)?;
carrier.watch_acknowledgements(index).map_err(named)?;
let inbox = carrier.subscribe(index).map_err(named)?;
let supervisor = supervisor.clone();
let services = services.clone();
workers.push(std::thread::Builder::new()
.name("link-udp-import".to_owned())
.spawn(move || serve(&carrier, index, peer, &inbox, &supervisor, &services))
.map_err(named)?);
peers.insert(key, index);
}
drop(connections);
Ok(workers)
}).map_err(named)?;
Ok(Self {
stop,
worker: Some(worker),
})
}
pub(crate) fn finish(&mut self) -> Result<(), OpenError> {
let stopped = (self.stop)();
let Some(worker) = self.worker.take() else {
return stopped;
};
let workers = worker
.join()
.map_err(|_| OpenError("udp_accept_worker_panicked".to_owned()))??;
let mut failure = stopped.err();
for worker in workers {
match worker.join() {
Ok(Ok(())) => {}
Ok(Err(error)) => {
tracing::error!(%error,"UDP importer ended by refusal");
if failure.is_none() {
failure = Some(error);
}
}
Err(_) => {
failure = Some(OpenError("udp_import_worker_panicked".to_owned()));
}
}
}
failure.map_or(Ok(()), Err)
}
}
impl Drop for UdpEndpoint {
fn drop(&mut self) {
if self.worker.is_some()
&& let Err(error) = self.finish()
{
tracing::error!(%error,"UDP endpoint shutdown failed");
}
}
}
fn named(error: impl std::fmt::Display) -> OpenError {
OpenError(error.to_string())
}
fn serve(
carrier: &UdpCarrier,
index: u32,
peer: MachineIdentity,
inbox: &Inbox,
supervisor: &ConnectionSupervisor,
services: &LiminalConnectionServices,
) -> Result<(), OpenError> {
let (mut connection, server) = LoopbackDuplex::bounded(crate::link_carrier::WIRE_FRAME_BOUND);
supervisor
.spawn_loopback_connection(server)
.map_err(named)?;
let mut assembled = Zeroizing::new(Vec::new());
let mut total = 0;
let mut connected = false;
let mut opened = false;
loop {
let datagram = match inbox.receive() {
Ok(datagram) => datagram,
Err(crate::link_udp::UdpError::Stopped) => return Ok(()),
Err(error) => return Err(named(error)),
};
let payload = &datagram.payload;
if payload.get(..4) != Some(b"LMUC") {
return Err(OpenError("invalid_udp_request".to_owned()));
}
let length = read_u32(payload.get(4..8))?;
let offset = read_u32(payload.get(8..12))?;
let body = payload
.get(12..)
.ok_or_else(|| OpenError("short_udp_request".to_owned()))?;
if length == 0
|| length > crate::link_carrier::WIRE_FRAME_BOUND
|| offset != assembled.len()
|| body.is_empty()
|| body.len() > length.saturating_sub(offset)
{
return Err(OpenError("invalid_udp_request_extent".to_owned()));
}
if offset == 0 {
total = length;
} else if total != length {
return Err(OpenError("udp_request_length_changed".to_owned()));
}
assembled.extend_from_slice(body);
if assembled.len() != total {
continue;
}
let (frame, read) = codec::decode(&assembled)
.map_err(|_| OpenError("invalid_udp_application_frame".to_owned()))?;
if read != assembled.len() {
return Err(OpenError("udp_request_has_trailing_bytes".to_owned()));
}
match &frame {
Frame::Connect { .. } if !connected => connected = true,
Frame::LinkOpen {
stream_id: 1,
offer,
..
} if connected
&& !opened
&& offer.machine == *peer.ed25519_public()
&& offer.end == LinkEnd::Export =>
{
opened = true;
}
Frame::LinkMessage { stream_id: 1, .. } if opened => {}
_ => return Err(OpenError("udp_frame_not_admitted_on_link".to_owned())),
}
connection.write_all(&assembled).map_err(named)?;
assembled.clear();
total = 0;
let response = read_frame(&mut connection)?;
let mut reply = b"LMUR".to_vec();
reply.extend_from_slice(&datagram.counter.to_be_bytes());
reply.extend(response);
carrier
.reply(index, datagram.counter, &reply)
.map_err(named)?;
services
.console()
.emit(crate::console::TopologyChange::Link {
link_id: crate::console::records::Link::to_peer(
services.console().device(),
&hex::encode(peer.ed25519_public()),
)
.id,
peer: hex::encode(peer.ed25519_public()),
carrier: Some(crate::console::records::Carrier::Direct),
state: crate::console::records::LinkState::Up,
reason: None,
counters: crate::console::records::LinkCounters::default(),
})
.map_err(named)?;
}
}
fn read_u32(bytes: Option<&[u8]>) -> Result<usize, OpenError> {
usize::try_from(u32::from_be_bytes(
bytes
.ok_or_else(|| OpenError("short_udp_request".to_owned()))?
.try_into()
.map_err(|_| OpenError("short_udp_request".to_owned()))?,
))
.map_err(named)
}
fn read_frame(connection: &mut LoopbackClientEnd) -> Result<Vec<u8>, OpenError> {
let mut header = [0; FrameHeader::WIRE_LEN];
read_exact(connection, &mut header)?;
let length = usize::try_from(u32::from_be_bytes(header[6..10].try_into().map_err(named)?))
.map_err(named)?;
if length > crate::link_carrier::WIRE_FRAME_BOUND {
return Err(OpenError("udp_route_response_too_large".to_owned()));
}
let mut bytes = header.to_vec();
bytes.resize(header.len() + length, 0);
read_exact(connection, &mut bytes[header.len()..])?;
Ok(bytes)
}
fn read_exact(connection: &mut LoopbackClientEnd, mut bytes: &mut [u8]) -> Result<(), OpenError> {
while !bytes.is_empty() {
let read = connection.read_timeout(bytes, None).map_err(named)?;
if read == 0 {
return Err(OpenError("udp_route_ended_before_answer".to_owned()));
}
bytes = &mut bytes[read..];
}
Ok(())
}