WIP

lib/dhcp_snooper.rs

@@ -96,7 +96,7 @@ impl Lease {
         coarsetime::Instant::recent() < self.valid_until
     }
 
-    pub fn valid_ip_source(&self, address: Ipv4Address) -> bool {
+    pub fn is_valid_for(&self, address: Ipv4Address) -> bool {
         self.address == address && self.valid()
     }
 }

lib/proxy/conntrack.rs

@@ -0,0 +1,386 @@
+#[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

@@ -0,0 +1,257 @@
+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

@@ -0,0 +1,131 @@
+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,6 +1,5 @@
-use super::{Action, Target};
+use super::{Rule, Rules, build_rules, rule_count};
 use anyhow::{Context, Result, bail};
-use ipnet::Ipv4Net;
 use jsonrpsee_types::{
     ErrorObjectOwned, Id, Request, Response, ResponsePayload,
     error::{
@@ -8,7 +7,6 @@ use jsonrpsee_types::{
         METHOD_NOT_FOUND_CODE as METHOD_NOT_FOUND, PARSE_ERROR_CODE as PARSE_ERROR,
     },
 };
-use prefix_trie::PrefixMap;
 use serde::Deserialize;
 use serde_json::{Value, json};
 use smoltcp::wire::Ipv4Address;
@@ -20,26 +18,26 @@ use std::os::unix::net::UnixStream;
 
 const MAX_REQUEST_BYTES: usize = 1024 * 1024;
 const MAX_PENDING_RESPONSE_BYTES: usize = 4 * MAX_REQUEST_BYTES;
-const MAX_TARGETS: usize = 4096;
+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<Target>,
-    block: Vec<Target>,
+    allow: Vec<Rule>,
+    block: Vec<Rule>,
     gateway_ip: Ipv4Address,
 }
 
 struct PolicyUpdate {
-    rules: PrefixMap<Ipv4Net, Action>,
-    allow: Vec<Target>,
-    block: Vec<Target>,
+    rules: Rules,
+    allow: Vec<Rule>,
+    block: Vec<Rule>,
 }
 
 impl Policy {
-    pub(super) fn new(gateway_ip: Ipv4Address, allow: Vec<Target>, block: Vec<Target>) -> Self {
-        let allow = normalize_targets(allow);
-        let block = normalize_targets(block);
+    pub(super) fn new(gateway_ip: Ipv4Address, allow: Vec<Rule>, block: Vec<Rule>) -> Self {
+        let allow = normalize_rules(allow);
+        let block = normalize_rules(block);
 
         Policy {
             allow,
@@ -53,15 +51,15 @@ impl Policy {
         allow: Vec<String>,
         block: Vec<String>,
     ) -> std::result::Result<PolicyUpdate, ErrorObjectOwned> {
-        if allow.len() + block.len() > MAX_TARGETS {
+        if allow.len() + block.len() > MAX_RULES {
             return Err(rpc_error(
                 INVALID_PARAMS,
-                format!("allow and block may contain at most {MAX_TARGETS} targets combined"),
+                format!("allow and block may contain at most {MAX_RULES} rules combined"),
             ));
         }
 
-        let allow = parse_targets(allow)?;
-        let block = parse_targets(block)?;
+        let allow = parse_rules(allow)?;
+        let block = parse_rules(block)?;
         let rules = build_rules(self.gateway_ip, &allow, &block);
 
         Ok(PolicyUpdate {
@@ -71,12 +69,16 @@ impl Policy {
         })
     }
 
-    fn apply(&mut self, update: PolicyUpdate) -> PrefixMap<Ipv4Net, Action> {
+    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;
+        }
+
         self.allow = update.allow;
         self.block = update.block;
-        update.rules
+        Some(update.rules)
     }
 
     fn result(&self, rule_count: usize) -> Value {
@@ -86,82 +88,38 @@ impl Policy {
 
 impl PolicyUpdate {
     fn result(&self) -> Value {
-        policy_result(&self.allow, &self.block, self.rules.len())
+        policy_result(&self.allow, &self.block, rule_count(&self.rules))
     }
 }
 
-fn policy_result(allow: &[Target], block: &[Target], rule_count: usize) -> Value {
+fn policy_result(allow: &[Rule], block: &[Rule], rule_count: usize) -> Value {
     json!({
-        "allow": allow.iter().map(target_string).collect::<Vec<_>>(),
-        "block": block.iter().map(target_string).collect::<Vec<_>>(),
+        "allow": allow.iter().map(ToString::to_string).collect::<Vec<_>>(),
+        "block": block.iter().map(ToString::to_string).collect::<Vec<_>>(),
         "ruleCount": rule_count,
     })
 }
 
-fn parse_targets(targets: Vec<String>) -> std::result::Result<Vec<Target>, ErrorObjectOwned> {
-    let mut parsed = Vec::with_capacity(targets.len());
+fn parse_rules(rules: Vec<String>) -> std::result::Result<Vec<Rule>, ErrorObjectOwned> {
+    let mut parsed = Vec::with_capacity(rules.len());
 
-    for target in targets {
-        let parsed_target = target.parse().map_err(|_| {
+    for rule in rules {
+        let parsed_rule = rule.parse().map_err(|_| {
             rpc_error(
                 INVALID_PARAMS,
-                format!("invalid target {target:?}: expected an IPv4 CIDR or @host"),
+                format!("invalid rule {rule:?}: expected TARGET, \"in TARGET\", or \"out TARGET\""),
             )
         })?;
-        parsed.push(parsed_target);
+        parsed.push(parsed_rule);
     }
 
-    Ok(normalize_targets(parsed))
+    Ok(normalize_rules(parsed))
 }
 
-fn normalize_targets(targets: Vec<Target>) -> Vec<Target> {
-    let mut targets = targets
-        .into_iter()
-        .map(|target| match target {
-            Target::Prefix(prefix) => Target::Prefix(prefix.trunc()),
-            Target::Host => Target::Host,
-        })
-        .collect::<Vec<_>>();
-
-    targets.sort_by_key(target_string);
-    targets.dedup();
-    targets
-}
-
-fn target_string(target: &Target) -> String {
-    match target {
-        Target::Prefix(prefix) => prefix.to_string(),
-        Target::Host => "@host".to_string(),
-    }
-}
-
-fn build_rules(
-    gateway_ip: Ipv4Address,
-    allow: &[Target],
-    block: &[Target],
-) -> PrefixMap<Ipv4Net, Action> {
-    let mut rules = PrefixMap::new();
-
-    for target in allow {
-        let prefix = match target {
-            Target::Prefix(prefix) => *prefix,
-            Target::Host => gateway_ip.into(),
-        };
-
-        rules.insert(prefix, Action::Allow);
-    }
-
-    // SECURITY: blocking rules must always take precedence over allowing rules when prefixes
-    // are identical, including @host and an explicit prefix for the gateway address.
-    for target in block {
-        let prefix = match target {
-            Target::Prefix(prefix) => *prefix,
-            Target::Host => gateway_ip.into(),
-        };
-
-        rules.insert(prefix, Action::Block);
-    }
-
+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();
     rules
 }
 
@@ -173,14 +131,15 @@ pub(super) struct Control {
     output_offset: usize,
     discarding_input: bool,
     input_closed: bool,
+    policy_changed: bool,
 }
 
 impl Control {
     pub(super) fn new(
         control_fd: RawFd,
         gateway_ip: Ipv4Address,
-        allow: Vec<Target>,
-        block: Vec<Target>,
+        allow: Vec<Rule>,
+        block: Vec<Rule>,
     ) -> Result<Self> {
         let control_fd = duplicate_control_fd(control_fd)?;
 
@@ -196,10 +155,11 @@ impl Control {
             output_offset: 0,
             discarding_input: false,
             input_closed: false,
+            policy_changed: false,
         })
     }
 
-    pub(super) fn service(&mut self, rules: &mut PrefixMap<Ipv4Net, Action>) -> Result<bool> {
+    pub(super) fn service(&mut self, rules: &mut Rules) -> Result<bool> {
         if !self.flush()? {
             return Ok(false);
         }
@@ -262,7 +222,12 @@ impl Control {
             .context("failed to shut down the control socket")
     }
 
-    fn process_input(&mut self, rules: &mut PrefixMap<Ipv4Net, Action>) -> Result<bool> {
+    /// 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') {
@@ -311,11 +276,15 @@ impl Control {
                 continue;
             }
 
-            let (response, update) = handle_request(&self.policy, rules.len(), &line[..newline]);
+            let (response, update) =
+                handle_request(&self.policy, rule_count(rules), &line[..newline]);
             self.enqueue(response)?;
 
-            if let Some(update) = update {
-                *rules = self.policy.apply(update);
+            if let Some(update) = update
+                && let Some(updated_rules) = self.policy.apply(update)
+            {
+                *rules = updated_rules;
+                self.policy_changed = true;
             }
 
             if !self.flush()? {
@@ -571,9 +540,9 @@ fn validate_control_fd(control_fd: RawFd) -> Result<()> {
 mod tests {
     use super::{
         Control, INVALID_PARAMS, INVALID_REQUEST, MAX_PENDING_RESPONSE_BYTES, MAX_REQUEST_BYTES,
-        MAX_TARGETS, METHOD_NOT_FOUND, PARSE_ERROR, Policy, build_rules, handle_request,
+        MAX_RULES, METHOD_NOT_FOUND, PARSE_ERROR, Policy, build_rules, handle_request, rule_count,
     };
-    use crate::proxy::{Action, Target};
+    use crate::proxy::{Action, Rule, Rules};
     use ipnet::Ipv4Net;
     use prefix_trie::PrefixMap;
     use serde_json::{Value, json};
@@ -588,12 +557,12 @@ mod tests {
 
     struct TestPolicy {
         state: Policy,
-        rules: PrefixMap<Ipv4Net, Action>,
+        rules: Rules,
     }
 
     impl TestPolicy {
         fn result(&self) -> Value {
-            self.state.result(self.rules.len())
+            self.state.result(rule_count(&self.rules))
         }
     }
 
@@ -605,17 +574,14 @@ mod tests {
         }
     }
 
-    fn targets(targets: &[&str]) -> Vec<Target> {
-        targets
-            .iter()
-            .map(|target| target.parse().unwrap())
-            .collect()
+    fn rules(rules: &[&str]) -> Vec<Rule> {
+        rules.iter().map(|rule| rule.parse().unwrap()).collect()
     }
 
     fn policy(allow: &[&str], block: &[&str]) -> TestPolicy {
         let gateway_ip = Ipv4Address::new(192, 168, 64, 1);
-        let allow = targets(allow);
-        let block = targets(block);
+        let allow = rules(allow);
+        let block = rules(block);
 
         TestPolicy {
             rules: build_rules(gateway_ip, &allow, &block),
@@ -637,9 +603,11 @@ mod tests {
     }
 
     fn raw_request(policy: &mut TestPolicy, line: &[u8]) -> Value {
-        let (response, update) = handle_request(&policy.state, policy.rules.len(), line);
-        if let Some(update) = update {
-            policy.rules = policy.state.apply(update);
+        let (response, update) = handle_request(&policy.state, rule_count(&policy.rules), line);
+        if let Some(update) = update
+            && let Some(rules) = policy.state.apply(update)
+        {
+            policy.rules = rules;
         }
 
         response
@@ -685,18 +653,18 @@ mod tests {
 
         assert_eq!(
             policy.rules.get(&Ipv4Net::from_str("10.0.0.0/8").unwrap()),
-            Some(&Action::Block)
+            Some(&vec![(Action::Block, "10.0.0.0/8".parse().unwrap())])
         );
         assert_eq!(
             policy
                 .rules
                 .get(&Ipv4Net::from_str("192.168.64.1/32").unwrap()),
-            Some(&Action::Block)
+            Some(&vec![(Action::Block, "192.168.64.1/32".parse().unwrap())])
         );
     }
 
     #[test]
-    fn set_normalizes_targets() {
+    fn set_normalizes_rules() {
         let mut policy = policy(&[], &[]);
 
         let first = request(
@@ -722,7 +690,39 @@ mod tests {
     }
 
     #[test]
-    fn invalid_targets_and_limits_leave_policy_unchanged() {
+    fn set_supports_directional_rules_and_counts_logical_rules() {
+        let mut policy = policy(&[], &[]);
+
+        let response = request(
+            &mut policy,
+            json!({
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "softnet.policy.set",
+                "params": {
+                    "allow": ["in 10.1.2.3/8", "out 10.0.0.0/8"],
+                    "block": ["in 10.0.0.0/8"]
+                }
+            }),
+        );
+
+        assert_eq!(
+            response["result"]["allow"],
+            json!(["in 10.0.0.0/8", "out 10.0.0.0/8"])
+        );
+        assert_eq!(response["result"]["block"], json!(["in 10.0.0.0/8"]));
+        assert_eq!(response["result"]["ruleCount"], 2);
+        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()),
+            ])
+        );
+    }
+
+    #[test]
+    fn invalid_rules_and_limits_leave_policy_unchanged() {
         let mut policy = policy(&["@host"], &["0.0.0.0/0"]);
         let before = policy.result();
 
@@ -738,14 +738,14 @@ mod tests {
         assert_eq!(invalid["error"]["code"], INVALID_PARAMS);
         assert_eq!(policy.result(), before);
 
-        let targets = vec!["10.0.0.0/8"; MAX_TARGETS + 1];
+        let rules = vec!["10.0.0.0/8"; MAX_RULES + 1];
         let too_many = request(
             &mut policy,
             json!({
                 "jsonrpc": "2.0",
                 "id": 2,
                 "method": "softnet.policy.set",
-                "params": {"allow": targets, "block": []}
+                "params": {"allow": rules, "block": []}
             }),
         );
         assert_eq!(too_many["error"]["code"], INVALID_PARAMS);
@@ -903,12 +903,31 @@ mod tests {
         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![Target::Host]);
+        assert_eq!(control.policy.allow, vec!["@host".parse().unwrap()]);
 
-        let before = control.policy.result(rules.len());
+        let before = control.policy.result(rule_count(&rules));
         drop(client);
         assert!(!control.service(&mut rules).unwrap());
-        assert_eq!(control.policy.result(rules.len()), before);
+        assert_eq!(control.policy.result(rule_count(&rules)), before);
+    }
+
+    #[test]
+    fn repeated_normalized_policy_is_not_reported_as_changed() {
+        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")
+            .unwrap();
+        assert!(control.service(&mut rules).unwrap());
+        assert!(control.policy_changed());
+
+        client
+            .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"in 10.0.0.0/8\"],\"block\":[]}}\n")
+            .unwrap();
+        assert!(control.service(&mut rules).unwrap());
+        assert!(!control.policy_changed());
     }
 
     #[test]
@@ -974,7 +993,7 @@ mod tests {
             .unwrap();
         let mut control = control(server.as_raw_fd()).unwrap();
         let mut rules = PrefixMap::new();
-        let before = control.policy.result(rules.len());
+        let before = control.policy.result(rule_count(&rules));
 
         client
             .write_all(
@@ -982,7 +1001,7 @@ mod tests {
             )
             .unwrap();
         assert!(control.service(&mut rules).unwrap());
-        assert_eq!(control.policy.result(rules.len()), before);
+        assert_eq!(control.policy.result(rule_count(&rules)), before);
         assert!(control.output.is_empty());
 
         client.write_all(b"\"block\":[]}}\n").unwrap();
@@ -993,7 +1012,7 @@ mod tests {
         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![Target::Host]);
+        assert_eq!(control.policy.allow, vec!["@host".parse().unwrap()]);
     }
 
     #[test]
@@ -1027,7 +1046,7 @@ mod tests {
         let (_client, server) = UnixStream::pair().unwrap();
         let mut control = control(server.as_raw_fd()).unwrap();
         let mut rules = PrefixMap::new();
-        let allow = (0..MAX_TARGETS)
+        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!({
@@ -1048,7 +1067,7 @@ mod tests {
 
         control.input = input;
         assert!(control.process_input(&mut rules).unwrap());
-        assert_eq!(rules.len(), MAX_TARGETS);
+        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);
@@ -1059,7 +1078,7 @@ mod tests {
         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(rules.len());
+        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();
@@ -1070,7 +1089,7 @@ mod tests {
                 .to_string()
                 .contains("control response queue exceeded")
         );
-        assert_eq!(control.policy.result(rules.len()), before);
+        assert_eq!(control.policy.result(rule_count(&rules)), before);
     }
 
     #[test]
@@ -1078,7 +1097,7 @@ mod tests {
         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(rules.len());
+        let before = control.policy.result(rule_count(&rules));
 
         control.output = vec![b'x'; MAX_REQUEST_BYTES];
         assert!(control.flush().unwrap());
@@ -1089,7 +1108,7 @@ mod tests {
             .unwrap();
 
         assert!(control.service(&mut rules).unwrap());
-        assert_eq!(control.policy.result(rules.len()), before);
+        assert_eq!(control.policy.result(rule_count(&rules)), before);
         assert!(control.input.is_empty());
     }
 

lib/proxy/host.rs

@@ -1,5 +1,6 @@
-use crate::proxy::Proxy;
+use crate::proxy::conntrack::ConntrackResult;
 use crate::proxy::udp_packet_helper::UdpPacketHelper;
+use crate::proxy::{Action, Direction, Proxy, Rule, select_rules};
 use anyhow::{Context, Result};
 use smoltcp::wire::{EthernetFrame, EthernetProtocol, Ipv4Packet, UdpPacket};
 
@@ -39,11 +40,69 @@ impl Proxy<'_> {
     fn allowed_from_host(&mut self, frame: &EthernetFrame<&[u8]>) -> Option<()> {
         match frame.ethertype() {
             EthernetProtocol::Arp => Some(()),
-            EthernetProtocol::Ipv4 => Some(()),
+            EthernetProtocol::Ipv4 => {
+                let ipv4_pkt = Ipv4Packet::new_checked(frame.payload()).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 {
+            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(_)))
+            {
+                return (*action == Action::Allow).then_some(());
+            }
+
+            // 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;
@@ -54,11 +113,7 @@ impl Proxy<'_> {
             _ => return,
         };
 
-        if ipv4_pkt.src_addr() != self.host.gateway_ip {
-            return;
-        }
-
-        if ipv4_pkt.next_header() != smoltcp::wire::IpProtocol::Udp {
+        if !self.is_allowed_dhcp_response(&ipv4_pkt) {
             return;
         }
 
@@ -67,10 +122,18 @@ impl Proxy<'_> {
             Err(_) => return,
         };
 
-        if !udp_pkt.is_dhcp_response() {
-            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
+        {
+            return false;
+        }
+
+        UdpPacket::new_checked(ipv4_pkt.payload())
+            .map(|udp_pkt| udp_pkt.is_dhcp_response())
+            .unwrap_or(false)
+    }
 }

lib/proxy/mod.rs

@@ -1,7 +1,10 @@
+mod conntrack;
 mod control;
 mod exposed_port;
 mod host;
 mod port_forwarder;
+mod rule;
+mod rules;
 mod udp_packet_helper;
 mod vm;
 
@@ -11,16 +14,17 @@ use crate::host::NetType;
 use crate::poller::Poller;
 use crate::vm::VM;
 use anyhow::Result;
+use conntrack::Conntrack;
 use control::Control;
 pub use exposed_port::ExposedPort;
 use ipnet::Ipv4Net;
 use mac_address::MacAddress;
 use port_forwarder::PortForwarder;
-use prefix_trie::PrefixMap;
+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};
-use std::str::FromStr;
 use std::time::Duration;
 use vmnet::Batch;
 
@@ -30,50 +34,28 @@ pub struct Proxy<'proxy> {
     poller: Poller<'proxy>,
     vm_mac_address: smoltcp::wire::EthernetAddress,
     dhcp_snooper: DhcpSnooper,
-    rules: PrefixMap<Ipv4Net, Action>,
+    rules: Rules,
+    stateful_policy: bool,
     control: Option<Control>,
+    conntrack: Conntrack,
     enobufs_encountered: bool,
     port_forwarder: PortForwarder,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub enum Target {
-    Prefix(Ipv4Net),
-    Host,
-}
-
-impl FromStr for Target {
-    type Err = ipnet::AddrParseError;
-
-    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
-        if s == "@host" {
-            return Ok(Target::Host);
-        }
-
-        Ipv4Net::from_str(s).map(Target::Prefix)
-    }
-}
-
-#[derive(Debug, Clone, PartialEq)]
-pub(crate) enum Action {
-    Block,
-    Allow,
-}
-
 impl Proxy<'_> {
     pub fn new<'proxy>(
         vm_fd: RawFd,
         vm_mac_address: MacAddress,
         vm_net_type: NetType,
-        allow: Vec<Target>,
-        block: Vec<Target>,
+        allow: Vec<Rule>,
+        block: Vec<Rule>,
         exposed_ports: Vec<ExposedPort>,
         control_fd: Option<RawFd>,
     ) -> Result<Proxy<'proxy>> {
         let vm = VM::new(vm_fd)?;
         let host = Host::new(
             vm_net_type,
-            !allow.contains(&Target::Prefix(Ipv4Net::default())),
+            !allow.contains(&Rule::Stateless(Target::Prefix(Ipv4Net::default()))),
         )?;
         let poller_timeout = Duration::from_millis(100);
         let control = control_fd
@@ -88,29 +70,8 @@ impl Proxy<'_> {
             poller_timeout,
         )?;
 
-        // Craft packet filter rules
-        //
-        // SECURITY: blocking rules must always take precedence
-        // over allowing rules when prefixes are identical.
-        let mut rules = PrefixMap::new();
-
-        for allow_target in allow {
-            let allow_prefix = match allow_target {
-                Target::Prefix(prefix) => prefix,
-                Target::Host => host.gateway_ip.into(),
-            };
-
-            rules.insert(allow_prefix, Action::Allow);
-        }
-
-        for block_target in block {
-            let block_prefix = match block_target {
-                Target::Prefix(prefix) => prefix,
-                Target::Host => host.gateway_ip.into(),
-            };
-
-            rules.insert(block_prefix, Action::Block);
-        }
+        let rules = build_rules(host.gateway_ip, &allow, &block);
+        let stateful_policy = has_stateful_rules(&rules);
 
         Ok(Proxy {
             vm,
@@ -119,7 +80,9 @@ impl Proxy<'_> {
             vm_mac_address: smoltcp::wire::EthernetAddress(vm_mac_address.bytes()),
             dhcp_snooper: DhcpSnooper::new(poller_timeout),
             rules,
+            stateful_policy,
             control,
+            conntrack: Conntrack::new(),
             enobufs_encountered: false,
             port_forwarder: PortForwarder::new(exposed_ports),
         })
@@ -141,9 +104,12 @@ impl Proxy<'_> {
         loop {
             let (vm_readable, host_readable, interrupt) = self.poller.wait()?;
 
-            // Update coarse time for the DHCP snooper
+            // 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();
@@ -177,7 +143,7 @@ impl Proxy<'_> {
         loop {
             match self.vm.read(buf) {
                 Ok(n) => {
-                    // Update coarse time for the DHCP snooper
+                    // Update coarse time for DHCP snooping and conntrack
                     coarsetime::Instant::update();
 
                     if let Ok(frame) = EthernetFrame::new_checked(&buf[..n]) {
@@ -205,7 +171,7 @@ impl Proxy<'_> {
         loop {
             match self.host.read(batch, bufs) {
                 Ok(pktcnt) => {
-                    // Update coarse time for the DHCP snooper
+                    // Update coarse time for DHCP snooping and conntrack
                     coarsetime::Instant::update();
 
                     for buf in batch.packet_sized_bufs(bufs).take(pktcnt) {
@@ -240,6 +206,11 @@ impl Proxy<'_> {
             }
         };
 
+        if control.policy_changed() {
+            self.stateful_policy = has_stateful_rules(&self.rules);
+            self.conntrack.clear();
+        }
+
         if keep_open {
             return;
         }
@@ -260,7 +231,7 @@ impl Proxy<'_> {
 mod tests {
     use crate::NetType;
     use crate::dhcp_snooper::Lease;
-    use crate::proxy::{Action, Proxy};
+    use crate::proxy::{Action, Proxy, Rule, Target};
     use ipnet::Ipv4Net;
     use mac_address::MacAddress;
     use nix::sys::socket::{AddressFamily, SockFlag, SockType, socketpair};
@@ -276,57 +247,73 @@ mod tests {
     #[serial]
     fn test_blocking_takes_precedence() {
         let vm_ip = Ipv4Address::from_str("192.168.0.2").unwrap();
-        let proxy = create_proxy(vm_ip, vec!["66.66.0.0/16"], vec!["66.66.0.0/16"]);
+        let mut proxy = create_proxy(vm_ip, vec!["66.66.0.0/16"], vec!["66.66.0.0/16"]);
 
         assert_eq!(
             proxy.rules,
-            PrefixMap::<Ipv4Net, Action>::from_iter(vec![(
+            PrefixMap::<Ipv4Net, Vec<(Action, Rule)>>::from_iter(vec![(
                 Ipv4Net::from_str("66.66.0.0/16").unwrap(),
-                Action::Block
+                vec![(Action::Block, "66.66.0.0/16".parse().unwrap(),)]
             ),])
         );
 
-        assert!(allowed_from_vm_ipv4(&proxy, vm_ip, "66.66.66.66").is_none());
+        assert!(allowed_from_vm_ipv4(&mut proxy, vm_ip, "66.66.66.66").is_none());
     }
 
     #[test]
     #[serial]
     fn test_longest_prefix_match_wins() {
         let vm_ip = Ipv4Address::from_str("192.168.0.2").unwrap();
-        let proxy = create_proxy(vm_ip, vec!["33.33.33.33/32"], vec!["33.33.33.0/24"]);
+        let mut proxy = create_proxy(vm_ip, vec!["33.33.33.33/32"], vec!["33.33.33.0/24"]);
 
         assert_eq!(
             proxy.rules,
-            PrefixMap::<Ipv4Net, Action>::from_iter(vec![
-                (Ipv4Net::from_str("33.33.33.33/32").unwrap(), Action::Allow),
-                (Ipv4Net::from_str("33.33.33.0/24").unwrap(), Action::Block),
+            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!(allowed_from_vm_ipv4(&proxy, vm_ip, "33.33.33.32").is_none());
-        assert!(allowed_from_vm_ipv4(&proxy, vm_ip, "33.33.33.33").is_some());
-        assert!(allowed_from_vm_ipv4(&proxy, vm_ip, "33.33.33.34").is_none());
+        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]
     #[serial]
     fn test_allow_host() {
         let vm_ip = Ipv4Address::from_str("192.168.0.2").unwrap();
-        let proxy = create_proxy(vm_ip, vec!["@host"], vec!["0.0.0.0/0"]);
+        let mut proxy = create_proxy(vm_ip, vec!["@host"], vec!["0.0.0.0/0"]);
 
         assert_eq!(
             proxy.rules,
             PrefixMap::from_iter(vec![
-                (proxy.host.gateway_ip.into(), Action::Allow),
-                (Ipv4Net::from_str("0.0.0.0/0").unwrap(), Action::Block),
+                (
+                    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(),)]
+                ),
             ])
         );
 
         // Access to global IPs should be disallowed because of --block=0.0.0.0/0
-        assert!(allowed_from_vm_ipv4(&proxy, vm_ip, "8.8.8.8").is_none());
+        assert!(allowed_from_vm_ipv4(&mut proxy, vm_ip, "8.8.8.8").is_none());
 
         // Despite the above, access to host IP address should be possible because of --allow=@host
-        assert!(allowed_from_vm_ipv4(&proxy, vm_ip, &proxy.host.gateway_ip.to_string()).is_some());
+        let gateway_ip = proxy.host.gateway_ip.to_string();
+        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> {
@@ -345,11 +332,11 @@ mod tests {
             NetType::Nat,
             allow
                 .into_iter()
-                .map(|cidr| cidr.parse().unwrap())
+                .map(|value| value.parse().unwrap())
                 .collect(),
             block
                 .into_iter()
-                .map(|cidr| cidr.parse().unwrap())
+                .map(|value| value.parse().unwrap())
                 .collect(),
             Vec::default(),
             None,
@@ -365,7 +352,7 @@ mod tests {
         proxy
     }
 
-    fn allowed_from_vm_ipv4(proxy: &Proxy, src: Ipv4Address, dst: &str) -> Option<()> {
+    fn allowed_from_vm_ipv4(proxy: &mut Proxy, src: Ipv4Address, dst: &str) -> Option<()> {
         let mut buf = vec![0; 1500];
 
         let mut ipv4_pkt_mut = Ipv4Packet::new_unchecked(&mut buf[..]);

lib/proxy/rule.rs

@@ -0,0 +1,171 @@
+use ipnet::Ipv4Net;
+use std::fmt::{self, Display, Formatter};
+use std::str::FromStr;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Rule {
+    Stateless(Target),
+    Stateful {
+        direction: Direction,
+        target: Target,
+    },
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Target {
+    Prefix(Ipv4Net),
+    Host,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Direction {
+    In,
+    Out,
+}
+
+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()),
+            Rule::Stateful { direction, target } => Rule::Stateful {
+                direction,
+                target: target.normalized(),
+            },
+        }
+    }
+}
+
+impl FromStr for Rule {
+    type Err = ipnet::AddrParseError;
+
+    fn from_str(input: &str) -> Result<Self, Self::Err> {
+        let (direction, target) = if let Some(target) = input.strip_prefix("in ") {
+            (Direction::In, target)
+        } else if let Some(target) = input.strip_prefix("out ") {
+            (Direction::Out, target)
+        } else {
+            return input.parse().map(Rule::Stateless);
+        };
+
+        let target = target.trim_start_matches(' ').parse()?;
+        Ok(Rule::Stateful { direction, target })
+    }
+}
+
+impl Display for Rule {
+    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
+        match self {
+            Rule::Stateless(target) => Display::fmt(target, formatter),
+            Rule::Stateful { direction, target } => match direction {
+                Direction::In => write!(formatter, "in {target}"),
+                Direction::Out => write!(formatter, "out {target}"),
+            },
+        }
+    }
+}
+
+impl Target {
+    fn normalized(self) -> Self {
+        match self {
+            Target::Prefix(prefix) => Target::Prefix(prefix.trunc()),
+            Target::Host => Target::Host,
+        }
+    }
+}
+
+impl FromStr for Target {
+    type Err = ipnet::AddrParseError;
+
+    fn from_str(input: &str) -> Result<Self, Self::Err> {
+        if input == "@host" {
+            Ok(Target::Host)
+        } else {
+            input.parse().map(Target::Prefix)
+        }
+    }
+}
+
+impl Display for Target {
+    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
+        match self {
+            Target::Prefix(prefix) => Display::fmt(prefix, formatter),
+            Target::Host => formatter.write_str("@host"),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{Direction, Rule, Target};
+    use ipnet::Ipv4Net;
+    use std::str::FromStr;
+
+    #[test]
+    fn parses_stateless_target() {
+        assert_eq!(
+            "@host".parse::<Rule>().unwrap(),
+            Rule::Stateless(Target::Host)
+        );
+    }
+
+    #[test]
+    fn parses_stateful_directions() {
+        let private_network = Target::Prefix(Ipv4Net::from_str("10.0.0.0/8").unwrap());
+
+        assert_eq!(
+            "in   @host".parse::<Rule>().unwrap(),
+            Rule::Stateful {
+                direction: Direction::In,
+                target: Target::Host,
+            }
+        );
+        assert_eq!(
+            "out 10.0.0.0/8".parse::<Rule>().unwrap(),
+            Rule::Stateful {
+                direction: Direction::Out,
+                target: private_network,
+            }
+        );
+    }
+
+    #[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"
+        );
+    }
+
+    #[test]
+    fn rejects_invalid_rules() {
+        for input in [
+            "",
+            "from @host",
+            "in",
+            "out",
+            "in from @host",
+            "out to @host",
+            "infrom @host",
+            " in @host",
+            "in @host ",
+            "in\t@host",
+        ] {
+            assert!(input.parse::<Rule>().is_err(), "{input:?} should fail");
+        }
+    }
+}

lib/proxy/rules.rs

@@ -0,0 +1,202 @@
+use super::{Direction, Rule, Target};
+use ipnet::Ipv4Net;
+use prefix_trie::PrefixMap;
+use smoltcp::wire::Ipv4Address;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum Action {
+    Block,
+    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()?;
+    }
+}
+
+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
+}
+
+pub(crate) fn rule_count(rules: &Rules) -> usize {
+    rules.into_iter().map(|(_, rules)| rules.len()).sum()
+}
+
+pub(crate) fn has_stateful_rules(rules: &Rules) -> bool {
+    rules.into_iter().any(|(_, rules)| {
+        rules
+            .iter()
+            .any(|(_, rule)| matches!(rule, Rule::Stateful { .. }))
+    })
+}
+
+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 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;
+        }
+        *existing = (action, rule);
+    } else {
+        prefix_rules.push((action, rule));
+    }
+}
+
+#[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 smoltcp::wire::Ipv4Address;
+    use std::str::FromStr;
+
+    #[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();
+
+        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));
+
+        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()),
+                    }
+                )][..]
+            )
+        );
+        assert_eq!(
+            select_rules(&rules, target, Direction::Out),
+            Some(&[(Action::Block, "0.0.0.0/0".parse().unwrap())][..])
+        );
+
+        insert_rule(
+            &mut rules,
+            "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
+        );
+    }
+
+    #[test]
+    fn test_directional_rules_share_a_prefix() {
+        let host = Ipv4Address::new(192, 168, 64, 1);
+        let mut rules = PrefixMap::new();
+
+        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);
+        }
+
+        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()),
+                        }
+                    ),
+                ][..]
+            )
+        );
+        assert_eq!(rule_count(&rules), 2);
+    }
+}

lib/proxy/vm.rs

@@ -1,12 +1,12 @@
 use crate::dhcp_snooper::Lease;
+use crate::proxy::conntrack::ConntrackResult;
 use crate::proxy::udp_packet_helper::UdpPacketHelper;
-use crate::proxy::{Action, Proxy};
+use crate::proxy::{Action, Direction, Proxy, Rule, select_rules};
 use anyhow::Context;
 use anyhow::Result;
-use ipnet::Ipv4Net;
 use smoltcp::wire::{
-    ArpOperation, ArpPacket, ArpRepr, EthernetFrame, EthernetProtocol, IpProtocol, Ipv4Packet,
-    UdpPacket,
+    ArpOperation, ArpPacket, ArpRepr, EthernetFrame, EthernetProtocol, IpProtocol, Ipv4Address,
+    Ipv4Packet, UdpPacket,
 };
 
 impl Proxy<'_> {
@@ -22,7 +22,7 @@ impl Proxy<'_> {
             .context("failed to write to the host")
     }
 
-    fn allowed_from_vm(&self, frame: &EthernetFrame<&[u8]>) -> Option<()> {
+    fn allowed_from_vm(&mut self, frame: &EthernetFrame<&[u8]>) -> Option<()> {
         if frame.src_addr() != self.vm_mac_address {
             return None;
         }
@@ -44,23 +44,52 @@ impl Proxy<'_> {
         vm_arp_allowed(arp_pkt, self.vm_mac_address, self.dhcp_snooper.lease())
     }
 
-    pub(crate) fn allowed_from_vm_ipv4(&self, ipv4_pkt: Ipv4Packet<&[u8]>) -> Option<()> {
+    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.valid_ip_source(ipv4_pkt.src_addr())
+            && lease.is_valid_for(ipv4_pkt.src_addr())
         {
             let dst_addr = ipv4_pkt.dst_addr();
 
             // Filter traffic based on user-specified rules first
-            if !self.rules.is_empty() {
-                let dst_net = Ipv4Net::from(dst_addr);
+            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(());
+                }
 
-                if let Some((_, action)) = self.rules.get_lpm(&dst_net) {
+                if let Some((action, _)) = rules
+                    .iter()
+                    .find(|(_, rule)| matches!(rule, Rule::Stateless(_)))
+                {
                     return match action {
                         Action::Allow => Some(()),
                         Action::Block => None,
                     };
                 }
+
+                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(())
+                    }
+                };
             }
 
             // When no user-specified rules matched, simply allow all global traffic
@@ -87,21 +116,38 @@ impl Proxy<'_> {
             }
         }
 
-        // Allow outgoing DHCP requests to broadcast addresses,
+        // Allow outgoing DHCP requests to the bootpd(8) broadcast address,
         // otherwise DHCP snooper will never be populated
-        if ipv4_pkt.next_header() == IpProtocol::Udp {
-            let udp_pkt = UdpPacket::new_checked(ipv4_pkt.payload()).ok()?;
-
-            // Allow DHCP communication with the bootpd(8) on host via broadcast address
-            if udp_pkt.is_dhcp_request() && ipv4_pkt.dst_addr().is_broadcast() {
-                return Some(());
-            }
+        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();
+
+    // Keep the common path cheap and inspect UDP only for a permitted DHCP target
+    if !dst_addr.is_broadcast() && unicast_target != Some(dst_addr) {
+        return false;
+    }
+
+    if ipv4_pkt.next_header() != IpProtocol::Udp {
+        return false;
+    }
+
+    let Ok(udp_pkt) = UdpPacket::new_checked(ipv4_pkt.payload()) else {
+        return false;
+    };
+
+    udp_pkt.is_dhcp_request()
+}
+
 fn vm_arp_allowed(
     arp_pkt: ArpPacket<&[u8]>,
     vm_mac_address: smoltcp::wire::EthernetAddress,
@@ -127,7 +173,7 @@ fn vm_arp_allowed(
     }
 
     if let Some(lease) = lease {
-        if lease.valid_ip_source(source_protocol_addr) {
+        if lease.is_valid_for(source_protocol_addr) {
             return Some(());
         }
     } else if source_protocol_addr.is_unspecified() {
@@ -141,11 +187,23 @@ fn vm_arp_allowed(
 mod tests {
     use crate::dhcp_snooper::Lease;
     use smoltcp::wire::{
-        ArpHardware, ArpOperation, ArpPacket, EthernetAddress, EthernetProtocol, Ipv4Address,
+        ArpHardware, ArpOperation, ArpPacket, EthernetAddress, EthernetProtocol, IpProtocol,
+        Ipv4Address, Ipv4Packet, UdpPacket,
     };
     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() {
         let vm_mac_address = EthernetAddress([0x02, 0x00, 0x00, 0x00, 0x00, 0x01]);
@@ -244,4 +302,22 @@ 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)
+    }
 }