use super::carriers::named;
use super::{ChoiceConfig, OpenError};
use crate::config::types::{MachineConfig, OutboundLinkConfig};
use crate::link::LinkHost;
use crate::link_carrier::{read_frame, write_frame};
use crate::link_handshake::{LocalMachine, MachineIdentity};
use crate::link_punch::{Puncher, WireBroker};
use crate::server::connection::services::LiminalConnectionServices;
use liminal::protocol::{Frame, ProtocolVersion, REVOCATION_CLOSE_VERSION};
use liminal_protocol::credential::WireCredential;
use std::net::{Shutdown, TcpStream, UdpSocket};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, Weak};
use std::time::Duration;
use zeroize::Zeroizing;
#[derive(Default)]
struct DirectState {
puncher: Option<Puncher>,
next: u32,
connections: Vec<Arc<TcpStream>>,
}
pub struct Network {
host: Arc<LinkHost>,
machine: MachineConfig,
choice: ChoiceConfig,
direct: Mutex<DirectState>,
sockets: Mutex<Vec<Weak<TcpStream>>>,
stopped: AtomicBool,
cancellation: tokio::sync::Notify,
udp: Mutex<Option<Weak<crate::link_udp::UdpCarrier>>>,
pending: Mutex<std::collections::BTreeSet<String>>,
handshakes: Mutex<()>,
}
impl std::fmt::Debug for Network {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.debug_struct("Network").finish_non_exhaustive()
}
}
impl Network {
#[must_use]
pub fn new(host: Arc<LinkHost>, machine: MachineConfig, choice: ChoiceConfig) -> Arc<Self> {
Arc::new(Self {
host,
machine,
choice,
direct: Mutex::new(DirectState {
next: 1,
..DirectState::default()
}),
sockets: Mutex::new(Vec::new()),
stopped: AtomicBool::new(false),
cancellation: tokio::sync::Notify::new(),
udp: Mutex::default(),
pending: Mutex::default(),
handshakes: Mutex::new(()),
})
}
fn retain(&self, socket: TcpStream) -> Result<Arc<TcpStream>, OpenError> {
self.require_running()?;
let socket = Arc::new(socket);
let mut sockets = self.sockets.lock().map_err(named)?;
self.require_running()?;
sockets.retain(|socket| socket.strong_count() != 0);
sockets.push(Arc::downgrade(&socket));
drop(sockets);
Ok(socket)
}
fn require_running(&self) -> Result<(), OpenError> {
if self.stopped.load(Ordering::Acquire) {
return Err(OpenError("network_stopped".to_owned()));
}
Ok(())
}
fn with_puncher<T>(
&self,
action: impl FnOnce(&mut Puncher) -> Result<T, OpenError>,
) -> Result<T, OpenError> {
self.require_running()?;
let mut direct = self.direct.lock().map_err(named)?;
if direct.puncher.is_none() {
let (puncher, connections) = self.boot_puncher()?;
let carrier = puncher.shared_carrier().map_err(named)?;
*self.udp.lock().map_err(named)? = Some(Arc::downgrade(&carrier));
if self.is_stopped() {
carrier.cancel_waits().map_err(named)?;
return Err(OpenError("network_stopped".to_owned()));
}
direct.puncher = Some(puncher);
direct.connections = connections;
} else {
let (broker, connections) = self.broker()?;
direct
.puncher
.as_mut()
.ok_or_else(|| OpenError("device_not_observed".to_owned()))?
.rebind_broker(broker)
.map_err(named)?;
direct.connections = connections;
}
action(
direct
.puncher
.as_mut()
.ok_or_else(|| OpenError("device_not_observed".to_owned()))?,
)
}
fn connect(&self, address: &str) -> Result<(TcpStream, Arc<TcpStream>), OpenError> {
self.require_running()?;
let pass = crate::link_pass::MachinePass::configured(&self.machine).map_err(named)?;
let token = pass.for_open().map_err(named)?;
let credential = WireCredential::lys_record(token.trim().as_bytes());
if token.trim().is_empty()
|| credential.len()
> crate::server::connection::pre_admission::MAX_CONNECT_AUTH_TOKEN_BYTES
{
return Err(OpenError("machine_pass_refused".to_owned()));
}
let (mut socket, interrupt) = self.wire_socket(address)?;
let bound = Some(Duration::from_millis(self.choice.crossing_deadline_ms));
socket.set_read_timeout(bound).map_err(named)?;
socket.set_write_timeout(bound).map_err(named)?;
write_frame(
&mut socket,
&Frame::Connect {
flags: 0,
min_version: ProtocolVersion::new(1, 0),
max_version: REVOCATION_CLOSE_VERSION,
auth_token: credential,
},
)
.map_err(named)?;
match read_frame(&mut socket).map_err(named)? {
Some(Frame::ConnectAck {
selected_version, ..
}) if (ProtocolVersion::new(1, 0)..=REVOCATION_CLOSE_VERSION)
.contains(&selected_version) =>
{
Ok((socket, interrupt))
}
Some(Frame::ConnectError { reason_code, .. }) => {
Err(OpenError(format!("service_connect_refused:{reason_code}")))
}
_ => Err(OpenError("unexpected_service_connect_answer".to_owned())),
}
}
fn broker(&self) -> Result<(WireBroker, Vec<Arc<TcpStream>>), OpenError> {
let address = self
.choice
.broker_address
.as_deref()
.ok_or_else(|| OpenError("broker_not_configured".to_owned()))?;
let (control, control_interrupt) = self.connect(address)?;
let (words, words_interrupt) = self.connect(address)?;
Ok((
WireBroker::new(control, words),
vec![control_interrupt, words_interrupt],
))
}
fn boot_puncher(&self) -> Result<(Puncher, Vec<Arc<TcpStream>>), OpenError> {
let echo = self
.choice
.broker_echo
.ok_or_else(|| OpenError("broker_echo_not_configured".to_owned()))?;
let digits = Zeroizing::new(
std::fs::read_to_string(&self.machine.identity_path)
.map_err(|_| OpenError("machine_identity_refused".to_owned()))?,
);
let mut seed = Zeroizing::new([0; 32]);
hex::decode_to_slice(digits.trim(), &mut *seed)
.map_err(|_| OpenError("machine_identity_refused".to_owned()))?;
let local = LocalMachine::new(lys_core::keys::Ed25519Identity::from_seed(&seed));
if local.machine() != self.host.local().machine() {
return Err(OpenError("machine_identity_changed".to_owned()));
}
let socket = Arc::new(UdpSocket::bind(self.choice.udp_listen).map_err(named)?);
let (broker, connections) = self.broker()?;
let mut puncher = Puncher::boot(
local,
socket,
broker,
echo,
Duration::from_millis(self.choice.crossing_deadline_ms),
)
.map_err(named)?;
puncher.observe().map_err(named)?;
puncher.register().map_err(named)?;
Ok((puncher, connections))
}
fn stop(&self) -> Result<(), OpenError> {
self.stopped.store(true, Ordering::Release);
self.cancellation.notify_waiters();
let mut failure = None;
let sockets = self.sockets.lock().map_err(named)?;
for socket in sockets.iter().filter_map(Weak::upgrade) {
if let Err(error) = socket.shutdown(Shutdown::Both)
&& error.kind() != std::io::ErrorKind::NotConnected
{
tracing::error!(%error, "owned carrier shutdown failed");
if failure.is_none() {
failure = Some(named(error));
}
}
}
drop(sockets);
let carrier = self
.udp
.lock()
.map_err(named)?
.as_ref()
.and_then(Weak::upgrade);
if let Some(carrier) = carrier {
carrier.cancel_waits().map_err(named)?;
}
let cancelled = self.with_puncher_if_present(|puncher| {
puncher
.carrier()
.map_err(named)?
.cancel_waits()
.map_err(named)
});
if let Err(error) = cancelled {
tracing::error!(%error, "direct crossing cancellation failed");
if failure.is_none() {
failure = Some(error);
}
}
failure.map_or(Ok(()), Err)
}
fn with_puncher_if_present(
&self,
action: impl FnOnce(&Puncher) -> Result<(), OpenError>,
) -> Result<(), OpenError> {
let mut direct = self.direct.lock().map_err(named)?;
let result = direct.puncher.as_ref().map_or(Ok(()), action);
direct.connections.clear();
drop(direct);
result
}
}
#[derive(Debug)]
pub struct NetworkBackend {
network: Arc<Network>,
peer: MachineIdentity,
services: Arc<LiminalConnectionServices>,
refused: u64,
link_id: String,
peer_holder: String,
}
impl NetworkBackend {
pub fn new(
network: Arc<Network>,
services: Arc<LiminalConnectionServices>,
config: &OutboundLinkConfig,
) -> Result<Self, OpenError> {
let peer = network
.host
.recorded_machine(&config.peer_holder)
.ok_or_else(|| OpenError("peer_has_no_recorded_machine".to_owned()))?;
Ok(Self {
network,
peer,
services,
refused: 0,
link_id: config.link_id.clone(),
peer_holder: config.peer_holder.clone(),
})
}
fn require_config(&self, config: &OutboundLinkConfig) -> Result<(), OpenError> {
if config.link_id != self.link_id || config.peer_holder != self.peer_holder {
return Err(OpenError(
"carrier_backend_bound_to_another_link".to_owned(),
));
}
Ok(())
}
}
mod backend;
mod dial;
mod incoming;