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,15 +27,16 @@ 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 = "3"
+serial_test = "4"
 coarsetime = "0.1.37"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

README.md

@@ -23,15 +23,39 @@ 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:
+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.
 
-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
+### Stateful flow authorization
 
-...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.
+Stateful `in`/`out` rules use a bounded authorization cache that records the
+direction and exact transport tuple of policy-approved flows, allowing matching
+return traffic without treating it as a new flow and thus requiring a separate
+policy entry. It is not a complete TCP connection tracker: endpoint transport
+stacks remain responsible for validating sequence numbers, receive windows,
+resets, and application-level traffic.
+
+This cache deliberately favors security, bounded resource use, and a simple
+implementation over availability. Only policy-authorized initiator traffic
+renews an entry; return traffic does not. Softnet does not maintain fairness
+quotas, eviction heuristics, or complete TCP lifecycle state.
+
+A packet admitted by a stateful rule is denied when Softnet cannot represent its
+flow, including when the cache is full. High flow churn, long idle connections,
+or ambiguous retransmissions may therefore interrupt networking and require the
+affected VM to reconnect.
+
+For TCP, a bare TCP SYN on an existing tuple is deliberately returned to policy
+because Softnet cannot distinguish a retransmission from tuple reuse without
+tracking TCP sequence state. If authorized, it may replace the tuple's previous
+cache lifetime; this can reduce availability but cannot grant traffic that
+policy did not permit.
+
+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
@@ -57,6 +81,6 @@ The supported methods are `softnet.policy.get` and `softnet.policy.set`. A compl
 
 Every request must include a non-null string (at most 256 bytes) or non-negative integer `id`; notifications are rejected so policy changes always have an acknowledgment. Policy updates are atomic: all rules are parsed and a new prefix map is built before the active policy changes. Longest-prefix matching and block precedence for identical rules are preserved. Rules are normalized and deduplicated. A policy update may contain at most 4096 combined allow/block rules, and a request frame may not exceed 1 MiB.
 
-When the effective normalized policy changes, Softnet clears existing conntrack state so the new policy applies to established flows immediately. This may interrupt active connections. Repeating an equivalent normalized policy is a no-op and preserves conntrack state.
+When the normalized allow or block policy changes, Softnet clears the flow table so the new policy applies to established flows immediately. This may interrupt active connections. Repeating the same normalized policy is a no-op and preserves the flow table.
 
-Use `block=["0.0.0.0/0"]` with specific allow rules for a default-deny policy. Closing the control socket leaves the last accepted policy active.
+Use `block=["0.0.0.0/0"]` with specific allow rules for a default-deny egress policy. Closing the control socket leaves the last accepted policy active.

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,102 +0,0 @@
-use dhcproto::Decodable;
-use dhcproto::v4::{DhcpOption, MessageType, OptionCode};
-use smoltcp::wire::Ipv4Address;
-use std::collections::HashSet;
-use std::time::Duration;
-
-#[derive(Default)]
-pub struct DhcpSnooper {
-    vm_lease: Option<Lease>,
-    uncertainty_duration: Duration,
-}
-
-impl DhcpSnooper {
-    pub fn new(uncertainty_duration: Duration) -> Self {
-        DhcpSnooper {
-            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,
-        };
-
-        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 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()
-    }
-}

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/conntrack.rs

@@ -1,386 +0,0 @@
-#[path = "conntrack_tcp.rs"]
-mod tcp;
-#[path = "conntrack_udp.rs"]
-mod udp;
-
-use coarsetime::{Duration, Instant};
-use smoltcp::wire::{IpProtocol, Ipv4Address, Ipv4Packet};
-use std::collections::HashMap;
-
-const MAX_FLOWS: usize = 4096;
-const MAX_VM_INITIATED_FLOWS: usize = 1024;
-const MAX_EMBRYONIC_FLOWS_PER_SOURCE: usize = 256;
-const SWEEP_INTERVAL: Duration = Duration::from_secs(1);
-
-/// Tracks permission to use an exact host/VM flow tuple.
-///
-/// This deliberately does not replace either endpoint's transport stack:
-/// TCP sequence/window validation remains the host's and VM's responsibility.
-/// Its security boundary is flow initiation: only an authorized first packet
-/// can create an entry, and VM traffic to the host must match the reverse.
-#[derive(Debug)]
-pub(crate) struct Conntrack {
-    flows: HashMap<FlowKey, Flow>,
-    next_sweep: Instant,
-}
-
-pub(crate) enum ConntrackResult {
-    Allowed,
-    New(PendingFlow),
-    Denied,
-}
-
-pub(crate) struct PendingFlow {
-    key: FlowKey,
-    flow: Flow,
-}
-
-#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
-enum FlowKey {
-    Tcp {
-        host_addr: Ipv4Address,
-        host_port: u16,
-        vm_addr: Ipv4Address,
-        vm_port: u16,
-    },
-    Udp {
-        host_addr: Ipv4Address,
-        host_port: u16,
-        vm_addr: Ipv4Address,
-        vm_port: u16,
-    },
-}
-
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
-enum Initiator {
-    Host,
-    Vm,
-}
-
-#[derive(Clone, Copy, Debug)]
-struct Flow {
-    initiator: Initiator,
-    state: FlowState,
-    last_seen: Instant,
-}
-
-#[derive(Clone, Copy, Debug)]
-enum FlowState {
-    Tcp(tcp::State),
-    Udp(udp::State),
-}
-
-#[derive(Clone, Copy)]
-enum Direction {
-    FromHost,
-    FromVm,
-}
-
-impl Conntrack {
-    pub(crate) fn new() -> Self {
-        let now = Instant::recent();
-        Self {
-            flows: HashMap::new(),
-            next_sweep: now + SWEEP_INTERVAL,
-        }
-    }
-
-    pub(crate) fn inspect_from_host(&mut self, packet: &Ipv4Packet<&[u8]>) -> ConntrackResult {
-        self.inspect(packet, Direction::FromHost, Instant::recent())
-    }
-
-    pub(crate) fn inspect_from_vm(&mut self, packet: &Ipv4Packet<&[u8]>) -> ConntrackResult {
-        self.inspect(packet, Direction::FromVm, Instant::recent())
-    }
-
-    pub(crate) fn commit(&mut self, mut pending: PendingFlow) -> bool {
-        let now = Instant::recent();
-        pending.flow.last_seen = now;
-        self.insert(pending.key, pending.flow, now)
-    }
-
-    pub(crate) fn tick(&mut self) {
-        let now = Instant::recent();
-        if now >= self.next_sweep {
-            self.expire(now);
-            self.next_sweep = now + SWEEP_INTERVAL;
-        }
-    }
-
-    pub(crate) fn clear(&mut self) {
-        self.flows.clear();
-    }
-
-    fn inspect(
-        &mut self,
-        packet: &Ipv4Packet<&[u8]>,
-        direction: Direction,
-        now: Instant,
-    ) -> ConntrackResult {
-        // Later fragments do not contain the transport header needed to bind them
-        // to a permitted flow. Fail closed instead of admitting them by IP alone.
-        if packet.more_frags() || packet.frag_offset() != 0 {
-            return ConntrackResult::Denied;
-        }
-
-        match packet.next_header() {
-            IpProtocol::Tcp => self.inspect_tcp(packet, direction, now),
-            IpProtocol::Udp => self.inspect_udp(packet, direction, now),
-            _ => ConntrackResult::Denied,
-        }
-    }
-
-    fn insert(&mut self, key: FlowKey, flow: Flow, now: Instant) -> bool {
-        self.expire(now);
-        if self.flows.len() >= MAX_FLOWS {
-            return false;
-        }
-        if flow.initiator == Initiator::Vm
-            && self
-                .flows
-                .values()
-                .filter(|flow| flow.initiator == Initiator::Vm)
-                .count()
-                >= MAX_VM_INITIATED_FLOWS
-        {
-            return false;
-        }
-        if let Some(initiator_addr) = embryonic_source(key, flow)
-            && self
-                .flows
-                .iter()
-                .filter(|(key, flow)| embryonic_source(**key, **flow) == Some(initiator_addr))
-                .count()
-                >= MAX_EMBRYONIC_FLOWS_PER_SOURCE
-        {
-            return false;
-        }
-        self.flows.insert(key, flow);
-        true
-    }
-
-    fn expire(&mut self, now: Instant) {
-        self.flows
-            .retain(|_, flow| now.duration_since(flow.last_seen) < flow.timeout());
-    }
-}
-
-impl Flow {
-    fn timeout(&self) -> Duration {
-        match self.state {
-            FlowState::Tcp(state) => state.timeout(),
-            FlowState::Udp(state) => state.timeout(),
-        }
-    }
-}
-
-fn embryonic_source(key: FlowKey, flow: Flow) -> Option<Ipv4Address> {
-    match (key, flow.initiator, flow.state) {
-        (
-            FlowKey::Tcp { host_addr, .. },
-            Initiator::Host,
-            FlowState::Tcp(tcp::State::SynSent | tcp::State::SynReceived),
-        ) => Some(host_addr),
-        (
-            FlowKey::Tcp { vm_addr, .. },
-            Initiator::Vm,
-            FlowState::Tcp(tcp::State::SynSent | tcp::State::SynReceived),
-        ) => Some(vm_addr),
-        _ => None,
-    }
-}
-
-fn oriented_transport_key(
-    packet: &Ipv4Packet<&[u8]>,
-    direction: Direction,
-    make_key: impl FnOnce(Ipv4Address, u16, Ipv4Address, u16) -> FlowKey,
-    src_port: u16,
-    dst_port: u16,
-) -> FlowKey {
-    match direction {
-        Direction::FromHost => make_key(packet.src_addr(), src_port, packet.dst_addr(), dst_port),
-        Direction::FromVm => make_key(packet.dst_addr(), dst_port, packet.src_addr(), src_port),
-    }
-}
-
-fn initiator(direction: Direction) -> Initiator {
-    match direction {
-        Direction::FromHost => Initiator::Host,
-        Direction::FromVm => Initiator::Vm,
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::test_support::{HOST, TcpFlags, VM, inspect_from_host, inspect_from_vm, tcp_packet};
-    use super::{Conntrack, ConntrackResult, MAX_EMBRYONIC_FLOWS_PER_SOURCE};
-    use smoltcp::wire::{Ipv4Address, Ipv4Packet};
-
-    #[test]
-    fn fragments_fail_closed() {
-        let mut packet = tcp_packet(HOST, 49152, VM, 22, TcpFlags::SYN);
-        Ipv4Packet::new_unchecked(packet.as_mut_slice()).set_more_frags(true);
-
-        let mut tracker = Conntrack::new();
-        assert!(matches!(
-            inspect_from_host(&mut tracker, &packet),
-            ConntrackResult::Denied
-        ));
-    }
-
-    #[test]
-    fn clear_removes_tracked_flows() {
-        let packet = tcp_packet(HOST, 49152, VM, 22, TcpFlags::SYN);
-        let mut tracker = Conntrack::new();
-
-        let ConntrackResult::New(pending) = inspect_from_host(&mut tracker, &packet) else {
-            panic!("expected a new flow");
-        };
-        assert!(tracker.commit(pending));
-        assert_eq!(tracker.flows.len(), 1);
-
-        tracker.clear();
-        assert!(tracker.flows.is_empty());
-    }
-
-    #[test]
-    fn embryonic_tcp_limit_is_per_source() {
-        let mut tracker = Conntrack::new();
-
-        for index in 0..MAX_EMBRYONIC_FLOWS_PER_SOURCE {
-            let syn = tcp_packet(HOST, 10000 + index as u16, VM, 22, TcpFlags::SYN);
-            let ConntrackResult::New(pending) = inspect_from_host(&mut tracker, &syn) else {
-                panic!("expected a new flow");
-            };
-            assert!(tracker.commit(pending));
-        }
-
-        let over_limit = tcp_packet(HOST, 20000, VM, 22, TcpFlags::SYN);
-        let ConntrackResult::New(pending) = inspect_from_host(&mut tracker, &over_limit) else {
-            panic!("expected a new flow");
-        };
-        assert!(!tracker.commit(pending));
-
-        let other_host = Ipv4Address::new(192, 168, 64, 3);
-        let other_source = tcp_packet(other_host, 20000, VM, 22, TcpFlags::SYN);
-        let ConntrackResult::New(pending) = inspect_from_host(&mut tracker, &other_source) else {
-            panic!("expected a new flow");
-        };
-        assert!(tracker.commit(pending));
-
-        let syn_ack = tcp_packet(VM, 22, HOST, 10000, TcpFlags::SYN_ACK);
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &syn_ack),
-            ConntrackResult::Allowed
-        ));
-        let ack = tcp_packet(HOST, 10000, VM, 22, TcpFlags::ACK);
-        assert!(matches!(
-            inspect_from_host(&mut tracker, &ack),
-            ConntrackResult::Allowed
-        ));
-
-        let ConntrackResult::New(pending) = inspect_from_host(&mut tracker, &over_limit) else {
-            panic!("expected a new flow");
-        };
-        assert!(tracker.commit(pending));
-    }
-}
-
-#[cfg(test)]
-mod test_support {
-    use super::{Conntrack, ConntrackResult};
-    use smoltcp::wire::{IpProtocol, Ipv4Address, Ipv4Packet, TcpPacket, UdpPacket};
-
-    pub(super) const HOST: Ipv4Address = Ipv4Address::new(192, 168, 64, 1);
-    pub(super) const VM: Ipv4Address = Ipv4Address::new(192, 168, 64, 2);
-
-    pub(super) fn inspect_from_host(tracker: &mut Conntrack, bytes: &[u8]) -> ConntrackResult {
-        let packet = Ipv4Packet::new_checked(bytes).unwrap();
-        tracker.inspect_from_host(&packet)
-    }
-
-    pub(super) fn inspect_from_vm(tracker: &mut Conntrack, bytes: &[u8]) -> ConntrackResult {
-        let packet = Ipv4Packet::new_checked(bytes).unwrap();
-        tracker.inspect_from_vm(&packet)
-    }
-
-    #[derive(Clone, Copy)]
-    pub(super) struct TcpFlags {
-        syn: bool,
-        ack: bool,
-        rst: bool,
-    }
-
-    impl TcpFlags {
-        pub(super) const SYN: Self = Self {
-            syn: true,
-            ack: false,
-            rst: false,
-        };
-        pub(super) const SYN_ACK: Self = Self {
-            syn: true,
-            ack: true,
-            rst: false,
-        };
-        pub(super) const ACK: Self = Self {
-            syn: false,
-            ack: true,
-            rst: false,
-        };
-        pub(super) const RST: Self = Self {
-            syn: false,
-            ack: false,
-            rst: true,
-        };
-    }
-
-    pub(super) fn tcp_packet(
-        src_addr: Ipv4Address,
-        src_port: u16,
-        dst_addr: Ipv4Address,
-        dst_port: u16,
-        flags: TcpFlags,
-    ) -> Vec<u8> {
-        let mut bytes = ipv4_packet(src_addr, dst_addr, IpProtocol::Tcp, 20);
-        let mut tcp = TcpPacket::new_unchecked(&mut bytes[20..]);
-        tcp.set_src_port(src_port);
-        tcp.set_dst_port(dst_port);
-        tcp.set_header_len(20);
-        tcp.set_syn(flags.syn);
-        tcp.set_ack(flags.ack);
-        tcp.set_rst(flags.rst);
-        bytes
-    }
-
-    pub(super) fn udp_packet(
-        src_addr: Ipv4Address,
-        src_port: u16,
-        dst_addr: Ipv4Address,
-        dst_port: u16,
-    ) -> Vec<u8> {
-        let mut bytes = ipv4_packet(src_addr, dst_addr, IpProtocol::Udp, 8);
-        let mut udp = UdpPacket::new_unchecked(&mut bytes[20..]);
-        udp.set_src_port(src_port);
-        udp.set_dst_port(dst_port);
-        udp.set_len(8);
-        bytes
-    }
-
-    fn ipv4_packet(
-        src_addr: Ipv4Address,
-        dst_addr: Ipv4Address,
-        protocol: IpProtocol,
-        payload_len: usize,
-    ) -> Vec<u8> {
-        let mut bytes = vec![0; 20 + payload_len];
-        let total_len = bytes.len() as u16;
-        let mut ipv4 = Ipv4Packet::new_unchecked(bytes.as_mut_slice());
-        ipv4.set_version(4);
-        ipv4.set_header_len(20);
-        ipv4.set_total_len(total_len);
-        ipv4.set_next_header(protocol);
-        ipv4.set_src_addr(src_addr);
-        ipv4.set_dst_addr(dst_addr);
-        bytes
-    }
-}

lib/proxy/conntrack_tcp.rs

@@ -1,257 +0,0 @@
-use super::{
-    Conntrack, ConntrackResult, Direction, Flow, FlowKey, FlowState, PendingFlow, initiator,
-    oriented_transport_key,
-};
-use coarsetime::{Duration, Instant};
-use smoltcp::wire::{Ipv4Packet, TcpPacket};
-
-const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(60);
-const ESTABLISHED_TIMEOUT: Duration = Duration::from_secs(5 * 24 * 60 * 60);
-const CLOSING_TIMEOUT: Duration = Duration::from_secs(2 * 60);
-
-#[derive(Clone, Copy, Debug)]
-pub(super) enum State {
-    SynSent,
-    SynReceived,
-    Established,
-    Closing { host_fin: bool, vm_fin: bool },
-}
-
-impl State {
-    pub(super) fn timeout(self) -> Duration {
-        match self {
-            Self::SynSent | Self::SynReceived => HANDSHAKE_TIMEOUT,
-            Self::Established => ESTABLISHED_TIMEOUT,
-            Self::Closing { .. } => CLOSING_TIMEOUT,
-        }
-    }
-}
-
-impl Conntrack {
-    pub(super) fn inspect_tcp(
-        &mut self,
-        packet: &Ipv4Packet<&[u8]>,
-        direction: Direction,
-        now: Instant,
-    ) -> ConntrackResult {
-        let Ok(tcp) = TcpPacket::new_checked(packet.payload()) else {
-            return ConntrackResult::Denied;
-        };
-        if tcp.src_port() == 0 || tcp.dst_port() == 0 {
-            return ConntrackResult::Denied;
-        }
-
-        let key = oriented_transport_key(
-            packet,
-            direction,
-            |host_addr, host_port, vm_addr, vm_port| FlowKey::Tcp {
-                host_addr,
-                host_port,
-                vm_addr,
-                vm_port,
-            },
-            tcp.src_port(),
-            tcp.dst_port(),
-        );
-
-        if let Some(flow) = self.flows.get_mut(&key) {
-            let from_initiator = matches!(
-                (flow.initiator, direction),
-                (super::Initiator::Host, Direction::FromHost)
-                    | (super::Initiator::Vm, Direction::FromVm)
-            );
-            let FlowState::Tcp(state) = &mut flow.state else {
-                return ConntrackResult::Denied;
-            };
-
-            if tcp.rst() {
-                self.flows.remove(&key);
-                return ConntrackResult::Allowed;
-            }
-
-            let allowed = match *state {
-                State::SynSent if from_initiator => is_initial_syn(&tcp),
-                State::SynSent => {
-                    if tcp.syn() && tcp.ack() && !tcp.fin() {
-                        *state = State::SynReceived;
-                        true
-                    } else {
-                        false
-                    }
-                }
-                State::SynReceived if from_initiator => {
-                    if is_initial_syn(&tcp) {
-                        true
-                    } else if tcp.ack() && !tcp.syn() {
-                        *state = if tcp.fin() {
-                            State::Closing {
-                                host_fin: matches!(direction, Direction::FromHost),
-                                vm_fin: matches!(direction, Direction::FromVm),
-                            }
-                        } else {
-                            State::Established
-                        };
-                        true
-                    } else {
-                        false
-                    }
-                }
-                State::SynReceived => tcp.syn() && tcp.ack() && !tcp.fin(),
-                State::Established | State::Closing { .. }
-                    if is_initial_syn(&tcp) && !from_initiator =>
-                {
-                    false
-                }
-                State::Established | State::Closing { .. } if is_initial_syn(&tcp) => {
-                    *state = State::SynSent;
-                    true
-                }
-                State::Established if tcp.fin() => {
-                    *state = State::Closing {
-                        host_fin: matches!(direction, Direction::FromHost),
-                        vm_fin: matches!(direction, Direction::FromVm),
-                    };
-                    true
-                }
-                State::Closing { .. } if tcp.fin() => {
-                    if let State::Closing { host_fin, vm_fin } = state {
-                        match direction {
-                            Direction::FromHost => *host_fin = true,
-                            Direction::FromVm => *vm_fin = true,
-                        }
-                    }
-                    true
-                }
-                _ => true,
-            };
-
-            if allowed {
-                flow.last_seen = now;
-            }
-            return if allowed {
-                ConntrackResult::Allowed
-            } else {
-                ConntrackResult::Denied
-            };
-        }
-
-        if !is_initial_syn(&tcp) {
-            return ConntrackResult::Denied;
-        }
-
-        ConntrackResult::New(PendingFlow {
-            key,
-            flow: Flow {
-                initiator: initiator(direction),
-                state: FlowState::Tcp(State::SynSent),
-                last_seen: now,
-            },
-        })
-    }
-}
-
-fn is_initial_syn(tcp: &TcpPacket<&[u8]>) -> bool {
-    tcp.syn() && !tcp.ack() && !tcp.fin() && !tcp.rst()
-}
-
-#[cfg(test)]
-mod tests {
-    use super::super::test_support::{
-        HOST, TcpFlags, VM, inspect_from_host, inspect_from_vm, tcp_packet,
-    };
-    use super::super::{Conntrack, ConntrackResult};
-
-    #[test]
-    fn tcp_flows_are_oriented() {
-        let mut tracker = Conntrack::new();
-
-        let vm_syn = tcp_packet(VM, 22, HOST, 49152, TcpFlags::SYN);
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &vm_syn),
-            ConntrackResult::New(_)
-        ));
-
-        let host_syn = tcp_packet(HOST, 49152, VM, 22, TcpFlags::SYN);
-        let ConntrackResult::New(pending) = inspect_from_host(&mut tracker, &host_syn) else {
-            panic!("expected a new flow");
-        };
-        assert!(tracker.commit(pending));
-
-        let premature_vm_ack = tcp_packet(VM, 22, HOST, 49152, TcpFlags::ACK);
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &premature_vm_ack),
-            ConntrackResult::Denied
-        ));
-
-        let wrong_vm_reply = tcp_packet(VM, 22, HOST, 49153, TcpFlags::SYN_ACK);
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &wrong_vm_reply),
-            ConntrackResult::Denied
-        ));
-
-        let vm_syn_ack = tcp_packet(VM, 22, HOST, 49152, TcpFlags::SYN_ACK);
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &vm_syn_ack),
-            ConntrackResult::Allowed
-        ));
-
-        let premature_vm_ack = tcp_packet(VM, 22, HOST, 49152, TcpFlags::ACK);
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &premature_vm_ack),
-            ConntrackResult::Denied
-        ));
-
-        let host_ack = tcp_packet(HOST, 49152, VM, 22, TcpFlags::ACK);
-        assert!(matches!(
-            inspect_from_host(&mut tracker, &host_ack),
-            ConntrackResult::Allowed
-        ));
-
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &premature_vm_ack),
-            ConntrackResult::Allowed
-        ));
-    }
-
-    #[test]
-    fn vm_can_initiate_tcp() {
-        let mut tracker = Conntrack::new();
-        let vm_syn = tcp_packet(VM, 49152, HOST, 22, TcpFlags::SYN);
-        let host_syn_ack = tcp_packet(HOST, 22, VM, 49152, TcpFlags::SYN_ACK);
-        let vm_ack = tcp_packet(VM, 49152, HOST, 22, TcpFlags::ACK);
-
-        let ConntrackResult::New(pending) = inspect_from_vm(&mut tracker, &vm_syn) else {
-            panic!("expected a new flow");
-        };
-        assert!(tracker.commit(pending));
-        assert!(matches!(
-            inspect_from_host(&mut tracker, &host_syn_ack),
-            ConntrackResult::Allowed
-        ));
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &vm_ack),
-            ConntrackResult::Allowed
-        ));
-    }
-
-    #[test]
-    fn tcp_rst_removes_permission() {
-        let mut tracker = Conntrack::new();
-        let host_syn = tcp_packet(HOST, 49152, VM, 22, TcpFlags::SYN);
-        let vm_rst = tcp_packet(VM, 22, HOST, 49152, TcpFlags::RST);
-        let vm_ack = tcp_packet(VM, 22, HOST, 49152, TcpFlags::ACK);
-
-        let ConntrackResult::New(pending) = inspect_from_host(&mut tracker, &host_syn) else {
-            panic!("expected a new flow");
-        };
-        assert!(tracker.commit(pending));
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &vm_rst),
-            ConntrackResult::Allowed
-        ));
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &vm_ack),
-            ConntrackResult::Denied
-        ));
-    }
-}

lib/proxy/conntrack_udp.rs

@@ -1,131 +0,0 @@
-use super::{
-    Conntrack, ConntrackResult, Direction, Flow, FlowKey, FlowState, Initiator, PendingFlow,
-    initiator, oriented_transport_key,
-};
-use coarsetime::{Duration, Instant};
-use smoltcp::wire::{Ipv4Packet, UdpPacket};
-
-const UNREPLIED_TIMEOUT: Duration = Duration::from_secs(30);
-const REPLIED_TIMEOUT: Duration = Duration::from_secs(3 * 60);
-
-#[derive(Clone, Copy, Debug)]
-pub(super) struct State {
-    replied: bool,
-}
-
-impl State {
-    pub(super) fn timeout(self) -> Duration {
-        if self.replied {
-            REPLIED_TIMEOUT
-        } else {
-            UNREPLIED_TIMEOUT
-        }
-    }
-}
-
-impl Conntrack {
-    pub(super) fn inspect_udp(
-        &mut self,
-        packet: &Ipv4Packet<&[u8]>,
-        direction: Direction,
-        now: Instant,
-    ) -> ConntrackResult {
-        let Ok(udp) = UdpPacket::new_checked(packet.payload()) else {
-            return ConntrackResult::Denied;
-        };
-        if udp.src_port() == 0 || udp.dst_port() == 0 {
-            return ConntrackResult::Denied;
-        }
-
-        let key = oriented_transport_key(
-            packet,
-            direction,
-            |host_addr, host_port, vm_addr, vm_port| FlowKey::Udp {
-                host_addr,
-                host_port,
-                vm_addr,
-                vm_port,
-            },
-            udp.src_port(),
-            udp.dst_port(),
-        );
-
-        if let Some(flow) = self.flows.get_mut(&key) {
-            let FlowState::Udp(state) = &mut flow.state else {
-                return ConntrackResult::Denied;
-            };
-
-            let is_reply = matches!(
-                (flow.initiator, direction),
-                (Initiator::Host, Direction::FromVm) | (Initiator::Vm, Direction::FromHost)
-            );
-            state.replied |= is_reply;
-            flow.last_seen = now;
-            return ConntrackResult::Allowed;
-        }
-
-        ConntrackResult::New(PendingFlow {
-            key,
-            flow: Flow {
-                initiator: initiator(direction),
-                state: FlowState::Udp(State { replied: false }),
-                last_seen: now,
-            },
-        })
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::super::test_support::{HOST, VM, inspect_from_host, inspect_from_vm, udp_packet};
-    use super::super::{Conntrack, ConntrackResult};
-
-    #[test]
-    fn udp_reply_requires_an_exact_host_request() {
-        let mut tracker = Conntrack::new();
-
-        let vm_datagram = udp_packet(VM, 5353, HOST, 50000);
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &vm_datagram),
-            ConntrackResult::New(_)
-        ));
-
-        let host_datagram = udp_packet(HOST, 50000, VM, 5353);
-        let ConntrackResult::New(pending) = inspect_from_host(&mut tracker, &host_datagram) else {
-            panic!("expected a new flow");
-        };
-        assert!(tracker.commit(pending));
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &vm_datagram),
-            ConntrackResult::Allowed
-        ));
-
-        let wrong_vm_datagram = udp_packet(VM, 5353, HOST, 50001);
-        assert!(matches!(
-            inspect_from_vm(&mut tracker, &wrong_vm_datagram),
-            ConntrackResult::New(_)
-        ));
-    }
-
-    #[test]
-    fn explicitly_allowed_vm_udp_gets_only_its_reply() {
-        let mut tracker = Conntrack::new();
-        let vm_dns = udp_packet(VM, 53000, HOST, 53);
-        let host_dns = udp_packet(HOST, 53, VM, 53000);
-
-        let ConntrackResult::New(pending) = inspect_from_vm(&mut tracker, &vm_dns) else {
-            panic!("expected a new flow");
-        };
-        assert!(tracker.commit(pending));
-        assert!(matches!(
-            inspect_from_host(&mut tracker, &host_dns),
-            ConntrackResult::Allowed
-        ));
-
-        let unsolicited_host_udp = udp_packet(HOST, 53, VM, 53001);
-        assert!(matches!(
-            inspect_from_host(&mut tracker, &unsolicited_host_udp),
-            ConntrackResult::New(_)
-        ));
-    }
-}

lib/proxy/control.rs

@@ -1,5 +1,7 @@
-use super::{Rule, Rules, build_rules, rule_count};
+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>,
@@ -60,7 +63,7 @@ impl Policy {
 
         let allow = parse_rules(allow)?;
         let block = parse_rules(block)?;
-        let rules = build_rules(self.gateway_ip, &allow, &block);
+        let rules = Rules::new(self.gateway_ip, &allow, &block);
 
         Ok(PolicyUpdate {
             rules,
@@ -71,14 +74,13 @@ impl Policy {
 
     fn apply(&mut self, update: PolicyUpdate) -> Option<Rules> {
         // Build and validate everything before updating any active state. The packet filter
-        // observes either the old PrefixMap or the complete new one.
-        if self.allow == update.allow && self.block == update.block {
-            return None;
-        }
+        // observes either the old rule set or the complete new one.
+        let changed = self.allow != update.allow || self.block != update.block;
 
         self.allow = update.allow;
         self.block = update.block;
-        Some(update.rules)
+
+        changed.then_some(update.rules)
     }
 
     fn result(&self, rule_count: usize) -> Value {
@@ -88,7 +90,7 @@ impl Policy {
 
 impl PolicyUpdate {
     fn result(&self) -> Value {
-        policy_result(&self.allow, &self.block, rule_count(&self.rules))
+        policy_result(&self.allow, &self.block, self.rules.len())
     }
 }
 
@@ -116,7 +118,7 @@ fn parse_rules(rules: Vec<String>) -> std::result::Result<Vec<Rule>, ErrorObject
     Ok(normalize_rules(parsed))
 }
 
-fn normalize_rules(mut rules: Vec<Rule>) -> Vec<Rule> {
+pub(super) fn normalize_rules(mut rules: Vec<Rule>) -> Vec<Rule> {
     rules.iter_mut().for_each(|rule| *rule = rule.normalized());
     rules.sort_by_key(ToString::to_string);
     rules.dedup();
@@ -125,13 +127,7 @@ 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 {
@@ -149,200 +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, rule_count(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(())
     }
 }
 
@@ -353,11 +267,7 @@ struct SetParams {
     block: Vec<String>,
 }
 
-fn handle_request(
-    policy: &Policy,
-    rule_count: usize,
-    line: &[u8],
-) -> (Value, Option<PolicyUpdate>) {
+fn handle_request(policy: &Policy, rules: &Rules, line: &[u8]) -> (Value, Option<PolicyUpdate>) {
     let value = match serde_json::from_slice::<Value>(line) {
         Ok(value) => value,
         Err(_) => {
@@ -394,7 +304,7 @@ fn handle_request(
                     "softnet.policy.get does not accept parameters",
                 ))
             } else {
-                Ok(policy.result(rule_count))
+                Ok(policy.result(rules.len()))
             }
         }
         "softnet.policy.set" => {
@@ -539,21 +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, build_rules, handle_request, rule_count,
+        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::{Action, Rule, Rules};
-    use ipnet::Ipv4Net;
-    use prefix_trie::PrefixMap;
+    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::str::FromStr;
     use std::time::Duration;
+    use tokio_util::codec::{Decoder, Encoder, LinesCodec};
 
     struct TestPolicy {
         state: Policy,
@@ -562,7 +472,7 @@ mod tests {
 
     impl TestPolicy {
         fn result(&self) -> Value {
-            self.state.result(rule_count(&self.rules))
+            self.state.result(self.rules.len())
         }
     }
 
@@ -584,7 +494,7 @@ mod tests {
         let block = rules(block);
 
         TestPolicy {
-            rules: build_rules(gateway_ip, &allow, &block),
+            rules: Rules::new(gateway_ip, &allow, &block),
             state: Policy::new(gateway_ip, allow, block),
         }
     }
@@ -603,7 +513,7 @@ mod tests {
     }
 
     fn raw_request(policy: &mut TestPolicy, line: &[u8]) -> Value {
-        let (response, update) = handle_request(&policy.state, rule_count(&policy.rules), line);
+        let (response, update) = handle_request(&policy.state, &policy.rules, line);
         if let Some(update) = update
             && let Some(rules) = policy.state.apply(update)
         {
@@ -652,14 +562,16 @@ mod tests {
         assert_eq!(response["result"]["ruleCount"], 2);
 
         assert_eq!(
-            policy.rules.get(&Ipv4Net::from_str("10.0.0.0/8").unwrap()),
-            Some(&vec![(Action::Block, "10.0.0.0/8".parse().unwrap())])
+            policy
+                .rules
+                .policy_decision(Ipv4Address::new(10, 0, 0, 1), Direction::Out),
+            Some(PolicyDecision::Block)
         );
         assert_eq!(
             policy
                 .rules
-                .get(&Ipv4Net::from_str("192.168.64.1/32").unwrap()),
-            Some(&vec![(Action::Block, "192.168.64.1/32".parse().unwrap())])
+                .policy_decision(Ipv4Address::new(192, 168, 64, 1), Direction::Out),
+            Some(PolicyDecision::Block)
         );
     }
 
@@ -712,12 +624,14 @@ mod tests {
         );
         assert_eq!(response["result"]["block"], json!(["in 10.0.0.0/8"]));
         assert_eq!(response["result"]["ruleCount"], 2);
+        let target = Ipv4Address::new(10, 1, 2, 3);
         assert_eq!(
-            policy.rules.get(&Ipv4Net::from_str("10.0.0.0/8").unwrap()),
-            Some(&vec![
-                (Action::Block, "in 10.0.0.0/8".parse::<Rule>().unwrap()),
-                (Action::Allow, "out 10.0.0.0/8".parse::<Rule>().unwrap()),
-            ])
+            policy.rules.policy_decision(target, Direction::In),
+            Some(PolicyDecision::Block)
+        );
+        assert_eq!(
+            policy.rules.policy_decision(target, Direction::Out),
+            Some(PolicyDecision::AllowStateful)
         );
     }
 
@@ -872,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 = PrefixMap::new();
-
+        let mut rules = Rules::default();
         client
             .write_all(
                 concat!(
@@ -890,93 +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(rule_count(&rules));
-        drop(client);
-        assert!(!control.service(&mut rules).unwrap());
-        assert_eq!(control.policy.result(rule_count(&rules)), 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 repeated_normalized_policy_is_not_reported_as_changed() {
+    #[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 = PrefixMap::new();
-
-        client
-            .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"in 10.1.2.3/8\"],\"block\":[]}}\n")
+        let mut rules = Rules::default();
+        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());
-
+            assert_eq!(control.service(&mut rules).await.unwrap(), Some(changed));
+        }
         client
-            .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"in 10.0.0.0/8\"],\"block\":[]}}\n")
+            .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 = PrefixMap::new();
-
+        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 = PrefixMap::new();
-
-        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 rules = Rules::default();
+        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<_>>();
@@ -985,146 +903,210 @@ 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 = PrefixMap::new();
-        let before = control.policy.result(rule_count(&rules));
-
+        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_eq!(control.policy.result(rule_count(&rules)), before);
-        assert!(control.output.is_empty());
+        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.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 = PrefixMap::new();
+        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 = PrefixMap::new();
-        let before = control.policy.result(rule_count(&rules));
-
-        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();
+        let mut rules = Rules::default();
+        let before = control.policy.result(rules.len());
+        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()
                 .contains("control response queue exceeded")
         );
-        assert_eq!(control.policy.result(rule_count(&rules)), before);
+        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 = PrefixMap::new();
-        let before = control.policy.result(rule_count(&rules));
-
-        control.output = vec![b'x'; MAX_REQUEST_BYTES];
-        assert!(control.flush().unwrap());
-        assert!(!control.output.is_empty());
-
+        let mut rules = Rules::default();
+        let before = control.policy.result(rules.len());
+        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_eq!(control.policy.result(rule_count(&rules)), before);
-        assert!(control.input.is_empty());
+        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.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();
@@ -1133,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/icmp.rs

@@ -0,0 +1,173 @@
+use super::{FlowDirection, FlowKey, FlowMatch, FlowTable};
+use coarsetime::{Duration, Instant};
+use smoltcp::wire::{Icmpv4Message, Icmpv4Packet, Ipv4Address, Ipv4Packet};
+
+const ECHO_TIMEOUT: Duration = Duration::from_secs(30);
+
+impl FlowTable {
+    pub(super) fn inspect_icmp(
+        &mut self,
+        ipv4_pkt: &Ipv4Packet<&[u8]>,
+        direction: FlowDirection,
+        now: Instant,
+    ) -> FlowMatch {
+        let Ok(icmp) = Icmpv4Packet::new_checked(ipv4_pkt.payload()) else {
+            return FlowMatch::Denied;
+        };
+        if !icmp.verify_checksum() {
+            return FlowMatch::Denied;
+        }
+
+        match (icmp.msg_type(), icmp.msg_code()) {
+            (Icmpv4Message::EchoRequest, 0) => {
+                self.inspect_echo(ipv4_pkt, direction, icmp.echo_ident(), true, now)
+            }
+            (Icmpv4Message::EchoReply, 0) => {
+                self.inspect_echo(ipv4_pkt, direction, icmp.echo_ident(), false, now)
+            }
+            // Other ICMP messages follow normal source policy: treating errors that quote
+            // tracked tuples as RELATED requires validation beyond this exact-tuple table
+            //
+            // Potential degradation of PMTU discovery and traceroute is an accepted tradeoff.
+            _ => FlowMatch::Untracked,
+        }
+    }
+
+    fn inspect_echo(
+        &mut self,
+        ipv4_pkt: &Ipv4Packet<&[u8]>,
+        direction: FlowDirection,
+        ident: u16,
+        is_request: bool,
+        now: Instant,
+    ) -> FlowMatch {
+        // We need to preserve the request direction so opposite-direction echo flows remain distinct
+        let initiating_direction = if is_request {
+            direction
+        } else {
+            direction.opposite()
+        };
+
+        let key = FlowKey::icmp_echo(
+            ipv4_pkt.src_addr(),
+            ipv4_pkt.dst_addr(),
+            direction,
+            initiating_direction,
+            ident,
+        );
+
+        if let Some(matched) = self.match_existing_flow(key, direction, now, ECHO_TIMEOUT) {
+            return matched;
+        }
+
+        if is_request {
+            FlowMatch::candidate(key, initiating_direction, now, ECHO_TIMEOUT)
+        } else {
+            // Only replies matching an admitted request are tracked
+            FlowMatch::Untracked
+        }
+    }
+}
+
+impl FlowKey {
+    fn icmp_echo(
+        src_addr: Ipv4Address,
+        dst_addr: Ipv4Address,
+        direction: FlowDirection,
+        initiating_direction: FlowDirection,
+        ident: u16,
+    ) -> Self {
+        let (host_addr, vm_addr) = direction.host_vm_pair(src_addr, dst_addr);
+        Self::IcmpEcho {
+            host_addr,
+            vm_addr,
+            ident,
+            initiating_direction,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::test_support::{HOST, VM, inspect_from_host, inspect_from_vm, ipv4_packet};
+    use super::super::{FlowMatch, FlowTable};
+    use smoltcp::wire::{Icmpv4Message, Icmpv4Packet, IpProtocol, Ipv4Address};
+
+    #[test]
+    fn echo_request_and_reply_are_tracked_in_both_directions() {
+        let mut tracker = FlowTable::new();
+        let vm_request = echo_packet(VM, HOST, Icmpv4Message::EchoRequest, 7, 1);
+
+        let FlowMatch::Candidate(pending) = inspect_from_vm(&mut tracker, &vm_request) else {
+            panic!("expected a new flow");
+        };
+        assert!(tracker.commit(pending));
+
+        let host_reply = echo_packet(HOST, VM, Icmpv4Message::EchoReply, 7, 1);
+        assert!(matches!(
+            inspect_from_host(&mut tracker, &host_reply),
+            FlowMatch::Allowed
+        ));
+
+        let wrong_ident = echo_packet(HOST, VM, Icmpv4Message::EchoReply, 8, 1);
+        assert!(matches!(
+            inspect_from_host(&mut tracker, &wrong_ident),
+            FlowMatch::Untracked
+        ));
+
+        // The opposite direction is a distinct flow even with the same identifier.
+        let host_request = echo_packet(HOST, VM, Icmpv4Message::EchoRequest, 7, 1);
+        let FlowMatch::Candidate(pending) = inspect_from_host(&mut tracker, &host_request) else {
+            panic!("expected a new flow");
+        };
+        assert!(tracker.commit(pending));
+
+        let vm_reply = echo_packet(VM, HOST, Icmpv4Message::EchoReply, 7, 1);
+        assert!(matches!(
+            inspect_from_vm(&mut tracker, &vm_reply),
+            FlowMatch::Allowed
+        ));
+    }
+
+    #[test]
+    fn invalid_echo_is_denied_and_unsolicited_echo_is_untracked() {
+        let mut tracker = FlowTable::new();
+        let unsolicited = echo_packet(HOST, VM, Icmpv4Message::EchoReply, 7, 1);
+        assert!(matches!(
+            inspect_from_host(&mut tracker, &unsolicited),
+            FlowMatch::Untracked
+        ));
+
+        let mut wrong_code = echo_packet(HOST, VM, Icmpv4Message::EchoRequest, 7, 1);
+        wrong_code[21] = 1;
+        Icmpv4Packet::new_unchecked(&mut wrong_code[20..]).fill_checksum();
+        assert!(matches!(
+            inspect_from_host(&mut tracker, &wrong_code),
+            FlowMatch::Untracked
+        ));
+
+        let mut bad_checksum = echo_packet(HOST, VM, Icmpv4Message::EchoRequest, 7, 1);
+        bad_checksum[27] ^= 1;
+        assert!(matches!(
+            inspect_from_host(&mut tracker, &bad_checksum),
+            FlowMatch::Denied
+        ));
+    }
+
+    fn echo_packet(
+        src_addr: Ipv4Address,
+        dst_addr: Ipv4Address,
+        message: Icmpv4Message,
+        ident: u16,
+        sequence: u16,
+    ) -> Vec<u8> {
+        let mut bytes = ipv4_packet(src_addr, dst_addr, IpProtocol::Icmp, 8);
+        let mut icmp = Icmpv4Packet::new_unchecked(&mut bytes[20..]);
+        icmp.set_msg_type(message);
+        icmp.set_msg_code(0);
+        icmp.set_echo_ident(ident);
+        icmp.set_echo_seq_no(sequence);
+        icmp.fill_checksum();
+        bytes
+    }
+}

lib/proxy/flows/mod.rs

@@ -0,0 +1,390 @@
+mod icmp;
+mod tcp;
+mod udp;
+
+use coarsetime::{Duration, Instant};
+use smoltcp::wire::{IpProtocol, Ipv4Address, Ipv4Packet};
+use std::collections::HashMap;
+
+const MAX_FLOWS: usize = 32_768;
+const SWEEP_INTERVAL: Duration = Duration::from_secs(1);
+
+/// A bounded cache of exact-tuple permissions for return traffic.
+///
+/// The table establishes only the direction in which a flow was authorized.
+/// This is deliberately not a TCP state machine. Endpoint transport stacks remain
+/// responsible for TCP handshakes, teardown, sequence numbers, and receive windows,
+/// as well as ICMP echo sequences.
+///
+/// Capacity is intentionally enforced with one fail-closed limit per VM.
+/// Exhaustion may deny further networking for that VM; this availability
+/// tradeoff is accepted to avoid fairness quotas, admission scans, and eviction.
+#[derive(Debug)]
+pub(crate) struct FlowTable {
+    flows: HashMap<FlowKey, Flow>,
+    next_sweep: Instant,
+}
+
+/// A proposed flow-table entry awaiting policy authorization.
+pub(crate) struct PendingFlow {
+    key: FlowKey,
+    flow: Flow,
+}
+
+/// Metadata stored for an admitted flow.
+#[derive(Clone, Copy, Debug)]
+struct Flow {
+    initiating_direction: FlowDirection,
+    expires_at: Instant,
+}
+
+/// The result of classifying a packet against the flow table.
+pub(crate) enum FlowMatch {
+    /// The packet is exact-tuple return traffic for an admitted flow.
+    Allowed,
+
+    /// Policy must authorize this initiator-side packet before committing it.
+    Candidate(PendingFlow),
+
+    /// The packet is malformed or violates a transport invariant that we deliberately enforce.
+    Denied,
+
+    /// The flow table has no stateful interpretation for this packet.
+    Untracked,
+}
+
+impl FlowMatch {
+    fn candidate(
+        key: FlowKey,
+        initiating_direction: FlowDirection,
+        now: Instant,
+        timeout: Duration,
+    ) -> Self {
+        Self::Candidate(PendingFlow {
+            key,
+            flow: Flow {
+                initiating_direction,
+                expires_at: now + timeout,
+            },
+        })
+    }
+}
+
+/// The canonical identity used to index a flow-table entry.
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+enum FlowKey {
+    IcmpEcho {
+        host_addr: Ipv4Address,
+        vm_addr: Ipv4Address,
+        ident: u16,
+        initiating_direction: FlowDirection,
+    },
+    Tcp {
+        host_addr: Ipv4Address,
+        host_port: u16,
+        vm_addr: Ipv4Address,
+        vm_port: u16,
+    },
+    Udp {
+        host_addr: Ipv4Address,
+        host_port: u16,
+        vm_addr: Ipv4Address,
+        vm_port: u16,
+    },
+}
+
+/// A packet direction across the host–VM boundary.
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub(super) enum FlowDirection {
+    FromHost,
+    FromVm,
+}
+
+impl FlowTable {
+    pub(crate) fn new() -> Self {
+        Self {
+            flows: HashMap::new(),
+            next_sweep: Instant::recent() + SWEEP_INTERVAL,
+        }
+    }
+
+    pub(super) fn inspect(
+        &mut self,
+        ipv4_pkt: &Ipv4Packet<&[u8]>,
+        direction: FlowDirection,
+    ) -> FlowMatch {
+        self.inspect_at(ipv4_pkt, direction, Instant::recent())
+    }
+
+    fn inspect_at(
+        &mut self,
+        ipv4_pkt: &Ipv4Packet<&[u8]>,
+        direction: FlowDirection,
+        now: Instant,
+    ) -> FlowMatch {
+        // Perform lazy garbage collection of expired flow entries
+        self.sweep_if_due(now);
+
+        // Later fragments do not contain enough transport information to bind
+        // them to an exact flow, so packet policy must decide
+        if ipv4_pkt.more_frags() || ipv4_pkt.frag_offset() != 0 {
+            return FlowMatch::Untracked;
+        }
+
+        match ipv4_pkt.next_header() {
+            IpProtocol::Icmp => self.inspect_icmp(ipv4_pkt, direction, now),
+            IpProtocol::Tcp => self.inspect_tcp(ipv4_pkt, direction, now),
+            IpProtocol::Udp => self.inspect_udp(ipv4_pkt, direction, now),
+            _ => FlowMatch::Untracked,
+        }
+    }
+
+    pub(crate) fn commit(&mut self, pending: PendingFlow) -> bool {
+        let PendingFlow { key, flow } = pending;
+
+        // Reject new entries when full while allowing existing entries to be updated
+        if !self.flows.contains_key(&key) && self.flows.len() >= MAX_FLOWS {
+            return false;
+        }
+
+        self.flows.insert(key, flow);
+
+        true
+    }
+
+    fn sweep_if_due(&mut self, now: Instant) {
+        if now < self.next_sweep {
+            return;
+        }
+
+        self.flows.retain(|_, flow| now < flow.expires_at);
+
+        self.next_sweep = now + SWEEP_INTERVAL;
+    }
+
+    fn match_existing_flow(
+        &self,
+        key: FlowKey,
+        direction: FlowDirection,
+        now: Instant,
+        timeout: Duration,
+    ) -> Option<FlowMatch> {
+        let existing = self.flows.get(&key).copied()?;
+
+        // Treat expired entries as missing even before the next sweep
+        if now >= existing.expires_at {
+            return None;
+        }
+
+        if direction == existing.initiating_direction {
+            Some(FlowMatch::candidate(
+                key,
+                existing.initiating_direction,
+                now,
+                timeout,
+            ))
+        } else {
+            Some(FlowMatch::Allowed)
+        }
+    }
+}
+
+impl FlowDirection {
+    fn host_vm_pair<T>(self, src: T, dst: T) -> (T, T) {
+        match self {
+            Self::FromHost => (src, dst),
+            Self::FromVm => (dst, src),
+        }
+    }
+
+    fn opposite(self) -> Self {
+        match self {
+            Self::FromHost => Self::FromVm,
+            Self::FromVm => Self::FromHost,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::test_support::{HOST, VM, inspect_from_host, ipv4_packet, udp_packet};
+    use super::{
+        Duration, FlowDirection, FlowMatch, FlowTable, Instant, MAX_FLOWS, SWEEP_INTERVAL,
+    };
+    use smoltcp::wire::{IpProtocol, Ipv4Packet};
+
+    #[test]
+    fn all_ipv4_fragments_are_untracked() {
+        for (more_fragments, offset) in [(true, 0), (false, 8)] {
+            let mut bytes = udp_packet(HOST, 50_000, VM, 53);
+            let mut packet = Ipv4Packet::new_unchecked(bytes.as_mut_slice());
+            packet.set_frag_offset(offset);
+            packet.set_more_frags(more_fragments);
+
+            let mut tracker = FlowTable::new();
+            assert!(matches!(
+                inspect_from_host(&mut tracker, &bytes),
+                FlowMatch::Untracked
+            ));
+        }
+    }
+
+    #[test]
+    fn expired_tuple_is_not_matched_before_the_next_sweep() {
+        let mut tracker = FlowTable::new();
+        let datagram = udp_packet(HOST, 50_000, VM, 53);
+        let FlowMatch::Candidate(pending) = inspect_from_host(&mut tracker, &datagram) else {
+            panic!("expected a candidate");
+        };
+        assert!(tracker.commit(pending));
+
+        let now = Instant::recent();
+        let flow = tracker.flows.values_mut().next().unwrap();
+        flow.expires_at = now - Duration::from_secs(1);
+        tracker.next_sweep = now + SWEEP_INTERVAL;
+
+        let reply = udp_packet(VM, 53, HOST, 50_000);
+        assert!(matches!(
+            tracker.inspect_at(
+                &Ipv4Packet::new_checked(reply.as_slice()).unwrap(),
+                FlowDirection::FromVm,
+                now,
+            ),
+            FlowMatch::Candidate(_)
+        ));
+    }
+
+    #[test]
+    fn global_limit_rejects_new_tuple_but_allows_replacement() {
+        let mut tracker = FlowTable::new();
+        for port in 10_000..10_000 + MAX_FLOWS as u16 {
+            let datagram = udp_packet(HOST, port, VM, 53);
+            let FlowMatch::Candidate(pending) = inspect_from_host(&mut tracker, &datagram) else {
+                panic!("expected a candidate");
+            };
+            assert!(tracker.commit(pending));
+        }
+
+        let over_limit = udp_packet(HOST, 50_000, VM, 53);
+        let FlowMatch::Candidate(pending) = inspect_from_host(&mut tracker, &over_limit) else {
+            panic!("expected a candidate");
+        };
+        assert!(!tracker.commit(pending));
+
+        let replacement = udp_packet(HOST, 10_000, VM, 53);
+        let FlowMatch::Candidate(pending) = inspect_from_host(&mut tracker, &replacement) else {
+            panic!("expected a candidate");
+        };
+        assert!(tracker.commit(pending));
+    }
+
+    #[test]
+    fn unsupported_protocol_is_untracked() {
+        let bytes = ipv4_packet(HOST, VM, IpProtocol::Unknown(253), 0);
+        let mut tracker = FlowTable::new();
+        assert!(matches!(
+            inspect_from_host(&mut tracker, &bytes),
+            FlowMatch::Untracked
+        ));
+    }
+}
+
+#[cfg(test)]
+mod test_support {
+    use super::{FlowDirection, FlowMatch, FlowTable};
+    use smoltcp::wire::{IpProtocol, Ipv4Address, Ipv4Packet, TcpPacket, UdpPacket};
+
+    pub(super) const HOST: Ipv4Address = Ipv4Address::new(192, 168, 64, 1);
+    pub(super) const VM: Ipv4Address = Ipv4Address::new(192, 168, 64, 2);
+
+    pub(super) fn inspect_from_host(tracker: &mut FlowTable, bytes: &[u8]) -> FlowMatch {
+        let packet = Ipv4Packet::new_checked(bytes).unwrap();
+        tracker.inspect(&packet, FlowDirection::FromHost)
+    }
+
+    pub(super) fn inspect_from_vm(tracker: &mut FlowTable, bytes: &[u8]) -> FlowMatch {
+        let packet = Ipv4Packet::new_checked(bytes).unwrap();
+        tracker.inspect(&packet, FlowDirection::FromVm)
+    }
+
+    #[derive(Clone, Copy)]
+    pub(super) struct TcpFlags {
+        syn: bool,
+        ack: bool,
+        rst: bool,
+        fin: bool,
+    }
+
+    impl TcpFlags {
+        pub(super) const SYN: Self = Self {
+            syn: true,
+            ack: false,
+            rst: false,
+            fin: false,
+        };
+        pub(super) const SYN_ACK: Self = Self {
+            syn: true,
+            ack: true,
+            rst: false,
+            fin: false,
+        };
+        pub(super) const ACK: Self = Self {
+            syn: false,
+            ack: true,
+            rst: false,
+            fin: false,
+        };
+    }
+
+    pub(super) fn tcp_packet(
+        src_addr: Ipv4Address,
+        src_port: u16,
+        dst_addr: Ipv4Address,
+        dst_port: u16,
+        flags: TcpFlags,
+    ) -> Vec<u8> {
+        let mut bytes = ipv4_packet(src_addr, dst_addr, IpProtocol::Tcp, 20);
+        let mut tcp = TcpPacket::new_unchecked(&mut bytes[20..]);
+        tcp.set_src_port(src_port);
+        tcp.set_dst_port(dst_port);
+        tcp.set_header_len(20);
+        tcp.set_syn(flags.syn);
+        tcp.set_ack(flags.ack);
+        tcp.set_rst(flags.rst);
+        tcp.set_fin(flags.fin);
+        tcp.set_window_len(u16::MAX);
+        bytes
+    }
+
+    pub(super) fn udp_packet(
+        src_addr: Ipv4Address,
+        src_port: u16,
+        dst_addr: Ipv4Address,
+        dst_port: u16,
+    ) -> Vec<u8> {
+        let mut bytes = ipv4_packet(src_addr, dst_addr, IpProtocol::Udp, 8);
+        let mut udp = UdpPacket::new_unchecked(&mut bytes[20..]);
+        udp.set_src_port(src_port);
+        udp.set_dst_port(dst_port);
+        udp.set_len(8);
+        bytes
+    }
+
+    pub(super) fn ipv4_packet(
+        src_addr: Ipv4Address,
+        dst_addr: Ipv4Address,
+        protocol: IpProtocol,
+        payload_len: usize,
+    ) -> Vec<u8> {
+        let mut bytes = vec![0; 20 + payload_len];
+        let total_len = bytes.len() as u16;
+        let mut ipv4 = Ipv4Packet::new_unchecked(bytes.as_mut_slice());
+        ipv4.set_version(4);
+        ipv4.set_header_len(20);
+        ipv4.set_total_len(total_len);
+        ipv4.set_next_header(protocol);
+        ipv4.set_src_addr(src_addr);
+        ipv4.set_dst_addr(dst_addr);
+        bytes
+    }
+}

lib/proxy/flows/tcp.rs

@@ -0,0 +1,156 @@
+use super::{FlowDirection, FlowKey, FlowMatch, FlowTable};
+use coarsetime::{Duration, Instant};
+use smoltcp::wire::{Ipv4Address, Ipv4Packet, TcpPacket};
+
+const SYN_TIMEOUT: Duration = Duration::from_secs(60);
+const TCP_TIMEOUT: Duration = Duration::from_secs(5 * 60);
+
+impl FlowTable {
+    pub(super) fn inspect_tcp(
+        &mut self,
+        ipv4_pkt: &Ipv4Packet<&[u8]>,
+        direction: FlowDirection,
+        now: Instant,
+    ) -> FlowMatch {
+        let Ok(tcp) = TcpPacket::new_checked(ipv4_pkt.payload()) else {
+            return FlowMatch::Denied;
+        };
+        if tcp.src_port() == 0 || tcp.dst_port() == 0 {
+            return FlowMatch::Denied;
+        }
+
+        let key = FlowKey::tcp(
+            direction,
+            (ipv4_pkt.src_addr(), tcp.src_port()),
+            (ipv4_pkt.dst_addr(), tcp.dst_port()),
+        );
+
+        // A bare SYN may be either a retransmission or a new connection reusing
+        // the tuple in either direction. Always return it to policy, and do not
+        // replace an existing permission until the candidate is committed.
+        let is_initial_syn = tcp.syn() && !tcp.ack() && !tcp.fin() && !tcp.rst();
+
+        if is_initial_syn {
+            return FlowMatch::candidate(key, direction, now, SYN_TIMEOUT);
+        }
+
+        self.match_existing_flow(key, direction, now, TCP_TIMEOUT)
+            .unwrap_or(FlowMatch::Untracked)
+    }
+}
+
+impl FlowKey {
+    fn tcp(direction: FlowDirection, src: (Ipv4Address, u16), dst: (Ipv4Address, u16)) -> Self {
+        let ((host_addr, host_port), (vm_addr, vm_port)) = direction.host_vm_pair(src, dst);
+        Self::Tcp {
+            host_addr,
+            host_port,
+            vm_addr,
+            vm_port,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::test_support::{
+        HOST, TcpFlags, VM, inspect_from_host, inspect_from_vm, tcp_packet,
+    };
+    use super::super::{FlowMatch, FlowTable};
+
+    fn admit_host_syn(tracker: &mut FlowTable) {
+        let syn = tcp_packet(HOST, 49_152, VM, 22, TcpFlags::SYN);
+        let FlowMatch::Candidate(pending) = inspect_from_host(tracker, &syn) else {
+            panic!("expected a candidate");
+        };
+        assert!(tracker.commit(pending));
+    }
+
+    #[test]
+    fn admitted_syn_creates_an_exact_reverse_permission() {
+        let mut tracker = FlowTable::new();
+        admit_host_syn(&mut tracker);
+        let original_expiry = tracker.flows.values().next().unwrap().expires_at;
+
+        let reply = tcp_packet(VM, 22, HOST, 49_152, TcpFlags::ACK);
+        assert!(matches!(
+            inspect_from_vm(&mut tracker, &reply),
+            FlowMatch::Allowed
+        ));
+        assert_eq!(
+            tracker.flows.values().next().unwrap().expires_at,
+            original_expiry,
+            "return traffic must not refresh a tuple"
+        );
+
+        let initiator_data = tcp_packet(HOST, 49_152, VM, 22, TcpFlags::ACK);
+        let FlowMatch::Candidate(pending) = inspect_from_host(&mut tracker, &initiator_data) else {
+            panic!("expected an initiator candidate");
+        };
+        assert_eq!(
+            tracker.flows.values().next().unwrap().expires_at,
+            original_expiry,
+            "inspection must not refresh a tuple before policy accepts it"
+        );
+        assert!(tracker.commit(pending));
+        assert!(tracker.flows.values().next().unwrap().expires_at > original_expiry);
+    }
+
+    #[test]
+    fn new_tuple_requires_a_clean_syn() {
+        let mut tracker = FlowTable::new();
+
+        for flags in [TcpFlags::ACK, TcpFlags::SYN_ACK] {
+            let packet = tcp_packet(HOST, 49_152, VM, 22, flags);
+            assert!(matches!(
+                inspect_from_host(&mut tracker, &packet),
+                FlowMatch::Untracked
+            ));
+        }
+
+        let syn = tcp_packet(HOST, 49_152, VM, 22, TcpFlags::SYN);
+        assert!(matches!(
+            inspect_from_host(&mut tracker, &syn),
+            FlowMatch::Candidate(_)
+        ));
+    }
+
+    #[test]
+    fn reply_requires_the_exact_tuple() {
+        let mut tracker = FlowTable::new();
+        admit_host_syn(&mut tracker);
+
+        let wrong_port = tcp_packet(VM, 22, HOST, 49_153, TcpFlags::ACK);
+        assert!(matches!(
+            inspect_from_vm(&mut tracker, &wrong_port),
+            FlowMatch::Untracked
+        ));
+    }
+
+    #[test]
+    fn reverse_bare_syn_is_a_new_policy_candidate() {
+        let mut tracker = FlowTable::new();
+        admit_host_syn(&mut tracker);
+
+        let reverse_syn = tcp_packet(VM, 22, HOST, 49_152, TcpFlags::SYN);
+        let FlowMatch::Candidate(pending) = inspect_from_vm(&mut tracker, &reverse_syn) else {
+            panic!("reverse SYN must return to policy");
+        };
+
+        // Inspection is provisional: a policy rejection leaves the admitted
+        // flow untouched.
+        let old_flow_reply = tcp_packet(VM, 22, HOST, 49_152, TcpFlags::ACK);
+        assert!(matches!(
+            inspect_from_vm(&mut tracker, &old_flow_reply),
+            FlowMatch::Allowed
+        ));
+
+        assert!(tracker.commit(pending));
+
+        let reverse_reply = tcp_packet(HOST, 49_152, VM, 22, TcpFlags::SYN_ACK);
+        assert!(matches!(
+            inspect_from_host(&mut tracker, &reverse_reply),
+            FlowMatch::Allowed
+        ));
+    }
+}

lib/proxy/flows/udp.rs

@@ -0,0 +1,92 @@
+use super::{FlowDirection, FlowKey, FlowMatch, FlowTable};
+use coarsetime::{Duration, Instant};
+use smoltcp::wire::{Ipv4Address, Ipv4Packet, UdpPacket};
+
+const UDP_TIMEOUT: Duration = Duration::from_secs(30);
+
+impl FlowTable {
+    pub(super) fn inspect_udp(
+        &mut self,
+        ipv4_pkt: &Ipv4Packet<&[u8]>,
+        direction: FlowDirection,
+        now: Instant,
+    ) -> FlowMatch {
+        let Ok(udp) = UdpPacket::new_checked(ipv4_pkt.payload()) else {
+            return FlowMatch::Denied;
+        };
+        if udp.dst_port() == 0 {
+            return FlowMatch::Denied;
+        }
+
+        // Deliberate policy, not merely a packet-format validation:
+        // RFC 8085, §5.1 says UDP senders SHOULD NOT use source port zero.
+        //
+        // We enforce this recommendation to retain source-port entropy
+        // and protection against off-path packet injection.
+        if udp.src_port() == 0 {
+            return FlowMatch::Denied;
+        }
+
+        let key = FlowKey::udp(
+            direction,
+            (ipv4_pkt.src_addr(), udp.src_port()),
+            (ipv4_pkt.dst_addr(), udp.dst_port()),
+        );
+
+        if let Some(matched) = self.match_existing_flow(key, direction, now, UDP_TIMEOUT) {
+            return matched;
+        }
+
+        FlowMatch::candidate(key, direction, now, UDP_TIMEOUT)
+    }
+}
+
+impl FlowKey {
+    fn udp(direction: FlowDirection, src: (Ipv4Address, u16), dst: (Ipv4Address, u16)) -> Self {
+        let ((host_addr, host_port), (vm_addr, vm_port)) = direction.host_vm_pair(src, dst);
+        Self::Udp {
+            host_addr,
+            host_port,
+            vm_addr,
+            vm_port,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::test_support::{HOST, VM, inspect_from_host, inspect_from_vm, udp_packet};
+    use super::super::{FlowMatch, FlowTable};
+
+    #[test]
+    fn rejects_zero_source_port_per_rfc_8085() {
+        let mut tracker = FlowTable::new();
+        let datagram = udp_packet(HOST, 0, VM, 5353);
+
+        assert!(matches!(
+            inspect_from_host(&mut tracker, &datagram),
+            FlowMatch::Denied
+        ));
+    }
+
+    #[test]
+    fn udp_reply_requires_an_exact_admitted_request() {
+        let mut tracker = FlowTable::new();
+        let request = udp_packet(HOST, 50_000, VM, 5353);
+        let reply = udp_packet(VM, 5353, HOST, 50_000);
+
+        assert!(matches!(
+            inspect_from_vm(&mut tracker, &reply),
+            FlowMatch::Candidate(_)
+        ));
+
+        let FlowMatch::Candidate(pending) = inspect_from_host(&mut tracker, &request) else {
+            panic!("expected a candidate");
+        };
+        assert!(tracker.commit(pending));
+        assert!(matches!(
+            inspect_from_vm(&mut tracker, &reply),
+            FlowMatch::Allowed
+        ));
+    }
+}

lib/proxy/host.rs

@@ -1,23 +1,22 @@
-use crate::proxy::conntrack::ConntrackResult;
-use crate::proxy::udp_packet_helper::UdpPacketHelper;
-use crate::proxy::{Action, Direction, Proxy, Rule, select_rules};
+use crate::proxy::flows::{FlowDirection, FlowMatch};
+use crate::proxy::{Direction, PolicyDecision, Proxy};
 use anyhow::{Context, Result};
-use smoltcp::wire::{EthernetFrame, EthernetProtocol, Ipv4Packet, UdpPacket};
+use dhcproto::v4::{CLIENT_PORT, SERVER_PORT};
+use smoltcp::phy::ChecksumCapabilities;
+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 {
-            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() {
@@ -41,99 +40,68 @@ impl Proxy<'_> {
         match frame.ethertype() {
             EthernetProtocol::Arp => Some(()),
             EthernetProtocol::Ipv4 => {
-                let ipv4_pkt = Ipv4Packet::new_checked(frame.payload()).ok()?;
+                let ipv4_pkt = Ipv4Packet::new_unchecked(frame.payload());
+                Ipv4Repr::parse(&ipv4_pkt, &ChecksumCapabilities::ignored()).ok()?;
+
                 self.allowed_from_host_ipv4(&ipv4_pkt)
             }
             _ => None,
         }
     }
 
-    fn allowed_from_host_ipv4(&mut self, ipv4_pkt: &Ipv4Packet<&[u8]>) -> Option<()> {
-        if !self.stateful_policy {
+    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(());
         }
 
-        if let Some(rules) = select_rules(&self.rules, ipv4_pkt.src_addr(), Direction::In) {
-            // 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(());
-            }
-
-            // Only process packets addressed to the VM's current IP
-            let Some(lease) = self.dhcp_snooper.lease() else {
-                return None;
-            };
-            if !lease.is_valid_for(ipv4_pkt.dst_addr()) {
-                return None;
-            }
-
-            // Stateless rules decide each packet immediately, without consulting the conntrack
-            if let Some((action, _)) = rules
-                .iter()
-                .find(|(_, rule)| matches!(rule, Rule::Stateless(_)))
+        // Consult the flow table before evaluating inbound policy
+        // so established flows are not treated as new traffic
+        let pending = if self.host.vm_ip == ipv4_pkt.dst_addr() {
+            match self
+                .flows
+                .as_mut()?
+                .inspect(ipv4_pkt, FlowDirection::FromHost)
             {
-                return (*action == Action::Allow).then_some(());
+                FlowMatch::Allowed => return Some(()),
+                FlowMatch::Denied => return None,
+                FlowMatch::Candidate(pending) => Some(pending),
+                FlowMatch::Untracked => None,
             }
-
-            // Existing connections follow conntrack; new ones require an inbound stateful allow rule
-            return match self.conntrack.inspect_from_host(ipv4_pkt) {
-                ConntrackResult::Allowed => Some(()),
-                ConntrackResult::Denied => None,
-                ConntrackResult::New(pending) => {
-                    let allow_new = rules.iter().any(|(action, rule)| {
-                        *action == Action::Allow
-                            && matches!(
-                                rule,
-                                Rule::Stateful {
-                                    direction: Direction::In,
-                                    ..
-                                }
-                            )
-                    });
-
-                    if !allow_new {
-                        return None;
-                    }
-
-                    self.conntrack.commit(pending).then_some(())
-                }
-            };
-        }
-
-        Some(())
-    }
-
-    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,
+        } else {
+            None
         };
 
-        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,
-        };
-
-        self.dhcp_snooper.register_dhcp_reply(udp_pkt.payload());
-    }
-
-    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
+        // The flow is either pending or untracked, evaluate it against inbound policy
+        match self
+            .rules
+            .policy_decision(ipv4_pkt.src_addr(), Direction::In)
         {
-            return false;
-        }
+            // Return traffic was handled above; enforce explicit inbound blocks here
+            Some(PolicyDecision::Block) => None,
 
-        UdpPacket::new_checked(ipv4_pkt.payload())
-            .map(|udp_pkt| udp_pkt.is_dhcp_response())
-            .unwrap_or(false)
+            // Stateless policy is outbound-only; fail closed if this invariant is violated
+            Some(PolicyDecision::AllowStateless) => None,
+
+            // Untracked packets cannot satisfy stateful policy
+            Some(PolicyDecision::AllowStateful) => self.admit_with_tracking(pending?),
+
+            // No inbound rule matched, so allow by default. Track the flow when needed
+            // so its reply is not treated as a new outbound flow
+            None => {
+                self.admit_with_tracking_if_stateful(pending, ipv4_pkt.src_addr(), Direction::Out)
+            }
+        }
     }
 }

lib/proxy/mod.rs

@@ -1,49 +1,44 @@
-mod conntrack;
 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 conntrack::Conntrack;
-use control::Control;
+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::{Action, Rules, build_rules, has_stateful_rules, rule_count, select_rules};
-use smoltcp::wire::EthernetFrame;
-use std::io::ErrorKind;
-use std::os::unix::io::{AsRawFd, RawFd};
+pub(crate) use rules::{PolicyDecision, Rules};
+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,
-    stateful_policy: bool,
     control: Option<Control>,
-    conntrack: Conntrack,
+    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,
@@ -51,44 +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 = build_rules(host.gateway_ip, &allow, &block);
-        let stateful_policy = has_stateful_rules(&rules);
+        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 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,
+            dhcp_server,
             rules,
-            stateful_policy,
             control,
-            conntrack: Conntrack::new(),
+            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];
 
@@ -99,130 +112,120 @@ 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?;
 
-            // Update coarse time for DHCP snooping and conntrack
-            coarsetime::Instant::update();
-
-            // Expire stale flows before processing packets
-            self.conntrack.tick();
-
-            // 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 conntrack
+                    // 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 conntrack
+                    // 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;
-        };
+    /// 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(())
+    }
 
-        let keep_open = match control.service(&mut self.rules) {
-            Ok(keep_open) => keep_open,
-            Err(err) => {
-                log::warn!("disabling Softnet control socket: {err:#}");
-                false
-            }
-        };
-
-        if control.policy_changed() {
-            self.stateful_policy = has_stateful_rules(&self.rules);
-            self.conntrack.clear();
+    /// Commits a pending flow for trackable packets; untracked packets proceed without one.
+    fn admit_with_tracking_if_trackable(&mut self, pending: Option<PendingFlow>) -> Option<()> {
+        match pending {
+            Some(pending) => self.admit_with_tracking(pending),
+            None => Some(()),
         }
+    }
 
-        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 a pending flow when the return-direction rule is stateful.
+    fn admit_with_tracking_if_stateful(
+        &mut self,
+        pending: Option<PendingFlow>,
+        peer_addr: Ipv4Address,
+        return_direction: Direction,
+    ) -> Option<()> {
+        if self.rules.is_stateful(peer_addr, return_direction) {
+            self.admit_with_tracking_if_trackable(pending)
+        } else {
+            Some(())
         }
     }
 }
@@ -230,83 +233,45 @@ impl Proxy<'_> {
 #[cfg(test)]
 mod tests {
     use crate::NetType;
-    use crate::dhcp_snooper::Lease;
-    use crate::proxy::{Action, Proxy, Rule, Target};
-    use ipnet::Ipv4Net;
+    use crate::proxy::Proxy;
     use mac_address::MacAddress;
     use nix::sys::socket::{AddressFamily, SockFlag, SockType, socketpair};
-    use prefix_trie::PrefixMap;
     use serial_test::serial;
-    use smoltcp::wire::{Ipv4Address, Ipv4Packet};
-    use std::collections::HashSet;
+    use smoltcp::wire::{IpProtocol, Ipv4Address, Ipv4Packet, UdpPacket};
     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,
-            PrefixMap::<Ipv4Net, Vec<(Action, Rule)>>::from_iter(vec![(
-                Ipv4Net::from_str("66.66.0.0/16").unwrap(),
-                vec![(Action::Block, "66.66.0.0/16".parse().unwrap(),)]
-            ),])
-        );
+        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,
-            PrefixMap::<Ipv4Net, Vec<(Action, Rule)>>::from_iter(vec![
-                (
-                    Ipv4Net::from_str("33.33.33.33/32").unwrap(),
-                    vec![(Action::Allow, "33.33.33.33/32".parse().unwrap(),)]
-                ),
-                (
-                    Ipv4Net::from_str("33.33.33.0/24").unwrap(),
-                    vec![(Action::Block, "33.33.33.0/24".parse().unwrap(),)]
-                ),
-            ])
-        );
+        assert_eq!(proxy.rules.len(), 2);
 
         assert!(allowed_from_vm_ipv4(&mut proxy, vm_ip, "33.33.33.32").is_none());
         assert!(allowed_from_vm_ipv4(&mut proxy, vm_ip, "33.33.33.33").is_some());
         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,
-            PrefixMap::from_iter(vec![
-                (
-                    proxy.host.gateway_ip.into(),
-                    vec![(
-                        Action::Allow,
-                        Rule::Stateless(Target::Prefix(proxy.host.gateway_ip.into())),
-                    )],
-                ),
-                (
-                    Ipv4Net::from_str("0.0.0.0/0").unwrap(),
-                    vec![(Action::Block, "0.0.0.0/0".parse().unwrap(),)]
-                ),
-            ])
-        );
+        assert_eq!(proxy.rules.len(), 2);
 
         // Access to global IPs should be disallowed because of --block=0.0.0.0/0
         assert!(allowed_from_vm_ipv4(&mut proxy, vm_ip, "8.8.8.8").is_none());
@@ -316,7 +281,64 @@ mod tests {
         assert!(allowed_from_vm_ipv4(&mut proxy, vm_ip, &gateway_ip).is_some());
     }
 
-    fn create_proxy<'test>(vm_ip: Ipv4Address, allow: Vec<&str>, block: Vec<&str>) -> Proxy<'test> {
+    #[tokio::test]
+    #[serial]
+    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 fallback_request = udp_packet(fallback_peer, 40_000, vm_ip, 1_234);
+        let fallback_request = Ipv4Packet::new_checked(fallback_request.as_slice()).unwrap();
+        assert!(proxy.allowed_from_host_ipv4(&fallback_request).is_none());
+
+        let fallback_reply = udp_packet(vm_ip, 1_234, fallback_peer, 40_000);
+        let fallback_reply = Ipv4Packet::new_checked(fallback_reply.as_slice()).unwrap();
+        assert!(proxy.allowed_from_vm_ipv4(fallback_reply).is_none());
+
+        let allowed_request = udp_packet(explicitly_allowed_peer, 40_000, vm_ip, 1_234);
+        let allowed_request = Ipv4Packet::new_checked(allowed_request.as_slice()).unwrap();
+        assert!(proxy.allowed_from_host_ipv4(&allowed_request).is_some());
+
+        let allowed_reply = udp_packet(vm_ip, 1_234, explicitly_allowed_peer, 40_000);
+        let allowed_reply = Ipv4Packet::new_checked(allowed_reply.as_slice()).unwrap();
+        assert!(proxy.allowed_from_vm_ipv4(allowed_reply).is_some());
+    }
+
+    #[tokio::test]
+    #[serial]
+    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 request = udp_packet(peer, 40_000, vm_ip, 1_234);
+        let request = Ipv4Packet::new_checked(request.as_slice()).unwrap();
+        assert!(proxy.allowed_from_host_ipv4(&request).is_some());
+
+        let reply = udp_packet(vm_ip, 1_234, peer, 40_000);
+        let reply = Ipv4Packet::new_checked(reply.as_slice()).unwrap();
+        assert!(proxy.allowed_from_vm_ipv4(reply).is_some());
+    }
+
+    #[tokio::test]
+    #[serial]
+    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 request = udp_packet(vm_ip, 1_234, peer, 40_000);
+        let request = Ipv4Packet::new_checked(request.as_slice()).unwrap();
+        assert!(proxy.allowed_from_vm_ipv4(request).is_some());
+
+        let reply = udp_packet(peer, 40_000, vm_ip, 1_234);
+        let reply = Ipv4Packet::new_checked(reply.as_slice()).unwrap();
+        assert!(proxy.allowed_from_host_ipv4(&reply).is_some());
+    }
+
+    fn create_proxy(allow: Vec<&str>, block: Vec<&str>) -> Proxy {
         let (vm_fd, _) = socketpair(
             AddressFamily::Unix,
             SockType::Datagram,
@@ -326,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()
@@ -341,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<()> {
@@ -363,4 +377,26 @@ mod tests {
 
         proxy.allowed_from_vm_ipv4(ipv4_pkt)
     }
+
+    fn udp_packet(
+        src_addr: Ipv4Address,
+        src_port: u16,
+        dst_addr: Ipv4Address,
+        dst_port: u16,
+    ) -> Vec<u8> {
+        let mut bytes = vec![0; 28];
+        let mut ipv4 = Ipv4Packet::new_unchecked(bytes.as_mut_slice());
+        ipv4.set_version(4);
+        ipv4.set_header_len(20);
+        ipv4.set_total_len(28);
+        ipv4.set_next_header(IpProtocol::Udp);
+        ipv4.set_src_addr(src_addr);
+        ipv4.set_dst_addr(dst_addr);
+
+        let mut udp = UdpPacket::new_unchecked(ipv4.payload_mut());
+        udp.set_src_port(src_port);
+        udp.set_dst_port(dst_port);
+        udp.set_len(8);
+        bytes
+    }
 }

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/rule.rs

@@ -24,13 +24,6 @@ pub enum Direction {
 }
 
 impl Rule {
-    pub(super) fn target(self) -> Target {
-        match self {
-            Rule::Stateless(target) => target,
-            Rule::Stateful { target, .. } => target,
-        }
-    }
-
     pub(super) fn normalized(self) -> Self {
         match self {
             Rule::Stateless(target) => Rule::Stateless(target.normalized()),
@@ -137,18 +130,12 @@ mod tests {
 
     #[test]
     fn displays_normalized_rules() {
-        assert_eq!(
-            "in 10.1.2.3/8"
-                .parse::<Rule>()
-                .unwrap()
-                .normalized()
-                .to_string(),
-            "in 10.0.0.0/8"
-        );
-        assert_eq!(
-            "@host".parse::<Rule>().unwrap().normalized().to_string(),
-            "@host"
-        );
+        for (input, expected) in [("in 10.1.2.3/8", "in 10.0.0.0/8"), ("@host", "@host")] {
+            assert_eq!(
+                input.parse::<Rule>().unwrap().normalized().to_string(),
+                expected
+            );
+        }
     }
 
     #[test]

lib/proxy/rules.rs

@@ -9,194 +9,271 @@ pub(crate) enum Action {
     Allow,
 }
 
-pub(crate) type Rules = PrefixMap<Ipv4Net, Vec<(Action, Rule)>>;
-
-pub(crate) fn select_rules(
-    rules: &Rules,
-    address: Ipv4Address,
-    direction: Direction,
-) -> Option<&[(Action, Rule)]> {
-    if rules.is_empty() {
-        return None;
-    }
-
-    let mut prefix = Ipv4Net::from(address);
-    loop {
-        let (matched_prefix, matched_rules) = rules.get_lpm(&prefix)?;
-        if matched_rules.iter().any(|(_, rule)| match rule {
-            Rule::Stateless(_) => true,
-            Rule::Stateful {
-                direction: rule_direction,
-                ..
-            } => *rule_direction == direction,
-        }) {
-            return Some(matched_rules.as_slice());
-        }
-        prefix = matched_prefix.supernet()?;
-    }
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum PolicyDecision {
+    Block,
+    AllowStateless,
+    AllowStateful,
 }
 
-pub(crate) fn build_rules(host_address: Ipv4Address, allow: &[Rule], block: &[Rule]) -> Rules {
-    let mut rules = PrefixMap::new();
-
-    for &rule in allow {
-        insert_rule(&mut rules, rule, Action::Allow, host_address);
-    }
-
-    for &rule in block {
-        insert_rule(&mut rules, rule, Action::Block, host_address);
-    }
-
-    rules
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+enum Mode {
+    #[default]
+    Legacy,
+    Stateful,
 }
 
-pub(crate) fn rule_count(rules: &Rules) -> usize {
-    rules.into_iter().map(|(_, rules)| rules.len()).sum()
+#[derive(Default)]
+pub(crate) struct Rules {
+    mode: Mode,
+    inbound: PrefixMap<Ipv4Net, Action>,
+    outbound: PrefixMap<Ipv4Net, Action>,
 }
 
-pub(crate) fn has_stateful_rules(rules: &Rules) -> bool {
-    rules.into_iter().any(|(_, rules)| {
-        rules
+impl Rules {
+    pub(crate) fn new(host_address: Ipv4Address, allow: &[Rule], block: &[Rule]) -> Self {
+        // Preserve legacy behavior for bare-only policies. Once a directional rule
+        // is present, compile the whole policy using directional semantics.
+        let mode = if allow
             .iter()
-            .any(|(_, rule)| matches!(rule, Rule::Stateful { .. }))
-    })
-}
+            .chain(block)
+            .any(|rule| matches!(rule, Rule::Stateful { .. }))
+        {
+            Mode::Stateful
+        } else {
+            Mode::Legacy
+        };
 
-fn insert_rule(rules: &mut Rules, rule: Rule, mut action: Action, host_address: Ipv4Address) {
-    let prefix = match rule.target() {
-        Target::Prefix(prefix) => prefix,
-        Target::Host => host_address.into(),
-    };
-    let rule = match rule {
-        Rule::Stateless(_) => Rule::Stateless(Target::Prefix(prefix)),
-        Rule::Stateful { direction, .. } => Rule::Stateful {
-            direction,
-            target: Target::Prefix(prefix),
-        },
-    };
+        let mut rules = Self {
+            mode,
+            ..Self::default()
+        };
 
-    let prefix_rules = rules.entry(prefix).or_default();
-
-    // SECURITY: blocking rules must always take precedence
-    // over allowing rules when prefixes are identical
-    if let Some(existing) = prefix_rules
-        .iter_mut()
-        .find(|(_, existing_rule)| *existing_rule == rule)
-    {
-        if existing.0 == Action::Block {
-            action = Action::Block;
+        for &rule in allow {
+            rules.insert(rule, Action::Allow, host_address);
         }
-        *existing = (action, rule);
-    } else {
-        prefix_rules.push((action, rule));
+
+        // SECURITY: blocking rules must always take precedence
+        // over allowing rules when the rules are identical.
+        for &rule in block {
+            rules.insert(rule, Action::Block, host_address);
+        }
+
+        rules
+    }
+
+    pub(crate) fn policy_decision(
+        &self,
+        address: Ipv4Address,
+        direction: Direction,
+    ) -> Option<PolicyDecision> {
+        match (self.select(address, direction)?, self.mode) {
+            (Action::Block, _) => Some(PolicyDecision::Block),
+            (Action::Allow, Mode::Legacy) => Some(PolicyDecision::AllowStateless),
+            (Action::Allow, Mode::Stateful) => Some(PolicyDecision::AllowStateful),
+        }
+    }
+
+    pub(crate) fn is_stateful(&self, address: Ipv4Address, direction: Direction) -> bool {
+        self.mode == Mode::Stateful && self.select(address, direction).is_some()
+    }
+
+    pub(crate) fn len(&self) -> usize {
+        self.inbound.len() + self.outbound.len()
+    }
+
+    pub(crate) fn has_stateful(&self) -> bool {
+        self.mode == Mode::Stateful
+    }
+
+    fn select(&self, address: Ipv4Address, direction: Direction) -> Option<Action> {
+        let entries = match direction {
+            Direction::In => &self.inbound,
+            Direction::Out => &self.outbound,
+        };
+
+        entries
+            .get_lpm(&Ipv4Net::from(address))
+            .map(|(_, action)| *action)
+    }
+
+    fn insert(&mut self, rule: Rule, action: Action, host_address: Ipv4Address) {
+        match rule {
+            Rule::Stateless(target) => {
+                // Bare rules apply in both directions in stateful mode
+                if self.mode == Mode::Stateful {
+                    self.insert_direction(Direction::In, target, action, host_address);
+                }
+
+                self.insert_direction(Direction::Out, target, action, host_address);
+            }
+            Rule::Stateful { direction, target } => {
+                self.insert_direction(direction, target, action, host_address);
+            }
+        }
+    }
+
+    fn insert_direction(
+        &mut self,
+        direction: Direction,
+        target: Target,
+        action: Action,
+        host_address: Ipv4Address,
+    ) {
+        let prefix = match target {
+            Target::Prefix(prefix) => prefix,
+            Target::Host => host_address.into(),
+        };
+        let entries = match direction {
+            Direction::In => &mut self.inbound,
+            Direction::Out => &mut self.outbound,
+        };
+
+        entries.insert(prefix, action);
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use super::{Action, has_stateful_rules, insert_rule, rule_count, select_rules};
-    use crate::proxy::{Direction, Rule, Target};
-    use ipnet::Ipv4Net;
-    use prefix_trie::PrefixMap;
+    use super::{Action, Mode, PolicyDecision, Rules};
+    use crate::proxy::Direction;
     use smoltcp::wire::Ipv4Address;
-    use std::str::FromStr;
+
+    const HOST: Ipv4Address = Ipv4Address::new(192, 168, 64, 1);
+
+    fn stateful_rules() -> Rules {
+        Rules {
+            mode: Mode::Stateful,
+            ..Rules::default()
+        }
+    }
 
     #[test]
     fn test_policy_precedence() {
-        let host = Ipv4Address::new(192, 168, 64, 1);
         let target = Ipv4Address::new(10, 0, 0, 1);
-        let mut rules = PrefixMap::new();
+        let mut rules = stateful_rules();
 
-        insert_rule(
-            &mut rules,
-            "0.0.0.0/0".parse().unwrap(),
-            Action::Block,
-            host,
-        );
-        assert!(!has_stateful_rules(&rules));
-
-        insert_rule(
-            &mut rules,
-            "in 10.0.0.0/8".parse().unwrap(),
-            Action::Allow,
-            host,
-        );
-        assert!(has_stateful_rules(&rules));
+        rules.insert("0.0.0.0/0".parse().unwrap(), Action::Block, HOST);
+        rules.insert("in 10.0.0.0/8".parse().unwrap(), Action::Allow, HOST);
 
         assert_eq!(
-            select_rules(&rules, target, Direction::In),
-            Some(
-                &[(
-                    Action::Allow,
-                    Rule::Stateful {
-                        direction: Direction::In,
-                        target: Target::Prefix(Ipv4Net::from_str("10.0.0.0/8").unwrap()),
-                    }
-                )][..]
-            )
+            rules.policy_decision(target, Direction::In),
+            Some(PolicyDecision::AllowStateful)
         );
         assert_eq!(
-            select_rules(&rules, target, Direction::Out),
-            Some(&[(Action::Block, "0.0.0.0/0".parse().unwrap())][..])
+            rules.policy_decision(target, Direction::Out),
+            Some(PolicyDecision::Block)
         );
 
-        insert_rule(
-            &mut rules,
-            "10.0.0.1/32".parse().unwrap(),
-            Action::Allow,
-            host,
-        );
+        rules.insert("10.0.0.1/32".parse().unwrap(), Action::Allow, HOST);
         assert_eq!(
-            select_rules(&rules, target, Direction::Out).unwrap().len(),
-            1
-        );
-
-        insert_rule(
-            &mut rules,
-            "out 10.0.0.1/32".parse().unwrap(),
-            Action::Allow,
-            host,
-        );
-        assert_eq!(
-            select_rules(&rules, target, Direction::Out).unwrap().len(),
-            2
+            rules.policy_decision(target, Direction::Out),
+            Some(PolicyDecision::AllowStateful)
         );
     }
 
     #[test]
-    fn test_directional_rules_share_a_prefix() {
-        let host = Ipv4Address::new(192, 168, 64, 1);
-        let mut rules = PrefixMap::new();
+    fn test_directional_rules_at_same_prefix_are_independent() {
+        let mut rules = stateful_rules();
 
         for (target, action) in [
             ("in @host", Action::Allow),
             ("out @host", Action::Allow),
             ("in @host", Action::Block),
         ] {
-            insert_rule(&mut rules, target.parse().unwrap(), action, host);
+            rules.insert(target.parse().unwrap(), action, HOST);
         }
 
         assert_eq!(
-            select_rules(&rules, host, Direction::In),
-            Some(
-                &[
-                    (
-                        Action::Block,
-                        Rule::Stateful {
-                            direction: Direction::In,
-                            target: Target::Prefix(host.into()),
-                        }
-                    ),
-                    (
-                        Action::Allow,
-                        Rule::Stateful {
-                            direction: Direction::Out,
-                            target: Target::Prefix(host.into()),
-                        }
-                    ),
-                ][..]
-            )
+            rules.policy_decision(HOST, Direction::In),
+            Some(PolicyDecision::Block)
         );
-        assert_eq!(rule_count(&rules), 2);
+        assert_eq!(
+            rules.policy_decision(HOST, Direction::Out),
+            Some(PolicyDecision::AllowStateful)
+        );
+        assert_eq!(rules.len(), 2);
+    }
+
+    #[test]
+    fn test_stateless_rules_are_outbound_only() {
+        let target = Ipv4Address::new(10, 1, 2, 3);
+        let mut rules = Rules::default();
+
+        rules.insert("10.0.0.0/8".parse().unwrap(), Action::Block, HOST);
+
+        assert!(rules.policy_decision(target, Direction::In).is_none());
+        assert_eq!(
+            rules.policy_decision(target, Direction::Out),
+            Some(PolicyDecision::Block)
+        );
+    }
+
+    #[test]
+    fn test_inbound_selection_uses_more_specific_bare_rule() {
+        let target = Ipv4Address::new(10, 1, 2, 3);
+        let mut rules = stateful_rules();
+
+        rules.insert("10.1.0.0/16".parse().unwrap(), Action::Allow, HOST);
+        rules.insert("in 10.0.0.0/8".parse().unwrap(), Action::Block, HOST);
+
+        assert_eq!(
+            rules.policy_decision(target, Direction::In),
+            Some(PolicyDecision::AllowStateful)
+        );
+    }
+
+    #[test]
+    fn test_block_wins_over_allow_at_same_outbound_prefix() {
+        let target = Ipv4Address::new(10, 1, 2, 3);
+
+        for (allow, block) in [
+            ("10.0.0.0/8", "out 10.0.0.0/8"),
+            ("out 10.0.0.0/8", "10.0.0.0/8"),
+        ] {
+            let mut rules = stateful_rules();
+            rules.insert(allow.parse().unwrap(), Action::Allow, HOST);
+            rules.insert(block.parse().unwrap(), Action::Block, HOST);
+
+            assert_eq!(
+                rules.policy_decision(target, Direction::Out),
+                Some(PolicyDecision::Block)
+            );
+        }
+    }
+
+    #[test]
+    fn test_directional_rule_makes_bare_rules_stateful() {
+        let allow = "out @host".parse().unwrap();
+        let block = "0.0.0.0/0".parse().unwrap();
+        let rules = Rules::new(HOST, &[allow], &[block]);
+
+        assert_eq!(rules.len(), 3);
+        assert!(rules.has_stateful());
+        assert_eq!(
+            rules.policy_decision(HOST, Direction::In),
+            Some(PolicyDecision::Block)
+        );
+        assert_eq!(
+            rules.policy_decision(HOST, Direction::Out),
+            Some(PolicyDecision::AllowStateful)
+        );
+    }
+
+    #[test]
+    fn test_return_tracking_applies_to_all_rules_in_stateful_mode() {
+        let stateful_target = Ipv4Address::new(10, 1, 2, 3);
+        let stateless_target = Ipv4Address::new(192, 0, 2, 1);
+        let mut rules = stateful_rules();
+
+        rules.insert("0.0.0.0/0".parse().unwrap(), Action::Block, HOST);
+        rules.insert("out 10.0.0.0/8".parse().unwrap(), Action::Block, HOST);
+
+        assert!(rules.is_stateful(stateful_target, Direction::Out));
+        assert!(rules.is_stateful(stateless_target, Direction::Out));
+        assert!(rules.is_stateful(stateful_target, Direction::In));
+
+        rules.insert("10.0.0.0/8".parse().unwrap(), Action::Allow, HOST);
+        assert!(rules.is_stateful(stateful_target, Direction::Out));
     }
 }

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,15 +1,17 @@
-use crate::dhcp_snooper::Lease;
-use crate::proxy::conntrack::ConntrackResult;
-use crate::proxy::udp_packet_helper::UdpPacketHelper;
-use crate::proxy::{Action, Direction, Proxy, Rule, select_rules};
+use crate::proxy::flows::{FlowDirection, FlowMatch};
+use crate::proxy::{Direction, PolicyDecision, Proxy};
 use anyhow::Context;
 use anyhow::Result;
+use dhcproto::v4::SERVER_PORT;
+use smoltcp::phy::ChecksumCapabilities;
 use smoltcp::wire::{
     ArpOperation, ArpPacket, ArpRepr, EthernetFrame, EthernetProtocol, IpProtocol, Ipv4Address,
-    Ipv4Packet, UdpPacket,
+    Ipv4Packet, Ipv4Repr, UdpPacket,
 };
 
-impl Proxy<'_> {
+const IPV4_HEADER_LEN_WITHOUT_OPTIONS: u8 = 20;
+
+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
@@ -33,7 +35,14 @@ impl Proxy<'_> {
                 self.allowed_from_vm_arp(arp_pkt)
             }
             EthernetProtocol::Ipv4 => {
-                let ipv4_pkt = Ipv4Packet::new_checked(frame.payload()).ok()?;
+                let ipv4_pkt = Ipv4Packet::new_unchecked(frame.payload());
+                Ipv4Repr::parse(&ipv4_pkt, &ChecksumCapabilities::ignored()).ok()?;
+
+                // Reject IPv4 options because source routing could bypass destination-based policy
+                if ipv4_pkt.header_len() != IPV4_HEADER_LEN_WITHOUT_OPTIONS {
+                    return None;
+                }
+
                 self.allowed_from_vm_ipv4(ipv4_pkt)
             }
             _ => None,
@@ -41,117 +50,105 @@ 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())
-        {
+        // 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
+                .flows
+                .as_mut()
+                .map(|flows| flows.inspect(&ipv4_pkt, FlowDirection::FromVm))
+                .unwrap_or(FlowMatch::Untracked)
+            {
+                FlowMatch::Allowed => return Some(()),
+                FlowMatch::Denied => return None,
+                FlowMatch::Candidate(pending) => Some(pending),
+                FlowMatch::Untracked => None,
+            };
+
+            // The flow is either pending or untracked, evaluate it against outbound policy
             let dst_addr = ipv4_pkt.dst_addr();
 
-            // Filter traffic based on user-specified rules first
-            if let Some(rules) = select_rules(&self.rules, dst_addr, Direction::Out) {
-                // DHCP 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)) {
-                    return Some(());
+            match self.rules.policy_decision(dst_addr, Direction::Out) {
+                // Return traffic was handled above; enforce explicit outbound blocks here
+                Some(PolicyDecision::Block) => return None,
+
+                // Track statelessly allowed traffic only when needed so its reply is not
+                // treated as a new inbound flow
+                Some(PolicyDecision::AllowStateless) => {
+                    return self.admit_with_tracking_if_stateful(pending, dst_addr, Direction::In);
                 }
 
-                if let Some((action, _)) = rules
-                    .iter()
-                    .find(|(_, rule)| matches!(rule, Rule::Stateless(_)))
-                {
-                    return match action {
-                        Action::Allow => Some(()),
-                        Action::Block => None,
-                    };
-                }
+                // Untracked packets cannot satisfy stateful policy
+                Some(PolicyDecision::AllowStateful) => return self.admit_with_tracking(pending?),
 
-                return match self.conntrack.inspect_from_vm(&ipv4_pkt) {
-                    ConntrackResult::Allowed => Some(()),
-                    ConntrackResult::Denied => None,
-                    ConntrackResult::New(pending) => {
-                        let allow_new = rules.iter().any(|(action, rule)| {
-                            *action == Action::Allow
-                                && matches!(
-                                    rule,
-                                    Rule::Stateful {
-                                        direction: Direction::Out,
-                                        ..
-                                    }
-                                )
-                        });
-
-                        if !allow_new {
-                            return None;
-                        }
-
-                        self.conntrack.commit(pending).then_some(())
-                    }
-                };
+                // No outbound rule matched; apply the built-in fallbacks below
+                None => {}
             }
 
             // When no user-specified rules matched, simply allow all global traffic
             if ip_network::IpNetwork::from(dst_addr).is_global() {
-                return Some(());
+                return self.admit_with_tracking_if_trackable(pending);
             }
 
             // Additionally, allow communication with the host,
             // otherwise things like SSH to a VM won't work
-            if ipv4_pkt.dst_addr() == self.host.gateway_ip {
-                return Some(());
+            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(&ipv4_pkt.dst_addr())
-                {
-                    return Some(());
-                }
-            }
-        }
-
-        // 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) {
-            return Some(());
         }
 
         None
     }
-}
 
-fn is_allowed_dhcp_request(
-    ipv4_pkt: &Ipv4Packet<&[u8]>,
-    unicast_target: Option<Ipv4Address>,
-) -> bool {
-    let dst_addr = ipv4_pkt.dst_addr();
+    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;
+        }
 
-    // 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;
+        // 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
     }
-
-    if ipv4_pkt.next_header() != IpProtocol::Udp {
-        return false;
-    }
-
-    let Ok(udp_pkt) = UdpPacket::new_checked(ipv4_pkt.payload()) else {
-        return false;
-    };
-
-    udp_pkt.is_dhcp_request()
 }
 
 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()? {
@@ -172,66 +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 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;
 
     #[test]
-    fn test_allowed_dhcp_request_targets() {
-        let gateway = Ipv4Address::new(192, 168, 64, 1);
-        let other = Ipv4Address::new(192, 168, 64, 2);
-
-        assert!(allowed_dhcp_request(Ipv4Address::BROADCAST, None));
-        assert!(allowed_dhcp_request(gateway, Some(gateway)));
-        assert!(!allowed_dhcp_request(gateway, None));
-        assert!(!allowed_dhcp_request(other, Some(gateway)));
-    }
-
-    #[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]
@@ -246,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]
@@ -257,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]
@@ -268,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]
@@ -277,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(
@@ -302,22 +275,4 @@ mod tests {
         arp_pkt.set_target_protocol_addr(&vec![0; protocol_len as usize]);
         buf
     }
-
-    fn allowed_dhcp_request(dst_addr: Ipv4Address, unicast_target: Option<Ipv4Address>) -> bool {
-        let mut buf = vec![0; 28];
-        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(28);
-        ipv4_pkt.set_next_header(IpProtocol::Udp);
-        ipv4_pkt.set_dst_addr(dst_addr);
-
-        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);
-
-        let ipv4_pkt = Ipv4Packet::new_checked(buf.as_slice()).unwrap();
-        super::is_allowed_dhcp_request(&ipv4_pkt, unicast_target)
-    }
 }

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,15 +18,31 @@ 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.get_ref().peer_addr() {
+            Ok(_) => Ok(true),
+            Err(error) if error.kind() == io::ErrorKind::NotConnected => Ok(false),
+            Err(error) => Err(error),
+        }
     }
 }
 
@@ -106,12 +124,6 @@ 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;
@@ -119,6 +131,7 @@ mod tests {
     use std::net::UdpSocket;
     use std::os::fd::AsRawFd;
     use std::os::unix::net::{UnixDatagram, UnixStream};
+    use std::time::Duration;
 
     #[test]
     fn test_new_rejects_negative_fd() {
@@ -170,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);
@@ -187,4 +200,53 @@ mod tests {
 
         assert!(socket_fd_is_open);
     }
+
+    #[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());
+    }
+
+    #[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 mut buf = [0];
+        drop(peer);
+
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), vm.read(&mut buf))
+                .await
+                .is_err()
+        );
+        assert!(!vm.is_connected().unwrap());
+    }
+
+    #[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();
+        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();
+
+        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,28 +1,37 @@
 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;
 use softnet::proxy::ExposedPort;
 use softnet::proxy::Proxy;
 use softnet::proxy::Rule;
-use std::borrow::Cow;
 use std::env;
 use std::os::raw::c_int;
 use std::os::unix::io::RawFd;
 use std::os::unix::process::CommandExt;
 use std::process::{Command, ExitCode};
 use system_configuration::core_foundation::base::TCFType;
+use system_configuration::core_foundation::boolean::CFBoolean;
 use system_configuration::core_foundation::dictionary::CFDictionary;
 use system_configuration::core_foundation::number::CFNumber;
 use system_configuration::core_foundation::string::CFString;
 use system_configuration::preferences::SCPreferences;
-use system_configuration::sys::preferences::{SCPreferencesCommitChanges, SCPreferencesSetValue};
+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,
@@ -59,14 +68,14 @@ struct Args {
 
     #[clap(
         long,
-        help = "Comma-separated list of rules for allowing traffic.\n\n\
-        Rule forms:\n\n\
-        * TARGET: traffic between TARGET and the VM in either direction\n\
+        help = "Comma-separated rules for allowing traffic, in the following forms:\n\n\
+        * TARGET: traffic sent from the VM to TARGET; reverse traffic is not filtered by this rule\n\
         * in TARGET: flows initiated from TARGET to the VM\n\
         * out TARGET: flows initiated from the VM to TARGET\n\n\
         Targets are:\n\n\
         * IPv4 CIDRs\n\
         * @host, which matches the vmnet bridge gateway IP\n\n\
+        Directional rules make bare TARGET rules stateful in both directions.\n\n\
         When used with --block, the longest prefix match wins. If an identical rule is both \
         allowed and blocked, blocking takes precedence.\n\n\
         --allow=0.0.0.0/0 additionally disables bridge isolation, even when \
@@ -84,21 +93,21 @@ struct Args {
 
     #[clap(
         long,
-        help = "Comma-separated list of rules for blocking traffic.\n\n\
-        Rule forms:\n\n\
-        * TARGET: traffic between TARGET and the VM in either direction\n\
+        help = "Comma-separated rules for blocking traffic, in the following forms:\n\n\
+        * TARGET: traffic sent from the VM to TARGET; reverse traffic is not filtered by this rule\n\
         * in TARGET: flows initiated from TARGET to the VM\n\
         * out TARGET: flows initiated from the VM to TARGET\n\n\
         Targets are:\n\n\
         * IPv4 CIDRs\n\
         * @host, which matches the vmnet bridge gateway IP\n\n\
+        Directional rules make bare TARGET rules stateful in both directions.\n\n\
         When used with --allow, the longest prefix match wins. If an identical rule is both \
         allowed and blocked, blocking takes precedence.\n\n\
         Examples:\n\n\
-        * --block=0.0.0.0/0 — establish a stateless default-deny policy\n\
-        * --block=\"in @host\" — block stateful flows initiated from @host\n\
+        * --block=0.0.0.0/0 — establish a stateless default-deny egress policy\n\
+        * --block=\"out @host\" — block stateful flows initiated toward @host\n\
         * --block=\"out 66.66.66.0/24\" — block stateful flows initiated toward this CIDR\n\
-        * --block=\"in @host,out 66.66.66.0/24\" — multiple rules may be comma-separated",
+        * --block=\"out @host,out 66.66.66.0/24\" — multiple rules may be comma-separated",
         value_name = "comma-separated rules",
         use_value_delimiter = true,
         action = clap::ArgAction::Set
@@ -130,10 +139,10 @@ fn main() -> ExitCode {
     }
 
     // Initialize Sentry
-    let _sentry = sentry::init(sentry::ClientOptions {
-        release: option_env!("CIRRUS_TAG").map(|tag| Cow::from(format!("softnet@{tag}"))),
-        ..Default::default()
-    });
+    let _sentry = sentry::init(
+        sentry::ClientOptions::default()
+            .maybe_release(option_env!("CIRRUS_TAG").map(|tag| format!("softnet@{tag}"))),
+    );
 
     // Enrich future events with Cirrus CI-specific tags
     if let Ok(tags) = env::var("CIRRUS_SENTRY_TAGS") {
@@ -159,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
@@ -216,8 +218,8 @@ fn try_main() -> anyhow::Result<()> {
         ));
     }
 
-    // Set bootpd(8) min/max lease time while still having the root privileges
-    set_bootpd_lease_time(args.bootpd_lease_time);
+    // Configure bootpd(8) while still having the root privileges
+    configure_bootpd(args.bootpd_lease_time)?;
 
     // Initialize the proxy while still having the root privileges
     let mut proxy = Proxy::new(
@@ -239,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> {
@@ -269,26 +276,59 @@ fn sudo_escalation_works() -> bool {
         .unwrap_or(false)
 }
 
-fn set_bootpd_lease_time(lease_time: u32) {
+fn configure_bootpd(lease_time: u32) -> anyhow::Result<()> {
     let prefs = SCPreferences::group(
         &CFString::new("softnet"),
         &CFString::new("com.apple.InternetSharing.default.plist"),
     );
 
-    let bootpd_dict = CFDictionary::from_CFType_pairs(&[(
-        CFString::new("DHCPLeaseTimeSecs"),
-        CFNumber::from(lease_time as i32),
-    )]);
+    let bootpd_dict = CFDictionary::from_CFType_pairs(&[
+        (
+            CFString::new("DHCPLeaseTimeSecs"),
+            CFNumber::from(lease_time as i32).as_CFType(),
+        ),
+        (
+            CFString::new("dhcp_ignore_client_identifier"),
+            CFBoolean::true_value().as_CFType(),
+        ),
+    ]);
 
     unsafe {
-        SCPreferencesSetValue(
-            prefs.as_concrete_TypeRef(),
-            CFString::new("bootpd").as_concrete_TypeRef(),
-            bootpd_dict.as_concrete_TypeRef().cast(),
+        let prefs = prefs.as_concrete_TypeRef();
+        anyhow::ensure!(
+            SCPreferencesLock(prefs, 1) != 0,
+            "failed to lock bootpd preferences"
         );
 
-        SCPreferencesCommitChanges(prefs.as_concrete_TypeRef());
+        let result = (|| -> anyhow::Result<()> {
+            anyhow::ensure!(
+                SCPreferencesSetValue(
+                    prefs,
+                    CFString::new("bootpd").as_concrete_TypeRef(),
+                    bootpd_dict.as_concrete_TypeRef().cast(),
+                ) != 0,
+                "failed to set bootpd preferences"
+            );
+
+            anyhow::ensure!(
+                SCPreferencesCommitChanges(prefs) != 0,
+                "failed to commit bootpd preferences"
+            );
+
+            anyhow::ensure!(
+                SCPreferencesApplyChanges(prefs) != 0,
+                "failed to apply bootpd preferences"
+            );
+
+            Ok(())
+        })();
+
+        let unlocked = SCPreferencesUnlock(prefs) != 0;
+        result?;
+        anyhow::ensure!(unlocked, "failed to unlock bootpd preferences");
     }
+
+    Ok(())
 }
 
 #[cfg(test)]