Add dynamic Softnet policy control (#181)

* Add dynamic Softnet policy control

* Simplify Softnet policy RPC methods

* Use jsonrpsee types for Softnet policy RPC

* Reject superseded Softnet policy revisions

* Apply Softnet policy updates only after enqueue

* Flush final Softnet response after input EOF

* Signal EOF when disabling Softnet control

* Normalize Softnet default-route isolation

* Keep policy rules on Proxy (#182)

* Simplify Softnet policy protocol

* Bound pipelined Softnet policy responses

* Keep Softnet port forwarding active on control wakes

---------

Co-authored-by: edi-oai <edi@openai.com>

Cargo.lock

@@ -1414,6 +1414,18 @@ dependencies = [
  "wasm-bindgen",
 ]
 
+[[package]]
+name = "jsonrpsee-types"
+version = "0.26.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bc88ff4688e43cc3fa9883a8a95c6fa27aa2e76c96e610b737b6554d650d7fd5"
+dependencies = [
+ "http 1.1.0",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.12",
+]
+
 [[package]]
 name = "language-tags"
 version = "0.3.2"
@@ -2417,14 +2429,15 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.128"
+version = "1.0.151"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6ff5456707a1de34e7e37f2a6fd3d3f808c318259cbd01ab6377795054b483d8"
+checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
 dependencies = [
  "itoa",
  "memchr",
- "ryu",
  "serde",
+ "serde_core",
+ "zmij",
 ]
 
 [[package]]
@@ -2568,6 +2581,7 @@ dependencies = [
  "dhcproto",
  "ip_network",
  "ipnet",
+ "jsonrpsee-types",
  "libc",
  "log",
  "mac_address",
@@ -2579,6 +2593,8 @@ dependencies = [
  "privdrop",
  "sentry",
  "sentry-anyhow",
+ "serde",
+ "serde_json",
  "serial_test",
  "smoltcp",
  "system-configuration",
@@ -3747,3 +3763,9 @@ dependencies = [
  "quote",
  "syn 2.0.117",
 ]
+
+[[package]]
+name = "zmij"
+version = "1.0.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b"

Cargo.toml

@@ -34,6 +34,9 @@ oslog = "0.2.0"
 log = "0.4.29"
 serial_test = "3"
 coarsetime = "0.1.37"
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+jsonrpsee-types = "0.26"
 
 [profile.release]
 debug = true

README.md

@@ -43,3 +43,18 @@ For proper functioning, Softnet binary requires two things:
 ## Running
 
 Softnet is started and managed automatically by Tart if `--net-softnet` flag is provided when calling `tart run`.
+
+### 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.
+
+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):
+
+```json
+{"jsonrpc":"2.0","id":"42","method":"softnet.policy.set","params":{"allow":["@host","10.0.0.0/8"],"block":["0.0.0.0/0"]}}
+{"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.
+
+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.

lib/poller.rs

@@ -12,6 +12,7 @@ pub struct Poller<'poller> {
     timeout: Duration,
     vm_fd: BorrowedFd<'poller>,
     host_fd: BorrowedFd<'poller>,
+    control_fd: Option<BorrowedFd<'poller>>,
 }
 
 #[derive(IntoPrimitive)]
@@ -19,6 +20,7 @@ pub struct Poller<'poller> {
 enum EventKey {
     VM,
     Host,
+    Control,
     Interrupt,
 }
 
@@ -26,6 +28,7 @@ impl Poller<'_> {
     pub fn new<'poller>(
         vm_fd: RawFd,
         host_fd: RawFd,
+        control_fd: Option<RawFd>,
         timeout: Duration,
     ) -> Result<Poller<'poller>> {
         let poller = polling::Poller::new()?;
@@ -36,6 +39,7 @@ impl Poller<'_> {
             timeout,
             vm_fd: unsafe { BorrowedFd::borrow_raw(vm_fd) },
             host_fd: unsafe { BorrowedFd::borrow_raw(host_fd) },
+            control_fd: control_fd.map(|fd| unsafe { BorrowedFd::borrow_raw(fd) }),
         })
     }
 
@@ -46,6 +50,14 @@ impl Poller<'_> {
                 self.vm_interest(),
                 PollMode::Edge,
             )?;
+
+            if let Some(control_fd) = self.control_fd {
+                self.poller.add_with_mode(
+                    control_fd.as_raw_fd(),
+                    polling::Event::all(EventKey::Control.into()),
+                    PollMode::Edge,
+                )?;
+            }
             self.poller.add_with_mode(
                 self.host_fd.as_raw_fd(),
                 self.host_interest(),
@@ -79,10 +91,17 @@ impl Poller<'_> {
             .events
             .iter()
             .any(|ev| ev.key == Into::<usize>::into(EventKey::Interrupt));
-
         Ok((vm_readable, host_readable, interrupt))
     }
 
+    pub fn remove_control(&mut self) -> Result<()> {
+        if let Some(control_fd) = self.control_fd.take() {
+            self.poller.delete(control_fd)?;
+        }
+
+        Ok(())
+    }
+
     fn vm_interest(&self) -> polling::Event {
         polling::Event::readable(EventKey::VM.into())
     }

lib/proxy/mod.rs

@@ -1,3 +1,4 @@
+mod control;
 mod exposed_port;
 mod host;
 mod port_forwarder;
@@ -10,6 +11,7 @@ use crate::host::NetType;
 use crate::poller::Poller;
 use crate::vm::VM;
 use anyhow::Result;
+use control::Control;
 pub use exposed_port::ExposedPort;
 use ipnet::Ipv4Net;
 use mac_address::MacAddress;
@@ -29,6 +31,7 @@ pub struct Proxy<'proxy> {
     vm_mac_address: smoltcp::wire::EthernetAddress,
     dhcp_snooper: DhcpSnooper,
     rules: PrefixMap<Ipv4Net, Action>,
+    control: Option<Control>,
     enobufs_encountered: bool,
     port_forwarder: PortForwarder,
 }
@@ -65,6 +68,7 @@ impl Proxy<'_> {
         allow: Vec<Target>,
         block: Vec<Target>,
         exposed_ports: Vec<ExposedPort>,
+        control_fd: Option<RawFd>,
     ) -> Result<Proxy<'proxy>> {
         let vm = VM::new(vm_fd)?;
         let host = Host::new(
@@ -72,7 +76,17 @@ impl Proxy<'_> {
             !allow.contains(&Target::Prefix(Ipv4Net::zero())),
         )?;
         let poller_timeout = Duration::from_millis(100);
-        let poller = Poller::new(vm.as_raw_fd(), host.as_raw_fd(), poller_timeout)?;
+        let control = control_fd
+            .map(|control_fd| {
+                Control::new(control_fd, host.gateway_ip, allow.clone(), block.clone())
+            })
+            .transpose()?;
+        let poller = Poller::new(
+            vm.as_raw_fd(),
+            host.as_raw_fd(),
+            control.as_ref().map(AsRawFd::as_raw_fd),
+            poller_timeout,
+        )?;
 
         // Craft packet filter rules
         //
@@ -105,6 +119,7 @@ impl Proxy<'_> {
             vm_mac_address: smoltcp::wire::EthernetAddress(vm_mac_address.bytes()),
             dhcp_snooper: DhcpSnooper::new(poller_timeout),
             rules,
+            control,
             enobufs_encountered: false,
             port_forwarder: PortForwarder::new(exposed_ports),
         })
@@ -129,6 +144,10 @@ impl Proxy<'_> {
             // Update coarse time for the DHCP snooper
             coarsetime::Instant::update();
 
+            // Service control on every wake (including timeouts) so a bounded read or a pending
+            // response continues making progress even when no new edge is generated.
+            self.service_control();
+
             if vm_readable {
                 self.read_from_vm(buf.as_mut_slice())?;
             }
@@ -153,6 +172,8 @@ impl Proxy<'_> {
     }
 
     fn read_from_vm(&mut self, buf: &mut [u8]) -> Result<()> {
+        let mut packets_read = 0;
+
         loop {
             match self.vm.read(buf) {
                 Ok(n) => {
@@ -162,6 +183,12 @@ impl Proxy<'_> {
                     if let Ok(frame) = EthernetFrame::new_checked(&buf[..n]) {
                         self.process_frame_from_vm(frame)?;
                     }
+
+                    packets_read += 1;
+                    if packets_read == 128 {
+                        self.service_control();
+                        packets_read = 0;
+                    }
                 }
                 Err(err) => {
                     if err.kind() == ErrorKind::WouldBlock {
@@ -186,6 +213,8 @@ impl Proxy<'_> {
                             self.process_frame_from_host(&pkt)?;
                         }
                     }
+
+                    self.service_control();
                 }
                 Err(err) => {
                     if let vmnet::Error::VmnetReadNothing = err {
@@ -197,6 +226,34 @@ impl Proxy<'_> {
             }
         }
     }
+
+    fn service_control(&mut self) {
+        let Some(control) = self.control.as_mut() else {
+            return;
+        };
+
+        let keep_open = match control.service(&mut self.rules) {
+            Ok(keep_open) => keep_open,
+            Err(err) => {
+                log::warn!("disabling Softnet control socket: {err:#}");
+                false
+            }
+        };
+
+        if keep_open {
+            return;
+        }
+
+        if let Err(err) = self.poller.remove_control() {
+            log::warn!("failed to remove Softnet control socket from the poller: {err:#}");
+        }
+
+        if let Some(control) = self.control.take()
+            && let Err(err) = control.shutdown()
+        {
+            log::warn!("failed to shut down Softnet control socket: {err:#}");
+        }
+    }
 }
 
 #[cfg(test)]
@@ -295,6 +352,7 @@ mod tests {
                 .map(|cidr| cidr.parse().unwrap())
                 .collect(),
             Vec::default(),
+            None,
         )
         .unwrap();
 

lib/proxy/udp_packet_helper.rs

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

lib/proxy/vm.rs

@@ -1,12 +1,13 @@
+use crate::dhcp_snooper::Lease;
 use crate::proxy::udp_packet_helper::UdpPacketHelper;
 use crate::proxy::{Action, Proxy};
 use anyhow::Context;
 use anyhow::Result;
 use ipnet::Ipv4Net;
 use smoltcp::wire::{
-    ArpPacket, EthernetFrame, EthernetProtocol, IpProtocol, Ipv4Packet, UdpPacket,
+    ArpOperation, ArpPacket, ArpRepr, EthernetFrame, EthernetProtocol, IpProtocol, Ipv4Packet,
+    UdpPacket,
 };
-use std::net::Ipv4Addr;
 
 impl Proxy<'_> {
     pub(crate) fn process_frame_from_vm(&mut self, frame: EthernetFrame<&[u8]>) -> Result<()> {
@@ -40,22 +41,7 @@ impl Proxy<'_> {
     }
 
     fn allowed_from_vm_arp(&self, arp_pkt: ArpPacket<&[u8]>) -> Option<()> {
-        if arp_pkt.source_hardware_addr() != self.vm_mac_address.0 {
-            return None;
-        }
-
-        let source_protocol_addr: [u8; 4] = arp_pkt.source_protocol_addr().try_into().unwrap();
-        let source_protocol_addr = Ipv4Addr::from(source_protocol_addr);
-
-        if let Some(lease) = self.dhcp_snooper.lease() {
-            if lease.valid_ip_source(source_protocol_addr) {
-                return Some(());
-            }
-        } else if source_protocol_addr.is_unspecified() {
-            return Some(());
-        }
-
-        None
+        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<()> {
@@ -115,3 +101,147 @@ impl Proxy<'_> {
         None
     }
 }
+
+fn vm_arp_allowed(
+    arp_pkt: ArpPacket<&[u8]>,
+    vm_mac_address: smoltcp::wire::EthernetAddress,
+    lease: &Option<Lease>,
+) -> Option<()> {
+    let (operation, source_hardware_addr, source_protocol_addr) =
+        match ArpRepr::parse(&arp_pkt).ok()? {
+            ArpRepr::EthernetIpv4 {
+                operation,
+                source_hardware_addr,
+                source_protocol_addr,
+                ..
+            } => (operation, source_hardware_addr, source_protocol_addr),
+            _ => return None,
+        };
+
+    if !matches!(operation, ArpOperation::Request | ArpOperation::Reply) {
+        return None;
+    }
+
+    if source_hardware_addr != vm_mac_address {
+        return None;
+    }
+
+    if let Some(lease) = lease {
+        if lease.valid_ip_source(source_protocol_addr) {
+            return Some(());
+        }
+    } else if source_protocol_addr.is_unspecified() {
+        return Some(());
+    }
+
+    None
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::dhcp_snooper::Lease;
+    use smoltcp::wire::{
+        ArpHardware, ArpOperation, ArpPacket, EthernetAddress, EthernetProtocol, Ipv4Address,
+    };
+    use std::collections::HashSet;
+    use std::time::Duration;
+
+    #[test]
+    fn test_allowed_from_vm_arp_allows_unspecified_request_without_lease() {
+        let vm_mac_address = EthernetAddress([0x02, 0x00, 0x00, 0x00, 0x00, 0x01]);
+        let buf = arp_packet(vm_mac_address.0, [0, 0, 0, 0], ArpOperation::Request, 6, 4);
+        let arp_pkt = ArpPacket::new_checked(buf.as_slice()).unwrap();
+
+        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, &None).is_some());
+    }
+
+    #[test]
+    fn test_allowed_from_vm_arp_allows_reply_for_leased_ip() {
+        let vm_mac_address = EthernetAddress([0x02, 0x00, 0x00, 0x00, 0x00, 0x01]);
+        let lease_ip = Ipv4Address::new(192, 168, 0, 2);
+        let lease = Some(Lease::new(
+            lease_ip,
+            Duration::from_secs(600),
+            HashSet::new(),
+        ));
+        let buf = arp_packet(
+            vm_mac_address.0,
+            lease_ip.octets(),
+            ArpOperation::Reply,
+            6,
+            4,
+        );
+        let arp_pkt = ArpPacket::new_checked(buf.as_slice()).unwrap();
+
+        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, &lease).is_some());
+    }
+
+    #[test]
+    fn test_allowed_from_vm_arp_rejects_unknown_operation() {
+        let vm_mac_address = EthernetAddress([0x02, 0x00, 0x00, 0x00, 0x00, 0x01]);
+        let buf = arp_packet(
+            vm_mac_address.0,
+            [0, 0, 0, 0],
+            ArpOperation::Unknown(3),
+            6,
+            4,
+        );
+        let arp_pkt = ArpPacket::new_checked(buf.as_slice()).unwrap();
+
+        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, &None).is_none());
+    }
+
+    #[test]
+    fn test_allowed_from_vm_arp_rejects_non_ethernet_hardware_type() {
+        let vm_mac_address = EthernetAddress([0x02, 0x00, 0x00, 0x00, 0x00, 0x01]);
+        let mut buf = arp_packet(vm_mac_address.0, [0, 0, 0, 0], ArpOperation::Request, 6, 4);
+        let mut arp_pkt = ArpPacket::new_unchecked(buf.as_mut_slice());
+        arp_pkt.set_hardware_type(ArpHardware::Unknown(2));
+        let arp_pkt = ArpPacket::new_checked(buf.as_slice()).unwrap();
+
+        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, &None).is_none());
+    }
+
+    #[test]
+    fn test_allowed_from_vm_arp_rejects_non_ipv4_protocol_type() {
+        let vm_mac_address = EthernetAddress([0x02, 0x00, 0x00, 0x00, 0x00, 0x01]);
+        let mut buf = arp_packet(vm_mac_address.0, [0, 0, 0, 0], ArpOperation::Request, 6, 4);
+        let mut arp_pkt = ArpPacket::new_unchecked(buf.as_mut_slice());
+        arp_pkt.set_protocol_type(EthernetProtocol::Ipv6);
+        let arp_pkt = ArpPacket::new_checked(buf.as_slice()).unwrap();
+
+        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, &None).is_none());
+    }
+
+    #[test]
+    fn test_allowed_from_vm_arp_rejects_non_ipv4_protocol_length() {
+        let vm_mac_address = EthernetAddress([0x02, 0x00, 0x00, 0x00, 0x00, 0x01]);
+        let buf = arp_packet(vm_mac_address.0, [0, 0, 0], ArpOperation::Request, 6, 3);
+        let arp_pkt = ArpPacket::new_checked(buf.as_slice()).unwrap();
+
+        assert!(super::vm_arp_allowed(arp_pkt, vm_mac_address, &None).is_none());
+    }
+
+    fn arp_packet(
+        source_hardware_addr: [u8; 6],
+        source_protocol_addr: impl AsRef<[u8]>,
+        operation: ArpOperation,
+        hardware_len: u8,
+        protocol_len: u8,
+    ) -> Vec<u8> {
+        let source_protocol_addr = source_protocol_addr.as_ref();
+        let payload_len = 8 + 2 * (hardware_len as usize + protocol_len as usize);
+        let mut buf = vec![0; payload_len];
+        let mut arp_pkt = ArpPacket::new_unchecked(buf.as_mut_slice());
+        arp_pkt.set_hardware_type(ArpHardware::Ethernet);
+        arp_pkt.set_protocol_type(EthernetProtocol::Ipv4);
+        arp_pkt.set_hardware_len(hardware_len);
+        arp_pkt.set_protocol_len(protocol_len);
+        arp_pkt.set_operation(operation);
+        arp_pkt.set_source_hardware_addr(&source_hardware_addr[..hardware_len as usize]);
+        arp_pkt.set_source_protocol_addr(source_protocol_addr);
+        arp_pkt.set_target_hardware_addr(&[0; 6][..hardware_len as usize]);
+        arp_pkt.set_target_protocol_addr(&vec![0; protocol_len as usize]);
+        buf
+    }
+}

lib/vm.rs

@@ -1,5 +1,7 @@
-use anyhow::Result;
-use std::os::unix::io::{AsRawFd, FromRawFd, RawFd};
+use anyhow::{Context, Result, bail};
+use std::io;
+use std::mem::{size_of, zeroed};
+use std::os::fd::{AsRawFd, FromRawFd, RawFd};
 use std::os::unix::net::UnixDatagram;
 
 pub struct VM {
@@ -8,6 +10,9 @@ pub struct VM {
 
 impl VM {
     pub fn new(vm_fd: RawFd) -> Result<VM> {
+        let vm_fd = duplicate_vm_fd(vm_fd)?;
+
+        // SAFETY: duplicate_vm_fd only returns a valid descriptor that it owns.
         let sock = unsafe { UnixDatagram::from_raw_fd(vm_fd) };
         sock.set_nonblocking(true)?;
 
@@ -23,8 +28,163 @@ impl VM {
     }
 }
 
+fn duplicate_vm_fd(vm_fd: RawFd) -> Result<RawFd> {
+    if vm_fd < 0 {
+        bail!("invalid VM file descriptor {vm_fd}: value must be non-negative");
+    }
+
+    // SAFETY: fcntl duplicates the descriptor without transferring ownership of vm_fd.
+    let duplicated_fd = unsafe { libc::fcntl(vm_fd, libc::F_DUPFD_CLOEXEC, 0) };
+    if duplicated_fd == -1 {
+        return Err(io::Error::last_os_error())
+            .with_context(|| format!("failed to duplicate VM file descriptor {vm_fd}"));
+    }
+
+    if let Err(error) = validate_vm_fd(duplicated_fd) {
+        // SAFETY: duplicated_fd is an open descriptor owned by this function.
+        unsafe { libc::close(duplicated_fd) };
+        return Err(error);
+    }
+
+    Ok(duplicated_fd)
+}
+
+fn validate_vm_fd(vm_fd: RawFd) -> Result<()> {
+    // SAFETY: fcntl only reads descriptor state and does not take ownership.
+    if unsafe { libc::fcntl(vm_fd, libc::F_GETFD) } == -1 {
+        return Err(io::Error::last_os_error())
+            .with_context(|| format!("failed to inspect VM file descriptor {vm_fd}"));
+    }
+
+    let mut socket_type = 0;
+    let mut socket_type_len = size_of::<libc::c_int>() as libc::socklen_t;
+
+    // SAFETY: socket_type and socket_type_len are valid writable buffers of the sizes given.
+    if unsafe {
+        libc::getsockopt(
+            vm_fd,
+            libc::SOL_SOCKET,
+            libc::SO_TYPE,
+            (&mut socket_type as *mut libc::c_int).cast(),
+            &mut socket_type_len,
+        )
+    } == -1
+    {
+        return Err(io::Error::last_os_error())
+            .with_context(|| format!("VM file descriptor {vm_fd} is not a socket"));
+    }
+
+    if socket_type != libc::SOCK_DGRAM {
+        bail!("VM file descriptor {vm_fd} is not a Unix datagram socket");
+    }
+
+    let mut address: libc::sockaddr_storage = unsafe { zeroed() };
+    let mut address_len = size_of::<libc::sockaddr_storage>() as libc::socklen_t;
+
+    // SAFETY: address and address_len are valid writable buffers of the sizes given.
+    if unsafe {
+        libc::getsockname(
+            vm_fd,
+            (&mut address as *mut libc::sockaddr_storage).cast(),
+            &mut address_len,
+        )
+    } == -1
+    {
+        return Err(io::Error::last_os_error()).with_context(|| {
+            format!("failed to inspect the address family of VM file descriptor {vm_fd}")
+        });
+    }
+
+    // macOS returns a zero-length address for unnamed UNIX-domain sockets,
+    // including socketpair descriptors. Other socket families return their
+    // address family when getsockname succeeds.
+    let is_unix_socket = address_len == 0 || address.ss_family as libc::c_int == libc::AF_UNIX;
+    if !is_unix_socket {
+        bail!("VM file descriptor {vm_fd} is not a Unix socket");
+    }
+
+    Ok(())
+}
+
 impl AsRawFd for VM {
     fn as_raw_fd(&self) -> RawFd {
         self.sock.as_raw_fd()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::VM;
+    use std::fs::File;
+    use std::net::UdpSocket;
+    use std::os::fd::AsRawFd;
+    use std::os::unix::net::{UnixDatagram, UnixStream};
+
+    #[test]
+    fn test_new_rejects_negative_fd() {
+        let error = VM::new(-1).err().unwrap();
+
+        assert_eq!(
+            error.to_string(),
+            "invalid VM file descriptor -1: value must be non-negative"
+        );
+    }
+
+    #[test]
+    fn test_new_rejects_non_socket_fd_without_taking_ownership() {
+        let file = File::open("/dev/null").unwrap();
+        let error = VM::new(file.as_raw_fd()).err().unwrap();
+
+        assert!(error.to_string().contains("is not a socket"));
+        assert!(file.metadata().is_ok());
+    }
+
+    #[test]
+    fn test_new_rejects_closed_fd() {
+        let (socket, _peer) = UnixDatagram::pair().unwrap();
+        let vm_fd = socket.as_raw_fd();
+        drop(socket);
+
+        let error = VM::new(vm_fd).err().unwrap();
+
+        assert!(
+            error
+                .to_string()
+                .contains("failed to duplicate VM file descriptor")
+        );
+    }
+
+    #[test]
+    fn test_new_rejects_non_datagram_socket() {
+        let (stream, _peer) = UnixStream::pair().unwrap();
+        let error = VM::new(stream.as_raw_fd()).err().unwrap();
+
+        assert!(error.to_string().contains("not a Unix datagram socket"));
+    }
+
+    #[test]
+    fn test_new_rejects_internet_datagram_socket() {
+        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+        let error = VM::new(socket.as_raw_fd()).err().unwrap();
+
+        assert!(error.to_string().contains("not a Unix socket"));
+    }
+
+    #[test]
+    fn test_new_does_not_close_original_fd_when_vm_is_dropped() {
+        let (socket, _peer) = UnixDatagram::pair().unwrap();
+        let vm = VM::new(socket.as_raw_fd()).unwrap();
+        drop(vm);
+
+        let socket_fd_is_open = unsafe { libc::fcntl(socket.as_raw_fd(), libc::F_GETFD) != -1 };
+
+        if socket_fd_is_open {
+            drop(socket);
+        } else {
+            // Avoid double-closing the descriptor if this test catches an unsafe implementation.
+            std::mem::forget(socket);
+        }
+
+        assert!(socket_fd_is_open);
+    }
+}

src/main.rs

@@ -26,10 +26,18 @@ use uzers::{get_current_groupname, get_current_username, get_effective_uid};
 struct Args {
     #[clap(
         long,
+        value_parser = parse_vm_fd,
         help = "FD number to use for communicating with the VM's networking stack"
     )]
     vm_fd: c_int,
 
+    #[clap(
+        long,
+        value_parser = parse_vm_fd,
+        help = "connected Unix stream FD for newline-delimited JSON-RPC policy control"
+    )]
+    control_fd: Option<c_int>,
+
     #[clap(long, help = "MAC address to enforce for the VM")]
     vm_mac_address: mac_address::MacAddress,
 
@@ -202,6 +210,7 @@ fn try_main() -> anyhow::Result<()> {
         args.allow,
         args.block,
         args.expose,
+        args.control_fd.map(|fd| fd as RawFd),
     )
     .context("failed to initialize proxy")?;
 
@@ -217,6 +226,18 @@ fn try_main() -> anyhow::Result<()> {
     proxy.run()
 }
 
+fn parse_vm_fd(value: &str) -> Result<c_int, String> {
+    let vm_fd = value
+        .parse::<c_int>()
+        .map_err(|err| format!("invalid file descriptor: {err}"))?;
+
+    if vm_fd < 0 {
+        return Err("file descriptor must be non-negative".to_string());
+    }
+
+    Ok(vm_fd)
+}
+
 fn sudo_escalation_works() -> bool {
     let exe = std::env::current_exe().unwrap();
     let args = std::env::args().skip(1);
@@ -252,3 +273,42 @@ fn set_bootpd_lease_time(lease_time: u32) {
         SCPreferencesCommitChanges(prefs.as_concrete_TypeRef());
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::Args;
+    use clap::Parser;
+
+    #[test]
+    fn test_cli_rejects_negative_vm_fd_before_startup() {
+        let error = Args::try_parse_from([
+            "softnet",
+            "--vm-fd=-1",
+            "--vm-mac-address=02:00:00:00:00:01",
+        ])
+        .unwrap_err();
+
+        assert!(
+            error
+                .to_string()
+                .contains("file descriptor must be non-negative")
+        );
+    }
+
+    #[test]
+    fn test_cli_rejects_negative_control_fd_before_startup() {
+        let error = Args::try_parse_from([
+            "softnet",
+            "--vm-fd=0",
+            "--control-fd=-1",
+            "--vm-mac-address=02:00:00:00:00:01",
+        ])
+        .unwrap_err();
+
+        assert!(
+            error
+                .to_string()
+                .contains("file descriptor must be non-negative")
+        );
+    }
+}