WIP

README.md

@@ -46,7 +46,7 @@ Softnet is started and managed automatically by Tart if `--net-softnet` flag is
 
 ### Dynamic network policy
 
-Softnet can update the running VM's IPv4 egress policy without restarting the VM. Pass a connected Unix stream socket as `--control-fd` to enable a newline-delimited [JSON-RPC 2.0](https://www.jsonrpc.org/specification) control channel. The socket is duplex and must be separate from `--vm-fd`, which carries VM packets.
+Softnet can update the running VM's IPv4 policy without restarting the VM. Pass a connected Unix stream socket as `--control-fd` to enable a newline-delimited [JSON-RPC 2.0](https://www.jsonrpc.org/specification) control channel. The socket is duplex and must be separate from `--vm-fd`, which carries VM packets.
 
 The supported methods are `softnet.policy.get` and `softnet.policy.set`. A complete policy update looks like this (each request and response occupies one line):
 
@@ -55,6 +55,8 @@ The supported methods are `softnet.policy.get` and `softnet.policy.set`. A compl
 {"jsonrpc":"2.0","id":"42","result":{"allow":["10.0.0.0/8","@host"],"block":["0.0.0.0/0"],"ruleCount":3}}
 ```
 
-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 targets are parsed and a new prefix map is built before the active policy changes. Longest-prefix matching and block precedence for identical prefixes are preserved. Targets are normalized and deduplicated. A policy update may contain at most 4096 combined allow/block targets, and a request frame may not exceed 1 MiB.
+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.
 
-Use `block=["0.0.0.0/0"]` with specific allow targets for a default-deny policy. Closing the control socket leaves the last accepted policy active.
+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.
+
+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.

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>,
+    allow: Vec<Rule>,
-    block: Vec<Target>,
+    block: Vec<Rule>,
     gateway_ip: Ipv4Address,
 }
 
 struct PolicyUpdate {
-    rules: PrefixMap<Ipv4Net, Action>,
+    rules: Rules,
-    allow: Vec<Target>,
+    allow: Vec<Rule>,
-    block: Vec<Target>,
+    block: Vec<Rule>,
 }
 
 impl Policy {
-    pub(super) fn new(gateway_ip: Ipv4Address, allow: Vec<Target>, block: Vec<Target>) -> Self {
+    pub(super) fn new(gateway_ip: Ipv4Address, allow: Vec<Rule>, block: Vec<Rule>) -> Self {
-        let allow = normalize_targets(allow);
+        let allow = normalize_rules(allow);
-        let block = normalize_targets(block);
+        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 allow = parse_rules(allow)?;
-        let block = parse_targets(block)?;
+        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<_>>(),
+        "allow": allow.iter().map(ToString::to_string).collect::<Vec<_>>(),
-        "block": block.iter().map(target_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> {
+fn parse_rules(rules: Vec<String>) -> std::result::Result<Vec<Rule>, ErrorObjectOwned> {
-    let mut parsed = Vec::with_capacity(targets.len());
+    let mut parsed = Vec::with_capacity(rules.len());
 
-    for target in targets {
+    for rule in rules {
-        let parsed_target = target.parse().map_err(|_| {
+        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> {
+fn normalize_rules(mut rules: Vec<Rule>) -> Vec<Rule> {
-    let mut targets = targets
+    rules.iter_mut().for_each(|rule| *rule = rule.normalized());
-        .into_iter()
+    rules.sort_by_key(ToString::to_string);
-        .map(|target| match target {
+    rules.dedup();
-            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);
-    }
-
     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>,
+        allow: Vec<Rule>,
-        block: Vec<Target>,
+        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 {
+            if let Some(update) = update
-                *rules = self.policy.apply(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> {
+    fn rules(rules: &[&str]) -> Vec<Rule> {
-        targets
+        rules.iter().map(|rule| rule.parse().unwrap()).collect()
-            .iter()
-            .map(|target| target.parse().unwrap())
-            .collect()
     }
 
     fn policy(allow: &[&str], block: &[&str]) -> TestPolicy {
         let gateway_ip = Ipv4Address::new(192, 168, 64, 1);
-        let allow = targets(allow);
+        let allow = rules(allow);
-        let block = targets(block);
+        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);
+        let (response, update) = handle_request(&policy.state, rule_count(&policy.rules), line);
-        if let Some(update) = update {
+        if let Some(update) = update
-            policy.rules = policy.state.apply(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 {
+        if !self.is_allowed_dhcp_response(&ipv4_pkt) {
-            return;
-        }
-
-        if ipv4_pkt.next_header() != smoltcp::wire::IpProtocol::Udp {
             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>,
+        allow: Vec<Rule>,
-        block: Vec<Target>,
+        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
+        let rules = build_rules(host.gateway_ip, &allow, &block);
-        //
+        let stateful_policy = has_stateful_rules(&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);
-        }
 
         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![
+            PrefixMap::<Ipv4Net, Vec<(Action, Rule)>>::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),
+                    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(&mut 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(&mut 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.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,
+    ArpOperation, ArpPacket, ArpRepr, EthernetFrame, EthernetProtocol, IpProtocol, Ipv4Address,
-    UdpPacket,
+    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() {
+            if let Some(rules) = select_rules(&self.rules, dst_addr, Direction::Out) {
-                let dst_net = Ipv4Net::from(dst_addr);
+                // 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 {
+        if is_allowed_dhcp_request(&ipv4_pkt, None) {
-            let udp_pkt = UdpPacket::new_checked(ipv4_pkt.payload()).ok()?;
+            return Some(());
-
-            // 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(());
-            }
         }
 
         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)
+    }
 }

src/main.rs

@@ -7,7 +7,7 @@ use privdrop::PrivDrop;
 use softnet::NetType;
 use softnet::proxy::ExposedPort;
 use softnet::proxy::Proxy;
-use softnet::proxy::Target;
+use softnet::proxy::Rule;
 use std::borrow::Cow;
 use std::env;
 use std::os::raw::c_int;
@@ -59,34 +59,51 @@ struct Args {
 
     #[clap(
         long,
-        help = "Comma-separated list of CIDRs to allow the traffic to \
+        help = "Comma-separated list of rules for allowing traffic.\n\n\
-        (e.g. --allow=192.168.0.0/24 may be used to allow a LAN access for a VM), \
+        Rule forms:\n\n\
-        plus supported @-aliases. Currently the only supported @-alias is @host, \
+        * TARGET: traffic between TARGET and the VM in either direction\n\
-        which matches the vmnet bridge gateway IP. \
+        * in TARGET: flows initiated from TARGET to the VM\n\
-        When used with --block, the longest prefix match always wins. \
+        * out TARGET: flows initiated from the VM to TARGET\n\n\
-        In case an identical prefix is both --allow'ed and --block'ed, \
+        Targets are:\n\n\
-        blocking will take precedence. --allow=0.0.0.0/0 is a special case, \
+        * IPv4 CIDRs\n\
-        it additionally disables bridge isolation (even when --block=0.0.0.0/0 is specified).",
+        * @host, which matches the vmnet bridge gateway IP\n\n\
-        value_name = "comma-separated CIDRs or @-alias",
+        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 \
+        --block=0.0.0.0/0 is specified.\n\n\
+        Examples:\n\n\
+        * --allow=192.168.0.0/24 — allow stateless traffic with this LAN\n\
+        * --allow=\"in @host\" — allow stateful flows initiated from @host\n\
+        * --allow=\"out 192.168.0.0/24\" — allow stateful flows initiated toward this LAN\n\
+        * --allow=\"in @host,out 192.168.0.0/24\" — multiple rules may be comma-separated",
+        value_name = "comma-separated rules",
         use_value_delimiter = true,
         action = clap::ArgAction::Set
     )]
-    allow: Vec<Target>,
+    allow: Vec<Rule>,
 
     #[clap(
         long,
-        help = "Comma-separated list of CIDRs to block the traffic to \
+        help = "Comma-separated list of rules for blocking traffic.\n\n\
-        (e.g. --block=0.0.0.0/0 may be used to establish a default deny policy \
+        Rule forms:\n\n\
-        that is further relaxed with --allow), plus supported @-aliases. \
+        * TARGET: traffic between TARGET and the VM in either direction\n\
-        Currently the only supported @-alias is @host, which matches the vmnet bridge gateway IP. \
+        * in TARGET: flows initiated from TARGET to the VM\n\
-        When used with --allow, the longest prefix match always wins. \
+        * out TARGET: flows initiated from the VM to TARGET\n\n\
-        In case an identical prefix is both --allow'ed and --block'ed, \
+        Targets are:\n\n\
-        blocking will take precedence.",
+        * IPv4 CIDRs\n\
-        value_name = "comma-separated CIDRs or @-alias",
+        * @host, which matches the vmnet bridge gateway IP\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=\"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",
+        value_name = "comma-separated rules",
         use_value_delimiter = true,
         action = clap::ArgAction::Set
     )]
-    block: Vec<Target>,
+    block: Vec<Rule>,
 
     #[clap(
         long,