Introduce stateful "in TARGET" and "out TARGET" rules (#188)

lib/proxy/flows/icmp.rs

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

lib/proxy/flows/mod.rs

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

lib/proxy/flows/tcp.rs

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

lib/proxy/flows/udp.rs

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