Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion crates/omsi-app/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -504,7 +504,7 @@ pub(crate) fn make_app(mut args: Args, server_cfg: Option<server::ServerCfg>) ->
// a dedicated server without its session has nobody to serve: why, and the end (it went
// on running alone, unreachable, with a warning among the log's lines)
if let (Some(cfg), None) = (server_cfg.as_ref(), lan.as_ref()) {
eprintln!("server: cannot open the session on UDP port {} (the line above says why): another server or program uses it, or the system refuses it. Stop the other one, or choose another `port` in {}", cfg.port, args.server.as_deref().map(|p| p.display().to_string()).unwrap_or_else(|| "server.cfg".into()));
eprintln!("server: cannot open the session on UDP/TCP port {} (the line above says why): another server or program uses it, or the system refuses it. Stop the other one, or choose another `port` in {}", cfg.port, args.server.as_deref().map(|p| p.display().to_string()).unwrap_or_else(|| "server.cfg".into()));
std::process::exit(3);
}
// turned away at the door (banned, full, another version): no game to play there
Expand Down
4 changes: 2 additions & 2 deletions crates/omsi-lan/src/game.rs
Original file line number Diff line number Diff line change
Expand Up @@ -572,8 +572,8 @@ pub fn share_mods(args: &mut LanConfig, lan: &mut LanSession) {
if !args.server {
crate::mods::upload::accept(crate::mods::upload::a_hosting_game_takes_uploads());
}
if let (Some(port), Some(key)) = (lan.local_addr().map(|a| a.port()), lan.host_key()) {
crate::mods::serve(port, key, &mod_source(args));
if let (Some(listener), Some(key)) = (lan.take_mods_listener(), lan.host_key()) {
crate::mods::serve(listener, key, &mod_source(args));
}
}
Role::Client => {
Expand Down
10 changes: 5 additions & 5 deletions crates/omsi-lan/src/mods.rs
Original file line number Diff line number Diff line change
Expand Up @@ -449,15 +449,15 @@ pub fn share_maps(on: bool) {
SHARE_MAPS.store(on, std::sync::atomic::Ordering::Relaxed);
}

/// Serve the session's mods on TCP `port` (a thread; the list is made in the background).
pub fn serve(port: u16, key: Arc<omsi_net::hostkey::HostKey>, args: &ModSource) {
/// Serve the session's mods on its already-reserved TCP listener.
pub fn serve(listener: TcpListener, key: Arc<omsi_net::hostkey::HostKey>, args: &ModSource) {
if omsi_cfg::flags::OMSI_NO_LAN_MODS.is_set() {
return;
}
let listener = match TcpListener::bind(("0.0.0.0", port)) {
Ok(l) => l,
let port = match listener.local_addr() {
Ok(address) => address.port(),
Err(e) => {
log::warn!("LAN mods: cannot serve on TCP port {port}: {e} (joining players need the host's mods installed)");
log::warn!("LAN mods: cannot inspect the reserved TCP listener: {e}");
return;
}
};
Expand Down
3 changes: 3 additions & 0 deletions crates/omsi-net/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -14,3 +14,6 @@ tungstenite = { version = "0.30", default-features = false, features = ["handsha

[target.'cfg(unix)'.dependencies]
libc.workspace = true

[target.'cfg(windows)'.dependencies]
windows = { workspace = true, features = ["Win32_Networking_WinSock"] }
72 changes: 72 additions & 0 deletions crates/omsi-net/src/discovery.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
//! Discovery covers the same bounded port range that hosts reserve.

use super::{PORT_RANGE, addrs::LocalAddr};
use std::net::{Ipv4Addr, SocketAddr};

pub(super) fn targets(
port: u16,
addresses: impl IntoIterator<Item = LocalAddr>,
) -> Vec<SocketAddr> {
// Include each interface's subnet broadcast, including VPNs and platforms
// that do not route the limited broadcast to every interface.
let mut ips = vec![Ipv4Addr::BROADCAST, Ipv4Addr::LOCALHOST];
for address in addresses {
let broadcast = address.broadcast.unwrap_or_else(|| {
let o = address.ip.octets();
Ipv4Addr::new(o[0], o[1], o[2], 255)
});
if !ips.contains(&broadcast) {
ips.push(broadcast);
}
}
ips.into_iter()
.flat_map(|ip| {
(0..PORT_RANGE).filter_map(move |offset| {
port.checked_add(offset)
.map(|port| SocketAddr::from((ip, port)))
})
})
.collect()
}

#[cfg(test)]
mod tests {
use super::*;
use crate::{DEFAULT_PORT, addrs::AddrKind};

#[test]
fn every_discovery_address_covers_the_host_range() {
let address = LocalAddr {
ip: Ipv4Addr::new(192, 168, 1, 20),
interface: "test".into(),
kind: AddrKind::Lan,
broadcast: Some(Ipv4Addr::new(192, 168, 3, 255)),
};
let fallback = LocalAddr {
broadcast: None,
..address.clone()
};
let targets = targets(DEFAULT_PORT, [address.clone(), address, fallback]);
assert_eq!(targets.len(), 4 * usize::from(PORT_RANGE));
for ip in [
Ipv4Addr::BROADCAST,
Ipv4Addr::LOCALHOST,
Ipv4Addr::new(192, 168, 3, 255),
Ipv4Addr::new(192, 168, 1, 255),
] {
for port in DEFAULT_PORT..DEFAULT_PORT + PORT_RANGE {
assert!(
targets.contains(&SocketAddr::from((ip, port))),
"{ip}:{port}"
);
}
}
}

#[test]
fn discovery_includes_the_highest_port_without_wrapping() {
let targets = targets(u16::MAX, []);
assert_eq!(targets.len(), 2);
assert!(targets.iter().all(|target| target.port() == u16::MAX));
}
}
223 changes: 223 additions & 0 deletions crates/omsi-net/src/host_ports.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,223 @@
//! Reserve the host's state and content transports before announcing the session.

use super::{
LanSession, PORT_RANGE, PROTOCOL, Role, WorldInfo, bridge, hostkey, local_addrs, session_hex,
};
use std::io::{self, Read};
use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream, UdpSocket};
use std::time::{Duration, Instant};

#[cfg(windows)]
mod windows;

pub(super) fn bind(port: u16) -> io::Result<(UdpSocket, Option<TcpListener>)> {
if omsi_cfg::flags::OMSI_NO_LAN_MODS.is_set() {
return Ok((udp_socket(port)?, None));
}
// Let TCP choose an ephemeral port first; Windows excludes different ranges
// for each transport. The same chosen port must still be reserved for UDP.
let listener = tcp_listener(port)
.map_err(|e| io::Error::new(e.kind(), format!("TCP port {port}: {e}")))?;
let port = listener.local_addr()?.port();
let udp =
udp_socket(port).map_err(|e| io::Error::new(e.kind(), format!("UDP port {port}: {e}")))?;
verify_loopback(&listener).map_err(probe_error)?;
Ok((udp, Some(listener)))
}

#[cfg(not(windows))]
fn udp_socket(port: u16) -> io::Result<UdpSocket> {
UdpSocket::bind(("0.0.0.0", port))
}

#[cfg(windows)]
fn udp_socket(port: u16) -> io::Result<UdpSocket> {
windows::udp_socket(port)
}

#[cfg(not(windows))]
fn tcp_listener(port: u16) -> io::Result<TcpListener> {
TcpListener::bind(("0.0.0.0", port))
}

#[cfg(windows)]
fn tcp_listener(port: u16) -> io::Result<TcpListener> {
windows::tcp_listener(port)
}

/// A wildcard bind alone can succeed while a more specific listener receives loopback.
/// Check the gateway's actual destination, without sending a protocol to another service.
fn verify_loopback(listener: &TcpListener) -> io::Result<()> {
let port = listener.local_addr()?.port();
let target = SocketAddr::from(([127, 0, 0, 1], port));
let probe = TcpStream::connect_timeout(&target, Duration::from_millis(250))?;
let source = probe.local_addr()?;
listener.set_nonblocking(true)?;
let deadline = Instant::now() + Duration::from_millis(250);
loop {
match listener.accept() {
Ok((mut stream, peer)) if peer == source => {
// Close the probe first so the host port does not enter TIME_WAIT.
probe.shutdown(Shutdown::Write)?;
stream.set_read_timeout(Some(Duration::from_millis(250)))?;
stream.read(&mut [0])?;
listener.set_nonblocking(false)?;
return Ok(());
}
Ok(_) => {}
Err(e) if e.kind() == io::ErrorKind::WouldBlock => {}
Err(e) => return Err(e),
}
if Instant::now() >= deadline {
return Err(io::Error::new(
io::ErrorKind::AddrInUse,
format!("TCP {target} does not reach this session's listener"),
));
}
std::thread::sleep(Duration::from_millis(1));
}
}

// A refused connection or failed handshake means this candidate did not reach
// our listener. Retry another port, but do not hide socket resource exhaustion.
fn probe_error(error: io::Error) -> io::Error {
match error.kind() {
io::ErrorKind::ConnectionRefused
| io::ErrorKind::ConnectionReset
| io::ErrorKind::ConnectionAborted
| io::ErrorKind::NotConnected
| io::ErrorKind::TimedOut
| io::ErrorKind::WouldBlock
| io::ErrorKind::UnexpectedEof
| io::ErrorKind::BrokenPipe => io::Error::new(
io::ErrorKind::AddrInUse,
format!("TCP loopback probe failed: {error}"),
),
_ => error,
}
}

// TCP and UDP have separate exclusions. Repeated bind(0) can keep choosing
// adjacent TCP ports inside one UDP exclusion, so spread retries across the
// dynamic range instead of exhausting the retry budget in the same block.
fn candidate(port: u16, offset: u16, seed: u16) -> Option<u16> {
if port != 0 {
return port.checked_add(offset);
}
if offset == 0 {
return Some(0);
}
let stride = (16384 / PORT_RANGE) | 1;
Some(49152 + seed.wrapping_add(offset.wrapping_mul(stride)) % 16384)
}

impl LanSession {
/// Host on a port reserved for both UDP state and TCP content transfers.
/// With `try_next`, a bounded range is tried when either transport is occupied.
pub fn host(
port: u16,
name: &str,
world: WorldInfo,
try_next: bool,
) -> std::io::Result<LanSession> {
let mut last_err = None;
let tries = if try_next || port == 0 { PORT_RANGE } else { 1 };
let seed = super::random_session_id() as u16;
for offset in 0..tries {
let Some(candidate) = candidate(port, offset, seed) else {
break;
};
match bind(candidate) {
Ok((socket, listener)) => {
let p = socket.local_addr()?.port();
socket.set_nonblocking(true)?;
socket.set_broadcast(true)?;
let mut s = LanSession::new(socket, Role::Host, name, world);
s.host_mods = listener;
let key = std::sync::Arc::new(
hostkey::HostKey::generate()
.ok_or_else(|| std::io::Error::other("no key for the session"))?,
);
s.session = key.session();
s.bridge = bridge::Bridge::start(
true,
s.session,
Some(key.clone()),
local_addrs(p),
p,
);
s.host_key = Some(key);
log::info!(
"LAN: hosting session {} on UDP/TCP port {p} as '{name}' (protocol {PROTOCOL}), code {}",
session_hex(s.session),
s.code().map(|c| c.encode()).unwrap_or_default()
);
if port != 0 && p != port {
log::info!("LAN: UDP/TCP port {port} is unavailable; using {p}");
}
return Ok(s);
}
Err(e)
if matches!(
e.kind(),
std::io::ErrorKind::AddrInUse | std::io::ErrorKind::PermissionDenied
) =>
{
log::info!("LAN: cannot reserve host port {candidate}: {e}");
last_err = Some(e);
}
Err(e) => return Err(e),
}
}
Err(last_err.unwrap_or_else(|| std::io::Error::other("no host port available")))
}

/// Transfer the reserved TCP listener without releasing its port between services.
pub fn take_mods_listener(&mut self) -> Option<std::net::TcpListener> {
self.host_mods.take()
}
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn loopback_connection_failures_allow_the_next_candidate() {
for kind in [
io::ErrorKind::ConnectionRefused,
io::ErrorKind::ConnectionReset,
io::ErrorKind::ConnectionAborted,
io::ErrorKind::NotConnected,
io::ErrorKind::TimedOut,
io::ErrorKind::WouldBlock,
io::ErrorKind::UnexpectedEof,
io::ErrorKind::BrokenPipe,
] {
assert_eq!(
probe_error(io::Error::from(kind)).kind(),
io::ErrorKind::AddrInUse
);
}
for kind in [io::ErrorKind::OutOfMemory, io::ErrorKind::Other] {
assert_eq!(probe_error(io::Error::from(kind)).kind(), kind);
}
}

#[test]
fn ephemeral_retries_spread_beyond_one_exclusion_block() {
for seed in [0, 55230, u16::MAX] {
assert_eq!(candidate(0, 0, seed), Some(0));
let ports: Vec<_> = (1..PORT_RANGE)
.map(|offset| candidate(0, offset, seed).unwrap())
.collect();
assert!(ports.iter().all(|&port| port >= 49152));
assert!(ports.windows(2).all(|p| p[0].abs_diff(p[1]) > 1000));
let unique: std::collections::BTreeSet<_> = ports.iter().collect();
assert_eq!(unique.len(), ports.len());
}
assert_eq!(candidate(27015, 1, 0), Some(27016));
assert_eq!(candidate(u16::MAX, 0, 0), Some(u16::MAX));
assert_eq!(candidate(u16::MAX, 1, 0), None);
}
}
Loading
Loading