diff --git a/CHANGELOG.md b/CHANGELOG.md index 4b09a88c..e843a567 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -147,6 +147,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 now share a single batch, which the kernel applies as one transaction, so a refused rebuild leaves the previous table in the packet path. The rules sent are unchanged. +- The gateway's NAT rebuild no longer fails once the table holds more than + about 105 mappings. Each rebuild is one netlink batch. From about 105 + mappings the default socket buffers could not hold its acknowledgements, so + rebuilds were logged as failed although they had taken effect. Past about + 313 mappings the buffers could not hold the batch itself, and new `.fips` + names past that count got a virtual IP with no translation. In releases with + the gateway through 0.5.1, a rebuild past about 313 mappings also deleted the + whole `fips_gateway` table, which stopped every mapping, the `fips0` + masquerade and the port forwards. The rebuild now sizes its send buffer to + the batch and requests one acknowledgement per batch, and NAT errors now + name the kernel errno. A rebuild that still fails is logged, and the next + successful rebuild installs the mapping. - A new OpenWrt install no longer enables and starts `fips-gateway`. The generated postinst turned it on unconditionally, contradicting the init script's own header, the package README and the deployment tutorial, all of diff --git a/src/gateway/nat.rs b/src/gateway/nat.rs index f0c11d60..1dade01b 100644 --- a/src/gateway/nat.rs +++ b/src/gateway/nat.rs @@ -4,7 +4,9 @@ //! for translating between virtual IPs and FIPS mesh addresses. use std::collections::HashMap; +use std::fmt; use std::net::Ipv6Addr; +use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; use tracing::{debug, info}; use rustables::expr::{ @@ -23,6 +25,73 @@ const POSTROUTING_CHAIN: &str = "postrouting"; const DSTNAT_PRIORITY: i32 = -100; const SRCNAT_PRIORITY: i32 = 100; +/// Largest value the kernel accepts for `SO_SNDBUFFORCE`. +/// +/// The kernel clamps the requested value to `i32::MAX / 2` and then doubles +/// it, so the socket's send buffer never exceeds `2 * MAX_SNDBUF`. +const MAX_SNDBUF: libc::c_int = libc::c_int::MAX / 2; + +/// Headroom added to half the batch length when sizing the send buffer. +const SNDBUF_HEADROOM: u64 = 64 * 1024; + +/// The kernel refuses a netlink message longer than the send buffer less +/// this many bytes. +const SNDBUF_OVERHEAD: u64 = 32; + +/// How long the rebuild waits for the kernel's acknowledgement. The rebuild +/// runs inside the gateway's event loop, so this bounds the stall there. +const ACK_TIMEOUT_SECS: libc::time_t = 5; + +/// Length of a `struct nlmsghdr`. +const NLMSG_HDRLEN: usize = 16; + +/// An errno value, displayed by name and number, e.g. `EMSGSIZE (90)`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Errno(pub i32); + +impl Errno { + /// The errno of the last failed libc call on this thread. + fn last() -> Self { + Errno(std::io::Error::last_os_error().raw_os_error().unwrap_or(0)) + } + + /// The symbolic name of the errno, for the values netlink can return. + fn name(self) -> &'static str { + match self.0 { + libc::EPERM => "EPERM", + libc::ENOENT => "ENOENT", + libc::EINTR => "EINTR", + libc::EBADF => "EBADF", + libc::EAGAIN => "EAGAIN", + libc::ENOMEM => "ENOMEM", + libc::EACCES => "EACCES", + libc::EFAULT => "EFAULT", + libc::EBUSY => "EBUSY", + libc::EEXIST => "EEXIST", + libc::ENODEV => "ENODEV", + libc::EINVAL => "EINVAL", + libc::ENFILE => "ENFILE", + libc::EMFILE => "EMFILE", + libc::ENOSPC => "ENOSPC", + libc::ERANGE => "ERANGE", + libc::ELOOP => "ELOOP", + libc::EMSGSIZE => "EMSGSIZE", + libc::EPROTONOSUPPORT => "EPROTONOSUPPORT", + libc::EOPNOTSUPP => "EOPNOTSUPP", + libc::EAFNOSUPPORT => "EAFNOSUPPORT", + libc::ENOBUFS => "ENOBUFS", + libc::ETIMEDOUT => "ETIMEDOUT", + _ => "errno", + } + } +} + +impl fmt::Display for Errno { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{} ({})", self.name(), self.0) + } +} + /// Errors from NAT operations. #[derive(Debug, thiserror::Error)] pub enum NatError { @@ -30,6 +99,18 @@ pub enum NatError { Nftables(String), #[error("rule not found for virtual IP {0}")] RuleNotFound(Ipv6Addr), + /// The kernel rejected a message of the NAT batch; the batch was aborted. + #[error("kernel rejected netlink message {seq} of the NAT batch: {errno}")] + Kernel { errno: Errno, seq: u32 }, + /// A netlink socket call failed. + #[error("netlink socket {op} failed: {errno}")] + Socket { op: &'static str, errno: Errno }, + /// The batch is larger than any netlink send buffer the kernel allows. + #[error( + "NAT batch of {bytes} bytes exceeds the kernel's netlink limit of {} bytes", + admissible_limit() + )] + BatchTooLarge { bytes: usize }, } impl From for NatError { @@ -54,8 +135,8 @@ struct NatMapping { /// One object a NAT rebuild sends, named rather than built. /// /// `rebuild_batches` decides what a rebuild sends and in what order; -/// `send_batches` turns that decision into rustables objects and hands each -/// batch to the kernel. The split is what lets a test see the delete and the +/// `encode_batch` turns that decision into netlink bytes and `send_batch` +/// hands them to the kernel. The split is what lets a test see the delete and the /// recreate share one transaction without a netlink socket, which is the /// property that keeps the table in the packet path. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -188,7 +269,7 @@ impl NatManager { batch.add(&self.table, MsgType::Del); batch .send() - .map_err(|e| NatError::Nftables(e.to_string()))?; + .map_err(|e| NatError::Nftables(error_chain(&e)))?; info!("Deleted nftables table '{TABLE_NAME}'"); Ok(()) @@ -238,133 +319,137 @@ impl NatManager { vec![ops] } - /// Build each op into its rustables object and send the batches in order. - fn send_batches(&self, batches: &[Vec]) -> Result<(), NatError> { - for ops in batches { - let mut batch = Batch::new(); - for op in ops { - match *op { - NatOp::Table(msg_type) => batch.add(&self.table, msg_type), - NatOp::PreChain => batch.add(&self.pre_chain, MsgType::Add), - NatOp::PostChain => batch.add(&self.post_chain, MsgType::Add), - NatOp::FipsMasquerade => { - // Rewrite the source address of traffic leaving fips0. - // Without this, LAN clients' source addresses (e.g. - // fd02::20) are not routable on the mesh, so return - // traffic would be black-holed. - let rule = Rule::new(&self.post_chain)? - .with_expr(Meta::new(MetaType::OifName)) - .with_expr(Cmp::new(CmpOp::Eq, b"fips0\0".to_vec())) - .with_expr(Masquerade::default()); - batch.add(&rule, MsgType::Add); - } - NatOp::Dnat(virtual_ip) => { - let mapping = self.mapping(virtual_ip)?; - let rule = Rule::new(&self.pre_chain)? - .with_expr(Meta::new(MetaType::NfProto)) - .with_expr(Cmp::new(CmpOp::Eq, [libc::NFPROTO_IPV6 as u8])) - .with_expr( - HighLevelPayload::Network(NetworkHeaderField::IPv6( - IPv6HeaderField::Daddr, - )) - .build(), - ) - .with_expr(Cmp::new(CmpOp::Eq, mapping.virtual_ip.octets())) - .with_expr(Immediate::new_data( - mapping.mesh_addr.octets().to_vec(), - Register::Reg1, + /// Build each op into its rustables object and encode the batch. + /// + /// Only the last object before the batch end requests an + /// acknowledgement. rustables sets `NLM_F_ACK` on every message, and one + /// ack per message overflows the socket's receive buffer from about a + /// hundred mappings, after the kernel has already committed the batch. + /// The kernel reports a failing message whatever its flags, so errors + /// stay attributable. + fn encode_batch(&self, ops: &[NatOp]) -> Result, NatError> { + let mut batch = Batch::new(); + for op in ops { + match *op { + NatOp::Table(msg_type) => batch.add(&self.table, msg_type), + NatOp::PreChain => batch.add(&self.pre_chain, MsgType::Add), + NatOp::PostChain => batch.add(&self.post_chain, MsgType::Add), + NatOp::FipsMasquerade => { + // Rewrite the source address of traffic leaving fips0. + // Without this, LAN clients' source addresses (e.g. + // fd02::20) are not routable on the mesh, so return + // traffic would be black-holed. + let rule = Rule::new(&self.post_chain)? + .with_expr(Meta::new(MetaType::OifName)) + .with_expr(Cmp::new(CmpOp::Eq, b"fips0\0".to_vec())) + .with_expr(Masquerade::default()); + batch.add(&rule, MsgType::Add); + } + NatOp::Dnat(virtual_ip) => { + let mapping = self.mapping(virtual_ip)?; + let rule = Rule::new(&self.pre_chain)? + .with_expr(Meta::new(MetaType::NfProto)) + .with_expr(Cmp::new(CmpOp::Eq, [libc::NFPROTO_IPV6 as u8])) + .with_expr( + HighLevelPayload::Network(NetworkHeaderField::IPv6( + IPv6HeaderField::Daddr, )) - .with_expr( - Nat::default() - .with_nat_type(NatType::DNat) - .with_family(ProtocolFamily::Ipv6) - .with_ip_register(Register::Reg1), - ); - batch.add(&rule, MsgType::Add); - } - NatOp::Snat(virtual_ip) => { - let mapping = self.mapping(virtual_ip)?; - let rule = Rule::new(&self.post_chain)? - .with_expr(Meta::new(MetaType::NfProto)) - .with_expr(Cmp::new(CmpOp::Eq, [libc::NFPROTO_IPV6 as u8])) - .with_expr( - HighLevelPayload::Network(NetworkHeaderField::IPv6( - IPv6HeaderField::Saddr, - )) - .build(), - ) - .with_expr(Cmp::new(CmpOp::Eq, mapping.mesh_addr.octets())) - .with_expr(Immediate::new_data( - mapping.virtual_ip.octets().to_vec(), - Register::Reg1, - )) - .with_expr( - Nat::default() - .with_nat_type(NatType::SNat) - .with_family(ProtocolFamily::Ipv6) - .with_ip_register(Register::Reg1), - ); - batch.add(&rule, MsgType::Add); - } - NatOp::PortForward(index) => { - let pf = self.port_forwards.get(index).expect( - "rebuild_batches only emits indices it read from port_forwards", + .build(), + ) + .with_expr(Cmp::new(CmpOp::Eq, mapping.virtual_ip.octets())) + .with_expr(Immediate::new_data( + mapping.mesh_addr.octets().to_vec(), + Register::Reg1, + )) + .with_expr( + Nat::default() + .with_nat_type(NatType::DNat) + .with_family(ProtocolFamily::Ipv6) + .with_ip_register(Register::Reg1), ); - let l4proto: u8 = match pf.proto { - Proto::Tcp => libc::IPPROTO_TCP as u8, - Proto::Udp => libc::IPPROTO_UDP as u8, - }; - let dport_field = match pf.proto { - Proto::Tcp => TransportHeaderField::Tcp(TCPHeaderField::Dport), - Proto::Udp => TransportHeaderField::Udp(UDPHeaderField::Dport), - }; - let target_ip = *pf.target.ip(); - let target_port_be = pf.target.port().to_be_bytes(); - - let rule = Rule::new(&self.pre_chain)? - .with_expr(Meta::new(MetaType::IifName)) - .with_expr(Cmp::new(CmpOp::Eq, b"fips0\0".to_vec())) - .with_expr(Meta::new(MetaType::NfProto)) - .with_expr(Cmp::new(CmpOp::Eq, [libc::NFPROTO_IPV6 as u8])) - .with_expr(Meta::new(MetaType::L4Proto)) - .with_expr(Cmp::new(CmpOp::Eq, [l4proto])) - .with_expr(HighLevelPayload::Transport(dport_field).build()) - .with_expr(Cmp::new(CmpOp::Eq, pf.listen_port.to_be_bytes().to_vec())) - .with_expr(Immediate::new_data( - target_ip.octets().to_vec(), - Register::Reg1, + batch.add(&rule, MsgType::Add); + } + NatOp::Snat(virtual_ip) => { + let mapping = self.mapping(virtual_ip)?; + let rule = Rule::new(&self.post_chain)? + .with_expr(Meta::new(MetaType::NfProto)) + .with_expr(Cmp::new(CmpOp::Eq, [libc::NFPROTO_IPV6 as u8])) + .with_expr( + HighLevelPayload::Network(NetworkHeaderField::IPv6( + IPv6HeaderField::Saddr, )) - .with_expr(Immediate::new_data(target_port_be.to_vec(), Register::Reg2)) - .with_expr( - Nat::default() - .with_nat_type(NatType::DNat) - .with_family(ProtocolFamily::Ipv6) - .with_ip_register(Register::Reg1) - .with_port_register(Register::Reg2), - ); - batch.add(&rule, MsgType::Add); - } - NatOp::LanMasquerade => { - let mut lan_iface = self.lan_interface.clone().into_bytes(); - lan_iface.push(0); - let rule = Rule::new(&self.post_chain)? - .with_expr(Meta::new(MetaType::IifName)) - .with_expr(Cmp::new(CmpOp::Eq, b"fips0\0".to_vec())) - .with_expr(Meta::new(MetaType::OifName)) - .with_expr(Cmp::new(CmpOp::Eq, lan_iface)) - .with_expr(Meta::new(MetaType::NfProto)) - .with_expr(Cmp::new(CmpOp::Eq, [libc::NFPROTO_IPV6 as u8])) - .with_expr(Masquerade::default()); - batch.add(&rule, MsgType::Add); - } + .build(), + ) + .with_expr(Cmp::new(CmpOp::Eq, mapping.mesh_addr.octets())) + .with_expr(Immediate::new_data( + mapping.virtual_ip.octets().to_vec(), + Register::Reg1, + )) + .with_expr( + Nat::default() + .with_nat_type(NatType::SNat) + .with_family(ProtocolFamily::Ipv6) + .with_ip_register(Register::Reg1), + ); + batch.add(&rule, MsgType::Add); + } + NatOp::PortForward(index) => { + let pf = self + .port_forwards + .get(index) + .expect("rebuild_batches only emits indices it read from port_forwards"); + let l4proto: u8 = match pf.proto { + Proto::Tcp => libc::IPPROTO_TCP as u8, + Proto::Udp => libc::IPPROTO_UDP as u8, + }; + let dport_field = match pf.proto { + Proto::Tcp => TransportHeaderField::Tcp(TCPHeaderField::Dport), + Proto::Udp => TransportHeaderField::Udp(UDPHeaderField::Dport), + }; + let target_ip = *pf.target.ip(); + let target_port_be = pf.target.port().to_be_bytes(); + + let rule = Rule::new(&self.pre_chain)? + .with_expr(Meta::new(MetaType::IifName)) + .with_expr(Cmp::new(CmpOp::Eq, b"fips0\0".to_vec())) + .with_expr(Meta::new(MetaType::NfProto)) + .with_expr(Cmp::new(CmpOp::Eq, [libc::NFPROTO_IPV6 as u8])) + .with_expr(Meta::new(MetaType::L4Proto)) + .with_expr(Cmp::new(CmpOp::Eq, [l4proto])) + .with_expr(HighLevelPayload::Transport(dport_field).build()) + .with_expr(Cmp::new(CmpOp::Eq, pf.listen_port.to_be_bytes().to_vec())) + .with_expr(Immediate::new_data( + target_ip.octets().to_vec(), + Register::Reg1, + )) + .with_expr(Immediate::new_data(target_port_be.to_vec(), Register::Reg2)) + .with_expr( + Nat::default() + .with_nat_type(NatType::DNat) + .with_family(ProtocolFamily::Ipv6) + .with_ip_register(Register::Reg1) + .with_port_register(Register::Reg2), + ); + batch.add(&rule, MsgType::Add); + } + NatOp::LanMasquerade => { + let mut lan_iface = self.lan_interface.clone().into_bytes(); + lan_iface.push(0); + let rule = Rule::new(&self.post_chain)? + .with_expr(Meta::new(MetaType::IifName)) + .with_expr(Cmp::new(CmpOp::Eq, b"fips0\0".to_vec())) + .with_expr(Meta::new(MetaType::OifName)) + .with_expr(Cmp::new(CmpOp::Eq, lan_iface)) + .with_expr(Meta::new(MetaType::NfProto)) + .with_expr(Cmp::new(CmpOp::Eq, [libc::NFPROTO_IPV6 as u8])) + .with_expr(Masquerade::default()); + batch.add(&rule, MsgType::Add); } } - - batch - .send() - .map_err(|e| NatError::Nftables(e.to_string()))?; } - Ok(()) + let mut bytes = batch.finalize(); + keep_last_ack(&mut bytes)?; + Ok(bytes) } /// The mapping an op names, or the error a caller can report. @@ -377,10 +462,363 @@ impl NatManager { /// Rebuild the entire nftables table with all current rules, in one /// netlink transaction. fn rebuild(&self) -> Result<(), NatError> { - self.send_batches(&self.rebuild_batches()) + for ops in self.rebuild_batches() { + send_batch(&self.encode_batch(&ops)?)?; + } + Ok(()) } } +/// Largest batch, in bytes, that the kernel can admit in one send. +fn admissible_limit() -> u64 { + 2 * MAX_SNDBUF as u64 - SNDBUF_OVERHEAD +} + +/// The `SO_SNDBUFFORCE` value that lets a batch of `len` bytes through. +/// +/// The kernel doubles the value it is given and refuses a message longer +/// than the result less 32 bytes, so half the length plus headroom is +/// enough. Saturates at the kernel's own clamp rather than wrapping. +fn sndbuf_for(len: usize) -> libc::c_int { + let want = (u64::try_from(len).unwrap_or(u64::MAX) / 2).saturating_add(SNDBUF_HEADROOM); + libc::c_int::try_from(want.min(MAX_SNDBUF as u64)).unwrap_or(MAX_SNDBUF) +} + +/// Refuse a batch the kernel could not accept at any send-buffer size. +fn check_admissible(len: usize) -> Result<(), NatError> { + if u64::try_from(len).unwrap_or(u64::MAX) > admissible_limit() { + return Err(NatError::BatchTooLarge { bytes: len }); + } + Ok(()) +} + +/// The fields of one `struct nlmsghdr` that the NAT batch code reads. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct NlHeader { + /// Offset of the header within the buffer. + offset: usize, + /// `nlmsg_len`: header plus payload, without alignment padding. + len: usize, + kind: u16, + flags: u16, + seq: u32, +} + +impl NlHeader { + /// The message's payload, after the header. + fn payload<'a>(&self, buf: &'a [u8]) -> &'a [u8] { + &buf[self.offset + NLMSG_HDRLEN..self.offset + self.len] + } +} + +/// Walk every netlink message header in `buf`. +/// +/// Fails on a header shorter than `struct nlmsghdr` or a length that runs +/// past the end of the buffer. +fn nl_headers(buf: &[u8]) -> Result, NatError> { + let mut headers = Vec::new(); + let mut offset = 0; + while offset < buf.len() { + let rest = &buf[offset..]; + if rest.len() < NLMSG_HDRLEN { + return Err(NatError::Nftables(format!( + "malformed netlink message at offset {offset}: {} bytes left, header needs {NLMSG_HDRLEN}", + rest.len() + ))); + } + let field = |at: usize, width: usize| &rest[at..at + width]; + let len = u32::from_ne_bytes(field(0, 4).try_into().expect("4-byte slice")) as usize; + if len < NLMSG_HDRLEN || len > rest.len() { + return Err(NatError::Nftables(format!( + "malformed netlink message at offset {offset}: length {len} with {} bytes left", + rest.len() + ))); + } + headers.push(NlHeader { + offset, + len, + kind: u16::from_ne_bytes(field(4, 2).try_into().expect("2-byte slice")), + flags: u16::from_ne_bytes(field(6, 2).try_into().expect("2-byte slice")), + seq: u32::from_ne_bytes(field(8, 4).try_into().expect("4-byte slice")), + }); + // Netlink messages are 4-byte aligned. + offset += (len + 3) & !3; + } + Ok(headers) +} + +/// The finalized batch's objects: every message between the batch begin +/// and the batch end. +fn batch_objects(headers: &[NlHeader]) -> Result<&[NlHeader], NatError> { + match headers { + [_begin, objects @ .., _end] if !objects.is_empty() => Ok(objects), + _ => Err(NatError::Nftables(format!( + "NAT batch holds {} messages; it needs a begin, an object and an end", + headers.len() + ))), + } +} + +/// Clear `NLM_F_ACK` on every message of a finalized batch except the last +/// object before the batch end. +fn keep_last_ack(buf: &mut [u8]) -> Result<(), NatError> { + let headers = nl_headers(buf)?; + let last = batch_objects(&headers)? + .last() + .expect("batch_objects returns a non-empty slice") + .offset; + let ack = libc::NLM_F_ACK as u16; + for header in &headers { + let flags = if header.offset == last { + header.flags | ack + } else { + header.flags & !ack + }; + buf[header.offset + 6..header.offset + 8].copy_from_slice(&flags.to_ne_bytes()); + } + Ok(()) +} + +/// What the kernel's replies to a NAT batch have shown so far. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum AckState { + /// No verdict yet; read another datagram. + Pending, + /// The acknowledgement of the batch's last object arrived with no error + /// before it. + Done, +} + +/// Reads the kernel's replies to a NAT batch, one datagram at a time. +/// +/// The kernel aborts the whole batch when any message fails, yet it still +/// acknowledges the last message after the error. So the last ack alone does +/// not prove success: any error that arrives before it fails the batch. +struct AckReader { + /// Sequence number of the one message that requested an ack. + last_seq: u32, +} + +impl AckReader { + /// Consume one received datagram, which may carry several messages. + fn feed(&self, datagram: &[u8]) -> Result { + for header in nl_headers(datagram)? { + if i32::from(header.kind) != libc::NLMSG_ERROR { + continue; + } + let payload = header.payload(datagram); + if payload.len() < 4 { + return Err(NatError::Nftables(format!( + "malformed netlink error message: {} payload bytes, error field needs 4", + payload.len() + ))); + } + let error = i32::from_ne_bytes(payload[..4].try_into().expect("4-byte slice")); + if error != 0 { + return Err(NatError::Kernel { + errno: Errno(error.saturating_neg()), + seq: header.seq, + }); + } + if header.seq == self.last_seq { + return Ok(AckState::Done); + } + } + Ok(AckState::Pending) + } +} + +/// Render an error with every source beneath it, so a wrapped errno is kept. +fn error_chain(error: &dyn std::error::Error) -> String { + let mut text = error.to_string(); + let mut source = error.source(); + while let Some(inner) = source { + text.push_str(": "); + text.push_str(&inner.to_string()); + source = inner.source(); + } + text +} + +/// Set an integer socket option. +fn set_int_opt( + sock: &OwnedFd, + level: libc::c_int, + name: libc::c_int, + value: libc::c_int, +) -> Result<(), Errno> { + // SAFETY: the descriptor is open for the life of `sock`, and the pointer + // and length describe `value`, a c_int. + let rc = unsafe { + libc::setsockopt( + sock.as_raw_fd(), + level, + name, + (&value as *const libc::c_int).cast(), + std::mem::size_of::() as libc::socklen_t, + ) + }; + if rc < 0 { Err(Errno::last()) } else { Ok(()) } +} + +/// Size of the buffer each reply datagram is read into. +/// +/// The largest message nftables sends back, as rustables computes it +/// (`nft_nlmsg_maxsize`, which it does not export), and at least 64 KiB. +fn recv_buffer_len() -> usize { + // SAFETY: sysconf has no preconditions. + let page = unsafe { libc::sysconf(libc::_SC_PAGESIZE) }; + (usize::from(u16::MAX) + usize::try_from(page).unwrap_or(0)).max(64 * 1024) +} + +/// Open a netfilter netlink socket sized for a batch of `len` bytes. +fn open_batch_socket(len: usize) -> Result { + let socket_err = |op| move |errno| NatError::Socket { op, errno }; + + // SAFETY: socket has no memory preconditions. + let fd = unsafe { + libc::socket( + libc::AF_NETLINK, + libc::SOCK_RAW | libc::SOCK_CLOEXEC, + libc::NETLINK_NETFILTER, + ) + }; + if fd < 0 { + return Err(socket_err("open")(Errno::last())); + } + // SAFETY: socket returned a new descriptor that nothing else owns. + let sock = unsafe { OwnedFd::from_raw_fd(fd) }; + + // SAFETY: an all-zero sockaddr_nl is valid; the family is set below. + let mut addr: libc::sockaddr_nl = unsafe { std::mem::zeroed() }; + addr.nl_family = libc::AF_NETLINK as libc::sa_family_t; + // SAFETY: the pointer and length describe `addr`, a sockaddr_nl. + let rc = unsafe { + libc::bind( + sock.as_raw_fd(), + (&addr as *const libc::sockaddr_nl).cast(), + std::mem::size_of::() as libc::socklen_t, + ) + }; + if rc < 0 { + return Err(socket_err("bind")(Errno::last())); + } + + // Without CAP_NET_ADMIN the forced size is refused; the plain option is + // then capped by wmem_max, and an oversized batch fails with EMSGSIZE. + let sndbuf = sndbuf_for(len); + match set_int_opt(&sock, libc::SOL_SOCKET, libc::SO_SNDBUFFORCE, sndbuf) { + Err(Errno(libc::EPERM)) => { + set_int_opt(&sock, libc::SOL_SOCKET, libc::SO_SNDBUF, sndbuf) + .map_err(socket_err("setsockopt SO_SNDBUF"))?; + } + other => other.map_err(socket_err("setsockopt SO_SNDBUFFORCE"))?, + } + // An error ack then carries only the failing header, not the message. + set_int_opt(&sock, libc::SOL_NETLINK, libc::NETLINK_CAP_ACK, 1) + .map_err(socket_err("setsockopt NETLINK_CAP_ACK"))?; + + let timeout = libc::timeval { + tv_sec: ACK_TIMEOUT_SECS, + tv_usec: 0, + }; + // SAFETY: the pointer and length describe `timeout`, a timeval. + let rc = unsafe { + libc::setsockopt( + sock.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_RCVTIMEO, + (&timeout as *const libc::timeval).cast(), + std::mem::size_of::() as libc::socklen_t, + ) + }; + if rc < 0 { + return Err(socket_err("setsockopt SO_RCVTIMEO")(Errno::last())); + } + Ok(sock) +} + +/// Send one encoded NAT batch and wait for the kernel's verdict. +/// +/// The batch goes out in a single send, so the kernel applies it as one +/// transaction. The send buffer is sized to the batch, because the default +/// one refuses a message past about 208 KiB, which is about 313 mappings. +fn send_batch(bytes: &[u8]) -> Result<(), NatError> { + check_admissible(bytes.len())?; + let headers = nl_headers(bytes)?; + let reader = AckReader { + last_seq: batch_objects(&headers)? + .last() + .expect("batch_objects returns a non-empty slice") + .seq, + }; + let sock = open_batch_socket(bytes.len())?; + + let sent = loop { + // SAFETY: the pointer and length describe `bytes`. + let rc = unsafe { libc::send(sock.as_raw_fd(), bytes.as_ptr().cast(), bytes.len(), 0) }; + if rc >= 0 { + break rc as usize; + } + let errno = Errno::last(); + if errno.0 != libc::EINTR { + return Err(NatError::Socket { op: "send", errno }); + } + }; + if sent != bytes.len() { + return Err(NatError::Nftables(format!( + "netlink send took {sent} of {} bytes", + bytes.len() + ))); + } + + let mut buf = vec![0u8; recv_buffer_len()]; + loop { + // MSG_TRUNC makes a netlink recv return the datagram's full length, + // so a reply larger than the buffer is detected rather than cut. + // SAFETY: the pointer and length describe `buf`. + let rc = unsafe { + libc::recv( + sock.as_raw_fd(), + buf.as_mut_ptr().cast(), + buf.len(), + libc::MSG_TRUNC, + ) + }; + if rc < 0 { + let errno = Errno::last(); + if errno.0 == libc::EINTR { + continue; + } + // EAGAIN is the receive timeout. ENOBUFS means error acks + // overflowed the receive buffer, since only one ack is requested. + return Err(NatError::Socket { op: "recv", errno }); + } + let got = rc as usize; + if got == 0 { + return Err(NatError::Nftables( + "netlink socket returned no reply".into(), + )); + } + if got > buf.len() { + return Err(NatError::Nftables(format!( + "netlink reply of {got} bytes truncated to {}", + buf.len() + ))); + } + if reader.feed(&buf[..got])? == AckState::Done { + return Ok(()); + } + } +} + +// Coverage gap. These tests run unprivileged and open no netlink socket, so +// three failure paths in `open_socket` and `send_batch` go unexercised here. +// The receive timeout firing and a reply longer than the buffer (seen through +// `MSG_TRUNC`) need a kernel fault to provoke, so nothing runs them. The +// `SO_SNDBUF` fallback after `SO_SNDBUFFORCE` returns `EPERM` needs a process +// without CAP_NET_ADMIN, and the gateway suite's container is privileged, so +// nothing runs that either. The gateway suite covers only the success path. #[cfg(test)] mod tests { use super::*; @@ -497,4 +935,176 @@ mod tests { 1 ); } + + /// The encoded rebuild of a manager holding `count` mappings. + fn encoded_rebuild(count: u16) -> Vec { + let mgr = manager_with_mappings(count); + let ops = mgr.rebuild_batches().remove(0); + mgr.encode_batch(&ops).expect("the rebuild encodes") + } + + /// One netlink message, padded to 4 bytes. + fn nlmsg(kind: u16, seq: u32, payload: &[u8]) -> Vec { + let len = (NLMSG_HDRLEN + payload.len()) as u32; + let mut msg = Vec::new(); + msg.extend_from_slice(&len.to_ne_bytes()); + msg.extend_from_slice(&kind.to_ne_bytes()); + msg.extend_from_slice(&0u16.to_ne_bytes()); + msg.extend_from_slice(&seq.to_ne_bytes()); + msg.extend_from_slice(&0u32.to_ne_bytes()); + msg.extend_from_slice(payload); + msg.resize(msg.len().div_ceil(4) * 4, 0); + msg + } + + /// The kernel's `NLMSG_ERROR` reply to message `seq`, as it sends it on a + /// socket with `NETLINK_CAP_ACK`: the error, then the request's header. + fn ack(seq: u32, error: i32) -> Vec { + let mut payload = error.to_ne_bytes().to_vec(); + payload.extend_from_slice(&nlmsg(0x0a00, seq, &[])[..NLMSG_HDRLEN]); + nlmsg(libc::NLMSG_ERROR as u16, seq, &payload) + } + + /// The largest batch the kernel admits: twice its send-buffer clamp, + /// less the 32 bytes netlink reserves. + const KERNEL_BATCH_LIMIT: usize = 2_147_483_614; + + #[test] + fn rebuild_for_2000_mappings_requests_exactly_one_ack_on_the_last_message() { + let encoded = encoded_rebuild(2000); + assert!( + encoded.len() > 212_960, + "the 2000-mapping batch ({} bytes) must be past the default \ + netlink send limit for this test to cover the large case", + encoded.len() + ); + + let headers = nl_headers(&encoded).expect("the batch parses"); + assert_eq!( + headers.first().map(|h| h.kind), + Some(libc::NFNL_MSG_BATCH_BEGIN as u16) + ); + assert_eq!( + headers.last().map(|h| h.kind), + Some(libc::NFNL_MSG_BATCH_END as u16) + ); + let acked: Vec = headers + .iter() + .enumerate() + .filter(|(_, h)| h.flags & libc::NLM_F_ACK as u16 != 0) + .map(|(i, _)| i) + .collect(); + assert_eq!( + acked, + vec![headers.len() - 2], + "only the last object before the batch end may request an ack; \ + one ack per message overflows the receive buffer after the \ + kernel has committed the batch" + ); + } + + #[test] + fn sndbuf_for_admits_the_2000_mapping_batch_and_small_batches_after_kernel_doubling() { + let large = encoded_rebuild(2000).len(); + for len in [large, 0, 1, 212_961] { + let sndbuf = sndbuf_for(len); + assert!( + 2 * sndbuf as u64 - 32 >= len as u64, + "a send buffer of {sndbuf}, doubled by the kernel, refuses a \ + {len}-byte batch" + ); + } + } + + #[test] + fn sndbuf_for_saturates_at_the_kernel_clamp_for_huge_batches() { + for len in [2 * MAX_SNDBUF as usize, usize::MAX] { + assert_eq!(sndbuf_for(len), i32::MAX / 2, "sndbuf_for({len})"); + } + } + + #[test] + fn check_admissible_refuses_a_batch_larger_than_the_kernel_can_accept() { + assert!(check_admissible(KERNEL_BATCH_LIMIT).is_ok()); + for len in [KERNEL_BATCH_LIMIT + 1, usize::MAX] { + match check_admissible(len) { + Err(e @ NatError::BatchTooLarge { bytes }) => { + assert_eq!(bytes, len); + assert!( + e.to_string().contains(&len.to_string()), + "the error names the batch size: {e}" + ); + } + other => panic!("a {len}-byte batch was admitted: {other:?}"), + } + } + } + + #[test] + fn ack_reader_fails_on_an_error_that_precedes_the_last_ack() { + // The kernel aborts the batch on a failing rule mid-batch, reports + // that rule's error, and still acknowledges the last message. + let reader = AckReader { last_seq: 4000 }; + let error = ack(1234, -libc::ENOENT); + let last = ack(4000, 0); + + let expect_error = |result: Result| match result { + Err(NatError::Kernel { errno, seq }) => { + assert_eq!(errno, Errno(libc::ENOENT)); + assert_eq!(seq, 1234); + } + other => panic!("the aborted batch was not reported: {other:?}"), + }; + + expect_error(reader.feed(&error)); + expect_error(reader.feed(&[error.clone(), last.clone()].concat())); + } + + #[test] + fn ack_reader_is_done_only_on_the_last_sequence_ack() { + let reader = AckReader { last_seq: 10 }; + + assert_eq!(reader.feed(&ack(10, 0)).expect("parses"), AckState::Done); + assert_eq!(reader.feed(&ack(5, 0)).expect("parses"), AckState::Pending); + assert_eq!( + reader + .feed(&nlmsg(libc::NLMSG_NOOP as u16, 10, &[])) + .expect("parses"), + AckState::Pending + ); + assert_eq!( + reader + .feed(&[ack(5, 0), ack(10, 0)].concat()) + .expect("parses"), + AckState::Done + ); + + let whole = ack(10, 0); + assert!( + reader.feed(&whole[..8]).is_err(), + "a header shorter than 16 bytes" + ); + let mut overlong = whole.clone(); + overlong[..4].copy_from_slice(&((whole.len() + 4) as u32).to_ne_bytes()); + assert!( + reader.feed(&overlong).is_err(), + "a length past the end of the datagram" + ); + let short = nlmsg(libc::NLMSG_ERROR as u16, 10, &[0, 0]); + assert!( + reader.feed(&short[..NLMSG_HDRLEN + 2]).is_err(), + "an NLMSG_ERROR payload shorter than its error field" + ); + } + + #[test] + fn kernel_error_display_names_the_errno() { + let text = NatError::Kernel { + errno: Errno(libc::EMSGSIZE), + seq: 7, + } + .to_string(); + assert!(text.contains("EMSGSIZE"), "{text}"); + assert!(text.contains(&format!("({})", libc::EMSGSIZE)), "{text}"); + } } diff --git a/testing/static/scripts/gateway-test.sh b/testing/static/scripts/gateway-test.sh index 9cb8a605..331037ea 100755 --- a/testing/static/scripts/gateway-test.sh +++ b/testing/static/scripts/gateway-test.sh @@ -477,6 +477,256 @@ else check "Gateway shutdown (no completion message in logs)" 1 fi +# Phase 11: NAT rebuild past the default netlink socket limits +# +# Every change rebuilds the whole fips_gateway table in one netlink batch. +# With the default socket buffers that batch failed from about 105 mappings +# (the acks overflowed the receive buffer, after the commit) and past about +# 313 (the batch overflowed the send buffer, and nothing was committed). +# Drive 400 new names through a gateway whose mappings outlive the phase and +# judge the result on the kernel's own table, not on the daemon's debug-level +# success line, which the suite's log level does not show. Runs after phase +# 10, so the gateway container is stopped when it starts, and it leaves it +# stopped. +echo "" +echo "Phase 11: NAT rebuild past default socket limits" + +NATBIG_NAMES=400 +NATBIG_CAP=180 +NATBIG_SETTLE=30 + +natbig_now() { + date -u +%s +} + +natbig_slice() { + NATBIG_LOG=$(docker logs --timestamps --since "$NATBIG_STARTED" "$GATEWAY" 2>&1) +} + +natbig_allocated() { + natbig_slice + NATBIG_ALLOCATED=$(grep -c "Allocated virtual IP" <<< "$NATBIG_LOG" || true) +} + +# Rules the kernel holds right now. A failed listing is recorded through +# NATBIG_RC, never read as a table with no rules. +natbig_kernel() { + NATBIG_RC=0 + NATBIG_NFT=$(docker exec "$GATEWAY" nft list table inet fips_gateway 2>&1) || NATBIG_RC=$? + NATBIG_DNAT=$(grep -cE "daddr fd01:[0-9a-f:]* .*dnat" <<< "$NATBIG_NFT" || true) + NATBIG_SNAT=$(grep -cE "saddr [0-9a-f:]+ .*snat" <<< "$NATBIG_NFT" || true) + NATBIG_MASQ=$(grep -c "masquerade" <<< "$NATBIG_NFT" || true) +} + +natbig_phase() { + local config_file="$GENERATED_DIR/gateway/node-a.yaml" + local expect_rev + expect_rev=$(git -C "$SCRIPT_DIR" rev-parse --short=10 HEAD) + + # Rewrite in place (same inode): the container sees the host file through + # a single-file bind mount, which a replace-by-rename would leave behind. + python3 - "$config_file" <<'PYEOF' +import sys, yaml +path = sys.argv[1] +with open(path, "r+") as f: + cfg = yaml.safe_load(f) + cfg["gateway"]["dns"]["ttl"] = 1800 + cfg["gateway"]["pool_grace_period"] = 1800 + f.seek(0) + yaml.dump(cfg, f, default_flow_style=False, sort_keys=False) + f.truncate() +PYEOF + + docker start "$GATEWAY" >/dev/null + NATBIG_STARTED=$(docker inspect -f '{{.State.StartedAt}}' "$GATEWAY") + local t0 + t0=$(natbig_now) + echo " Gateway started at $NATBIG_STARTED (expect rev $expect_rev)" + + local seen_ttl seen_grace + seen_ttl=$(docker exec "$GATEWAY" grep -c "ttl: 1800" /etc/fips/fips.yaml || true) + seen_grace=$(docker exec "$GATEWAY" grep -c "pool_grace_period: 1800" /etc/fips/fips.yaml || true) + if [ "$seen_ttl" -ge 1 ] && [ "$seen_grace" -ge 1 ]; then + check "NAT batch: container sees ttl 1800 and grace 1800" 0 + else + check "NAT batch: container config rewrite (ttl: $seen_ttl, grace: $seen_grace)" 1 + return 0 + fi + + if wait_for_peers "$GATEWAY" 2 60; then + check "NAT batch: gateway peers after restart" 0 + else + check "NAT batch: gateway peers after restart" 1 + return 0 + fi + local ready=false probe + for _ in $(seq 1 60); do + probe=$(docker exec "$CLIENT" dig +short AAAA "${NPUB_B}.fips" @${GW_DNS} 2>/dev/null || true) + if echo "$probe" | grep -q "^fd01::"; then + ready=true + break + fi + sleep 1 + done + if [ "$ready" = true ]; then + check "NAT batch: gateway DNS answers after restart" 0 + else + check "NAT batch: gateway DNS answers after restart" 1 + return 0 + fi + + sleep 1 + natbig_allocated + local rev_lines + rev_lines=$(grep -cE "fips-gateway [^ ]+ \(rev ${expect_rev}\) starting" <<< "$NATBIG_LOG" || true) + if [ "$rev_lines" -eq 1 ]; then + check "NAT batch: startup line reads rev ${expect_rev}) with no -dirty" 0 + else + check "NAT batch: startup line for rev ${expect_rev} (found $rev_lines)" 1 + return 0 + fi + local baseline="$NATBIG_ALLOCATED" + local target=$((baseline + NATBIG_NAMES)) + echo " Baseline allocations after readiness: $baseline; target $target" + if [ "$baseline" -ge 1 ]; then + check "NAT batch: readiness probe allocated (baseline $baseline)" 0 + else + check "NAT batch: readiness probe allocated (baseline $baseline)" 1 + return 0 + fi + + # Names: real keys, since the daemon parses each one as a public key. + local names_file have_names + names_file=$(mktemp) + docker exec "$GATEWAY" bash -c \ + "for i in \$(seq 1 $NATBIG_NAMES); do fipsctl keygen --stdout; done" \ + | grep '^npub1' >"$names_file" || true + have_names=$(wc -l <"$names_file") + if [ "$have_names" -lt "$NATBIG_NAMES" ]; then + check "NAT batch: generated $NATBIG_NAMES names (got $have_names)" 1 + rm -f "$names_file" + return 0 + fi + + # Closed-loop AAAA driver, 4 workers, one fresh socket per query. + docker exec -i "$CLIENT" sh -c 'cat > /tmp/gw_natbig.py' <<'PYEOF' +import random, socket, struct, sys, threading +server = sys.argv[1] +names = [n.strip() for n in sys.stdin if n.strip()] +lock = threading.Lock() +counts = {"answered": 0, "servfail": 0, "timeout": 0, "other": 0} +def query(name): + qid = random.getrandbits(16) + pkt = struct.pack(">HHHHHH", qid, 0x0100, 1, 0, 0, 0) + for label in (name + ".fips").split("."): + raw = label.encode() + pkt += bytes([len(raw)]) + raw + pkt += b"\x00" + struct.pack(">HH", 28, 1) + s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) + s.settimeout(6) + try: + s.sendto(pkt, (server, 53)) + while True: + data, _ = s.recvfrom(4096) + if len(data) >= 12 and struct.unpack(">H", data[:2])[0] == qid: + break + except socket.timeout: + return "timeout" + finally: + s.close() + flags, _, ancount = struct.unpack(">HHH", data[2:8]) + rcode = flags & 0xF + if rcode == 2: + return "servfail" + if rcode == 0 and ancount > 0: + return "answered" + return "other" +def worker(): + while True: + with lock: + if not names: + return + name = names.pop() + outcome = query(name) + with lock: + counts[outcome] += 1 +threads = [threading.Thread(target=worker) for _ in range(4)] +for t in threads: + t.start() +for t in threads: + t.join() +print(" ".join(f"{k}={v}" for k, v in counts.items())) +PYEOF + + local remaining out rc=0 + remaining=$((NATBIG_CAP - ($(natbig_now) - t0))) + # timeout reads 0 as no limit and refuses a negative duration. + if [ "$remaining" -le 0 ]; then + check "NAT batch: setup exceeded ${NATBIG_CAP}s cap" 1 + rm -f "$names_file" + return 0 + fi + out=$(docker exec -i "$CLIENT" timeout "$remaining" \ + python3 /tmp/gw_natbig.py "$GW_DNS" <"$names_file" 2>&1) || rc=$? + rm -f "$names_file" + echo " [$(($(natbig_now) - t0))s] sent $NATBIG_NAMES names: $out (rc=$rc)" + natbig_allocated + if [ "$rc" -eq 0 ] && [ "$NATBIG_ALLOCATED" -eq "$target" ]; then + check "NAT batch: $NATBIG_ALLOCATED live mappings allocated" 0 + else + check "NAT batch: live mappings allocated ($NATBIG_ALLOCATED of $target, rc $rc)" 1 + fi + + # Settle on the kernel: wait until it holds a DNAT rule per allocation, + # or the cap expires. Reaching the cap decides nothing by itself; the + # checks below do. + local settle=0 + natbig_kernel + while [ "$NATBIG_DNAT" -ne "$NATBIG_ALLOCATED" ] && [ "$settle" -lt "$NATBIG_SETTLE" ]; do + sleep 1 + settle=$((settle + 1)) + natbig_kernel + done + # An error logged just after the last commit still counts. + sleep 2 + natbig_kernel + natbig_allocated + local nat_fail + nat_fail=$(grep -c "Failed to add NAT rules" <<< "$NATBIG_LOG" || true) + echo " [$(($(natbig_now) - t0))s] allocated=$NATBIG_ALLOCATED nat_add_fail=$nat_fail" \ + "table_rc=$NATBIG_RC dnat=$NATBIG_DNAT snat=$NATBIG_SNAT masquerade=$NATBIG_MASQ settle=${settle}s" + if [ "$nat_fail" -gt 0 ]; then + grep "Failed to add NAT rules" <<< "$NATBIG_LOG" | sed 's/\x1b\[[0-9;]*m//g' \ + | sed -n '1p;$p' | sed 's/^/ /' + fi + + if [ "$nat_fail" -eq 0 ]; then + check "NAT batch: no NAT rebuild failed" 0 + else + check "NAT batch: NAT rebuilds failed ($nat_fail)" 1 + fi + if [ "$NATBIG_RC" -eq 0 ]; then + check "NAT batch: nft lists the fips_gateway table" 0 + else + check "NAT batch: nft list table failed (rc $NATBIG_RC)" 1 + fi + if [ "$NATBIG_DNAT" -eq "$NATBIG_ALLOCATED" ] && [ "$NATBIG_SNAT" -eq "$NATBIG_ALLOCATED" ]; then + check "NAT batch: kernel holds a DNAT and SNAT rule per mapping ($NATBIG_DNAT)" 0 + else + check "NAT batch: kernel rules (dnat $NATBIG_DNAT, snat $NATBIG_SNAT, allocated $NATBIG_ALLOCATED)" 1 + fi + if [ "$NATBIG_MASQ" -eq 2 ]; then + check "NAT batch: fips0 and LAN masquerades present" 0 + else + check "NAT batch: masquerade rules ($NATBIG_MASQ, expected 2)" 1 + fi + + docker stop --time=10 "$GATEWAY" >/dev/null 2>&1 || true + echo " Phase time: $(($(natbig_now) - t0))s" +} + +natbig_phase + echo "" echo "=== Results: $PASSED passed, $FAILED failed ===" [ "$FAILED" -eq 0 ] && exit 0 || exit 1