Migrate to Tokio asynchronous runtime

Bumps the all-updates group with 2 updates in the / directory: [libc](https://github.com/rust-lang/libc) and [serde_json](https://github.com/serde-rs/json).


Updates `libc` from 0.2.189 to 0.2.190
- [Release notes](https://github.com/rust-lang/libc/releases)
- [Changelog](https://github.com/rust-lang/libc/blob/0.2.190/CHANGELOG.md)
- [Commits](https://github.com/rust-lang/libc/compare/0.2.189...0.2.190)

Updates `serde_json` from 1.0.151 to 1.0.154
- [Release notes](https://github.com/serde-rs/json/releases)
- [Commits](https://github.com/serde-rs/json/compare/v1.0.151...v1.0.154)

---
updated-dependencies:
- dependency-name: libc
  dependency-version: 0.2.190
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: all-updates
- dependency-name: serde_json
  dependency-version: 1.0.154
  dependency-type: direct:production
  update-type: version-update:semver-patch
  dependency-group: all-updates
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.cargo/config.toml

@@ -1,2 +1,5 @@
+[env]
+MACOSX_DEPLOYMENT_TARGET = "26.0"
+
 [target.aarch64-apple-darwin]
 runner = 'sudo -E'

.github/workflows/ci.yml

@@ -17,7 +17,7 @@ jobs:
     runs-on: macos-26
     timeout-minutes: 30
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0
       - name: Install Rust
         run: |
           curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --profile minimal
@@ -34,7 +34,7 @@ jobs:
     runs-on: macos-26
     timeout-minutes: 30
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0
       - name: Install Rust
         run: |
           curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --profile minimal

.github/workflows/release.yml

@@ -17,7 +17,7 @@ jobs:
     environment: publish
     timeout-minutes: 60
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0
         with:
           fetch-depth: 0
           persist-credentials: false
@@ -45,7 +45,7 @@ jobs:
           permission-contents: write
           permission-pull-requests: write
       - name: Release
-        uses: goreleaser/goreleaser-action@v7
+        uses: goreleaser/goreleaser-action@f06c13b6b1a9625abc9e6e439d9c05a8f2190e94 # v7.2.3
         with:
           distribution: goreleaser-pro
           version: "~> v2"
@@ -61,7 +61,7 @@ jobs:
     runs-on: macos-26
     timeout-minutes: 60
     steps:
-      - uses: actions/checkout@v6
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0
         with:
           fetch-depth: 0
           persist-credentials: false
@@ -72,13 +72,13 @@ jobs:
       - name: Install release targets
         run: rustup target add aarch64-apple-darwin x86_64-apple-darwin
       - name: Release dry run
-        uses: goreleaser/goreleaser-action@v7
+        uses: goreleaser/goreleaser-action@f06c13b6b1a9625abc9e6e439d9c05a8f2190e94 # v7.2.3
         with:
           distribution: goreleaser-pro
           version: "~> v2"
           args: release --skip=publish --snapshot --clean
       - name: Upload dry-run artifacts
-        uses: actions/upload-artifact@v6
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6.0.0
         with:
           name: softnet-snapshot
           path: dist/**

.goreleaser.yml

@@ -5,6 +5,8 @@ project_name: softnet
 builds:
   - builder: rust
     command: build
+    env:
+      - SOFTNET_VERSION={{ .Version }}-{{ .ShortCommit }}
     targets:
       - aarch64-apple-darwin
       - x86_64-apple-darwin
@@ -36,5 +38,5 @@ brews:
     skip_upload: auto
     custom_block: |
       on_macos do
-        depends_on :macos => :sequoia
+        depends_on :macos => :tahoe
       end

Cargo.toml

@@ -14,9 +14,12 @@ debug = true
 [dependencies]
 smoltcp = "0"
 libc = "0"
-polling = "3"
+tokio = { version = "1", features = ["rt", "sync", "net", "signal", "time", "macros", "io-util"] }
+tokio-util = { version = "0.7", features = ["codec"] }
+bytes = "1"
+futures-util = { version = "0.3", default-features = false, features = ["std", "sink"] }
 dhcproto = { git = "https://github.com/bluecatengineering/dhcproto.git", branch = "master" }
-vmnet = "0.5.1"
+vmnet = "0.7.0"
 clap = { version = "4", features = ["derive"] }
 mac_address = "1"
 privdrop = "0"
@@ -24,12 +27,13 @@ anyhow = { version = "1", features = ["backtrace"] }
 ip_network = "0"
 uzers = "0"
 system-configuration = "0"
-num_enum = "0"
 sentry = { version = "0", features = ["debug-images"] }
 sentry-anyhow = { version = "0", features = ["backtrace"] }
-nix = { version = "0", features = ["signal", "socket"] }
+nix = { version = "0", features = ["socket"] }
+network-interface = "2"
 prefix-trie = "0"
 ipnet = "2"
+rand = "0.10"
 oslog = "0.2.0"
 log = "0.4.29"
 serial_test = "4"

README.md

@@ -23,12 +23,7 @@ Softnet solves two problems:
 2. DHCP exhaustion
     * macOS built-in DHCP-server allocates a `/24` subnet with 86400 seconds lease time by default, which only allows for ~253 VMs a day (or 1 VM every ~6 minutes) to be spawned without causing a denial-of-service, which is pretty limiting for CI services like Cirrus CI
 
-And assumes that:
-
-1. Tart gives it's VMs unique MAC-addresses
-2. macOS built-in DHCP-server won't re-use the IP-addresses from it's pool until their lease expire
-
-...otherwise it's possible for two VMs to receive an identical IP-address from the macOS built-in DHCP-server (even in the presence of Softnet's packet filtering) and thus bypass the protections offered by Softnet.
+Each VM gets a separate [`vmnet` network](https://developer.apple.com/documentation/vmnet/vmnet_network_create(_:_:)) with a private `/29` subnet selected to avoid overlap with existing host interface subnets. Softnet reserves an IP-address for the VM's MAC-address, delivers it through its own DHCP-server, and uses the reserved address for packet filtering.
 
 ### Stateful flow authorization
 
@@ -59,6 +54,8 @@ For ICMP, stateful flow authorization supports only echo requests and replies.
 
 ## Installing
 
+Softnet requires macOS 26 or newer; building also requires the macOS 26 SDK or newer.
+
 For proper functioning, Softnet binary requires two things:
 
 * a [SUID-bit](https://en.wikipedia.org/wiki/Setuid#SUID) to be set on the binary or a [passwordless sudo](https://serverfault.com/questions/160581/how-to-setup-passwordless-sudo-on-linux) to be configured, which effectively gives the binary `root` privileges

lib/dhcp_server/mod.rs

@@ -0,0 +1,389 @@
+use anyhow::Result;
+use dhcproto::v4::{
+    CLIENT_PORT, DhcpOption, Flags, HType, Message, MessageType, Opcode, OptionCode, SERVER_PORT,
+};
+use dhcproto::{Decodable, Encodable};
+use smoltcp::phy::ChecksumCapabilities;
+use smoltcp::wire::{
+    ETHERNET_HEADER_LEN, EthernetAddress, EthernetFrame, EthernetProtocol, IpAddress, IpProtocol,
+    Ipv4Address, Ipv4Packet, Ipv4Repr, UDP_HEADER_LEN, UdpPacket, UdpRepr,
+};
+
+pub(crate) const DHCP_SERVER_MAC: EthernetAddress = EthernetAddress([0x02, 0, 0, 0, 0, 1]);
+
+pub(crate) struct DhcpServer {
+    vm_mac: EthernetAddress,
+    vm_ip: Ipv4Address,
+    subnet_mask: Ipv4Address,
+    gateway_ip: Ipv4Address,
+}
+
+impl DhcpServer {
+    /// Creates a DHCP server for the VM's reserved IPv4 configuration.
+    pub fn new(
+        vm_mac: EthernetAddress,
+        vm_ip: Ipv4Address,
+        subnet_mask: Ipv4Address,
+        gateway_ip: Ipv4Address,
+    ) -> Self {
+        Self {
+            vm_mac,
+            vm_ip,
+            subnet_mask,
+            gateway_ip,
+        }
+    }
+
+    /// Returns a reply for a request sent to the DHCP server port.
+    pub fn receive_vm(
+        &self,
+        ip: &Ipv4Packet<&[u8]>,
+        udp: &UdpPacket<&[u8]>,
+    ) -> Result<Option<Vec<u8>>> {
+        // Accept only intact DHCP client datagrams with valid checksums
+        if udp.src_port() != CLIENT_PORT
+            || !ip.verify_checksum()
+            || ip.more_frags()
+            || ip.frag_offset() != 0
+            || UdpRepr::parse(
+                udp,
+                &IpAddress::Ipv4(ip.src_addr()),
+                &IpAddress::Ipv4(ip.dst_addr()),
+                &ChecksumCapabilities::default(),
+            )
+            .is_err()
+        {
+            return Ok(None);
+        }
+
+        // Only accept broadcasts or requests addressed to our gateway
+        if !(ip.dst_addr().is_broadcast() || ip.dst_addr() == self.gateway_ip) {
+            return Ok(None);
+        }
+
+        // Decode the DHCP request
+        let Ok(request) = Message::from_bytes(udp.payload()) else {
+            return Ok(None);
+        };
+
+        // Allow REQUESTs from an obsolete address so the VM can receive a NAK
+        if request.opts().msg_type() != Some(MessageType::Request)
+            && !ip.src_addr().is_unspecified()
+            && self.vm_ip != ip.src_addr()
+        {
+            return Ok(None);
+        }
+
+        // Build a reply if the request is supported
+        if let Some(reply) = self.receive(&request) {
+            // Broadcast NAKs and requested broadcasts; otherwise unicast to the VM
+            let (destination_mac, destination_ip) =
+                if reply.opts().msg_type() == Some(MessageType::Nak) || request.flags().broadcast()
+                {
+                    (EthernetAddress::BROADCAST, Ipv4Address::BROADCAST)
+                } else if !request.ciaddr().is_unspecified() {
+                    (self.vm_mac, request.ciaddr())
+                } else {
+                    (self.vm_mac, reply.yiaddr())
+                };
+
+            // Encode the reply for delivery to the VM
+            let packet = encode_packet(
+                &reply,
+                DHCP_SERVER_MAC,
+                destination_mac,
+                self.gateway_ip,
+                destination_ip,
+                SERVER_PORT,
+                CLIENT_PORT,
+            )?;
+            return Ok(Some(packet));
+        }
+
+        Ok(None)
+    }
+
+    /// Handles a DHCP request from the attached VM.
+    fn receive(&self, request: &Message) -> Option<Message> {
+        // Only accept direct Ethernet requests from the attached VM
+        if request.opcode() != Opcode::BootRequest
+            || request.htype() != HType::Eth
+            || request.hlen() != self.vm_mac.0.len() as u8
+            || request.chaddr() != self.vm_mac.0
+            || !request.giaddr().is_unspecified()
+        {
+            return None;
+        }
+
+        // Ignore requests that select another DHCP server
+        let server_id = match request.opts().get(OptionCode::ServerIdentifier) {
+            Some(DhcpOption::ServerIdentifier(address)) => Some(*address),
+            _ => None,
+        };
+        if server_id.is_some_and(|address| address != self.gateway_ip) {
+            return None;
+        }
+
+        // Read the requested address, if supplied
+        let requested_ip = match request.opts().get(OptionCode::RequestedIpAddress) {
+            Some(DhcpOption::RequestedIpAddress(address)) => Some(*address),
+            _ => None,
+        };
+
+        // Offer the reserved address or validate a request for it
+        match request.opts().msg_type()? {
+            MessageType::Discover if request.ciaddr().is_unspecified() && server_id.is_none() => {
+                Some(self.address_reply(request, MessageType::Offer))
+            }
+            MessageType::Request => {
+                // Use option 50 for SELECTING/INIT-REBOOT and ciaddr for RENEWING/REBINDING
+                let address = match (server_id, requested_ip, request.ciaddr().is_unspecified()) {
+                    (_, Some(address), true) => address,
+                    (None, None, false) => request.ciaddr(),
+                    _ => return None,
+                };
+
+                // Reject requests for any address other than the VM's reservation
+                if address != self.vm_ip {
+                    return Some(self.reply(request, MessageType::Nak));
+                }
+
+                // Acknowledge the reserved address
+                Some(self.address_reply(request, MessageType::Ack))
+            }
+            _ => None,
+        }
+    }
+
+    /// Builds an OFFER or ACK with the VM's reserved address and network configuration.
+    fn address_reply(&self, request: &Message, message_type: MessageType) -> Message {
+        // Include the reserved address and network configuration with an infinite lease
+        let mut reply = self.reply(request, message_type);
+        reply.set_yiaddr(self.vm_ip);
+        let options = reply.opts_mut();
+        options.insert(DhcpOption::SubnetMask(self.subnet_mask));
+        options.insert(DhcpOption::Router(vec![self.gateway_ip]));
+        options.insert(DhcpOption::DomainNameServer(vec![self.gateway_ip]));
+        options.insert(DhcpOption::AddressLeaseTime(u32::MAX));
+
+        reply
+    }
+
+    /// Builds a DHCP reply with the request's transaction and client identifiers.
+    fn reply(&self, request: &Message, message_type: MessageType) -> Message {
+        // Create a reply for the same client and transaction
+        let nak = message_type == MessageType::Nak;
+        let mut reply = Message::default();
+        reply
+            .set_xid(request.xid())
+            .set_chaddr(request.chaddr())
+            .set_opcode(Opcode::BootReply);
+
+        // Broadcast NAKs; otherwise retain the client's address and flags
+        if nak {
+            reply.set_flags(Flags::default().set_broadcast());
+        } else {
+            reply
+                .set_ciaddr(request.ciaddr())
+                .set_flags(request.flags());
+        }
+
+        // Identify the reply type and this DHCP server
+        let options = reply.opts_mut();
+        options.insert(DhcpOption::MessageType(message_type));
+        options.insert(DhcpOption::ServerIdentifier(self.gateway_ip));
+
+        // RFC 6842 requires echoing the client's identifier in every reply
+        if let Some(identifier) = request.opts().get(OptionCode::ClientIdentifier) {
+            options.insert(identifier.clone());
+        }
+
+        reply
+    }
+}
+
+/// Encodes a DHCP message in an Ethernet frame with IPv4 and UDP headers.
+fn encode_packet(
+    message: &Message,
+    source_mac: EthernetAddress,
+    destination_mac: EthernetAddress,
+    source_ip: Ipv4Address,
+    destination_ip: Ipv4Address,
+    source_port: u16,
+    destination_port: u16,
+) -> Result<Vec<u8>> {
+    // Pad the encoded message to the minimum BOOTP size, including short NAKs
+    let mut payload = message.to_vec()?;
+    payload.resize(payload.len().max(dhcproto::v4::MIN_PACKET_SIZE), 0);
+
+    // Allocate space for Ethernet, IPv4, UDP, and the DHCP payload
+    let ip_repr = Ipv4Repr {
+        src_addr: source_ip,
+        dst_addr: destination_ip,
+        next_header: IpProtocol::Udp,
+        payload_len: UDP_HEADER_LEN + payload.len(),
+        hop_limit: 64,
+    };
+    let mut bytes = vec![0; ETHERNET_HEADER_LEN + ip_repr.buffer_len() + ip_repr.payload_len];
+
+    // Write the Ethernet header
+    let mut ethernet = EthernetFrame::new_unchecked(bytes.as_mut_slice());
+    ethernet.set_src_addr(source_mac);
+    ethernet.set_dst_addr(destination_mac);
+    ethernet.set_ethertype(EthernetProtocol::Ipv4);
+
+    // Write the IPv4 header and checksum
+    let mut ip = Ipv4Packet::new_unchecked(ethernet.payload_mut());
+    ip_repr.emit(&mut ip, &ChecksumCapabilities::default());
+
+    // Write the UDP header, DHCP payload, and checksum
+    let mut udp = UdpPacket::new_unchecked(ip.payload_mut());
+    UdpRepr {
+        src_port: source_port,
+        dst_port: destination_port,
+    }
+    .emit(
+        &mut udp,
+        &IpAddress::Ipv4(source_ip),
+        &IpAddress::Ipv4(destination_ip),
+        payload.len(),
+        |buffer| buffer.copy_from_slice(&payload),
+        &ChecksumCapabilities::default(),
+    );
+
+    Ok(bytes)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{DHCP_SERVER_MAC, DhcpServer, encode_packet};
+    use dhcproto::Decodable;
+    use dhcproto::v4::{
+        CLIENT_PORT, DhcpOption, Message, MessageType, Opcode, OptionCode, SERVER_PORT,
+    };
+    use smoltcp::phy::ChecksumCapabilities;
+    use smoltcp::wire::{
+        EthernetAddress, EthernetFrame, Ipv4Address, Ipv4Packet, Ipv4Repr, UdpPacket, UdpRepr,
+    };
+
+    const VM: EthernetAddress = EthernetAddress([2, 0, 0, 0, 0, 2]);
+    const ADDRESS: Ipv4Address = Ipv4Address::new(192, 168, 1, 2);
+    const GATEWAY: Ipv4Address = Ipv4Address::new(192, 168, 1, 1);
+    const MASK: Ipv4Address = Ipv4Address::new(255, 255, 255, 252);
+    const CLIENT_ID: &[u8] = &[1, 2, 0, 0, 0, 0, 2];
+
+    #[test]
+    fn assigns_reserved_address() {
+        // Create a server and a client request
+        let dhcp = DhcpServer::new(VM, ADDRESS, MASK, GATEWAY);
+        let mut message = request(MessageType::Discover);
+
+        // Discover the reserved address
+        let (offer, destination) = exchange(&dhcp, &message, Ipv4Address::BROADCAST);
+        assert_eq!(offer.opts().msg_type(), Some(MessageType::Offer));
+        assert_eq!(offer.yiaddr(), ADDRESS);
+        assert_eq!(destination, ADDRESS);
+
+        // Request the offered address
+        message
+            .opts_mut()
+            .insert(DhcpOption::MessageType(MessageType::Request));
+        message
+            .opts_mut()
+            .insert(DhcpOption::RequestedIpAddress(offer.yiaddr()));
+        message
+            .opts_mut()
+            .insert(DhcpOption::ServerIdentifier(GATEWAY));
+        let (ack, destination) = exchange(&dhcp, &message, Ipv4Address::BROADCAST);
+
+        // Check the address and configuration in the acknowledgement
+        assert_eq!(ack.opcode(), Opcode::BootReply);
+        assert_eq!(ack.opts().msg_type(), Some(MessageType::Ack));
+        assert_eq!(ack.xid(), message.xid());
+        assert_eq!(ack.chaddr(), VM.0);
+        assert_eq!(ack.yiaddr(), ADDRESS);
+        assert_eq!(destination, ADDRESS);
+        for option in [
+            DhcpOption::ServerIdentifier(GATEWAY),
+            DhcpOption::SubnetMask(MASK),
+            DhcpOption::Router(vec![GATEWAY]),
+            DhcpOption::DomainNameServer(vec![GATEWAY]),
+            DhcpOption::AddressLeaseTime(u32::MAX),
+            DhcpOption::ClientIdentifier(CLIENT_ID.to_vec()),
+        ] {
+            assert_eq!(ack.opts().get(OptionCode::from(&option)), Some(&option));
+        }
+    }
+
+    #[test]
+    fn renews_reserved_address_and_rejects_stale_address() {
+        // Renew the reserved address
+        let dhcp = DhcpServer::new(VM, ADDRESS, MASK, GATEWAY);
+        let mut message = request(MessageType::Request);
+        message.set_ciaddr(ADDRESS);
+        let (ack, destination) = exchange(&dhcp, &message, GATEWAY);
+        assert_eq!(ack.opts().msg_type(), Some(MessageType::Ack));
+        assert_eq!(ack.yiaddr(), ADDRESS);
+        assert_eq!(destination, ADDRESS);
+
+        // Broadcast a NAK when the client renews an old address
+        message.set_ciaddr(Ipv4Address::new(192, 168, 2, 2));
+        let (nak, destination) = exchange(&dhcp, &message, GATEWAY);
+        assert_eq!(nak.opts().msg_type(), Some(MessageType::Nak));
+        assert!(nak.yiaddr().is_unspecified());
+        assert_eq!(destination, Ipv4Address::BROADCAST);
+    }
+
+    fn request(kind: MessageType) -> Message {
+        let mut message = Message::default();
+        message.set_xid(42).set_chaddr(&VM.0);
+        message.opts_mut().insert(DhcpOption::MessageType(kind));
+        message
+            .opts_mut()
+            .insert(DhcpOption::ClientIdentifier(CLIENT_ID.to_vec()));
+        message
+    }
+
+    fn exchange(
+        dhcp: &DhcpServer,
+        request: &Message,
+        destination: Ipv4Address,
+    ) -> (Message, Ipv4Address) {
+        // Encode the client request
+        let frame = encode_packet(
+            request,
+            VM,
+            EthernetAddress::BROADCAST,
+            request.ciaddr(),
+            destination,
+            CLIENT_PORT,
+            SERVER_PORT,
+        )
+        .unwrap();
+
+        // Pass the packet to the DHCP server
+        let ethernet = EthernetFrame::new_checked(frame.as_slice()).unwrap();
+        let ip = Ipv4Packet::new_checked(ethernet.payload()).unwrap();
+        let udp = UdpPacket::new_unchecked(ip.payload());
+        let frame = dhcp
+            .receive_vm(&ip, &udp)
+            .unwrap()
+            .expect("expected a DHCP reply");
+
+        // Decode and validate the reply
+        let ethernet = EthernetFrame::new_checked(frame.as_slice()).unwrap();
+        let ip = Ipv4Packet::new_unchecked(ethernet.payload());
+        Ipv4Repr::parse(&ip, &ChecksumCapabilities::default()).unwrap();
+        let udp = UdpPacket::new_unchecked(ip.payload());
+        UdpRepr::parse(
+            &udp,
+            &ip.src_addr().into(),
+            &ip.dst_addr().into(),
+            &ChecksumCapabilities::default(),
+        )
+        .unwrap();
+        assert_eq!(ethernet.src_addr(), DHCP_SERVER_MAC);
+        assert_eq!(ip.src_addr(), GATEWAY);
+        (Message::from_bytes(udp.payload()).unwrap(), ip.dst_addr())
+    }
+}

lib/dhcp_snooper.rs

@@ -1,182 +0,0 @@
-use dhcproto::Decodable;
-use dhcproto::v4::{DhcpOption, HType, Message, MessageType, Opcode, OptionCode};
-use smoltcp::wire::Ipv4Address;
-use std::collections::HashSet;
-use std::time::Duration;
-
-#[derive(Default)]
-pub struct DhcpSnooper {
-    vm_mac_address: [u8; 6],
-    vm_lease: Option<Lease>,
-    uncertainty_duration: Duration,
-}
-
-impl DhcpSnooper {
-    pub fn new(uncertainty_duration: Duration, vm_mac_address: [u8; 6]) -> Self {
-        DhcpSnooper {
-            vm_mac_address,
-            uncertainty_duration,
-            ..Default::default()
-        }
-    }
-
-    pub fn register_dhcp_reply(&mut self, dhcp_packet: &[u8]) {
-        let mut decoder = dhcproto::v4::Decoder::new(dhcp_packet);
-
-        let message = match dhcproto::v4::Message::decode(&mut decoder) {
-            Ok(message) => message,
-            Err(_) => return,
-        };
-
-        // Decoded DHCP replies may be broadcast[1], so additionally validate the BOOTP client
-        // hardware address to avoid acting on another VM's lease transition
-        //
-        // [1]: https://datatracker.ietf.org/doc/html/rfc2131#section-4.1
-        if !message_matches_bootp_client(&message, Opcode::BootReply, self.vm_mac_address) {
-            return;
-        }
-
-        match message.opts().msg_type() {
-            Some(MessageType::Ack) => {
-                let lease_time = match message.opts().get(OptionCode::AddressLeaseTime) {
-                    Some(DhcpOption::AddressLeaseTime(lease_time)) => lease_time,
-                    _ => return,
-                };
-
-                let dns_ips = match message.opts().get(OptionCode::DomainNameServer) {
-                    Some(DhcpOption::DomainNameServer(dns_ips)) => {
-                        HashSet::from_iter(dns_ips.iter().cloned())
-                    }
-                    _ => HashSet::new(),
-                };
-
-                let mut lease_duration = Duration::from_secs(*lease_time as u64);
-
-                // Adjust for uncertainty caused by using a coarse clock
-                lease_duration = lease_duration.saturating_sub(self.uncertainty_duration);
-
-                self.vm_lease = Some(Lease::new(message.yiaddr(), lease_duration, dns_ips))
-            }
-            Some(MessageType::Nak) => {
-                self.vm_lease = None;
-            }
-            _ => {}
-        };
-    }
-
-    #[cfg(test)]
-    pub(crate) fn set_lease(&mut self, vm_lease: Option<Lease>) {
-        self.vm_lease = vm_lease
-    }
-
-    pub fn lease(&self) -> &Option<Lease> {
-        &self.vm_lease
-    }
-
-    pub(crate) fn address_and_dns_ips(&self) -> Option<(Ipv4Address, HashSet<Ipv4Address>)> {
-        let lease = self.vm_lease.as_ref().filter(|lease| lease.valid())?;
-        Some((lease.address(), lease.dns_ips.clone()))
-    }
-
-    pub fn valid_dns_target(&self, addr: &Ipv4Address) -> bool {
-        if let Some(lease) = &self.vm_lease {
-            return lease.dns_ips.contains(addr);
-        }
-
-        false
-    }
-}
-
-#[derive(Debug)]
-pub struct Lease {
-    address: Ipv4Address,
-    valid_until: coarsetime::Instant,
-    dns_ips: HashSet<Ipv4Address>,
-}
-
-impl Lease {
-    pub fn new(address: Ipv4Address, lease_time: Duration, dns_ips: HashSet<Ipv4Address>) -> Lease {
-        Lease {
-            address,
-            valid_until: coarsetime::Instant::recent() + lease_time.into(),
-            dns_ips,
-        }
-    }
-
-    pub fn address(&self) -> Ipv4Address {
-        self.address
-    }
-
-    pub fn valid(&self) -> bool {
-        coarsetime::Instant::recent() < self.valid_until
-    }
-
-    pub fn is_valid_for(&self, address: Ipv4Address) -> bool {
-        self.address == address && self.valid()
-    }
-}
-
-pub(crate) fn message_matches_bootp_client(
-    message: &Message,
-    opcode: Opcode,
-    mac: [u8; 6],
-) -> bool {
-    message.opcode() == opcode
-        && message.htype() == HType::Eth
-        && message.hlen() == mac.len() as u8
-        && message.chaddr() == mac
-}
-
-#[cfg(test)]
-mod tests {
-    use super::{DhcpSnooper, Lease};
-    use dhcproto::v4::{DhcpOption, Message, MessageType, Opcode};
-    use dhcproto::{Encodable, Encoder};
-    use smoltcp::wire::Ipv4Address;
-    use std::collections::HashSet;
-    use std::time::Duration;
-
-    const VM_MAC: [u8; 6] = [0x02, 0x00, 0x00, 0x00, 0x00, 0x01];
-    const OTHER_MAC: [u8; 6] = [0x02, 0x00, 0x00, 0x00, 0x00, 0x02];
-    const OLD_ADDRESS: Ipv4Address = Ipv4Address::new(192, 168, 64, 2);
-
-    #[test]
-    fn processes_replies_only_for_matching_client() {
-        // Start with an active lease
-        let mut snooper = DhcpSnooper::new(Duration::ZERO, VM_MAC);
-        snooper.set_lease(Some(Lease::new(
-            OLD_ADDRESS,
-            Duration::from_secs(600),
-            HashSet::new(),
-        )));
-
-        // Ignore a NAK for another client
-        let mut message = Message::new(
-            Ipv4Address::UNSPECIFIED,
-            Ipv4Address::UNSPECIFIED,
-            Ipv4Address::UNSPECIFIED,
-            Ipv4Address::UNSPECIFIED,
-            &OTHER_MAC,
-        );
-        message.set_opcode(Opcode::BootReply);
-        message
-            .opts_mut()
-            .insert(DhcpOption::MessageType(MessageType::Nak));
-
-        let mut encoded = Vec::new();
-        message.encode(&mut Encoder::new(&mut encoded)).unwrap();
-
-        snooper.register_dhcp_reply(&encoded);
-
-        assert_eq!(snooper.lease().as_ref().unwrap().address(), OLD_ADDRESS);
-
-        // Process a NAK for the matching client
-        message.set_chaddr(&VM_MAC);
-        encoded.clear();
-        message.encode(&mut Encoder::new(&mut encoded)).unwrap();
-
-        snooper.register_dhcp_reply(&encoded);
-
-        assert!(snooper.lease().is_none());
-    }
-}

lib/host.rs

@@ -1,15 +1,16 @@
+use crate::subnet_finder;
 use anyhow::{Context, Result, anyhow};
 use clap::ValueEnum;
 use log::info;
+use smoltcp::wire::EthernetAddress;
 use std::net::Ipv4Addr;
-use std::os::unix::io::{AsRawFd, RawFd};
-use std::os::unix::net::UnixDatagram;
-use std::str::FromStr;
+use std::sync::Arc;
 use std::sync::mpsc::{SyncSender, sync_channel};
-use vmnet::mode::Mode;
+use tokio::sync::Notify;
+use vmnet::network::Mode;
 use vmnet::parameters::{Parameter, ParameterKind};
 use vmnet::port_forwarding::{AddressFamily, Protocol};
-use vmnet::{Batch, Events, Options};
+use vmnet::{Batch, Events, Network, NetworkConfiguration, Options};
 
 #[derive(ValueEnum, Clone, Debug)]
 pub enum NetType {
@@ -25,8 +26,11 @@ pub enum NetType {
 
 pub struct Host {
     interface: vmnet::Interface,
-    new_packets_rx: UnixDatagram,
-    callback_can_continue_tx: SyncSender<()>,
+    ready: Arc<Notify>,
+    draining: bool,
+    callback_can_continue_tx: Option<SyncSender<()>>,
+    pub vm_ip: Ipv4Addr,
+    pub subnet_mask: Ipv4Addr,
     pub gateway_ip: smoltcp::wire::Ipv4Address,
     pub max_packet_size: u64,
     pub read_max_packets: u64,
@@ -34,31 +38,49 @@ pub struct Host {
 }
 
 impl Host {
-    pub fn new(vm_net_type: NetType, enable_isolation: bool) -> Result<Host> {
-        // Initialize a vmnet.framework NAT or Host interface with isolation enabled
-        let mut interface = vmnet::Interface::new(
-            match vm_net_type {
-                NetType::Nat => Mode::Shared(Default::default()),
-                NetType::Host => Mode::Host(Default::default()),
-            },
+    pub fn new(
+        vm_net_type: NetType,
+        vm_mac: EthernetAddress,
+        enable_isolation: bool,
+    ) -> Result<Host> {
+        // Find an unused /29 subnet for the gateway and VM
+        let (gateway_ip, vm_ip, subnet) = subnet_finder::find_available_subnet(29)?;
+
+        // Create vmnet's network configuration
+        let mut configuration = NetworkConfiguration::new(match vm_net_type {
+            NetType::Nat => Mode::Shared,
+            NetType::Host => Mode::Host,
+        })
+        .context("failed to create vmnet network configuration")?;
+
+        // Configure the gateway address and subnet mask
+        let subnet_mask = subnet.netmask();
+        configuration
+            .set_ipv4_subnet(gateway_ip, subnet_mask)
+            .context("failed to configure IPv4 subnet")?;
+
+        // Reserve the VM's address for its MAC
+        configuration
+            .add_dhcp_reservation(vm_mac.0, vm_ip)
+            .context("failed to add DHCP address reservation")?;
+
+        // Create vmnet's network
+        let network = Network::new(&configuration)
+            .with_context(|| format!("failed to create vmnet network {subnet}"))?;
+
+        // Instantiate vmnet's interface
+        //
+        // The interface retains the network reservation until it is stopped
+        let mut interface = vmnet::Interface::with_network(
+            &network,
             Options {
+                allocate_mac_address: Some(false),
                 enable_isolation: Some(enable_isolation),
                 ..Default::default()
             },
         )
         .context("failed to initialize vmnet interface")?;
 
-        // Retrieve first IP (gateway) used for this interface
-        let Some(Parameter::StartAddress(gateway_ip)) =
-            interface.parameters().get(ParameterKind::StartAddress)
-        else {
-            return Err(anyhow!(
-                "failed to retrieve vmnet's interface start address"
-            ));
-        };
-        let gateway_ip = Ipv4Addr::from_str(&gateway_ip)
-            .context("failed to parse vmnet's interface start address")?;
-
         // Retrieve max packet size for this interface
         let Some(Parameter::MaxPacketSize(max_packet_size)) =
             interface.parameters().get(ParameterKind::MaxPacketSize)
@@ -77,33 +99,28 @@ impl Host {
             ));
         };
 
-        // Set up a socketpair() to emulate polling of the vmnet interface
-        let (new_packets_tx, new_packets_rx) = UnixDatagram::pair()?;
-        new_packets_rx.set_nonblocking(true)?;
-
-        let (callback_can_continue_tx, callback_can_continue_rx) = sync_channel(0);
+        // Coordinate packet readiness and callback acknowledgments
+        let ready = Arc::new(Notify::new());
+        let ready_for_callback = Arc::clone(&ready);
+        let (callback_can_continue_tx, callback_can_continue_rx) = sync_channel(1);
 
         interface
             .set_event_callback(Events::PACKETS_AVAILABLE, move |_mask, _params| {
-                // Send a dummy datagram to make the other end of socketpair() readable
-                // and ignore the error as this merely a signalling channel to wake up
-                // the poller
-                new_packets_tx.send(&[0; 1]).ok();
+                // Wake the Tokio reader to drain the available packets
+                ready_for_callback.notify_one();
 
-                // Wait for the permission to continue to avoid
-                // wasting CPU cycles or in case of termination,
-                // to unblock this Block[1] and allow
-                // vmnet.framework to terminate
-                //
-                // [1]: https://en.wikipedia.org/wiki/Blocks_(C_language_extension)
-                callback_can_continue_rx.recv().unwrap();
+                // Park this GCD worker until drained; returning early makes vmnet repeat callbacks
+                let _ = callback_can_continue_rx.recv();
             })
             .context("failed to set vmnet interface's event callback")?;
 
         Ok(Host {
             interface,
-            new_packets_rx,
-            callback_can_continue_tx,
+            ready,
+            draining: false,
+            callback_can_continue_tx: Some(callback_can_continue_tx),
+            vm_ip,
+            subnet_mask,
             gateway_ip,
             max_packet_size,
             read_max_packets,
@@ -135,31 +152,26 @@ impl Host {
             .map_err(|err| anyhow!("failed to add port forwarding rule {details}: {err}"))
     }
 
-    pub fn port_forwarding_remove_rule(&mut self, external_port: u16) -> Result<()> {
-        let details = format!("external_port={external_port}");
+    pub async fn read(&mut self, batch: &mut Batch, bufs: &mut [Vec<u8>]) -> vmnet::Result<usize> {
+        loop {
+            // One callback can cover several batches; stay ready until an empty read
+            if !self.draining {
+                self.ready.notified().await;
+                self.draining = true;
+            }
 
-        self.interface
-            .port_forwarding_rule_remove(AddressFamily::Ipv4, Protocol::Tcp, external_port)
-            .map(|_| info!("removed port forwarding rule {details}"))
-            .map_err(|err| anyhow!("failed to remove port forwarding rule {details}: {err}"))
-    }
+            match self.interface.read_batch(batch, bufs) {
+                Err(vmnet::Error::VmnetReadNothing) => {
+                    self.draining = false;
 
-    pub fn read(&mut self, batch: &mut Batch, bufs: &mut [Vec<u8>]) -> vmnet::Result<usize> {
-        // Dequeue dummy datagram from the socket (if any)
-        // to free up buffer space and reduce false-positives
-        // when polling
-        let mut buf_to_be_discarded: [u8; 1] = [0; 1];
-        let _ = self.new_packets_rx.recv(&mut buf_to_be_discarded);
-
-        let result = self.interface.read_batch(batch, bufs);
-
-        if let Err(vmnet::Error::VmnetReadNothing) = result {
-            // We've emptied everything, unlock the callback
-            // so that it will be able to pick up new events
-            let _ = self.callback_can_continue_tx.send(());
+                    if let Some(tx) = &self.callback_can_continue_tx {
+                        // Buffer the acknowledgment so the reader never waits for the callback
+                        let _ = tx.try_send(());
+                    }
+                }
+                result => return result,
+            }
         }
-
-        result
     }
 
     pub fn write(&mut self, buf: &[u8]) -> vmnet::Result<usize> {
@@ -167,21 +179,21 @@ impl Host {
     }
 
     pub fn finalize(&mut self) -> Result<()> {
+        if self.finalized {
+            return Ok(());
+        }
+
         // First make sure our callback won't be scheduled again after it finishes
-        self.interface
-            .clear_event_callback()
-            .context("failed to clear vmnet interface's event callback")?;
+        let clear_result = self.interface.clear_event_callback();
 
-        // Now let the callback finish
-        let _ = self.callback_can_continue_tx.send(());
+        // Disconnecting releases the callback during shutdown without waiting for a receiver
+        self.callback_can_continue_tx.take();
 
-        self.interface
-            .finalize()
-            .context("failed to finalize vmnet's interface")?;
+        let stop_result = self.interface.finalize();
+        self.finalized = stop_result.is_ok();
 
-        self.finalized = true;
-
-        Ok(())
+        clear_result.context("failed to clear vmnet interface's event callback")?;
+        stop_result.context("failed to finalize vmnet's interface")
     }
 }
 
@@ -192,9 +204,3 @@ impl Drop for Host {
         }
     }
 }
-
-impl AsRawFd for Host {
-    fn as_raw_fd(&self) -> RawFd {
-        self.new_packets_rx.as_raw_fd()
-    }
-}

lib/mod.rs

@@ -1,6 +1,6 @@
-mod dhcp_snooper;
+mod dhcp_server;
 mod host;
 pub use host::NetType;
-mod poller;
 pub mod proxy;
+mod subnet_finder;
 mod vm;

lib/poller.rs

@@ -1,112 +0,0 @@
-use anyhow::Result;
-use num_enum::IntoPrimitive;
-use polling::PollMode;
-use polling::os::kqueue::PollerKqueueExt;
-use std::os::fd::{AsRawFd, BorrowedFd};
-use std::os::unix::io::RawFd;
-use std::time::Duration;
-
-pub struct Poller<'poller> {
-    poller: polling::Poller,
-    events: polling::Events,
-    timeout: Duration,
-    vm_fd: BorrowedFd<'poller>,
-    host_fd: BorrowedFd<'poller>,
-    control_fd: Option<BorrowedFd<'poller>>,
-}
-
-#[derive(IntoPrimitive)]
-#[repr(usize)]
-enum EventKey {
-    VM,
-    Host,
-    Control,
-    Interrupt,
-}
-
-impl Poller<'_> {
-    pub fn new<'poller>(
-        vm_fd: RawFd,
-        host_fd: RawFd,
-        control_fd: Option<RawFd>,
-        timeout: Duration,
-    ) -> Result<Poller<'poller>> {
-        let poller = polling::Poller::new()?;
-
-        Ok(Poller {
-            poller,
-            events: polling::Events::new(),
-            timeout,
-            vm_fd: unsafe { BorrowedFd::borrow_raw(vm_fd) },
-            host_fd: unsafe { BorrowedFd::borrow_raw(host_fd) },
-            control_fd: control_fd.map(|fd| unsafe { BorrowedFd::borrow_raw(fd) }),
-        })
-    }
-
-    pub fn arm(&self) -> Result<()> {
-        unsafe {
-            self.poller.add_with_mode(
-                self.vm_fd.as_raw_fd(),
-                self.vm_interest(),
-                PollMode::Edge,
-            )?;
-
-            if let Some(control_fd) = self.control_fd {
-                self.poller.add_with_mode(
-                    control_fd.as_raw_fd(),
-                    polling::Event::all(EventKey::Control.into()),
-                    PollMode::Edge,
-                )?;
-            }
-            self.poller.add_with_mode(
-                self.host_fd.as_raw_fd(),
-                self.host_interest(),
-                PollMode::Edge,
-            )?;
-        }
-
-        let interrupt_signal = polling::os::kqueue::Signal(libc::SIGINT);
-        self.poller
-            .add_filter(interrupt_signal, EventKey::Interrupt.into(), PollMode::Edge)?;
-
-        Ok(())
-    }
-
-    pub fn rearm(&mut self) {
-        self.events.clear();
-    }
-
-    pub fn wait(&mut self) -> Result<(bool, bool, bool)> {
-        self.poller.wait(&mut self.events, Some(self.timeout))?;
-
-        let vm_readable = self
-            .events
-            .iter()
-            .any(|ev| ev.key == Into::<usize>::into(EventKey::VM));
-        let host_readable = self
-            .events
-            .iter()
-            .any(|ev| ev.key == Into::<usize>::into(EventKey::Host));
-        let interrupt = self
-            .events
-            .iter()
-            .any(|ev| ev.key == Into::<usize>::into(EventKey::Interrupt));
-        Ok((vm_readable, host_readable, interrupt))
-    }
-
-    pub fn remove_control(&mut self) -> Result<()> {
-        if let Some(control_fd) = self.control_fd.take() {
-            self.poller.delete(control_fd)?;
-        }
-
-        Ok(())
-    }
-
-    fn vm_interest(&self) -> polling::Event {
-        polling::Event::readable(EventKey::VM.into())
-    }
-
-    fn host_interest(&self) -> polling::Event {
-        polling::Event::readable(EventKey::Host.into())
-    }
-}

lib/proxy/control.rs

@@ -1,5 +1,7 @@
 use super::{Rule, Rules};
 use anyhow::{Context, Result, bail};
+use bytes::BytesMut;
+use futures_util::{SinkExt, StreamExt};
 use jsonrpsee_types::{
     ErrorObjectOwned, Id, Request, Response, ResponsePayload,
     error::{
@@ -7,20 +9,21 @@ use jsonrpsee_types::{
         METHOD_NOT_FOUND_CODE as METHOD_NOT_FOUND, PARSE_ERROR_CODE as PARSE_ERROR,
     },
 };
+use nix::sys::socket::{Shutdown, shutdown};
 use serde::Deserialize;
 use serde_json::{Value, json};
 use smoltcp::wire::Ipv4Address;
-use std::io::{self, ErrorKind, Read, Write};
+use std::future::poll_fn;
+use std::io;
 use std::mem::{size_of, zeroed};
-use std::net::Shutdown;
 use std::os::fd::{AsRawFd, FromRawFd, RawFd};
 use std::os::unix::net::UnixStream;
+use tokio_util::codec::{Decoder, Encoder, Framed, LinesCodec, LinesCodecError};
 
 const MAX_REQUEST_BYTES: usize = 1024 * 1024;
 const MAX_PENDING_RESPONSE_BYTES: usize = 4 * MAX_REQUEST_BYTES;
 const MAX_RULES: usize = 4096;
 const MAX_IDENTIFIER_BYTES: usize = 256;
-const MAX_SERVICE_BYTES: usize = MAX_REQUEST_BYTES;
 
 pub(super) struct Policy {
     allow: Vec<Rule>,
@@ -124,13 +127,7 @@ pub(super) fn normalize_rules(mut rules: Vec<Rule>) -> Vec<Rule> {
 
 pub(super) struct Control {
     policy: Policy,
-    stream: UnixStream,
-    input: Vec<u8>,
-    output: Vec<u8>,
-    output_offset: usize,
-    discarding_input: bool,
-    input_closed: bool,
-    policy_changed: bool,
+    stream: Framed<tokio::net::UnixStream, ControlCodec>,
 }
 
 impl Control {
@@ -148,199 +145,118 @@ impl Control {
 
         Ok(Control {
             policy: Policy::new(gateway_ip, allow, block),
-            stream,
-            input: Vec::new(),
-            output: Vec::new(),
-            output_offset: 0,
-            discarding_input: false,
-            input_closed: false,
-            policy_changed: false,
+            stream: Framed::new(
+                tokio::net::UnixStream::from_std(stream)?,
+                ControlCodec(LinesCodec::new_with_max_length(MAX_REQUEST_BYTES)),
+            ),
         })
     }
 
-    pub(super) fn service(&mut self, rules: &mut Rules) -> Result<bool> {
-        if !self.flush()? {
-            return Ok(false);
-        }
+    /// Cancellation only interrupts buffered I/O; policy changes finish in the same poll.
+    /// Returns None on EOF, or whether the completed request changed the policy.
+    pub(super) async fn service(&mut self, rules: &mut Rules) -> Result<Option<bool>> {
+        // Flush pending responses before reading; Framed retains partial I/O across cancellation
+        self.stream.flush().await?;
+        poll_fn(|cx| self.stream.poll_ready_unpin(cx)).await?;
 
-        if !self.output.is_empty() {
-            return Ok(true);
-        }
+        let Some(frame) = self.stream.next().await.transpose()? else {
+            return Ok(None);
+        };
 
-        if !self.process_input(rules)? {
-            return Ok(false);
-        }
-
-        if !self.output.is_empty() {
-            return Ok(true);
-        }
-
-        if self.input_closed {
-            return Ok(false);
-        }
-
-        let mut buf = [0; 8192];
-        let mut bytes_read = 0;
-
-        while bytes_read < MAX_SERVICE_BYTES {
-            match self.stream.read(&mut buf) {
-                Ok(0) => {
-                    self.input_closed = true;
-                    break;
-                }
-                Ok(n) => {
-                    bytes_read += n;
-                    self.input.extend_from_slice(&buf[..n]);
-                    if !self.process_input(rules)? {
-                        return Ok(false);
-                    }
-
-                    if !self.output.is_empty() {
-                        return Ok(true);
-                    }
-                }
-                Err(err) if err.kind() == ErrorKind::WouldBlock => break,
-                Err(err)
-                    if matches!(
-                        err.kind(),
-                        ErrorKind::BrokenPipe | ErrorKind::ConnectionReset
-                    ) =>
-                {
-                    return Ok(false);
-                }
-                Err(err) => return Err(err).context("failed to read the control socket"),
-            }
-        }
-
-        Ok(!self.input_closed)
+        self.respond(frame, rules).map(Some)
     }
 
-    pub(super) fn shutdown(&self) -> Result<()> {
-        self.stream
-            .shutdown(Shutdown::Both)
-            .context("failed to shut down the control socket")
-    }
-
-    /// Returns whether the policy changed and clears the change flag.
-    pub(super) fn policy_changed(&mut self) -> bool {
-        std::mem::take(&mut self.policy_changed)
-    }
-
-    fn process_input(&mut self, rules: &mut Rules) -> Result<bool> {
-        loop {
-            if self.discarding_input {
-                if let Some(newline) = self.input.iter().position(|byte| *byte == b'\n') {
-                    self.input.drain(..=newline);
-                    self.discarding_input = false;
-                    continue;
-                }
-
-                self.input.clear();
-                return Ok(true);
-            }
-
-            let Some(newline) = self.input.iter().position(|byte| *byte == b'\n') else {
-                if self.input.len() > MAX_REQUEST_BYTES {
-                    self.input.clear();
-                    self.discarding_input = true;
-                    self.enqueue(error_response(
-                        Id::Null,
-                        PARSE_ERROR,
-                        "request exceeds the maximum frame size",
-                    ))?;
-
-                    return self.flush();
-                }
-
-                return Ok(true);
-            };
-
-            let line = self.input.drain(..=newline).collect::<Vec<_>>();
-
-            if newline > MAX_REQUEST_BYTES {
-                self.enqueue(error_response(
+    fn respond(
+        &mut self,
+        frame: Result<String, LinesCodecError>,
+        rules: &mut Rules,
+    ) -> Result<bool> {
+        // Reply to malformed frames without changing the active policy
+        let (response, update) = match frame {
+            Ok(line) => handle_request(&self.policy, rules, line.as_bytes()),
+            Err(error) => (
+                error_response(
                     Id::Null,
                     PARSE_ERROR,
-                    "request exceeds the maximum frame size",
-                ))?;
+                    match error {
+                        LinesCodecError::MaxLineLengthExceeded => {
+                            "request exceeds the maximum frame size"
+                        }
+                        LinesCodecError::Io(_) => "invalid JSON-RPC frame",
+                    },
+                ),
+                None,
+            ),
+        };
 
-                if !self.flush()? {
-                    return Ok(false);
-                }
+        // service() reserved sink capacity; encoding checks the byte limit before any update
+        self.stream.start_send_unpin(response)?;
 
-                if !self.output.is_empty() {
-                    return Ok(true);
-                }
-
-                continue;
-            }
-
-            let (response, update) = handle_request(&self.policy, rules, &line[..newline]);
-            self.enqueue(response)?;
-
-            if let Some(update) = update
-                && let Some(updated_rules) = self.policy.apply(update)
-            {
-                *rules = updated_rules;
-                self.policy_changed = true;
-            }
-
-            if !self.flush()? {
-                return Ok(false);
-            }
-
-            if !self.output.is_empty() {
-                return Ok(true);
-            }
-        }
-    }
-
-    fn enqueue(&mut self, response: Value) -> Result<()> {
-        if self.output_offset != 0 {
-            self.output.drain(..self.output_offset);
-            self.output_offset = 0;
+        // Keep the update in this poll with no intervening await
+        if let Some(update) = update
+            && let Some(updated_rules) = self.policy.apply(update)
+        {
+            *rules = updated_rules;
+            return Ok(true);
         }
 
-        let mut response =
-            serde_json::to_vec(&response).context("failed to encode RPC response")?;
-        response.push(b'\n');
-
-        if self.output.len() + response.len() > MAX_PENDING_RESPONSE_BYTES {
-            bail!("control response queue exceeded {MAX_PENDING_RESPONSE_BYTES} bytes");
-        }
-
-        self.output.extend(response);
-        Ok(())
-    }
-
-    fn flush(&mut self) -> Result<bool> {
-        while self.output_offset < self.output.len() {
-            match self.stream.write(&self.output[self.output_offset..]) {
-                Ok(0) => return Ok(false),
-                Ok(n) => self.output_offset += n,
-                Err(err) if err.kind() == ErrorKind::WouldBlock => return Ok(true),
-                Err(err)
-                    if matches!(
-                        err.kind(),
-                        ErrorKind::BrokenPipe | ErrorKind::ConnectionReset
-                    ) =>
-                {
-                    return Ok(false);
-                }
-                Err(err) => return Err(err).context("failed to write the control socket"),
-            }
-        }
-
-        self.output.clear();
-        self.output_offset = 0;
-
-        Ok(true)
+        Ok(false)
     }
 }
 
-impl AsRawFd for Control {
-    fn as_raw_fd(&self) -> RawFd {
-        self.stream.as_raw_fd()
+impl Drop for Control {
+    fn drop(&mut self) {
+        // Shut down the inherited connection even while the original descriptor remains open
+        let fd = self.stream.get_ref().as_raw_fd();
+
+        // Shut down writes first because macOS can reject SHUT_RDWR after a peer half-close
+        let _ = shutdown(fd, Shutdown::Write);
+        let _ = shutdown(fd, Shutdown::Read);
+    }
+}
+
+struct ControlCodec(LinesCodec);
+
+impl Decoder for ControlCodec {
+    type Item = Result<String, LinesCodecError>;
+    type Error = io::Error;
+
+    fn decode(&mut self, input: &mut BytesMut) -> io::Result<Option<Self::Item>> {
+        // Return malformed frames as RPC errors so later requests can still be processed
+        Ok(match self.0.decode(input) {
+            Ok(line) => line.map(Ok),
+            Err(error) => Some(Err(error)),
+        })
+    }
+
+    fn decode_eof(&mut self, input: &mut BytesMut) -> io::Result<Option<Self::Item>> {
+        let frame = self.decode(input)?;
+
+        // Unlike LinesCodec, discard an unterminated final request without applying it
+        if frame.is_none() {
+            input.clear();
+        }
+
+        Ok(frame)
+    }
+}
+
+impl Encoder<Value> for ControlCodec {
+    type Error = io::Error;
+
+    fn encode(&mut self, response: Value, output: &mut BytesMut) -> io::Result<()> {
+        let mut response = serde_json::to_vec(&response).map_err(io::Error::other)?;
+        response.push(b'\n');
+
+        // Bound the complete encoded response before appending it
+        if output.len() + response.len() > MAX_PENDING_RESPONSE_BYTES {
+            return Err(io::Error::other(format!(
+                "control response queue exceeded {MAX_PENDING_RESPONSE_BYTES} bytes"
+            )));
+        }
+
+        output.extend_from_slice(&response);
+        Ok(())
     }
 }
 
@@ -533,18 +449,21 @@ fn validate_control_fd(control_fd: RawFd) -> Result<()> {
 #[cfg(test)]
 mod tests {
     use super::{
-        Control, INVALID_PARAMS, INVALID_REQUEST, MAX_PENDING_RESPONSE_BYTES, MAX_REQUEST_BYTES,
-        MAX_RULES, METHOD_NOT_FOUND, PARSE_ERROR, Policy, handle_request,
+        Control, ControlCodec, INVALID_PARAMS, INVALID_REQUEST, MAX_PENDING_RESPONSE_BYTES,
+        MAX_REQUEST_BYTES, MAX_RULES, METHOD_NOT_FOUND, PARSE_ERROR, Policy, handle_request,
     };
     use crate::proxy::{Direction, PolicyDecision, Rule, Rules};
+    use bytes::BytesMut;
+    use futures_util::SinkExt;
     use serde_json::{Value, json};
     use smoltcp::wire::Ipv4Address;
     use std::fs::File;
     use std::io::{Read, Write};
-    use std::net::{Shutdown, TcpListener};
+    use std::net::Shutdown;
     use std::os::fd::{AsRawFd, RawFd};
     use std::os::unix::net::{UnixDatagram, UnixStream};
     use std::time::Duration;
+    use tokio_util::codec::{Decoder, Encoder, LinesCodec};
 
     struct TestPolicy {
         state: Policy,
@@ -867,15 +786,11 @@ mod tests {
         assert_eq!(policy.result(), before);
     }
 
-    #[test]
-    fn newline_delimited_control_socket_handles_multiple_requests_and_eof() {
+    #[tokio::test(flavor = "current_thread")]
+    async fn newline_delimited_control_socket_handles_multiple_requests_and_eof() {
         let (mut client, server) = UnixStream::pair().unwrap();
-        client
-            .set_read_timeout(Some(Duration::from_secs(1)))
-            .unwrap();
         let mut control = control(server.as_raw_fd()).unwrap();
         let mut rules = Rules::default();
-
         client
             .write_all(
                 concat!(
@@ -885,100 +800,101 @@ mod tests {
                 .as_bytes(),
             )
             .unwrap();
+        client.shutdown(Shutdown::Write).unwrap();
 
-        assert!(control.service(&mut rules).unwrap());
-        let mut response = [0; 2048];
-        let n = client.read(&mut response).unwrap();
-        let lines = std::str::from_utf8(&response[..n])
-            .unwrap()
+        assert_eq!(control.service(&mut rules).await.unwrap(), Some(false));
+        assert_eq!(control.service(&mut rules).await.unwrap(), Some(true));
+        assert_eq!(control.service(&mut rules).await.unwrap(), None);
+        assert_eq!(control.policy.allow, vec!["@host".parse().unwrap()]);
+        drop(control);
+
+        let mut output = String::new();
+        client.read_to_string(&mut output).unwrap();
+        let responses = output
             .lines()
             .map(|line| serde_json::from_str::<Value>(line).unwrap())
             .collect::<Vec<_>>();
-        assert_eq!(lines.len(), 2);
-        assert_eq!(lines[0]["id"], 1);
-        assert_eq!(lines[1]["id"], 2);
-        assert_eq!(lines[1]["result"]["allow"], json!(["@host"]));
-        assert_eq!(control.policy.allow, vec!["@host".parse().unwrap()]);
-
-        let before = control.policy.result(rules.len());
-        drop(client);
-        assert!(!control.service(&mut rules).unwrap());
-        assert_eq!(control.policy.result(rules.len()), before);
+        assert_eq!(responses.len(), 2);
+        assert_eq!(responses[0]["id"], 1);
+        assert_eq!(responses[1]["id"], 2);
+        assert_eq!(responses[1]["result"]["allow"], json!(["@host"]));
     }
 
-    #[test]
-    fn policy_change_detection_uses_normalized_policy() {
+    #[tokio::test(flavor = "current_thread")]
+    async fn policy_change_detection_uses_normalized_policy() {
         let (mut client, server) = UnixStream::pair().unwrap();
         let mut control = control(server.as_raw_fd()).unwrap();
         let mut rules = Rules::default();
-
-        client
-            .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"out 10.1.2.3/8\"],\"block\":[\"in 10.0.0.0/8\",\"out 10.0.0.0/8\"]}}\n")
+        for (allow, changed) in [("out 10.1.2.3/8", true), ("out 10.0.0.0/8", false)] {
+            writeln!(
+                client,
+                "{}",
+                json!({
+                    "jsonrpc": "2.0", "id": 1, "method": "softnet.policy.set",
+                    "params": {"allow": [allow], "block": ["in 10.0.0.0/8", "out 10.0.0.0/8"]}
+                })
+            )
             .unwrap();
-        assert!(control.service(&mut rules).unwrap());
-        assert!(control.policy_changed());
-
-        client
-            .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"out 10.0.0.0/8\"],\"block\":[\"in 10.0.0.0/8\",\"out 10.0.0.0/8\"]}}\n")
-            .unwrap();
-        assert!(control.service(&mut rules).unwrap());
-        assert!(!control.policy_changed());
-
+            assert_eq!(control.service(&mut rules).await.unwrap(), Some(changed));
+        }
         client
             .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[],\"block\":[\"in 10.0.0.0/8\",\"out 10.0.0.0/8\"]}}\n")
             .unwrap();
-        assert!(control.service(&mut rules).unwrap());
-        assert!(control.policy_changed());
+        assert_eq!(control.service(&mut rules).await.unwrap(), Some(true));
         assert!(control.policy.allow.is_empty());
     }
 
-    #[test]
-    fn write_side_eof_flushes_the_final_policy_response() {
+    #[tokio::test(flavor = "current_thread")]
+    async fn write_side_eof_flushes_the_final_policy_response() {
         let (mut client, server) = UnixStream::pair().unwrap();
-        client
-            .set_read_timeout(Some(Duration::from_secs(1)))
-            .unwrap();
         let mut control = control(server.as_raw_fd()).unwrap();
         let mut rules = Rules::default();
-
         client
             .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"10.0.0.0/8\"],\"block\":[]}}\n")
             .unwrap();
         client.shutdown(Shutdown::Write).unwrap();
 
-        assert!(!control.service(&mut rules).unwrap());
+        assert_eq!(control.service(&mut rules).await.unwrap(), Some(true));
+        assert_eq!(control.service(&mut rules).await.unwrap(), None);
+        assert!(control.stream.write_buffer().is_empty());
+        drop(control);
 
-        let mut response = [0; 1024];
-        let n = client.read(&mut response).unwrap();
-        let response = serde_json::from_slice::<Value>(&response[..n - 1]).unwrap();
+        let mut response = String::new();
+        client.read_to_string(&mut response).unwrap();
+        let response = serde_json::from_str::<Value>(&response).unwrap();
         assert_eq!(response["id"], 1);
         assert_eq!(response["result"]["allow"], json!(["10.0.0.0/8"]));
-        assert!(control.output.is_empty());
     }
 
-    #[test]
-    fn oversized_frame_is_discarded_and_following_frame_is_processed() {
-        let (mut client, server) = UnixStream::pair().unwrap();
-        client
-            .set_read_timeout(Some(Duration::from_secs(1)))
-            .unwrap();
+    #[tokio::test(flavor = "current_thread")]
+    async fn oversized_frame_is_discarded_and_following_frame_is_processed() {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+        let (client, server) = UnixStream::pair().unwrap();
+        client.set_nonblocking(true).unwrap();
+        let mut client = tokio::net::UnixStream::from_std(client).unwrap();
         let mut control = control(server.as_raw_fd()).unwrap();
         let mut rules = Rules::default();
-
-        control.input = vec![b'x'; MAX_REQUEST_BYTES + 1];
-        control.process_input(&mut rules).unwrap();
-        assert!(control.discarding_input);
-
-        control.input.extend_from_slice(
+        let mut input = vec![b'x'; MAX_REQUEST_BYTES + 1];
+        input.extend_from_slice(
             b"still-too-long\n{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"softnet.policy.get\"}\n",
         );
-        control.process_input(&mut rules).unwrap();
-        assert!(!control.discarding_input);
-
-        let mut output = [0; 2048];
-        let n = client.read(&mut output).unwrap();
-        let responses = std::str::from_utf8(&output[..n])
-            .unwrap()
+        let (output, ()) = tokio::join!(
+            async {
+                client.write_all(&input).await.unwrap();
+                client.shutdown().await.unwrap();
+                let mut output = String::new();
+                client.read_to_string(&mut output).await.unwrap();
+                output
+            },
+            async {
+                assert_eq!(control.service(&mut rules).await.unwrap(), Some(false));
+                assert_eq!(control.service(&mut rules).await.unwrap(), Some(false));
+                assert_eq!(control.service(&mut rules).await.unwrap(), None);
+                drop(control);
+            }
+        );
+        let responses = output
             .lines()
             .map(|line| serde_json::from_str::<Value>(line).unwrap())
             .collect::<Vec<_>>();
@@ -987,105 +903,112 @@ mod tests {
         assert_eq!(responses[1]["id"], 2);
     }
 
-    #[test]
-    fn fragmented_frame_does_not_apply_until_the_newline_arrives() {
+    #[tokio::test(flavor = "current_thread")]
+    async fn fragmented_frame_survives_cancelled_service() {
         let (mut client, server) = UnixStream::pair().unwrap();
-        client
-            .set_read_timeout(Some(Duration::from_secs(1)))
-            .unwrap();
         let mut control = control(server.as_raw_fd()).unwrap();
         let mut rules = Rules::default();
         let before = control.policy.result(rules.len());
-
         client
-            .write_all(
-                b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"@host\"],",
-            )
+            .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"@host\"],")
             .unwrap();
-        assert!(control.service(&mut rules).unwrap());
+        assert!(
+            tokio::time::timeout(Duration::from_millis(10), control.service(&mut rules))
+                .await
+                .is_err()
+        );
         assert_eq!(control.policy.result(rules.len()), before);
-        assert!(control.output.is_empty());
+        assert!(control.stream.write_buffer().is_empty());
 
         client.write_all(b"\"block\":[]}}\n").unwrap();
-        assert!(control.service(&mut rules).unwrap());
-
+        assert_eq!(control.service(&mut rules).await.unwrap(), Some(true));
+        control.stream.flush().await.unwrap();
         let mut response = [0; 1024];
         let n = client.read(&mut response).unwrap();
         let response = serde_json::from_slice::<Value>(&response[..n - 1]).unwrap();
         assert_eq!(response["id"], 1);
         assert_eq!(response["result"]["allow"], json!(["@host"]));
-        assert_eq!(control.policy.allow, vec!["@host".parse().unwrap()]);
     }
 
     #[test]
-    fn response_backpressure_keeps_the_pending_queue_bounded() {
-        let (_client, server) = UnixStream::pair().unwrap();
-        let mut control = control(server.as_raw_fd()).unwrap();
-        let response = json!({"jsonrpc": "2.0", "id": 1, "result": "x".repeat(MAX_REQUEST_BYTES)});
-        let mut bounded = false;
+    fn codec_enforces_exact_request_and_response_limits() {
+        let mut codec = ControlCodec(LinesCodec::new_with_max_length(MAX_REQUEST_BYTES));
+        let mut input = BytesMut::from(vec![b'x'; MAX_REQUEST_BYTES].as_slice());
+        input.extend_from_slice(b"\n");
+        assert_eq!(
+            codec.decode(&mut input).unwrap().unwrap().unwrap().len(),
+            MAX_REQUEST_BYTES
+        );
+        input.resize(MAX_REQUEST_BYTES + 1, b'x');
+        assert!(codec.decode(&mut input).unwrap().unwrap().is_err());
+        input.extend_from_slice(b"discarded\nvalid\n");
+        assert_eq!(codec.decode(&mut input).unwrap().unwrap().unwrap(), "valid");
 
-        for _ in 0..8 {
-            match control.enqueue(response.clone()) {
-                Ok(()) => assert!(control.flush().unwrap()),
-                Err(error) => {
-                    assert!(
-                        error
-                            .to_string()
-                            .contains("control response queue exceeded")
-                    );
-                    bounded = true;
-                    break;
-                }
-            }
-        }
-
-        assert!(bounded);
-        assert!(control.output.len() - control.output_offset <= 4 * MAX_REQUEST_BYTES);
+        let response = json!({"result": "x".repeat(MAX_REQUEST_BYTES)});
+        let mut output = BytesMut::new();
+        while codec.encode(response.clone(), &mut output).is_ok() {}
+        assert!(output.len() <= MAX_PENDING_RESPONSE_BYTES);
+        assert!(output.len() >= 3 * MAX_REQUEST_BYTES);
     }
 
-    #[test]
-    fn pipelined_policy_responses_stop_before_queue_overflow() {
-        let (_client, server) = UnixStream::pair().unwrap();
+    #[tokio::test(flavor = "current_thread")]
+    async fn pipelined_policy_responses_stop_before_queue_overflow() {
+        use tokio::io::AsyncWriteExt;
+
+        let (client, server) = UnixStream::pair().unwrap();
+        client.set_nonblocking(true).unwrap();
+        let mut client = tokio::net::UnixStream::from_std(client).unwrap();
         let mut control = control(server.as_raw_fd()).unwrap();
         let mut rules = Rules::default();
         let allow = (0..MAX_RULES)
             .map(|index| format!("10.{}.{}.0/24", index / 256, index % 256))
             .collect::<Vec<_>>();
         let mut input = serde_json::to_vec(&json!({
-            "jsonrpc": "2.0",
-            "id": 0,
-            "method": "softnet.policy.set",
+            "jsonrpc": "2.0", "id": 0, "method": "softnet.policy.set",
             "params": {"allow": allow, "block": []}
         }))
         .unwrap();
         input.push(b'\n');
-
         for id in 1..=100 {
             input.extend_from_slice(
                 format!("{{\"jsonrpc\":\"2.0\",\"id\":{id},\"method\":\"softnet.policy.get\"}}\n")
                     .as_bytes(),
             );
         }
-
-        control.input = input;
-        assert!(control.process_input(&mut rules).unwrap());
+        let (written, processed) = tokio::join!(client.write_all(&input), async {
+            let mut processed = 0;
+            while let Ok(result) =
+                tokio::time::timeout(Duration::from_millis(50), control.service(&mut rules)).await
+            {
+                assert!(result.unwrap().is_some());
+                processed += 1;
+                assert!(processed <= 101);
+            }
+            processed
+        });
+        written.unwrap();
+        assert!((1..101).contains(&processed));
         assert_eq!(rules.len(), MAX_RULES);
-        assert!(!control.input.is_empty());
-        assert!(!control.output.is_empty());
-        assert!(control.output.len() - control.output_offset <= MAX_PENDING_RESPONSE_BYTES);
+        assert!(!control.stream.write_buffer().is_empty());
+        assert!(control.stream.write_buffer().len() <= MAX_PENDING_RESPONSE_BYTES);
     }
 
-    #[test]
-    fn response_queue_overflow_does_not_apply_a_policy_update() {
+    #[tokio::test(flavor = "current_thread")]
+    async fn response_queue_overflow_does_not_apply_a_policy_update() {
         let (_client, server) = UnixStream::pair().unwrap();
         let mut control = control(server.as_raw_fd()).unwrap();
         let mut rules = Rules::default();
         let before = control.policy.result(rules.len());
-
-        control.output = vec![b'x'; MAX_PENDING_RESPONSE_BYTES];
-        control.input = b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"10.0.0.0/8\"],\"block\":[]}}\n".to_vec();
-
-        let error = control.process_input(&mut rules).unwrap_err();
+        control
+            .stream
+            .write_buffer_mut()
+            .resize(MAX_PENDING_RESPONSE_BYTES, b'x');
+        let error = control
+            .respond(
+                Ok("{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"10.0.0.0/8\"],\"block\":[]}}".into()),
+                &mut rules,
+            )
+            .unwrap_err();
         assert!(
             error
                 .to_string()
@@ -1094,39 +1017,96 @@ mod tests {
         assert_eq!(control.policy.result(rules.len()), before);
     }
 
-    #[test]
-    fn response_backpressure_stops_consuming_policy_updates() {
+    #[tokio::test(flavor = "current_thread")]
+    async fn backpressure_survives_cancellation_and_stops_policy_updates() {
+        use tokio::io::{AsyncBufReadExt, BufReader};
+
         let (mut client, server) = UnixStream::pair().unwrap();
         let mut control = control(server.as_raw_fd()).unwrap();
         let mut rules = Rules::default();
         let before = control.policy.result(rules.len());
-
-        control.output = vec![b'x'; MAX_REQUEST_BYTES];
-        assert!(control.flush().unwrap());
-        assert!(!control.output.is_empty());
-
+        let response = json!({"jsonrpc": "2.0", "id": 0, "result": "x".repeat(MAX_REQUEST_BYTES)});
+        control.stream.start_send_unpin(response.clone()).unwrap();
         client
             .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"10.0.0.0/8\"],\"block\":[]}}\n")
             .unwrap();
-
-        assert!(control.service(&mut rules).unwrap());
+        assert!(
+            tokio::time::timeout(Duration::from_millis(10), control.service(&mut rules))
+                .await
+                .is_err()
+        );
         assert_eq!(control.policy.result(rules.len()), before);
-        assert!(control.input.is_empty());
+        assert!(control.stream.read_buffer().is_empty());
+        assert!(!control.stream.write_buffer().is_empty());
+        client.set_nonblocking(true).unwrap();
+        let mut client = BufReader::new(tokio::net::UnixStream::from_std(client).unwrap());
+        let mut line = String::new();
+        let (changed, _) = tokio::join!(control.service(&mut rules), client.read_line(&mut line));
+        assert_eq!(changed.unwrap(), Some(true));
+        assert_eq!(serde_json::from_str::<Value>(&line).unwrap(), response);
+        line.clear();
+        let (flushed, read) = tokio::join!(control.stream.flush(), client.read_line(&mut line));
+        flushed.unwrap();
+        read.unwrap();
+        assert_eq!(serde_json::from_str::<Value>(&line).unwrap()["id"], 1);
     }
 
-    #[test]
-    fn validates_control_descriptor_without_taking_ownership() {
+    #[tokio::test(flavor = "current_thread")]
+    async fn unterminated_request_at_eof_does_not_change_policy() {
+        let (mut client, server) = UnixStream::pair().unwrap();
+        let mut control = control(server.as_raw_fd()).unwrap();
+        let mut rules = Rules::default();
+        let before = control.policy.result(rules.len());
+        client
+            .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"@host\"],\"block\":[]}}")
+            .unwrap();
+        client.shutdown(Shutdown::Write).unwrap();
+        assert_eq!(control.service(&mut rules).await.unwrap(), None);
+        assert_eq!(control.policy.result(rules.len()), before);
+        assert!(control.stream.write_buffer().is_empty());
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn invalid_utf8_frame_does_not_prevent_following_request() {
+        let (mut client, server) = UnixStream::pair().unwrap();
+        let mut control = control(server.as_raw_fd()).unwrap();
+        let mut rules = Rules::default();
+        client
+            .write_all(b"\xff\n{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"softnet.policy.get\"}\n")
+            .unwrap();
+        client.shutdown(Shutdown::Write).unwrap();
+        assert_eq!(control.service(&mut rules).await.unwrap(), Some(false));
+        assert_eq!(control.service(&mut rules).await.unwrap(), Some(false));
+        assert_eq!(control.service(&mut rules).await.unwrap(), None);
+        drop(control);
+        let mut output = String::new();
+        client.read_to_string(&mut output).unwrap();
+        let responses = output
+            .lines()
+            .map(|line| serde_json::from_str::<Value>(line).unwrap())
+            .collect::<Vec<_>>();
+        assert_eq!(responses.len(), 2);
+        assert_eq!(responses[0]["error"]["code"], PARSE_ERROR);
+        assert_eq!(responses[1]["id"], 2);
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn validates_control_descriptor_without_taking_ownership() {
         let file = File::open("/dev/null").unwrap();
         let error = control(file.as_raw_fd()).err().unwrap();
         assert!(error.to_string().contains("is not a socket"));
         assert!(file.metadata().is_ok());
-
         let (datagram, _) = UnixDatagram::pair().unwrap();
         let error = control(datagram.as_raw_fd()).err().unwrap();
         assert!(error.to_string().contains("not a Unix stream socket"));
-
-        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
-        let error = control(listener.as_raw_fd()).err().unwrap();
+        let internet = nix::sys::socket::socket(
+            nix::sys::socket::AddressFamily::Inet,
+            nix::sys::socket::SockType::Stream,
+            nix::sys::socket::SockFlag::empty(),
+            None,
+        )
+        .unwrap();
+        let error = control(internet.as_raw_fd()).err().unwrap();
         assert!(error.to_string().contains("not a Unix socket"));
 
         let (stream, _peer) = UnixStream::pair().unwrap();
@@ -1135,17 +1115,11 @@ mod tests {
         assert!(unsafe { libc::fcntl(stream.as_raw_fd(), libc::F_GETFD) != -1 });
     }
 
-    #[test]
-    fn shutdown_signals_eof_while_the_original_descriptor_remains_open() {
+    #[tokio::test(flavor = "current_thread")]
+    async fn dropping_control_signals_eof_while_the_original_descriptor_remains_open() {
         let (mut client, server) = UnixStream::pair().unwrap();
-        client
-            .set_read_timeout(Some(Duration::from_secs(1)))
-            .unwrap();
         let control = control(server.as_raw_fd()).unwrap();
-
-        control.shutdown().unwrap();
         drop(control);
-
         assert!(unsafe { libc::fcntl(server.as_raw_fd(), libc::F_GETFD) != -1 });
         let mut response = [0; 1];
         assert_eq!(client.read(&mut response).unwrap(), 0);

lib/proxy/flows/mod.rs

@@ -152,10 +152,6 @@ impl FlowTable {
         true
     }
 
-    pub(crate) fn clear(&mut self) {
-        self.flows.clear();
-    }
-
     fn sweep_if_due(&mut self, now: Instant) {
         if now < self.next_sweep {
             return;

lib/proxy/host.rs

@@ -1,27 +1,22 @@
-use crate::dhcp_snooper::message_matches_bootp_client;
 use crate::proxy::flows::{FlowDirection, FlowMatch};
-use crate::proxy::udp_packet_helper::UdpPacketHelper;
 use crate::proxy::{Direction, PolicyDecision, Proxy};
 use anyhow::{Context, Result};
-use dhcproto::Decodable;
-use dhcproto::v4::Opcode;
+use dhcproto::v4::{CLIENT_PORT, SERVER_PORT};
 use smoltcp::phy::ChecksumCapabilities;
-use smoltcp::wire::{EthernetFrame, EthernetProtocol, Ipv4Packet, Ipv4Repr, UdpPacket};
+use smoltcp::wire::{EthernetFrame, EthernetProtocol, IpProtocol, Ipv4Packet, Ipv4Repr, UdpPacket};
 
-impl Proxy<'_> {
+impl Proxy {
     pub(crate) fn process_frame_from_host(&mut self, frame: &EthernetFrame<&[u8]>) -> Result<()> {
         if self.allowed_from_host(frame).is_none() {
             // Block packet by not forwarding it to the VM
             return Ok(());
         }
 
-        // Snoop bootpd(8) replies from the host to
-        // figure out the IP assigned to the VM
-        if frame.dst_addr() == self.vm_mac_address || frame.dst_addr().is_broadcast() {
-            self.snoop(frame);
-        }
+        self.write_to_vm(frame.as_ref())
+    }
 
-        match self.vm.write(frame.as_ref()) {
+    pub(super) fn write_to_vm(&mut self, packet: &[u8]) -> Result<()> {
+        match self.vm.write(packet) {
             Ok(_) => Ok(()),
             Err(err) => {
                 if let Some(libc::ENOBUFS) = err.raw_os_error() {
@@ -55,24 +50,25 @@ impl Proxy<'_> {
     }
 
     pub(super) fn allowed_from_host_ipv4(&mut self, ipv4_pkt: &Ipv4Packet<&[u8]>) -> Option<()> {
+        // Drop external DHCP replies, including malformed and fragmented replies
+        if ipv4_pkt.next_header() == IpProtocol::Udp
+            && ipv4_pkt.frag_offset() == 0
+            && ipv4_pkt.payload().len() >= 4
+        {
+            let udp = UdpPacket::new_unchecked(ipv4_pkt.payload());
+            if udp.src_port() == SERVER_PORT && udp.dst_port() == CLIENT_PORT {
+                return None;
+            }
+        }
+
         // Backwards compatibility with Softnet consumers that only use stateless rules
         if self.flows.is_none() {
             return Some(());
         }
 
-        // DHCP is required to maintain the VM's lease and must bypass user-specified rules
-        if self.is_allowed_dhcp_response(ipv4_pkt) {
-            return Some(());
-        }
-
         // Consult the flow table before evaluating inbound policy
         // so established flows are not treated as new traffic
-        let pending = if self
-            .dhcp_snooper
-            .lease()
-            .as_ref()
-            .is_some_and(|lease| lease.is_valid_for(ipv4_pkt.dst_addr()))
-        {
+        let pending = if self.host.vm_ip == ipv4_pkt.dst_addr() {
             match self
                 .flows
                 .as_mut()?
@@ -108,111 +104,4 @@ impl Proxy<'_> {
             }
         }
     }
-
-    fn snoop(&mut self, frame: &EthernetFrame<&[u8]>) {
-        if frame.ethertype() != EthernetProtocol::Ipv4 {
-            return;
-        }
-
-        let ipv4_pkt = match Ipv4Packet::new_checked(frame.payload()) {
-            Ok(ipv4_pkt) => ipv4_pkt,
-            _ => return,
-        };
-
-        if !self.is_allowed_dhcp_response(&ipv4_pkt) {
-            return;
-        }
-
-        let udp_pkt = match UdpPacket::new_checked(ipv4_pkt.payload()) {
-            Ok(udp_pkt) => udp_pkt,
-            Err(_) => return,
-        };
-
-        let address_and_dns_ips_saved = self.dhcp_snooper.address_and_dns_ips();
-        self.dhcp_snooper.register_dhcp_reply(udp_pkt.payload());
-        if address_and_dns_ips_saved != self.dhcp_snooper.address_and_dns_ips()
-            && let Some(flows) = &mut self.flows
-        {
-            flows.clear();
-        }
-    }
-
-    fn is_allowed_dhcp_response(&self, ipv4_pkt: &Ipv4Packet<&[u8]>) -> bool {
-        if ipv4_pkt.src_addr() != self.host.gateway_ip
-            || ipv4_pkt.next_header() != smoltcp::wire::IpProtocol::Udp
-        {
-            return false;
-        }
-
-        let Ok(udp_pkt) = UdpPacket::new_checked(ipv4_pkt.payload()) else {
-            return false;
-        };
-
-        // Require the standard DHCP server and client ports
-        if !udp_pkt.is_dhcp_response() {
-            return false;
-        }
-
-        // Require the BOOTP client hardware address to match this VM
-        // (symmetric with is_allowed_dhcp_request / #191 on the VM→host path)
-        let mut decoder = dhcproto::v4::Decoder::new(udp_pkt.payload());
-        let Ok(message) = dhcproto::v4::Message::decode(&mut decoder) else {
-            return false;
-        };
-
-        message_matches_bootp_client(&message, Opcode::BootReply, self.vm_mac_address.0)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use crate::dhcp_snooper::message_matches_bootp_client;
-    use dhcproto::Decodable;
-    use dhcproto::v4::{DhcpOption, Message, MessageType, Opcode};
-    use dhcproto::{Encodable, Encoder};
-    use smoltcp::wire::Ipv4Address;
-
-    const VM_MAC: [u8; 6] = [0x02, 0x00, 0x00, 0x00, 0x00, 0x01];
-    const OTHER_MAC: [u8; 6] = [0x02, 0x00, 0x00, 0x00, 0x00, 0x02];
-
-    #[test]
-    fn dhcp_boot_reply_chaddr_must_match_vm() {
-        let own = encode_boot_reply(VM_MAC);
-        let foreign = encode_boot_reply(OTHER_MAC);
-
-        let mut dec = dhcproto::v4::Decoder::new(&own);
-        let own_msg = Message::decode(&mut dec).unwrap();
-        let mut dec = dhcproto::v4::Decoder::new(&foreign);
-        let foreign_msg = Message::decode(&mut dec).unwrap();
-
-        assert!(message_matches_bootp_client(
-            &own_msg,
-            Opcode::BootReply,
-            VM_MAC
-        ));
-        assert!(!message_matches_bootp_client(
-            &foreign_msg,
-            Opcode::BootReply,
-            VM_MAC
-        ));
-    }
-
-    fn encode_boot_reply(chaddr: [u8; 6]) -> Vec<u8> {
-        let mut message = Message::new(
-            Ipv4Address::UNSPECIFIED,
-            Ipv4Address::new(192, 168, 64, 2),
-            Ipv4Address::UNSPECIFIED,
-            Ipv4Address::UNSPECIFIED,
-            &chaddr,
-        );
-        message.set_opcode(Opcode::BootReply);
-        message
-            .opts_mut()
-            .insert(DhcpOption::MessageType(MessageType::Ack));
-        message.opts_mut().insert(DhcpOption::AddressLeaseTime(600));
-
-        let mut encoded = Vec::new();
-        message.encode(&mut Encoder::new(&mut encoded)).unwrap();
-        encoded
-    }
 }

lib/proxy/mod.rs

@@ -2,47 +2,43 @@ mod control;
 mod exposed_port;
 mod flows;
 mod host;
-mod port_forwarder;
 mod rule;
 mod rules;
-mod udp_packet_helper;
 mod vm;
 
-use crate::dhcp_snooper::DhcpSnooper;
+use crate::dhcp_server::{DHCP_SERVER_MAC, DhcpServer};
 use crate::host::Host;
 use crate::host::NetType;
-use crate::poller::Poller;
 use crate::vm::VM;
-use anyhow::Result;
+use anyhow::{Context, Result, bail};
 use control::{Control, normalize_rules};
 pub use exposed_port::ExposedPort;
 use flows::{FlowTable, PendingFlow};
+use futures_util::future::OptionFuture;
 use ipnet::Ipv4Net;
 use mac_address::MacAddress;
-use port_forwarder::PortForwarder;
 pub use rule::{Direction, Rule, Target};
 pub(crate) use rules::{PolicyDecision, Rules};
-use smoltcp::wire::{EthernetFrame, Ipv4Address};
-use std::io::ErrorKind;
-use std::os::unix::io::{AsRawFd, RawFd};
+use smoltcp::wire::{EthernetAddress, EthernetFrame, Ipv4Address};
+use std::os::unix::io::RawFd;
 use std::time::Duration;
+use tokio::time::{MissedTickBehavior, interval};
 use vmnet::Batch;
 
-pub struct Proxy<'proxy> {
+pub struct Proxy {
     vm: VM,
     host: Host,
-    poller: Poller<'proxy>,
     vm_mac_address: smoltcp::wire::EthernetAddress,
-    dhcp_snooper: DhcpSnooper,
+    dhcp_server: DhcpServer,
     rules: Rules,
     control: Option<Control>,
     flows: Option<FlowTable>,
     enobufs_encountered: bool,
-    port_forwarder: PortForwarder,
 }
 
-impl Proxy<'_> {
-    pub fn new<'proxy>(
+impl Proxy {
+    /// Requires an entered Tokio runtime during construction.
+    pub fn new(
         vm_fd: RawFd,
         vm_mac_address: MacAddress,
         vm_net_type: NetType,
@@ -50,49 +46,62 @@ impl Proxy<'_> {
         block: Vec<Rule>,
         exposed_ports: Vec<ExposedPort>,
         control_fd: Option<RawFd>,
-    ) -> Result<Proxy<'proxy>> {
+    ) -> Result<Proxy> {
+        // Ensure that VM's MAC address won't conflict with our DHCP server's MAC address
+        let vm_mac_address = EthernetAddress(vm_mac_address.bytes());
+        if vm_mac_address == DHCP_SERVER_MAC {
+            bail!("VM MAC address {vm_mac_address} is reserved for the DHCP server");
+        }
+
         let allow = normalize_rules(allow);
         let block = normalize_rules(block);
 
         let vm = VM::new(vm_fd)?;
-        let host = Host::new(
+        let mut host = Host::new(
             vm_net_type,
+            vm_mac_address,
             !allow.contains(&Rule::Stateless(Target::Prefix(Ipv4Net::default()))),
         )?;
-        let poller_timeout = Duration::from_millis(100);
         let control = control_fd
             .map(|control_fd| {
                 Control::new(control_fd, host.gateway_ip, allow.clone(), block.clone())
             })
             .transpose()?;
-        let poller = Poller::new(
-            vm.as_raw_fd(),
-            host.as_raw_fd(),
-            control.as_ref().map(AsRawFd::as_raw_fd),
-            poller_timeout,
-        )?;
 
         let rules = Rules::new(host.gateway_ip, &allow, &block);
 
         // Any stateful rule enables flow inspection for the whole VM, including
-        // traffic admitted through implicit global, gateway, and DNS fallbacks
+        // traffic admitted through implicit global and gateway fallbacks
         let flows = rules.has_stateful().then(FlowTable::new);
 
+        // vmnet owns the gateway (including ARP and DNS); the VM address is
+        // reserved for this MAC for the lifetime of the attached interface.
+        let dhcp_server = DhcpServer::new(
+            vm_mac_address,
+            host.vm_ip,
+            host.subnet_mask,
+            host.gateway_ip,
+        );
+
+        // Install port forwardings for the reserved VM's IP address
+        for port in exposed_ports {
+            host.port_forwarding_add_rule(port.external_port, host.vm_ip, port.internal_port)
+                .context("failed to configure port forwarding")?;
+        }
+
         Ok(Proxy {
             vm,
             host,
-            poller,
-            vm_mac_address: smoltcp::wire::EthernetAddress(vm_mac_address.bytes()),
-            dhcp_snooper: DhcpSnooper::new(poller_timeout, vm_mac_address.bytes()),
+            vm_mac_address,
+            dhcp_server,
             rules,
             control,
             flows,
             enobufs_encountered: false,
-            port_forwarder: PortForwarder::new(exposed_ports),
         })
     }
 
-    pub fn run(&mut self) -> Result<()> {
+    pub async fn run(&mut self) -> Result<()> {
         // Create a single buffer from reading from the VM
         let mut buf: Vec<u8> = vec![0; self.host.max_packet_size as usize];
 
@@ -103,135 +112,96 @@ impl Proxy<'_> {
         ];
         let mut batch = Batch::preallocate(bufs.len());
 
-        self.poller.arm()?;
+        // Check VM connectivity periodically, skipping missed ticks
+        let mut maintenance = interval(Duration::from_millis(100));
+        maintenance.set_missed_tick_behavior(MissedTickBehavior::Skip);
 
+        // Service packet I/O, policy changes, and maintenance on the same runtime
         loop {
-            let (vm_readable, host_readable, interrupt) = self.poller.wait()?;
+            tokio::select! {
+                result = self.vm.read(&mut buf) => {
+                    let n = result?;
 
-            // kqueue does not report peer disconnects for Unix datagram sockets.
-            if !self.vm.is_connected()? {
-                return Ok(());
-            }
-
-            // Update coarse time for DHCP snooping and flows
-            coarsetime::Instant::update();
-
-            // Service control on every wake (including timeouts) so a bounded read or a pending
-            // response continues making progress even when no new edge is generated.
-            self.service_control();
-
-            if vm_readable {
-                self.read_from_vm(buf.as_mut_slice())?;
-            }
-
-            if host_readable {
-                self.read_from_host(&mut batch, &mut bufs)?;
-            }
-
-            // Graceful termination
-            if interrupt {
-                return Ok(());
-            }
-
-            // Timeout
-            if !vm_readable && !host_readable && !interrupt {
-                self.port_forwarder
-                    .tick(&mut self.host, self.dhcp_snooper.lease());
-            }
-
-            self.poller.rearm();
-        }
-    }
-
-    fn read_from_vm(&mut self, buf: &mut [u8]) -> Result<()> {
-        let mut packets_read = 0;
-
-        loop {
-            match self.vm.read(buf) {
-                Ok(n) => {
-                    // Update coarse time for DHCP snooping and flows
+                    // Update the coarse time once the packet arrives
                     coarsetime::Instant::update();
 
                     if let Ok(frame) = EthernetFrame::new_checked(&buf[..n]) {
                         self.process_frame_from_vm(frame)?;
                     }
 
-                    packets_read += 1;
-                    if packets_read == 128 {
-                        self.service_control();
-                        packets_read = 0;
+                    // Limit VM bursts to 128 packets (including the one we read above),
+                    // so that other select branches will have a chance to run
+                    for _ in 0..127 {
+                        match self.vm.try_read(&mut buf) {
+                            Ok(n) => {
+                                coarsetime::Instant::update();
+
+                                if let Ok(frame) = EthernetFrame::new_checked(&buf[..n]) {
+                                    self.process_frame_from_vm(frame)?;
+                                }
+
+                                tokio::task::consume_budget().await;
+                            }
+                            Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => break,
+                            Err(err) => return Err(err.into()),
+                        }
                     }
                 }
-                Err(err) => {
-                    if err.kind() == ErrorKind::WouldBlock {
-                        return Ok(());
-                    }
 
-                    return Err(err.into());
-                }
-            }
-        }
-    }
+                result = self.host.read(&mut batch, &mut bufs) => {
+                    let pktcnt = result?;
 
-    fn read_from_host(&mut self, batch: &mut Batch, bufs: &mut [Vec<u8>]) -> Result<()> {
-        loop {
-            match self.host.read(batch, bufs) {
-                Ok(pktcnt) => {
-                    // Update coarse time for DHCP snooping and flows
+                    // Update the coarse time once the packet arrives
                     coarsetime::Instant::update();
 
-                    for buf in batch.packet_sized_bufs(bufs).take(pktcnt) {
+                    for buf in batch.packet_sized_bufs(&bufs).take(pktcnt) {
                         if let Ok(pkt) = EthernetFrame::new_checked(buf) {
                             self.process_frame_from_host(&pkt)?;
                         }
+
+                        tokio::task::consume_budget().await;
+                    }
+                }
+
+                // Skip this branch when no control socket is present
+                Some(result) = OptionFuture::from(
+                    self.control
+                        .as_mut()
+                        .map(|control| control.service(&mut self.rules))
+                ) => {
+                    match result {
+                        Ok(Some(true)) => {
+                            // Invalidate tracked flows whenever the policy changes
+                            self.flows = self.rules.has_stateful().then(FlowTable::new);
+                        }
+                        Ok(Some(false)) => {
+                            // Policy unchanged; keep tracked flows intact
+                        }
+                        Ok(None) => {
+                            // Control stream reached EOF; stop polling it
+                            self.control = None;
+                        }
+                        Err(err) => {
+                            // Control service failed; stop polling it
+                            self.control = None;
+
+                            log::warn!("disabling Softnet control socket: {err:#}");
+                        }
                     }
 
-                    self.service_control();
+                    coarsetime::Instant::update();
                 }
-                Err(err) => {
-                    if let vmnet::Error::VmnetReadNothing = err {
+
+                _ = maintenance.tick() => {
+                    // Unix datagram sockets do not report peer closure as readable
+                    if !self.vm.is_connected()? {
                         return Ok(());
                     }
-
-                    return Err(err.into());
                 }
             }
         }
     }
 
-    fn service_control(&mut self) {
-        let Some(control) = self.control.as_mut() else {
-            return;
-        };
-
-        let keep_open = match control.service(&mut self.rules) {
-            Ok(keep_open) => keep_open,
-            Err(err) => {
-                log::warn!("disabling Softnet control socket: {err:#}");
-                false
-            }
-        };
-
-        // Invalidate tracked flows whenever the policy changes
-        if control.policy_changed() {
-            self.flows = self.rules.has_stateful().then(FlowTable::new);
-        }
-
-        if keep_open {
-            return;
-        }
-
-        if let Err(err) = self.poller.remove_control() {
-            log::warn!("failed to remove Softnet control socket from the poller: {err:#}");
-        }
-
-        if let Some(control) = self.control.take()
-            && let Err(err) = control.shutdown()
-        {
-            log::warn!("failed to shut down Softnet control socket: {err:#}");
-        }
-    }
-
     /// Commits the pending flow, rejecting the packet if the table cannot store it.
     fn admit_with_tracking(&mut self, pending: PendingFlow) -> Option<()> {
         self.flows.as_mut()?.commit(pending).then_some(())
@@ -263,33 +233,30 @@ impl Proxy<'_> {
 #[cfg(test)]
 mod tests {
     use crate::NetType;
-    use crate::dhcp_snooper::Lease;
     use crate::proxy::Proxy;
     use mac_address::MacAddress;
     use nix::sys::socket::{AddressFamily, SockFlag, SockType, socketpair};
     use serial_test::serial;
     use smoltcp::wire::{IpProtocol, Ipv4Address, Ipv4Packet, UdpPacket};
-    use std::collections::HashSet;
     use std::os::fd::AsRawFd;
     use std::str::FromStr;
-    use std::time::Duration;
 
-    #[test]
+    #[tokio::test]
     #[serial]
-    fn test_blocking_takes_precedence() {
-        let vm_ip = Ipv4Address::from_str("192.168.0.2").unwrap();
-        let mut proxy = create_proxy(vm_ip, vec!["66.66.0.0/16"], vec!["66.66.0.0/16"]);
+    async fn test_blocking_takes_precedence() {
+        let mut proxy = create_proxy(vec!["66.66.0.0/16"], vec!["66.66.0.0/16"]);
+        let vm_ip = proxy.host.vm_ip;
 
         assert_eq!(proxy.rules.len(), 1);
 
         assert!(allowed_from_vm_ipv4(&mut proxy, vm_ip, "66.66.66.66").is_none());
     }
 
-    #[test]
+    #[tokio::test]
     #[serial]
-    fn test_longest_prefix_match_wins() {
-        let vm_ip = Ipv4Address::from_str("192.168.0.2").unwrap();
-        let mut proxy = create_proxy(vm_ip, vec!["33.33.33.33/32"], vec!["33.33.33.0/24"]);
+    async fn test_longest_prefix_match_wins() {
+        let mut proxy = create_proxy(vec!["33.33.33.33/32"], vec!["33.33.33.0/24"]);
+        let vm_ip = proxy.host.vm_ip;
 
         assert_eq!(proxy.rules.len(), 2);
 
@@ -298,11 +265,11 @@ mod tests {
         assert!(allowed_from_vm_ipv4(&mut proxy, vm_ip, "33.33.33.34").is_none());
     }
 
-    #[test]
+    #[tokio::test]
     #[serial]
-    fn test_allow_host() {
-        let vm_ip = Ipv4Address::from_str("192.168.0.2").unwrap();
-        let mut proxy = create_proxy(vm_ip, vec!["@host"], vec!["0.0.0.0/0"]);
+    async fn test_allow_host() {
+        let mut proxy = create_proxy(vec!["@host"], vec!["0.0.0.0/0"]);
+        let vm_ip = proxy.host.vm_ip;
 
         assert_eq!(proxy.rules.len(), 2);
 
@@ -314,13 +281,13 @@ mod tests {
         assert!(allowed_from_vm_ipv4(&mut proxy, vm_ip, &gateway_ip).is_some());
     }
 
-    #[test]
+    #[tokio::test]
     #[serial]
-    fn test_bare_default_block_applies_in_both_directions_in_stateful_mode() {
-        let vm_ip = Ipv4Address::new(192, 168, 0, 2);
+    async fn test_bare_default_block_applies_in_both_directions_in_stateful_mode() {
+        let mut proxy = create_proxy(vec!["in 192.0.2.0/24"], vec!["0.0.0.0/0"]);
+        let vm_ip = proxy.host.vm_ip;
         let fallback_peer = Ipv4Address::new(203, 0, 113, 1);
         let explicitly_allowed_peer = Ipv4Address::new(192, 0, 2, 1);
-        let mut proxy = create_proxy(vm_ip, vec!["in 192.0.2.0/24"], vec!["0.0.0.0/0"]);
 
         let fallback_request = udp_packet(fallback_peer, 40_000, vm_ip, 1_234);
         let fallback_request = Ipv4Packet::new_checked(fallback_request.as_slice()).unwrap();
@@ -339,12 +306,12 @@ mod tests {
         assert!(proxy.allowed_from_vm_ipv4(allowed_reply).is_some());
     }
 
-    #[test]
+    #[tokio::test]
     #[serial]
-    fn test_directional_egress_block_does_not_block_reply_to_unmatched_inbound_flow() {
-        let vm_ip = Ipv4Address::new(192, 168, 0, 2);
+    async fn test_directional_egress_block_does_not_block_reply_to_unmatched_inbound_flow() {
+        let mut proxy = create_proxy(vec![], vec!["out 203.0.113.0/24"]);
+        let vm_ip = proxy.host.vm_ip;
         let peer = Ipv4Address::new(203, 0, 113, 1);
-        let mut proxy = create_proxy(vm_ip, vec![], vec!["out 203.0.113.0/24"]);
 
         let request = udp_packet(peer, 40_000, vm_ip, 1_234);
         let request = Ipv4Packet::new_checked(request.as_slice()).unwrap();
@@ -355,12 +322,12 @@ mod tests {
         assert!(proxy.allowed_from_vm_ipv4(reply).is_some());
     }
 
-    #[test]
+    #[tokio::test]
     #[serial]
-    fn test_directional_ingress_block_does_not_block_reply_to_bare_outbound_allow() {
-        let vm_ip = Ipv4Address::new(192, 168, 0, 2);
+    async fn test_directional_ingress_block_does_not_block_reply_to_bare_outbound_allow() {
+        let mut proxy = create_proxy(vec!["203.0.113.0/24"], vec!["in 203.0.113.0/24"]);
+        let vm_ip = proxy.host.vm_ip;
         let peer = Ipv4Address::new(203, 0, 113, 1);
-        let mut proxy = create_proxy(vm_ip, vec!["203.0.113.0/24"], vec!["in 203.0.113.0/24"]);
 
         let request = udp_packet(vm_ip, 1_234, peer, 40_000);
         let request = Ipv4Packet::new_checked(request.as_slice()).unwrap();
@@ -371,7 +338,7 @@ mod tests {
         assert!(proxy.allowed_from_host_ipv4(&reply).is_some());
     }
 
-    fn create_proxy<'test>(vm_ip: Ipv4Address, allow: Vec<&str>, block: Vec<&str>) -> Proxy<'test> {
+    fn create_proxy(allow: Vec<&str>, block: Vec<&str>) -> Proxy {
         let (vm_fd, _) = socketpair(
             AddressFamily::Unix,
             SockType::Datagram,
@@ -381,9 +348,9 @@ mod tests {
         .unwrap();
         let vm_fd = Box::leak(Box::new(vm_fd));
 
-        let mut proxy = Proxy::new(
+        Proxy::new(
             vm_fd.as_raw_fd(),
-            MacAddress::from_str("02:00:00:00:00:01").unwrap(),
+            MacAddress::from_str("02:00:00:00:00:02").unwrap(),
             NetType::Nat,
             allow
                 .into_iter()
@@ -396,15 +363,7 @@ mod tests {
             Vec::default(),
             None,
         )
-        .unwrap();
-
-        proxy.dhcp_snooper.set_lease(Some(Lease::new(
-            vm_ip,
-            Duration::from_secs(600),
-            HashSet::new(),
-        )));
-
-        proxy
+        .unwrap()
     }
 
     fn allowed_from_vm_ipv4(proxy: &mut Proxy, src: Ipv4Address, dst: &str) -> Option<()> {

lib/proxy/port_forwarder.rs

@@ -1,99 +0,0 @@
-use crate::dhcp_snooper::Lease;
-use crate::host::Host;
-use crate::proxy::exposed_port::ExposedPort;
-use anyhow::Result;
-use log::error;
-use std::net::Ipv4Addr;
-
-#[derive(Default)]
-pub struct PortForwarder {
-    port_forwardings: Vec<PortForwarding>,
-    failed: bool,
-}
-
-#[derive(Debug, Clone, Copy, Default)]
-struct PortForwarding {
-    exposed_port: ExposedPort,
-    forwarding_to_addr: Option<Ipv4Addr>,
-}
-
-impl PortForwarder {
-    pub fn new(exposed_ports: Vec<ExposedPort>) -> PortForwarder {
-        let port_forwardings = exposed_ports
-            .into_iter()
-            .map(|exposed_port| PortForwarding {
-                exposed_port,
-                ..Default::default()
-            })
-            .collect();
-
-        PortForwarder {
-            port_forwardings,
-            ..Default::default()
-        }
-    }
-
-    pub fn tick(&mut self, host: &mut Host, lease: &Option<Lease>) {
-        if self.failed {
-            return;
-        }
-
-        if let Err(err) = self.tick_inner(host, lease) {
-            error!("port-forwarding failed: {}", err);
-
-            self.failed = true;
-        }
-    }
-
-    fn tick_inner(&mut self, host: &mut Host, lease: &Option<Lease>) -> Result<()> {
-        if let Some(lease) = lease {
-            // Lease exists, but is not valid, remove all port forwardings
-            if !lease.valid() {
-                self.remove_all_port_forwardings(host)?;
-
-                return Ok(());
-            }
-
-            // Lease exists and is valid, install/re-install port forwardings
-            for port_forwarding in &mut self.port_forwardings {
-                if let Some(installed_addr) = port_forwarding.forwarding_to_addr {
-                    // Port forwarding already installed, perhaps it's outdated?
-                    if installed_addr == lease.address() {
-                        // Nope, the port forwarding is up to date
-                        continue;
-                    }
-
-                    // Remove port forwarding since the lease address had changed
-                    host.port_forwarding_remove_rule(port_forwarding.exposed_port.external_port)?;
-                    port_forwarding.forwarding_to_addr = None;
-                }
-
-                // Install new port forwarding
-                host.port_forwarding_add_rule(
-                    port_forwarding.exposed_port.external_port,
-                    lease.address(),
-                    port_forwarding.exposed_port.internal_port,
-                )?;
-                port_forwarding.forwarding_to_addr = Some(lease.address());
-            }
-        } else {
-            // Lease does not exist, remove all port forwardings
-            self.remove_all_port_forwardings(host)?;
-        }
-
-        Ok(())
-    }
-
-    fn remove_all_port_forwardings(&mut self, host: &mut Host) -> Result<()> {
-        for port_forwarding in &mut self.port_forwardings {
-            if port_forwarding.forwarding_to_addr.is_none() {
-                continue;
-            }
-
-            host.port_forwarding_remove_rule(port_forwarding.exposed_port.external_port)?;
-            port_forwarding.forwarding_to_addr = None;
-        }
-
-        Ok(())
-    }
-}

lib/proxy/udp_packet_helper.rs

@@ -1,66 +0,0 @@
-use smoltcp::wire::UdpPacket;
-
-pub(crate) trait UdpPacketHelper {
-    const DNS_PORT: u16 = 53;
-    const BOOTPS_PORT: u16 = 67;
-    const BOOTPC_PORT: u16 = 68;
-
-    fn is_dns_request(&self) -> bool;
-
-    fn is_dhcp_request(&self) -> bool;
-    fn is_dhcp_response(&self) -> bool;
-}
-
-impl UdpPacketHelper for UdpPacket<&[u8]> {
-    fn is_dns_request(&self) -> bool {
-        self.dst_port() == Self::DNS_PORT
-    }
-
-    fn is_dhcp_request(&self) -> bool {
-        self.src_port() == Self::BOOTPC_PORT && self.dst_port() == Self::BOOTPS_PORT
-    }
-
-    fn is_dhcp_response(&self) -> bool {
-        self.src_port() == Self::BOOTPS_PORT && self.dst_port() == Self::BOOTPC_PORT
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::UdpPacketHelper;
-    use smoltcp::wire::UdpPacket;
-
-    #[test]
-    fn test_is_dhcp_request_requires_both_standard_ports() {
-        assert!(is_dhcp_request(68, 67));
-        assert!(!is_dhcp_request(68, 9999));
-        assert!(!is_dhcp_request(9999, 67));
-    }
-
-    #[test]
-    fn test_is_dhcp_response_requires_both_standard_ports() {
-        assert!(is_dhcp_response(67, 68));
-        assert!(!is_dhcp_response(67, 9999));
-        assert!(!is_dhcp_response(9999, 68));
-    }
-
-    fn is_dhcp_request(src_port: u16, dst_port: u16) -> bool {
-        let buffer = udp_packet_buffer(src_port, dst_port);
-        let udp_pkt = UdpPacket::new_unchecked(&buffer[..]);
-        udp_pkt.is_dhcp_request()
-    }
-
-    fn is_dhcp_response(src_port: u16, dst_port: u16) -> bool {
-        let buffer = udp_packet_buffer(src_port, dst_port);
-        let udp_pkt = UdpPacket::new_unchecked(&buffer[..]);
-        udp_pkt.is_dhcp_response()
-    }
-
-    fn udp_packet_buffer(src_port: u16, dst_port: u16) -> [u8; 8] {
-        let mut buffer = [0; 8];
-        let mut udp_pkt = UdpPacket::new_unchecked(&mut buffer[..]);
-        udp_pkt.set_src_port(src_port);
-        udp_pkt.set_dst_port(dst_port);
-        buffer
-    }
-}

lib/proxy/vm.rs

@@ -1,11 +1,8 @@
-use crate::dhcp_snooper::{Lease, message_matches_bootp_client};
 use crate::proxy::flows::{FlowDirection, FlowMatch};
-use crate::proxy::udp_packet_helper::UdpPacketHelper;
 use crate::proxy::{Direction, PolicyDecision, Proxy};
 use anyhow::Context;
 use anyhow::Result;
-use dhcproto::Decodable;
-use dhcproto::v4::Opcode;
+use dhcproto::v4::SERVER_PORT;
 use smoltcp::phy::ChecksumCapabilities;
 use smoltcp::wire::{
     ArpOperation, ArpPacket, ArpRepr, EthernetFrame, EthernetProtocol, IpProtocol, Ipv4Address,
@@ -14,7 +11,7 @@ use smoltcp::wire::{
 
 const IPV4_HEADER_LEN_WITHOUT_OPTIONS: u8 = 20;
 
-impl Proxy<'_> {
+impl Proxy {
     pub(crate) fn process_frame_from_vm(&mut self, frame: EthernetFrame<&[u8]>) -> Result<()> {
         if self.allowed_from_vm(&frame).is_none() {
             // Block packet by not forwarding it to the host
@@ -53,25 +50,17 @@ impl Proxy<'_> {
     }
 
     fn allowed_from_vm_arp(&self, arp_pkt: ArpPacket<&[u8]>) -> Option<()> {
-        vm_arp_allowed(arp_pkt, self.vm_mac_address, self.dhcp_snooper.lease())
+        vm_arp_allowed(arp_pkt, self.vm_mac_address, self.host.vm_ip)
     }
 
     pub(crate) fn allowed_from_vm_ipv4(&mut self, ipv4_pkt: Ipv4Packet<&[u8]>) -> Option<()> {
-        // Is this packet coming from VM's IP address that we've learned from DHCP snooping?
-        if let Some(lease) = &self.dhcp_snooper.lease()
-            && lease.is_valid_for(ipv4_pkt.src_addr())
-        {
-            // Unicast DHCP renewal is required to maintain the VM's lease
-            // and must bypass user-specified rules
-            if is_allowed_dhcp_request(
-                &ipv4_pkt,
-                Some(self.host.gateway_ip),
-                self.vm_mac_address,
-                self.dhcp_snooper.lease(),
-            ) {
-                return Some(());
-            }
+        // Consume DHCP before enforcing the reserved source address
+        if self.consume_dhcp(&ipv4_pkt) {
+            return None;
+        }
 
+        // Is this packet coming from the VM's reserved IP address?
+        if self.host.vm_ip == ipv4_pkt.src_addr() {
             // Consult the flow table before evaluating outbound policy
             // so established flows are not treated as new traffic
             let pending = match self
@@ -116,83 +105,50 @@ impl Proxy<'_> {
             if dst_addr == self.host.gateway_ip {
                 return self.admit_with_tracking_if_trackable(pending);
             }
-
-            // Additionally, allow DNS requests to DNS-servers
-            // provided to a VM by the host's DHCP server
-            if ipv4_pkt.next_header() == IpProtocol::Udp {
-                let udp_pkt = UdpPacket::new_checked(ipv4_pkt.payload()).ok()?;
-
-                if udp_pkt.is_dns_request() && self.dhcp_snooper.valid_dns_target(&dst_addr) {
-                    return self.admit_with_tracking_if_trackable(pending);
-                }
-            }
-        }
-
-        // Allow outgoing DHCP requests to the bootpd(8) broadcast address,
-        // otherwise DHCP snooper will never be populated
-        if is_allowed_dhcp_request(
-            &ipv4_pkt,
-            None,
-            self.vm_mac_address,
-            self.dhcp_snooper.lease(),
-        ) {
-            return Some(());
         }
 
         None
     }
-}
 
-fn is_allowed_dhcp_request(
-    ipv4_pkt: &Ipv4Packet<&[u8]>,
-    unicast_target: Option<Ipv4Address>,
-    vm_mac_address: smoltcp::wire::EthernetAddress,
-    lease: &Option<Lease>,
-) -> bool {
-    // Require the source address to be either:
-    // * covered by the VM's current lease
-    // * unspecified on the broadcast DHCP path
-    let src_addr = ipv4_pkt.src_addr();
-    let src_has_valid_lease = lease
-        .as_ref()
-        .is_some_and(|lease| lease.is_valid_for(src_addr));
-    if !src_has_valid_lease && !(unicast_target.is_none() && src_addr.is_unspecified()) {
-        return false;
+    fn consume_dhcp(&mut self, ipv4_pkt: &Ipv4Packet<&[u8]>) -> bool {
+        // Only inspect UDP packets that contain the source and destination ports
+        if ipv4_pkt.next_header() != IpProtocol::Udp
+            || ipv4_pkt.frag_offset() != 0
+            || ipv4_pkt.payload().len() < 4
+        {
+            return false;
+        }
+
+        // Only consume packets addressed to the DHCP server port
+        let udp = UdpPacket::new_unchecked(ipv4_pkt.payload());
+        if udp.dst_port() != SERVER_PORT {
+            return false;
+        }
+
+        // Pass the request to the DHCP server and send any reply
+        let result = match self.dhcp_server.receive_vm(ipv4_pkt, &udp) {
+            Ok(Some(reply)) => self.write_to_vm(&reply),
+            Ok(None) => Ok(()),
+            Err(err) => Err(err),
+        };
+
+        // Report failures to generate or send the reply
+        if let Err(err) = result {
+            sentry::capture_message(
+                &format!("Failed to reply to DHCP request: {err:#}"),
+                sentry::Level::Warning,
+            );
+        }
+
+        // Consume the packet even if the request was rejected or the reply failed
+        true
     }
-
-    let dst_addr = ipv4_pkt.dst_addr();
-
-    // Keep the common path cheap and inspect UDP only for a permitted DHCP target
-    if !dst_addr.is_broadcast() && unicast_target != Some(dst_addr) {
-        return false;
-    }
-
-    if ipv4_pkt.next_header() != IpProtocol::Udp {
-        return false;
-    }
-
-    let Ok(udp_pkt) = UdpPacket::new_checked(ipv4_pkt.payload()) else {
-        return false;
-    };
-
-    // Require the standard DHCP client and server ports
-    if !udp_pkt.is_dhcp_request() {
-        return false;
-    }
-
-    // Require the BOOTP client hardware address to match this VM
-    let mut decoder = dhcproto::v4::Decoder::new(udp_pkt.payload());
-    let Ok(message) = dhcproto::v4::Message::decode(&mut decoder) else {
-        return false;
-    };
-
-    message_matches_bootp_client(&message, Opcode::BootRequest, vm_mac_address.0)
 }
 
 fn vm_arp_allowed(
     arp_pkt: ArpPacket<&[u8]>,
     vm_mac_address: smoltcp::wire::EthernetAddress,
-    lease: &Option<Lease>,
+    address: Ipv4Address,
 ) -> Option<()> {
     let (operation, source_hardware_addr, source_protocol_addr) =
         match ArpRepr::parse(&arp_pkt).ok()? {
@@ -213,84 +169,34 @@ fn vm_arp_allowed(
         return None;
     }
 
-    if let Some(lease) = lease {
-        if lease.is_valid_for(source_protocol_addr) {
-            return Some(());
-        }
-    } else if source_protocol_addr.is_unspecified() {
-        return Some(());
-    }
-
-    None
+    (source_protocol_addr == address || source_protocol_addr.is_unspecified()).then_some(())
 }
 
 #[cfg(test)]
 mod tests {
-    use crate::dhcp_snooper::Lease;
-    use dhcproto::v4::{DhcpOption, Message, MessageType};
-    use dhcproto::{Encodable, Encoder};
     use smoltcp::wire::{
-        ArpHardware, ArpOperation, ArpPacket, EthernetAddress, EthernetProtocol, IpProtocol,
-        Ipv4Address, Ipv4Packet, UdpPacket,
+        ArpHardware, ArpOperation, ArpPacket, EthernetAddress, EthernetProtocol, Ipv4Address,
     };
-    use std::collections::HashSet;
-    use std::time::Duration;
-
-    const VM_MAC: EthernetAddress = EthernetAddress([0x02, 0x00, 0x00, 0x00, 0x00, 0x01]);
 
     #[test]
-    fn test_allowed_dhcp_request_policy() {
-        let gateway = Ipv4Address::new(192, 168, 64, 1);
-        let lease_ip = Ipv4Address::new(192, 168, 64, 2);
-        let other = Ipv4Address::new(192, 168, 64, 3);
-        let no_lease = None;
-        let lease = Some(Lease::new(
-            lease_ip,
-            Duration::from_secs(600),
-            HashSet::new(),
-        ));
-        let initial = |src, chaddr| {
-            allowed_dhcp_request(src, Ipv4Address::BROADCAST, None, chaddr, &no_lease)
-        };
-        let renewal = |src, dst| allowed_dhcp_request(src, dst, Some(gateway), VM_MAC.0, &lease);
-        let other_mac = [0x02, 0x00, 0x00, 0x00, 0x00, 0x02];
-
-        assert!(initial(Ipv4Address::UNSPECIFIED, VM_MAC.0));
-        assert!(renewal(lease_ip, gateway));
-        assert!(!renewal(other, gateway));
-        assert!(!renewal(Ipv4Address::UNSPECIFIED, gateway));
-        assert!(!renewal(lease_ip, other));
-        assert!(!initial(Ipv4Address::UNSPECIFIED, other_mac));
-    }
-
-    #[test]
-    fn test_allowed_from_vm_arp_allows_unspecified_request_without_lease() {
+    fn test_allowed_from_vm_arp_allows_unspecified_request() {
         let vm_mac_address = EthernetAddress([0x02, 0x00, 0x00, 0x00, 0x00, 0x01]);
         let buf = arp_packet(vm_mac_address.0, [0, 0, 0, 0], ArpOperation::Request, 6, 4);
         let arp_pkt = ArpPacket::new_checked(buf.as_slice()).unwrap();
 
-        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, &None).is_some());
+        assert!(
+            super::vm_arp_allowed(arp_pkt, vm_mac_address, Ipv4Address::new(1, 2, 3, 4)).is_some()
+        );
     }
 
     #[test]
     fn test_allowed_from_vm_arp_allows_reply_for_leased_ip() {
         let vm_mac_address = EthernetAddress([0x02, 0x00, 0x00, 0x00, 0x00, 0x01]);
-        let lease_ip = Ipv4Address::new(192, 168, 0, 2);
-        let lease = Some(Lease::new(
-            lease_ip,
-            Duration::from_secs(600),
-            HashSet::new(),
-        ));
-        let buf = arp_packet(
-            vm_mac_address.0,
-            lease_ip.octets(),
-            ArpOperation::Reply,
-            6,
-            4,
-        );
+        let vm_ip = Ipv4Address::new(192, 168, 0, 2);
+        let buf = arp_packet(vm_mac_address.0, vm_ip.octets(), ArpOperation::Reply, 6, 4);
         let arp_pkt = ArpPacket::new_checked(buf.as_slice()).unwrap();
 
-        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, &lease).is_some());
+        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, vm_ip).is_some());
     }
 
     #[test]
@@ -305,7 +211,9 @@ mod tests {
         );
         let arp_pkt = ArpPacket::new_checked(buf.as_slice()).unwrap();
 
-        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, &None).is_none());
+        assert!(
+            super::vm_arp_allowed(arp_pkt, vm_mac_address, Ipv4Address::new(1, 2, 3, 4)).is_none()
+        );
     }
 
     #[test]
@@ -316,7 +224,9 @@ mod tests {
         arp_pkt.set_hardware_type(ArpHardware::Unknown(2));
         let arp_pkt = ArpPacket::new_checked(buf.as_slice()).unwrap();
 
-        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, &None).is_none());
+        assert!(
+            super::vm_arp_allowed(arp_pkt, vm_mac_address, Ipv4Address::new(1, 2, 3, 4)).is_none()
+        );
     }
 
     #[test]
@@ -327,7 +237,9 @@ mod tests {
         arp_pkt.set_protocol_type(EthernetProtocol::Ipv6);
         let arp_pkt = ArpPacket::new_checked(buf.as_slice()).unwrap();
 
-        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, &None).is_none());
+        assert!(
+            super::vm_arp_allowed(arp_pkt, vm_mac_address, Ipv4Address::new(1, 2, 3, 4)).is_none()
+        );
     }
 
     #[test]
@@ -336,7 +248,9 @@ mod tests {
         let buf = arp_packet(vm_mac_address.0, [0, 0, 0], ArpOperation::Request, 6, 3);
         let arp_pkt = ArpPacket::new_checked(buf.as_slice()).unwrap();
 
-        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, &None).is_none());
+        assert!(
+            super::vm_arp_allowed(arp_pkt, vm_mac_address, Ipv4Address::new(1, 2, 3, 4)).is_none()
+        );
     }
 
     fn arp_packet(
@@ -361,55 +275,4 @@ mod tests {
         arp_pkt.set_target_protocol_addr(&vec![0; protocol_len as usize]);
         buf
     }
-
-    fn allowed_dhcp_request(
-        src_addr: Ipv4Address,
-        dst_addr: Ipv4Address,
-        unicast_target: Option<Ipv4Address>,
-        chaddr: [u8; 6],
-        lease: &Option<Lease>,
-    ) -> bool {
-        let mut buf = dhcp_request(chaddr);
-        let mut ipv4_pkt = Ipv4Packet::new_unchecked(buf.as_mut_slice());
-        ipv4_pkt.set_src_addr(src_addr);
-        ipv4_pkt.set_dst_addr(dst_addr);
-
-        let ipv4_pkt = Ipv4Packet::new_checked(buf.as_slice()).unwrap();
-        super::is_allowed_dhcp_request(&ipv4_pkt, unicast_target, VM_MAC, lease)
-    }
-
-    fn dhcp_request(chaddr: [u8; 6]) -> Vec<u8> {
-        let mut message = Message::new(
-            Ipv4Address::UNSPECIFIED,
-            Ipv4Address::UNSPECIFIED,
-            Ipv4Address::UNSPECIFIED,
-            Ipv4Address::UNSPECIFIED,
-            &chaddr,
-        );
-        message
-            .opts_mut()
-            .insert(DhcpOption::MessageType(MessageType::Discover));
-
-        let mut dhcp_payload = Vec::new();
-        message
-            .encode(&mut Encoder::new(&mut dhcp_payload))
-            .unwrap();
-
-        let total_len = 20 + 8 + dhcp_payload.len();
-        let mut buf = vec![0; total_len];
-        let mut ipv4_pkt = Ipv4Packet::new_unchecked(buf.as_mut_slice());
-        ipv4_pkt.set_version(4);
-        ipv4_pkt.set_header_len(20);
-        ipv4_pkt.set_total_len(total_len as u16);
-        ipv4_pkt.set_next_header(IpProtocol::Udp);
-        ipv4_pkt.set_src_addr(Ipv4Address::UNSPECIFIED);
-        ipv4_pkt.set_dst_addr(Ipv4Address::BROADCAST);
-
-        let mut udp_pkt = UdpPacket::new_unchecked(ipv4_pkt.payload_mut());
-        udp_pkt.set_src_port(68);
-        udp_pkt.set_dst_port(67);
-        udp_pkt.set_len((8 + dhcp_payload.len()) as u16);
-        udp_pkt.payload_mut().copy_from_slice(&dhcp_payload);
-        buf
-    }
 }

lib/subnet_finder/mod.rs

@@ -0,0 +1,92 @@
+use anyhow::{Context, Result};
+use ipnet::{Ipv4AddrRange, Ipv4Net};
+use network_interface::{Addr, NetworkInterface, NetworkInterfaceConfig};
+use rand::seq::{IndexedRandom, IteratorRandom};
+use std::net::Ipv4Addr;
+
+const MAX_ATTEMPTS: usize = 128;
+
+/// Find an available private subnet and its first two usable host addresses
+pub(crate) fn find_available_subnet(prefix_len: u8) -> Result<(Ipv4Addr, Ipv4Addr, Ipv4Net)> {
+    // Add private address space (as defined in RFC 1918[1]) with one exception
+    //
+    // We narrow the 192.168.0.0/16 allocation pool to work around a macOS bug
+    // where long subnet address strings lose their CIDR prefix in the generated
+    // firewall rules, causing connection issues.
+    //
+    // [1]: https://datatracker.ietf.org/doc/html/rfc1918#section-3
+    let private_subnets: Vec<Ipv4Net> = vec![
+        "10.0.0.0/8".parse()?,
+        "172.16.0.0/12".parse()?,
+        "192.168.0.0/18".parse()?,  // third octet 0–63
+        "192.168.64.0/19".parse()?, // third octet 64–95
+        "192.168.96.0/22".parse()?, // third octet 96–99
+    ];
+
+    // Figure out which address space is already utilized on the host
+    let mut used_subnets = Vec::new();
+
+    for interface in NetworkInterface::show()? {
+        for addr in interface.addr {
+            // We only support IPv4 for now
+            if let Addr::V4(addr) = addr {
+                let mask = addr.netmask.context("IPv4 interface has no netmask")?;
+                used_subnets.push(Ipv4Net::with_netmask(addr.ip, mask)?);
+            }
+        }
+    }
+
+    // Try up to MAX_ATTEMPTS random subnets until we're able to
+    // get an available subnet of the desired length
+    for _ in 0..MAX_ATTEMPTS {
+        // Give each private range an equal chance, even if a larger one is occupied
+        let private_subnet = private_subnets.choose(&mut rand::rng()).unwrap();
+
+        // Skip private ranges too small to contain the requested subnet
+        if prefix_len < private_subnet.prefix_len() {
+            continue;
+        }
+
+        // Pick a subnet of the desired length within this private range
+        let address = Ipv4AddrRange::new(private_subnet.network(), private_subnet.broadcast())
+            .choose(&mut rand::rng())
+            .unwrap();
+        let candidate = Ipv4Net::new(address, prefix_len)?.trunc();
+
+        // Only use the subnet if it doesn't overlap address space
+        // that is already utilized on the host
+        let overlaps = used_subnets
+            .iter()
+            .any(|used_subnet| candidate.contains(used_subnet) || used_subnet.contains(&candidate));
+        if overlaps {
+            continue;
+        }
+
+        // Take the first two hosts from the subnet
+        let mut hosts = candidate.hosts();
+        if let (Some(first_host), Some(second_host)) = (hosts.next(), hosts.next()) {
+            return Ok((first_host, second_host, candidate));
+        }
+    }
+
+    anyhow::bail!(
+        "failed to find an unused private IPv4 /{prefix_len} subnet after {MAX_ATTEMPTS} attempts"
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    #[serial_test::serial]
+    fn finds_two_usable_addresses() {
+        // Find a subnet on the current host
+        let (gateway_ip, vm_ip, subnet) = find_available_subnet(29).unwrap();
+
+        // Check the subnet and its first two usable addresses
+        assert_eq!(subnet.prefix_len(), 29);
+        assert!(subnet.network().is_private());
+        assert_eq!(subnet.hosts().collect::<Vec<_>>()[..2], [gateway_ip, vm_ip]);
+    }
+}

lib/vm.rs

@@ -1,11 +1,13 @@
 use anyhow::{Context, Result, bail};
 use std::io;
 use std::mem::{size_of, zeroed};
-use std::os::fd::{AsRawFd, FromRawFd, RawFd};
+use std::os::fd::{FromRawFd, RawFd};
 use std::os::unix::net::UnixDatagram;
+use tokio::io::Interest;
+use tokio::io::unix::AsyncFd;
 
 pub struct VM {
-    sock: UnixDatagram,
+    sock: AsyncFd<UnixDatagram>,
 }
 
 impl VM {
@@ -16,19 +18,27 @@ impl VM {
         let sock = unsafe { UnixDatagram::from_raw_fd(vm_fd) };
         sock.set_nonblocking(true)?;
 
-        Ok(VM { sock })
+        Ok(VM {
+            sock: AsyncFd::new(sock)?,
+        })
     }
 
     pub fn write(&self, pkt: &[u8]) -> std::io::Result<usize> {
-        self.sock.send(pkt)
+        self.sock.get_ref().send(pkt)
     }
 
-    pub fn read(&self, buf: &mut [u8]) -> std::io::Result<usize> {
-        self.sock.recv(buf)
+    pub fn try_read(&self, buf: &mut [u8]) -> io::Result<usize> {
+        self.sock.get_ref().recv(buf)
+    }
+
+    pub async fn read(&self, buf: &mut [u8]) -> io::Result<usize> {
+        self.sock
+            .async_io(Interest::READABLE, |sock| sock.recv(buf))
+            .await
     }
 
     pub fn is_connected(&self) -> io::Result<bool> {
-        match self.sock.peer_addr() {
+        match self.sock.get_ref().peer_addr() {
             Ok(_) => Ok(true),
             Err(error) if error.kind() == io::ErrorKind::NotConnected => Ok(false),
             Err(error) => Err(error),
@@ -114,16 +124,9 @@ fn validate_vm_fd(vm_fd: RawFd) -> Result<()> {
     Ok(())
 }
 
-impl AsRawFd for VM {
-    fn as_raw_fd(&self) -> RawFd {
-        self.sock.as_raw_fd()
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::VM;
-    use polling::{Event, Events, PollMode, Poller};
     use std::fs::File;
     use std::net::UdpSocket;
     use std::os::fd::AsRawFd;
@@ -180,8 +183,8 @@ mod tests {
         assert!(error.to_string().contains("not a Unix socket"));
     }
 
-    #[test]
-    fn test_new_does_not_close_original_fd_when_vm_is_dropped() {
+    #[tokio::test]
+    async fn test_new_does_not_close_original_fd_when_vm_is_dropped() {
         let (socket, _peer) = UnixDatagram::pair().unwrap();
         let vm = VM::new(socket.as_raw_fd()).unwrap();
         drop(vm);
@@ -198,61 +201,52 @@ mod tests {
         assert!(socket_fd_is_open);
     }
 
-    #[test]
-    fn test_connected_socket_has_peer() {
+    #[tokio::test]
+    async fn test_connected_socket_has_peer() {
         let (socket, _peer) = UnixDatagram::pair().unwrap();
         let vm = VM::new(socket.as_raw_fd()).unwrap();
 
         assert!(vm.is_connected().unwrap());
     }
 
-    #[test]
-    fn test_disconnected_peer_is_detected_without_kqueue_event() {
+    #[tokio::test]
+    async fn test_disconnected_peer_is_detected_without_readiness() {
         let (socket, peer) = UnixDatagram::pair().unwrap();
         let vm = VM::new(socket.as_raw_fd()).unwrap();
-        let poller = Poller::new().unwrap();
-        let mut events = Events::new();
-
-        unsafe {
-            poller
-                .add_with_mode(vm.as_raw_fd(), Event::readable(0), PollMode::Edge)
-                .unwrap();
-        }
+        let mut buf = [0];
         drop(peer);
 
-        poller
-            .wait(&mut events, Some(Duration::from_millis(20)))
-            .unwrap();
-
-        assert!(events.is_empty());
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), vm.read(&mut buf))
+                .await
+                .is_err()
+        );
         assert!(!vm.is_connected().unwrap());
     }
 
-    #[test]
-    fn test_disconnected_peer_is_detected_when_another_socket_wakes_kqueue() {
+    #[tokio::test]
+    async fn test_disconnected_peer_is_detected_when_another_socket_is_readable() {
         let (socket, peer) = UnixDatagram::pair().unwrap();
         let (host, host_peer) = UnixDatagram::pair().unwrap();
         let vm = VM::new(socket.as_raw_fd()).unwrap();
-        let poller = Poller::new().unwrap();
-        let mut events = Events::new();
-
-        unsafe {
-            poller
-                .add_with_mode(vm.as_raw_fd(), Event::readable(0), PollMode::Edge)
-                .unwrap();
-            poller
-                .add_with_mode(host.as_raw_fd(), Event::readable(1), PollMode::Edge)
-                .unwrap();
-        }
+        host.set_nonblocking(true).unwrap();
+        let host = tokio::net::UnixDatagram::from_std(host).unwrap();
+        let mut buf = [0];
         drop(peer);
         host_peer.send(&[1]).unwrap();
 
-        poller
-            .wait(&mut events, Some(Duration::from_millis(20)))
-            .unwrap();
-
-        assert!(events.iter().any(|event| event.key == 1));
-        assert!(!events.iter().any(|event| event.key == 0));
+        assert_eq!(
+            tokio::time::timeout(Duration::from_secs(2), host.recv(&mut buf))
+                .await
+                .unwrap()
+                .unwrap(),
+            1
+        );
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), vm.read(&mut buf))
+                .await
+                .is_err()
+        );
         assert!(!vm.is_connected().unwrap());
     }
 }

src/main.rs

@@ -1,7 +1,6 @@
 use anyhow::{Context, anyhow};
 use clap::Parser;
 use log::LevelFilter;
-use nix::sys::signal::{SigHandler, Signal, signal};
 use oslog::OsLogger;
 use privdrop::PrivDrop;
 use softnet::NetType;
@@ -23,9 +22,16 @@ use system_configuration::sys::preferences::{
     SCPreferencesApplyChanges, SCPreferencesCommitChanges, SCPreferencesLock,
     SCPreferencesSetValue, SCPreferencesUnlock,
 };
+use tokio::signal::unix::{SignalKind, signal};
 use uzers::{get_current_groupname, get_current_username, get_effective_uid};
 
+const VERSION: &str = match option_env!("SOFTNET_VERSION") {
+    Some(version) => version,
+    None => "unknown-unknown",
+};
+
 #[derive(Parser, Debug)]
+#[command(version = VERSION)]
 struct Args {
     #[clap(
         long,
@@ -162,20 +168,13 @@ fn main() -> ExitCode {
     }
 }
 
-fn try_main() -> anyhow::Result<()> {
+#[tokio::main(flavor = "current_thread")]
+async fn try_main() -> anyhow::Result<()> {
     // Initialize logger
     OsLogger::new("org.cirruslabs.softnet")
         .level_filter(LevelFilter::Info)
         .init()?;
 
-    // The default signal(3)[1] action for SIGINT is to interrupt program,
-    // but we want to handle SIGINT ourselves, so we ignore it. The kqueue(2)'s[2]
-    // EVFILT_SIGNAL will receive it anyways, because it has lower precedence.
-    //
-    // [1]: https://developer.apple.com/library/archive/documentation/System/Conceptual/ManPages_iPhoneOS/man3/signal.3.html
-    // [2]: https://developer.apple.com/library/archive/documentation/System/Conceptual/ManPages_iPhoneOS/man2/kqueue.2.html
-    unsafe { signal(Signal::SIGINT, SigHandler::SigIgn) }?;
-
     let args: Args = Args::parse();
 
     // No need to run anything, just return
@@ -242,8 +241,13 @@ fn try_main() -> anyhow::Result<()> {
         .apply()
         .context("failed to drop privileges")?;
 
-    // Run proxy
-    proxy.run()
+    // Run the proxy until it finishes or the process receives SIGINT
+    let mut interrupt = signal(SignalKind::interrupt())?;
+
+    tokio::select! {
+        result = proxy.run() => result,
+        _ = interrupt.recv() => Ok(()),
+    }
 }
 
 fn parse_vm_fd(value: &str) -> Result<c_int, String> {