diff --git a/Cargo.lock b/Cargo.lock index e082d89..5194fe4 100644 --- a/Cargo.lock +++ b/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" diff --git a/Cargo.toml b/Cargo.toml index 00d701b..cf551bf 100644 --- a/Cargo.toml +++ b/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 diff --git a/README.md b/README.md index 529ff96..8a4e708 100644 --- a/README.md +++ b/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. diff --git a/lib/poller.rs b/lib/poller.rs index 27e38f4..0785689 100644 --- a/lib/poller.rs +++ b/lib/poller.rs @@ -12,6 +12,7 @@ pub struct Poller<'poller> { timeout: Duration, vm_fd: BorrowedFd<'poller>, host_fd: BorrowedFd<'poller>, + control_fd: Option>, } #[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, timeout: Duration, ) -> Result> { 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::::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()) } diff --git a/lib/proxy/control.rs b/lib/proxy/control.rs new file mode 100644 index 0000000..56173a5 --- /dev/null +++ b/lib/proxy/control.rs @@ -0,0 +1,1132 @@ +use super::{Action, Target}; +use anyhow::{Context, Result, bail}; +use ipnet::Ipv4Net; +use jsonrpsee_types::{ + ErrorObjectOwned, Id, Request, Response, ResponsePayload, + error::{ + INVALID_PARAMS_CODE as INVALID_PARAMS, INVALID_REQUEST_CODE as INVALID_REQUEST, + 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; +use std::io::{self, ErrorKind, Read, Write}; +use std::mem::{size_of, zeroed}; +use std::net::Shutdown; +use std::os::fd::{AsRawFd, FromRawFd, RawFd}; +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_IDENTIFIER_BYTES: usize = 256; +const MAX_SERVICE_BYTES: usize = MAX_REQUEST_BYTES; + +pub(super) struct Policy { + allow: Vec, + block: Vec, + gateway_ip: Ipv4Address, +} + +struct PolicyUpdate { + rules: PrefixMap, + allow: Vec, + block: Vec, +} + +impl Policy { + pub(super) fn new(gateway_ip: Ipv4Address, allow: Vec, block: Vec) -> Self { + let allow = normalize_targets(allow); + let block = normalize_targets(block); + + Policy { + allow, + block, + gateway_ip, + } + } + + fn set( + &self, + allow: Vec, + block: Vec, + ) -> std::result::Result { + if allow.len() + block.len() > MAX_TARGETS { + return Err(rpc_error( + INVALID_PARAMS, + format!("allow and block may contain at most {MAX_TARGETS} targets combined"), + )); + } + + let allow = parse_targets(allow)?; + let block = parse_targets(block)?; + let rules = build_rules(self.gateway_ip, &allow, &block); + + Ok(PolicyUpdate { + rules, + allow, + block, + }) + } + + fn apply(&mut self, update: PolicyUpdate) -> PrefixMap { + // Build and validate everything before updating any active state. The packet filter + // observes either the old PrefixMap or the complete new one. + self.allow = update.allow; + self.block = update.block; + update.rules + } + + fn result(&self, rule_count: usize) -> Value { + policy_result(&self.allow, &self.block, rule_count) + } +} + +impl PolicyUpdate { + fn result(&self) -> Value { + policy_result(&self.allow, &self.block, self.rules.len()) + } +} + +fn policy_result(allow: &[Target], block: &[Target], rule_count: usize) -> Value { + json!({ + "allow": allow.iter().map(target_string).collect::>(), + "block": block.iter().map(target_string).collect::>(), + "ruleCount": rule_count, + }) +} + +fn parse_targets(targets: Vec) -> std::result::Result, ErrorObjectOwned> { + let mut parsed = Vec::with_capacity(targets.len()); + + for target in targets { + let parsed_target = target.parse().map_err(|_| { + rpc_error( + INVALID_PARAMS, + format!("invalid target {target:?}: expected an IPv4 CIDR or @host"), + ) + })?; + parsed.push(parsed_target); + } + + Ok(normalize_targets(parsed)) +} + +fn normalize_targets(targets: Vec) -> Vec { + let mut targets = targets + .into_iter() + .map(|target| match target { + Target::Prefix(prefix) => Target::Prefix(prefix.trunc()), + Target::Host => Target::Host, + }) + .collect::>(); + + 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 { + 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 +} + +pub(super) struct Control { + policy: Policy, + stream: UnixStream, + input: Vec, + output: Vec, + output_offset: usize, + discarding_input: bool, + input_closed: bool, +} + +impl Control { + pub(super) fn new( + control_fd: RawFd, + gateway_ip: Ipv4Address, + allow: Vec, + block: Vec, + ) -> Result { + let control_fd = duplicate_control_fd(control_fd)?; + + // SAFETY: duplicate_control_fd returns an open Unix stream descriptor that it owns. + let stream = unsafe { UnixStream::from_raw_fd(control_fd) }; + stream.set_nonblocking(true)?; + + Ok(Control { + policy: Policy::new(gateway_ip, allow, block), + stream, + input: Vec::new(), + output: Vec::new(), + output_offset: 0, + discarding_input: false, + input_closed: false, + }) + } + + pub(super) fn service(&mut self, rules: &mut PrefixMap) -> Result { + if !self.flush()? { + return Ok(false); + } + + if !self.output.is_empty() { + return Ok(true); + } + + if !self.process_input(rules)? { + return Ok(false); + } + + if !self.output.is_empty() { + return Ok(true); + } + + if self.input_closed { + return Ok(false); + } + + let mut buf = [0; 8192]; + let mut bytes_read = 0; + + while bytes_read < MAX_SERVICE_BYTES { + match self.stream.read(&mut buf) { + Ok(0) => { + self.input_closed = true; + break; + } + Ok(n) => { + bytes_read += n; + self.input.extend_from_slice(&buf[..n]); + if !self.process_input(rules)? { + return Ok(false); + } + + if !self.output.is_empty() { + return Ok(true); + } + } + Err(err) if err.kind() == ErrorKind::WouldBlock => break, + Err(err) + if matches!( + err.kind(), + ErrorKind::BrokenPipe | ErrorKind::ConnectionReset + ) => + { + return Ok(false); + } + Err(err) => return Err(err).context("failed to read the control socket"), + } + } + + Ok(!self.input_closed) + } + + pub(super) fn shutdown(&self) -> Result<()> { + self.stream + .shutdown(Shutdown::Both) + .context("failed to shut down the control socket") + } + + fn process_input(&mut self, rules: &mut PrefixMap) -> Result { + loop { + if self.discarding_input { + if let Some(newline) = self.input.iter().position(|byte| *byte == b'\n') { + self.input.drain(..=newline); + self.discarding_input = false; + continue; + } + + self.input.clear(); + return Ok(true); + } + + let Some(newline) = self.input.iter().position(|byte| *byte == b'\n') else { + if self.input.len() > MAX_REQUEST_BYTES { + self.input.clear(); + self.discarding_input = true; + self.enqueue(error_response( + Id::Null, + PARSE_ERROR, + "request exceeds the maximum frame size", + ))?; + + return self.flush(); + } + + return Ok(true); + }; + + let line = self.input.drain(..=newline).collect::>(); + + if newline > MAX_REQUEST_BYTES { + self.enqueue(error_response( + Id::Null, + PARSE_ERROR, + "request exceeds the maximum frame size", + ))?; + + if !self.flush()? { + return Ok(false); + } + + if !self.output.is_empty() { + return Ok(true); + } + + continue; + } + + let (response, update) = handle_request(&self.policy, rules.len(), &line[..newline]); + self.enqueue(response)?; + + if let Some(update) = update { + *rules = self.policy.apply(update); + } + + if !self.flush()? { + return Ok(false); + } + + if !self.output.is_empty() { + return Ok(true); + } + } + } + + fn enqueue(&mut self, response: Value) -> Result<()> { + if self.output_offset != 0 { + self.output.drain(..self.output_offset); + self.output_offset = 0; + } + + let mut response = + serde_json::to_vec(&response).context("failed to encode RPC response")?; + response.push(b'\n'); + + if self.output.len() + response.len() > MAX_PENDING_RESPONSE_BYTES { + bail!("control response queue exceeded {MAX_PENDING_RESPONSE_BYTES} bytes"); + } + + self.output.extend(response); + Ok(()) + } + + fn flush(&mut self) -> Result { + while self.output_offset < self.output.len() { + match self.stream.write(&self.output[self.output_offset..]) { + Ok(0) => return Ok(false), + Ok(n) => self.output_offset += n, + Err(err) if err.kind() == ErrorKind::WouldBlock => return Ok(true), + Err(err) + if matches!( + err.kind(), + ErrorKind::BrokenPipe | ErrorKind::ConnectionReset + ) => + { + return Ok(false); + } + Err(err) => return Err(err).context("failed to write the control socket"), + } + } + + self.output.clear(); + self.output_offset = 0; + + Ok(true) + } +} + +impl AsRawFd for Control { + fn as_raw_fd(&self) -> RawFd { + self.stream.as_raw_fd() + } +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct SetParams { + allow: Vec, + block: Vec, +} + +fn handle_request( + policy: &Policy, + rule_count: usize, + line: &[u8], +) -> (Value, Option) { + let value = match serde_json::from_slice::(line) { + Ok(value) => value, + Err(_) => { + return ( + error_response(Id::Null, PARSE_ERROR, "invalid JSON-RPC frame"), + None, + ); + } + }; + + let strict_envelope = value.as_object().is_some_and(|object| { + object + .keys() + .all(|key| matches!(key.as_str(), "jsonrpc" | "id" | "method" | "params")) + }); + let response_id = value.get("id").map_or(Id::Null, response_id); + + let request = match serde_json::from_slice::>(line) { + Ok(request) if strict_envelope && valid_id(&request.id) => request, + _ => { + return ( + error_response(response_id, INVALID_REQUEST, "invalid JSON-RPC request"), + None, + ); + } + }; + + let mut update = None; + let result = match request.method_name() { + "softnet.policy.get" => { + if !request.params().parse::().is_ok_and(empty_params) { + Err(rpc_error( + INVALID_PARAMS, + "softnet.policy.get does not accept parameters", + )) + } else { + Ok(policy.result(rule_count)) + } + } + "softnet.policy.set" => { + let params = request.params(); + let params = params.parse::().map_err(|_| { + rpc_error( + INVALID_PARAMS, + "softnet.policy.set requires allow and block", + ) + }); + + params.and_then(|params| { + let next = policy.set(params.allow, params.block)?; + let result = next.result(); + update = Some(next); + Ok(result) + }) + } + _ => Err(rpc_error(METHOD_NOT_FOUND, "method not found")), + }; + + (response(request.id(), result), update) +} + +fn response_id(value: &Value) -> Id<'_> { + match value { + Value::String(value) if value.len() <= MAX_IDENTIFIER_BYTES => Id::Str(value.into()), + Value::Number(value) => value.as_u64().map_or(Id::Null, Id::Number), + _ => Id::Null, + } +} + +fn valid_id(id: &Id<'_>) -> bool { + matches!(id, Id::Number(_)) + || matches!(id, Id::Str(value) if value.len() <= MAX_IDENTIFIER_BYTES) +} + +fn empty_params(value: Value) -> bool { + value.is_null() || value.as_object().is_some_and(|object| object.is_empty()) +} + +fn response(id: Id<'_>, result: std::result::Result) -> Value { + let payload = result.map_or_else(ResponsePayload::error, ResponsePayload::success); + serde_json::to_value(Response::new(payload, id)).expect("JSON-RPC response is serializable") +} + +fn error_response(id: Id<'_>, code: i32, message: impl Into) -> Value { + response(id, Err(rpc_error(code, message))) +} + +fn rpc_error(code: i32, message: impl Into) -> ErrorObjectOwned { + ErrorObjectOwned::owned(code, message, None::<()>) +} + +fn duplicate_control_fd(control_fd: RawFd) -> Result { + if control_fd < 0 { + bail!("invalid control file descriptor {control_fd}: value must be non-negative"); + } + + // SAFETY: fcntl duplicates the descriptor without transferring ownership of control_fd. + let duplicated_fd = unsafe { libc::fcntl(control_fd, libc::F_DUPFD_CLOEXEC, 0) }; + if duplicated_fd == -1 { + return Err(io::Error::last_os_error()) + .with_context(|| format!("failed to duplicate control file descriptor {control_fd}")); + } + + if let Err(error) = validate_control_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_control_fd(control_fd: RawFd) -> Result<()> { + let mut socket_type = 0; + let mut socket_type_len = size_of::() as libc::socklen_t; + + // SAFETY: socket_type and socket_type_len are valid writable buffers of the sizes given. + if unsafe { + libc::getsockopt( + control_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!("control file descriptor {control_fd} is not a socket")); + } + + if socket_type != libc::SOCK_STREAM { + bail!("control file descriptor {control_fd} is not a Unix stream socket"); + } + + let mut address: libc::sockaddr_storage = unsafe { zeroed() }; + let mut address_len = size_of::() as libc::socklen_t; + + // SAFETY: address and address_len are valid writable buffers of the sizes given. + if unsafe { + libc::getsockname( + control_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 control file descriptor {control_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. + if address_len != 0 && address.ss_family as libc::c_int != libc::AF_UNIX { + bail!("control file descriptor {control_fd} is not a Unix socket"); + } + + let mut peer_address: libc::sockaddr_storage = unsafe { zeroed() }; + let mut peer_address_len = size_of::() as libc::socklen_t; + + // SAFETY: peer_address and peer_address_len are valid writable buffers of the sizes given. + if unsafe { + libc::getpeername( + control_fd, + (&mut peer_address as *mut libc::sockaddr_storage).cast(), + &mut peer_address_len, + ) + } == -1 + { + return Err(io::Error::last_os_error()).with_context(|| { + format!("control file descriptor {control_fd} is not a connected Unix stream socket") + }); + } + + Ok(()) +} + +#[cfg(test)] +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, + }; + use crate::proxy::{Action, Target}; + use ipnet::Ipv4Net; + use prefix_trie::PrefixMap; + use serde_json::{Value, json}; + use smoltcp::wire::Ipv4Address; + use std::fs::File; + use std::io::{Read, Write}; + use std::net::{Shutdown, TcpListener}; + use std::os::fd::{AsRawFd, RawFd}; + use std::os::unix::net::{UnixDatagram, UnixStream}; + use std::str::FromStr; + use std::time::Duration; + + struct TestPolicy { + state: Policy, + rules: PrefixMap, + } + + impl TestPolicy { + fn result(&self) -> Value { + self.state.result(self.rules.len()) + } + } + + impl std::ops::Deref for TestPolicy { + type Target = Policy; + + fn deref(&self) -> &Self::Target { + &self.state + } + } + + fn targets(targets: &[&str]) -> Vec { + targets + .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 block = targets(block); + + TestPolicy { + rules: build_rules(gateway_ip, &allow, &block), + state: Policy::new(gateway_ip, allow, block), + } + } + + fn control(control_fd: RawFd) -> anyhow::Result { + Control::new( + control_fd, + Ipv4Address::new(192, 168, 64, 1), + Vec::new(), + Vec::new(), + ) + } + + fn request(policy: &mut TestPolicy, value: Value) -> Value { + raw_request(policy, &serde_json::to_vec(&value).unwrap()) + } + + fn raw_request(policy: &mut TestPolicy, line: &[u8]) -> Value { + let (response, update) = handle_request(&policy.state, policy.rules.len(), line); + if let Some(update) = update { + policy.rules = policy.state.apply(update); + } + + response + } + + #[test] + fn get_reports_initial_policy() { + let mut policy = policy(&["@host"], &["0.0.0.0/0"]); + + let response = request( + &mut policy, + json!({"jsonrpc": "2.0", "id": 1, "method": "softnet.policy.get", "params": {}}), + ); + assert_eq!(response["result"]["allow"], json!(["@host"])); + assert_eq!(response["result"]["block"], json!(["0.0.0.0/0"])); + assert_eq!(response["result"]["ruleCount"], 2); + assert_eq!(response["result"].as_object().unwrap().len(), 3); + } + + #[test] + fn set_applies_complete_policy_and_preserves_block_precedence() { + let mut policy = policy(&[], &[]); + + let response = request( + &mut policy, + json!({ + "jsonrpc": "2.0", + "id": "set", + "method": "softnet.policy.set", + "params": { + "allow": ["@host", "10.0.0.0/8", "10.0.0.0/8"], + "block": ["192.168.64.1/32", "10.0.0.0/8"] + } + }), + ); + + assert_eq!(response["result"]["allow"], json!(["10.0.0.0/8", "@host"])); + assert_eq!( + response["result"]["block"], + json!(["10.0.0.0/8", "192.168.64.1/32"]) + ); + assert_eq!(response["result"]["ruleCount"], 2); + + assert_eq!( + policy.rules.get(&Ipv4Net::from_str("10.0.0.0/8").unwrap()), + Some(&Action::Block) + ); + assert_eq!( + policy + .rules + .get(&Ipv4Net::from_str("192.168.64.1/32").unwrap()), + Some(&Action::Block) + ); + } + + #[test] + fn set_normalizes_targets() { + let mut policy = policy(&[], &[]); + + let first = request( + &mut policy, + json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "softnet.policy.set", + "params": {"allow": ["@host", "10.1.2.3/8"], "block": []} + }), + ); + let retry = request( + &mut policy, + json!({ + "jsonrpc": "2.0", + "id": 2, + "method": "softnet.policy.set", + "params": {"allow": ["10.0.0.0/8", "@host", "@host"], "block": []} + }), + ); + assert_eq!(first["result"], retry["result"]); + assert_eq!(first["result"]["allow"], json!(["10.0.0.0/8", "@host"])); + } + + #[test] + fn invalid_targets_and_limits_leave_policy_unchanged() { + let mut policy = policy(&["@host"], &["0.0.0.0/0"]); + let before = policy.result(); + + let invalid = request( + &mut policy, + json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "softnet.policy.set", + "params": {"allow": ["2001:db8::/32"], "block": []} + }), + ); + assert_eq!(invalid["error"]["code"], INVALID_PARAMS); + assert_eq!(policy.result(), before); + + let targets = vec!["10.0.0.0/8"; MAX_TARGETS + 1]; + let too_many = request( + &mut policy, + json!({ + "jsonrpc": "2.0", + "id": 2, + "method": "softnet.policy.set", + "params": {"allow": targets, "block": []} + }), + ); + assert_eq!(too_many["error"]["code"], INVALID_PARAMS); + assert_eq!(policy.result(), before); + } + + #[test] + fn validates_json_rpc_envelope_method_and_parameters() { + let mut policy = policy(&[], &[]); + + let parse = raw_request(&mut policy, b"not-json"); + assert_eq!(parse["error"]["code"], PARSE_ERROR); + assert!(parse["id"].is_null()); + + let invalid = request( + &mut policy, + json!({"jsonrpc": "1.0", "id": {}, "method": "softnet.policy.get"}), + ); + assert_eq!(invalid["error"]["code"], INVALID_REQUEST); + assert!(invalid["id"].is_null()); + + let method = request( + &mut policy, + json!({"jsonrpc": "2.0", "id": 1, "method": "softnet.policy.patch"}), + ); + assert_eq!(method["error"]["code"], METHOD_NOT_FOUND); + + let params = request( + &mut policy, + json!({"jsonrpc": "2.0", "id": 2, "method": "softnet.policy.get", "params": {"unexpected": true}}), + ); + assert_eq!(params["error"]["code"], INVALID_PARAMS); + + let before = policy.result(); + let missing = request( + &mut policy, + json!({ + "jsonrpc": "2.0", + "id": 3, + "method": "softnet.policy.set", + "params": {"allow": []} + }), + ); + assert_eq!(missing["error"]["code"], INVALID_PARAMS); + assert_eq!(policy.result(), before); + + let missing_id = request( + &mut policy, + json!({ + "jsonrpc": "2.0", + "method": "softnet.policy.set", + "params": {"allow": ["@host"], "block": []} + }), + ); + assert_eq!(missing_id["error"]["code"], INVALID_REQUEST); + assert!(missing_id["id"].is_null()); + assert_eq!(policy.result(), before); + + let null_id = request( + &mut policy, + json!({ + "jsonrpc": "2.0", + "id": null, + "method": "softnet.policy.set", + "params": {"allow": ["@host"], "block": []} + }), + ); + assert_eq!(null_id["error"]["code"], INVALID_REQUEST); + assert!(null_id["id"].is_null()); + assert_eq!(policy.result(), before); + + let negative_id = request( + &mut policy, + json!({"jsonrpc": "2.0", "id": -1, "method": "softnet.policy.get"}), + ); + assert_eq!(negative_id["error"]["code"], INVALID_REQUEST); + assert!(negative_id["id"].is_null()); + + let fractional_id = request( + &mut policy, + json!({"jsonrpc": "2.0", "id": 1.5, "method": "softnet.policy.get"}), + ); + assert_eq!(fractional_id["error"]["code"], INVALID_REQUEST); + assert!(fractional_id["id"].is_null()); + + let oversized_id = request( + &mut policy, + json!({"jsonrpc": "2.0", "id": "x".repeat(257), "method": "softnet.policy.get"}), + ); + assert_eq!(oversized_id["error"]["code"], INVALID_REQUEST); + assert!(oversized_id["id"].is_null()); + + let maximum_id = request( + &mut policy, + json!({"jsonrpc": "2.0", "id": u64::MAX, "method": "softnet.policy.get"}), + ); + assert_eq!(maximum_id["id"], u64::MAX); + assert!(maximum_id.get("result").is_some()); + + let unexpected_field = request( + &mut policy, + json!({"jsonrpc": "2.0", "id": 4, "method": "softnet.policy.get", "unexpected": true}), + ); + assert_eq!(unexpected_field["error"]["code"], INVALID_REQUEST); + assert_eq!(unexpected_field["id"], 4); + + let duplicate_field = raw_request( + &mut policy, + br#"{"jsonrpc":"2.0","id":6,"method":"softnet.policy.get","method":"softnet.policy.set"}"#, + ); + assert_eq!(duplicate_field["error"]["code"], INVALID_REQUEST); + assert_eq!(duplicate_field["id"], 6); + + let unexpected_param = request( + &mut policy, + json!({ + "jsonrpc": "2.0", + "id": 5, + "method": "softnet.policy.set", + "params": {"allow": [], "block": [], "unexpected": true} + }), + ); + assert_eq!(unexpected_param["error"]["code"], INVALID_PARAMS); + assert_eq!(policy.result(), before); + } + + #[test] + fn newline_delimited_control_socket_handles_multiple_requests_and_eof() { + let (mut client, server) = UnixStream::pair().unwrap(); + client + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + let mut control = control(server.as_raw_fd()).unwrap(); + let mut rules = PrefixMap::new(); + + client + .write_all( + concat!( + "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.get\"}\n", + "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"@host\"],\"block\":[\"0.0.0.0/0\"]}}\n" + ) + .as_bytes(), + ) + .unwrap(); + + assert!(control.service(&mut rules).unwrap()); + let mut response = [0; 2048]; + let n = client.read(&mut response).unwrap(); + let lines = std::str::from_utf8(&response[..n]) + .unwrap() + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .collect::>(); + assert_eq!(lines.len(), 2); + assert_eq!(lines[0]["id"], 1); + assert_eq!(lines[1]["id"], 2); + assert_eq!(lines[1]["result"]["allow"], json!(["@host"])); + assert_eq!(control.policy.allow, vec![Target::Host]); + + let before = control.policy.result(rules.len()); + drop(client); + assert!(!control.service(&mut rules).unwrap()); + assert_eq!(control.policy.result(rules.len()), before); + } + + #[test] + fn write_side_eof_flushes_the_final_policy_response() { + let (mut client, server) = UnixStream::pair().unwrap(); + client + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + let mut control = control(server.as_raw_fd()).unwrap(); + let mut rules = PrefixMap::new(); + + client + .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"10.0.0.0/8\"],\"block\":[]}}\n") + .unwrap(); + client.shutdown(Shutdown::Write).unwrap(); + + assert!(!control.service(&mut rules).unwrap()); + + let mut response = [0; 1024]; + let n = client.read(&mut response).unwrap(); + let response = serde_json::from_slice::(&response[..n - 1]).unwrap(); + assert_eq!(response["id"], 1); + assert_eq!(response["result"]["allow"], json!(["10.0.0.0/8"])); + assert!(control.output.is_empty()); + } + + #[test] + fn oversized_frame_is_discarded_and_following_frame_is_processed() { + let (mut client, server) = UnixStream::pair().unwrap(); + client + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + let mut control = control(server.as_raw_fd()).unwrap(); + let mut rules = PrefixMap::new(); + + control.input = vec![b'x'; MAX_REQUEST_BYTES + 1]; + control.process_input(&mut rules).unwrap(); + assert!(control.discarding_input); + + control.input.extend_from_slice( + b"still-too-long\n{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"softnet.policy.get\"}\n", + ); + control.process_input(&mut rules).unwrap(); + assert!(!control.discarding_input); + + let mut output = [0; 2048]; + let n = client.read(&mut output).unwrap(); + let responses = std::str::from_utf8(&output[..n]) + .unwrap() + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .collect::>(); + assert_eq!(responses.len(), 2); + assert_eq!(responses[0]["error"]["code"], PARSE_ERROR); + assert_eq!(responses[1]["id"], 2); + } + + #[test] + fn fragmented_frame_does_not_apply_until_the_newline_arrives() { + let (mut client, server) = UnixStream::pair().unwrap(); + client + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + let mut control = control(server.as_raw_fd()).unwrap(); + let mut rules = PrefixMap::new(); + let before = control.policy.result(rules.len()); + + client + .write_all( + b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"@host\"],", + ) + .unwrap(); + assert!(control.service(&mut rules).unwrap()); + assert_eq!(control.policy.result(rules.len()), before); + assert!(control.output.is_empty()); + + client.write_all(b"\"block\":[]}}\n").unwrap(); + assert!(control.service(&mut rules).unwrap()); + + let mut response = [0; 1024]; + let n = client.read(&mut response).unwrap(); + let response = serde_json::from_slice::(&response[..n - 1]).unwrap(); + assert_eq!(response["id"], 1); + assert_eq!(response["result"]["allow"], json!(["@host"])); + assert_eq!(control.policy.allow, vec![Target::Host]); + } + + #[test] + fn response_backpressure_keeps_the_pending_queue_bounded() { + let (_client, server) = UnixStream::pair().unwrap(); + let mut control = control(server.as_raw_fd()).unwrap(); + let response = json!({"jsonrpc": "2.0", "id": 1, "result": "x".repeat(MAX_REQUEST_BYTES)}); + let mut bounded = false; + + for _ in 0..8 { + match control.enqueue(response.clone()) { + Ok(()) => assert!(control.flush().unwrap()), + Err(error) => { + assert!( + error + .to_string() + .contains("control response queue exceeded") + ); + bounded = true; + break; + } + } + } + + assert!(bounded); + assert!(control.output.len() - control.output_offset <= 4 * MAX_REQUEST_BYTES); + } + + #[test] + fn pipelined_policy_responses_stop_before_queue_overflow() { + 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) + .map(|index| format!("10.{}.{}.0/24", index / 256, index % 256)) + .collect::>(); + let mut input = serde_json::to_vec(&json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "softnet.policy.set", + "params": {"allow": allow, "block": []} + })) + .unwrap(); + input.push(b'\n'); + + for id in 1..=100 { + input.extend_from_slice( + format!("{{\"jsonrpc\":\"2.0\",\"id\":{id},\"method\":\"softnet.policy.get\"}}\n") + .as_bytes(), + ); + } + + control.input = input; + assert!(control.process_input(&mut rules).unwrap()); + assert_eq!(rules.len(), MAX_TARGETS); + assert!(!control.input.is_empty()); + assert!(!control.output.is_empty()); + assert!(control.output.len() - control.output_offset <= MAX_PENDING_RESPONSE_BYTES); + } + + #[test] + fn response_queue_overflow_does_not_apply_a_policy_update() { + let (_client, server) = UnixStream::pair().unwrap(); + let mut control = control(server.as_raw_fd()).unwrap(); + let mut rules = PrefixMap::new(); + let before = control.policy.result(rules.len()); + + control.output = vec![b'x'; MAX_PENDING_RESPONSE_BYTES]; + control.input = b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"10.0.0.0/8\"],\"block\":[]}}\n".to_vec(); + + let error = control.process_input(&mut rules).unwrap_err(); + assert!( + error + .to_string() + .contains("control response queue exceeded") + ); + assert_eq!(control.policy.result(rules.len()), before); + } + + #[test] + fn response_backpressure_stops_consuming_policy_updates() { + 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()); + + control.output = vec![b'x'; MAX_REQUEST_BYTES]; + assert!(control.flush().unwrap()); + assert!(!control.output.is_empty()); + + client + .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"softnet.policy.set\",\"params\":{\"allow\":[\"10.0.0.0/8\"],\"block\":[]}}\n") + .unwrap(); + + assert!(control.service(&mut rules).unwrap()); + assert_eq!(control.policy.result(rules.len()), before); + assert!(control.input.is_empty()); + } + + #[test] + fn validates_control_descriptor_without_taking_ownership() { + let file = File::open("/dev/null").unwrap(); + let error = control(file.as_raw_fd()).err().unwrap(); + assert!(error.to_string().contains("is not a socket")); + assert!(file.metadata().is_ok()); + + let (datagram, _) = UnixDatagram::pair().unwrap(); + let error = control(datagram.as_raw_fd()).err().unwrap(); + assert!(error.to_string().contains("not a Unix stream socket")); + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let error = control(listener.as_raw_fd()).err().unwrap(); + assert!(error.to_string().contains("not a Unix socket")); + + let (stream, _peer) = UnixStream::pair().unwrap(); + let control = control(stream.as_raw_fd()).unwrap(); + drop(control); + assert!(unsafe { libc::fcntl(stream.as_raw_fd(), libc::F_GETFD) != -1 }); + } + + #[test] + fn shutdown_signals_eof_while_the_original_descriptor_remains_open() { + let (mut client, server) = UnixStream::pair().unwrap(); + client + .set_read_timeout(Some(Duration::from_secs(1))) + .unwrap(); + let control = control(server.as_raw_fd()).unwrap(); + + control.shutdown().unwrap(); + drop(control); + + assert!(unsafe { libc::fcntl(server.as_raw_fd(), libc::F_GETFD) != -1 }); + let mut response = [0; 1]; + assert_eq!(client.read(&mut response).unwrap(), 0); + } +} diff --git a/lib/proxy/mod.rs b/lib/proxy/mod.rs index e32b338..449165b 100644 --- a/lib/proxy/mod.rs +++ b/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, + control: Option, enobufs_encountered: bool, port_forwarder: PortForwarder, } @@ -65,6 +68,7 @@ impl Proxy<'_> { allow: Vec, block: Vec, exposed_ports: Vec, + control_fd: Option, ) -> Result> { 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(); diff --git a/src/main.rs b/src/main.rs index f974784..2a1788b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -31,6 +31,13 @@ struct Args { )] 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, + #[clap(long, help = "MAC address to enforce for the VM")] vm_mac_address: mac_address::MacAddress, @@ -203,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")?; @@ -286,4 +294,21 @@ mod tests { .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") + ); + } }