use std::{
io::{self, IoSlice, IoSliceMut},
net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV6},
os::fd::AsRawFd,
time::SystemTime,
};
use nix::{
libc,
sys::socket::{
cmsg_space, recvmsg, sendmsg, setsockopt, sockopt, ControlMessage, ControlMessageOwned,
MsgFlags, RecvMsg, SockaddrStorage,
},
};
use tokio::{io::Interest, net::UdpSocket};
#[cfg(target_os = "linux")]
use std::time::{Duration, UNIX_EPOCH};
#[cfg(target_os = "linux")]
use nix::sys::time::TimeSpec;
use super::{ReceivedDatagram, ReplySource};
#[cfg(target_os = "linux")]
const DESTINATION_V4_LEN: usize = cmsg_space::<libc::in_pktinfo>();
#[cfg(any(target_os = "macos", target_os = "freebsd"))]
const DESTINATION_V4_LEN: usize = cmsg_space::<libc::in_addr>();
const DESTINATION_V6_LEN: usize = cmsg_space::<libc::in6_pktinfo>();
#[cfg(target_os = "linux")]
const TIMESTAMP_LEN: usize = cmsg_space::<libc::timespec>();
#[cfg(not(target_os = "linux"))]
const TIMESTAMP_LEN: usize = 0;
const CONTROL_LEN: usize = DESTINATION_V4_LEN + DESTINATION_V6_LEN + TIMESTAMP_LEN;
const _: () = {
assert!(CONTROL_LEN >= DESTINATION_V4_LEN + TIMESTAMP_LEN);
assert!(CONTROL_LEN >= DESTINATION_V6_LEN + TIMESTAMP_LEN);
};
pub(super) fn configure_destination_metadata(socket: &UdpSocket, is_ipv4: bool) -> io::Result<()> {
if is_ipv4 {
configure_ipv4_destination(socket)
} else {
setsockopt(socket, sockopt::Ipv6RecvPacketInfo, &true).map_err(io::Error::from)
}
}
#[cfg(target_os = "linux")]
fn configure_ipv4_destination(socket: &UdpSocket) -> io::Result<()> {
setsockopt(socket, sockopt::Ipv4PacketInfo, &true).map_err(io::Error::from)
}
#[cfg(any(target_os = "macos", target_os = "freebsd"))]
fn configure_ipv4_destination(socket: &UdpSocket) -> io::Result<()> {
setsockopt(socket, sockopt::Ipv4RecvDstAddr, &true).map_err(io::Error::from)
}
#[cfg(target_os = "linux")]
pub(super) fn try_configure_kernel_rx_timestamp(socket: &UdpSocket) -> bool {
setsockopt(socket, sockopt::ReceiveTimestampns, &true).is_ok()
}
pub(super) async fn receive_with_destination(
socket: &UdpSocket,
buffer: &mut [u8],
) -> io::Result<Option<ReceivedDatagram>> {
receive_ancillary(socket, buffer, true).await
}
#[cfg(target_os = "linux")]
pub(super) async fn receive_bound(
socket: &UdpSocket,
buffer: &mut [u8],
) -> io::Result<Option<ReceivedDatagram>> {
receive_ancillary(socket, buffer, false).await
}
const INTERRUPTED_YIELD_THRESHOLD: u32 = 64;
async fn receive_ancillary(
socket: &UdpSocket,
buffer: &mut [u8],
require_destination: bool,
) -> io::Result<Option<ReceivedDatagram>> {
let mut consecutive_interrupted = 0_u32;
loop {
socket.readable().await?;
match socket.try_io(Interest::READABLE, || {
receive_once(socket, buffer, require_destination)
}) {
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
consecutive_interrupted = 0;
continue;
}
Err(error) if error.kind() == io::ErrorKind::Interrupted => {
consecutive_interrupted += 1;
if consecutive_interrupted >= INTERRUPTED_YIELD_THRESHOLD {
consecutive_interrupted = 0;
tokio::task::yield_now().await;
}
continue;
}
result => return result,
}
}
}
fn receive_once(
socket: &UdpSocket,
buffer: &mut [u8],
require_destination: bool,
) -> io::Result<Option<ReceivedDatagram>> {
let mut control = ControlBuffer::new();
let mut iov = [IoSliceMut::new(buffer)];
let message = recvmsg::<SockaddrStorage>(
socket.as_raw_fd(),
&mut iov,
Some(control.as_mut_slice()),
MsgFlags::MSG_DONTWAIT,
)?;
Ok(received_datagram(&message, require_destination))
}
fn received_datagram(
message: &RecvMsg<'_, '_, SockaddrStorage>,
require_destination: bool,
) -> Option<ReceivedDatagram> {
if is_unusable(message.flags, require_destination) {
return None;
}
let metadata = receive_metadata(message);
Some(ReceivedDatagram {
len: message.bytes,
peer: peer_endpoint(message.address.as_ref()?)?,
reply_source: resolved_reply_source(metadata.reply_source, require_destination)?,
kernel_rx_timestamp: metadata.kernel_rx_timestamp,
})
}
fn is_unusable(flags: MsgFlags, require_destination: bool) -> bool {
flags.contains(MsgFlags::MSG_TRUNC)
|| (require_destination && flags.contains(MsgFlags::MSG_CTRUNC))
}
fn resolved_reply_source(
recovered: Option<ReplySource>,
require_destination: bool,
) -> Option<ReplySource> {
if require_destination {
recovered
} else {
Some(ReplySource::Bound)
}
}
fn peer_endpoint(address: &SockaddrStorage) -> Option<SocketAddr> {
if let Some(peer) = address.as_sockaddr_in() {
return Some(SocketAddr::from(*peer));
}
address
.as_sockaddr_in6()
.map(|peer| SocketAddr::from(*peer))
}
#[derive(Default)]
struct ReceiveMetadata {
reply_source: Option<ReplySource>,
kernel_rx_timestamp: Option<SystemTime>,
}
fn receive_metadata(message: &RecvMsg<'_, '_, SockaddrStorage>) -> ReceiveMetadata {
let mut metadata = ReceiveMetadata::default();
let Ok(control_messages) = message.cmsgs() else {
return metadata;
};
for control_message in control_messages {
match control_message {
#[cfg(target_os = "linux")]
ControlMessageOwned::Ipv4PacketInfo(info) => {
metadata.reply_source = ipv4_reply_source(ipv4_addr(info.ipi_addr));
}
#[cfg(any(target_os = "macos", target_os = "freebsd"))]
ControlMessageOwned::Ipv4RecvDstAddr(address) => {
metadata.reply_source = ipv4_reply_source(ipv4_addr(address));
}
ControlMessageOwned::Ipv6PacketInfo(info) => {
metadata.reply_source = ipv6_reply_source(info);
}
#[cfg(target_os = "linux")]
ControlMessageOwned::ScmTimestampns(timestamp) => {
metadata.kernel_rx_timestamp = system_time_from_timespec(timestamp);
}
_ => {}
}
}
metadata
}
fn ipv6_reply_source(info: libc::in6_pktinfo) -> Option<ReplySource> {
let address = Ipv6Addr::from(info.ipi6_addr.s6_addr);
(!address.is_unspecified()).then_some(ReplySource::V6 {
address,
interface_index: info.ipi6_ifindex,
})
}
#[cfg(target_os = "linux")]
fn system_time_from_timespec(timespec: TimeSpec) -> Option<SystemTime> {
let seconds = u64::try_from(timespec.tv_sec()).ok()?;
let nanos = u32::try_from(timespec.tv_nsec()).ok()?;
if nanos >= 1_000_000_000 {
return None;
}
UNIX_EPOCH.checked_add(Duration::new(seconds, nanos))
}
fn ipv4_reply_source(address: Ipv4Addr) -> Option<ReplySource> {
(!address.is_unspecified()).then_some(ReplySource::V4 { address })
}
fn ipv4_addr(address: libc::in_addr) -> Ipv4Addr {
Ipv4Addr::from(address.s_addr.to_ne_bytes())
}
fn in_addr(address: Ipv4Addr) -> libc::in_addr {
libc::in_addr {
s_addr: u32::from_ne_bytes(address.octets()),
}
}
pub(super) async fn send_from_source(
socket: &UdpSocket,
bytes: &[u8],
peer: SocketAddr,
reply_source: ReplySource,
) -> io::Result<usize> {
loop {
socket.writable().await?;
match socket.try_io(Interest::WRITABLE, || {
send_once(socket, bytes, peer, reply_source)
}) {
Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue,
result => return result,
}
}
}
fn send_once(
socket: &UdpSocket,
bytes: &[u8],
peer: SocketAddr,
reply_source: ReplySource,
) -> io::Result<usize> {
let destination = SockaddrStorage::from(peer);
let iov = [IoSlice::new(bytes)];
let sent = match reply_source {
ReplySource::Bound => sendmsg(
socket.as_raw_fd(),
&iov,
&[],
MsgFlags::MSG_DONTWAIT,
Some(&destination),
)?,
ReplySource::V4 { address } => send_from_ipv4(socket, &iov, &destination, address)?,
ReplySource::V6 {
address,
interface_index,
} => {
let info = libc::in6_pktinfo {
ipi6_addr: libc::in6_addr {
s6_addr: address.octets(),
},
ipi6_ifindex: egress_interface_index(address, interface_index),
};
let destination = SockaddrStorage::from(as_ipv6_endpoint(peer));
sendmsg(
socket.as_raw_fd(),
&iov,
&[ControlMessage::Ipv6PacketInfo(&info)],
MsgFlags::MSG_DONTWAIT,
Some(&destination),
)?
}
};
Ok(sent)
}
fn egress_interface_index(address: Ipv6Addr, interface_index: u32) -> u32 {
if address.is_unicast_link_local() {
interface_index
} else {
0
}
}
fn as_ipv6_endpoint(peer: SocketAddr) -> SocketAddrV6 {
match peer {
SocketAddr::V6(peer) => peer,
SocketAddr::V4(peer) => SocketAddrV6::new(peer.ip().to_ipv6_mapped(), peer.port(), 0, 0),
}
}
#[cfg(any(target_os = "linux", target_os = "macos"))]
fn send_from_ipv4(
socket: &UdpSocket,
iov: &[IoSlice<'_>],
destination: &SockaddrStorage,
address: Ipv4Addr,
) -> io::Result<usize> {
let info = libc::in_pktinfo {
ipi_ifindex: 0,
ipi_spec_dst: in_addr(address),
ipi_addr: in_addr(Ipv4Addr::UNSPECIFIED),
};
sendmsg(
socket.as_raw_fd(),
iov,
&[ControlMessage::Ipv4PacketInfo(&info)],
MsgFlags::MSG_DONTWAIT,
Some(destination),
)
.map_err(io::Error::from)
}
#[cfg(target_os = "freebsd")]
fn send_from_ipv4(
socket: &UdpSocket,
iov: &[IoSlice<'_>],
destination: &SockaddrStorage,
address: Ipv4Addr,
) -> io::Result<usize> {
let source = in_addr(address);
sendmsg(
socket.as_raw_fd(),
iov,
&[ControlMessage::Ipv4SendSrcAddr(&source)],
MsgFlags::MSG_DONTWAIT,
Some(destination),
)
.map_err(io::Error::from)
}
#[repr(align(8))]
struct ControlBuffer([u8; CONTROL_LEN]);
impl ControlBuffer {
fn new() -> Self {
Self([0; CONTROL_LEN])
}
fn as_mut_slice(&mut self) -> &mut [u8] {
&mut self.0
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn only_a_link_local_source_pins_the_egress_interface() {
const ARRIVED_ON: u32 = 7;
for (address, expected, why) in [
(
"fe80::1",
ARRIVED_ON,
"a link-local source needs its link named",
),
(
"fe80::7858:baff:fe91:183d",
ARRIVED_ON,
"and so does any other one",
),
("2001:db8::1", 0, "a global source identifies itself"),
("fd00::1", 0, "so does a unique-local one"),
("::1", 0, "and loopback"),
("::ffff:127.0.0.2", 0, "and an IPv4-mapped destination"),
] {
assert_eq!(
egress_interface_index(address.parse().unwrap(), ARRIVED_ON),
expected,
"{address}: {why}"
);
}
}
#[test]
fn only_a_listener_that_needs_a_destination_is_stopped_by_a_missing_one() {
let recovered = ReplySource::V4 {
address: Ipv4Addr::LOCALHOST,
};
assert_eq!(
resolved_reply_source(Some(recovered), true),
Some(recovered),
"a wildcard listener replies from the destination it recovered"
);
assert_eq!(
resolved_reply_source(None, true),
None,
"and cannot answer a request whose destination did not arrive"
);
assert_eq!(
resolved_reply_source(None, false),
Some(ReplySource::Bound),
"an explicit listener never needed one"
);
assert_eq!(
resolved_reply_source(Some(recovered), false),
Some(ReplySource::Bound),
"and sends from its bind even if one turned up"
);
}
#[test]
fn control_truncation_only_condemns_a_datagram_whose_destination_it_may_have_taken() {
for (flags, require_destination, unusable, why) in [
(MsgFlags::empty(), true, false, "a clean wildcard receive"),
(MsgFlags::empty(), false, false, "a clean explicit receive"),
(
MsgFlags::MSG_TRUNC,
false,
true,
"a cut payload is not what the client sent",
),
(
MsgFlags::MSG_TRUNC,
true,
true,
"and no less so for a wildcard listener",
),
(
MsgFlags::MSG_CTRUNC,
true,
true,
"cut control data may have taken the destination",
),
(
MsgFlags::MSG_CTRUNC,
false,
false,
"but for an explicit listener it can only have cost a timestamp",
),
(
MsgFlags::MSG_TRUNC | MsgFlags::MSG_CTRUNC,
false,
true,
"the payload rule still applies",
),
] {
assert_eq!(is_unusable(flags, require_destination), unusable, "{why}");
}
}
#[cfg(target_os = "linux")]
#[cfg_attr(target_env = "musl", allow(deprecated))]
#[test]
fn a_kernel_timestamp_converts_only_where_it_describes_an_instant() {
assert_eq!(
system_time_from_timespec(TimeSpec::new(1, 2))
.unwrap()
.duration_since(UNIX_EPOCH)
.unwrap(),
Duration::new(1, 2)
);
assert_eq!(
system_time_from_timespec(TimeSpec::new(0, 0)),
Some(UNIX_EPOCH),
"the epoch itself is representable"
);
assert_eq!(system_time_from_timespec(TimeSpec::new(-1, 0)), None);
assert_eq!(system_time_from_timespec(TimeSpec::new(0, -1)), None);
assert_eq!(
system_time_from_timespec(TimeSpec::new(0, 1_000_000_000)),
None
);
assert_eq!(
system_time_from_timespec(TimeSpec::new(libc::time_t::MIN, libc::c_long::MIN)),
None,
"and the extremes convert rather than panic"
);
let _ = system_time_from_timespec(TimeSpec::new(libc::time_t::MAX, 999_999_999));
}
}