Connection monitoring

Cargo.toml

@@ -23,9 +23,10 @@ log = "0.4"
 prost = "0.14"
 prost-types = "0.14"
 serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
 syslog = "7.0"
 thiserror = "2.0"
-tokio = { version = "1", features = ["fs", "macros", "process", "rt-multi-thread", "signal", "sync", "time"] }
+tokio = { version = "1", features = ["fs", "io-util", "macros", "net", "process", "rt-multi-thread", "signal", "sync", "time"] }
 tokio-stream = { version = "0.1", features = [] }
 toml = { version = "1.0", default-features = false, features = [
     "parse",

build.rs

@@ -20,6 +20,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
             &[
                 "proto/v2/gateway.proto",
                 "proto/enterprise/v2/firewall/firewall.proto",
+                "proto/enterprise/v2/monitor/monitor.proto",
             ],
             &["proto"],
         )?;

example-config.toml

@@ -51,3 +51,9 @@ syslog_socket = "/var/run/log"
 
 # Optional: Clean the network interface and VPN configuration on quit.
 #clean_on_quit = true
+
+# Start connection monitoring (Linux, FreeBSD and macOS).
+#connection_monitor = true
+
+# pflog interface for connection monitoring (FreeBSD and macOS); created if missing.
+#pflog_ifname = "pflog0"

src/bin/defguard-gateway.rs

@@ -17,7 +17,7 @@ use defguard_gateway::{
     config::get_config,
     error::GatewayError,
     execute_command,
-    gateway::{Gateway, TlsConfig, run_gateway_loop, run_stats},
+    gateway::{Gateway, TlsConfig, run_connection_events, run_gateway_loop, run_stats},
     init_syslog,
     logging::init_tracing,
     server::run_http_server,
@@ -177,6 +177,11 @@ async fn main() -> Result<(), GatewayError> {
     // Launch statistics gathering task.
     let gateway = Arc::new(Mutex::new(gateway));
     tasks.spawn(run_stats(Arc::clone(&gateway), config.stats_period()));
+    // Launch task sending connection monitoring events to Core.
+    tasks.spawn(run_connection_events(
+        Arc::clone(&gateway),
+        config.stats_period(),
+    ));
 
     let (shutdown_tx, shutdown_rx) = watch::channel(false);
     let mut gateway_task = tokio::spawn(run_gateway_loop(

src/config.rs

@@ -1,8 +1,9 @@
 use std::{
-    fs,
+    fmt, fs,
     net::{IpAddr, Ipv4Addr, SocketAddr},
     num::NonZeroU64,
     path::PathBuf,
+    str::FromStr,
     time::Duration,
 };
 
@@ -205,6 +206,16 @@ pub struct Config {
     #[arg(long, env = "DEFGUARD_CLEAN_ON_QUIT")]
     #[serde(default)]
     pub clean_on_quit: bool,
+
+    /// Enables connection monitoring (Linux, FreeBSD and macOS).
+    #[arg(long, env = "DEFGUARD_CONNECTION_MONITOR")]
+    #[serde(default)]
+    pub connection_monitor: bool,
+
+    /// pflog interface for connection monitoring (FreeBSD and macOS); created if missing.
+    #[arg(long, env = "DEFGUARD_PFLOG_IFNAME", default_value = "pflog0")]
+    #[serde(default)]
+    pub pflog_ifname: PflogIfname,
 }
 
 impl Config {
@@ -263,6 +274,8 @@ impl Default for Config {
             cert_dir: PathBuf::from("/etc/defguard/certs"),
             adoption_timeout: 5,
             clean_on_quit: false,
+            connection_monitor: false,
+            pflog_ifname: PflogIfname::default(),
         }
     }
 }
@@ -285,6 +298,39 @@ pub fn get_config() -> Result<Config, GatewayError> {
     Ok(cli_config)
 }
 
+/// pflog interface name, `pflog<unit>`.
+#[derive(Clone, Copy, Debug, Default, Deserialize)]
+#[serde(try_from = "String")]
+pub struct PflogIfname(pub u8);
+
+impl FromStr for PflogIfname {
+    type Err = String;
+
+    fn from_str(name: &str) -> Result<Self, Self::Err> {
+        name.strip_prefix("pflog")
+            .and_then(|unit| unit.parse().ok())
+            .map(Self)
+            .filter(|ifname| ifname.to_string() == name)
+            .ok_or_else(|| {
+                format!("invalid pflog interface \"{name}\": expected pflog0 to pflog255")
+            })
+    }
+}
+
+impl TryFrom<String> for PflogIfname {
+    type Error = String;
+
+    fn try_from(name: String) -> Result<Self, Self::Error> {
+        name.parse()
+    }
+}
+
+impl fmt::Display for PflogIfname {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "pflog{}", self.0)
+    }
+}
+
 fn validate_ifname(ifname: &str) -> Result<(), GatewayError> {
     if ifname.len() > 15 {
         return Err(GatewayError::InvalidConfigFile(format!(
@@ -295,6 +341,19 @@ fn validate_ifname(ifname: &str) -> Result<(), GatewayError> {
     Ok(())
 }
 
+#[test]
+fn test_pflog_ifname() {
+    let parse = |name: &str| name.parse::<PflogIfname>().ok().map(|ifname| ifname.0);
+    assert_eq!(parse("pflog0"), Some(0));
+    assert_eq!(parse("pflog12"), Some(12));
+    assert_eq!(parse("pflog255"), Some(255));
+    assert_eq!(parse("pflog256"), None);
+    assert_eq!(parse("pflog"), None);
+    assert_eq!(parse("pflog01"), None);
+    assert_eq!(parse("pflog+1"), None);
+    assert_eq!(parse("wg0"), None);
+}
+
 #[test]
 fn verify_cli() {
     use clap::CommandFactory;

src/enterprise/firewall/api.rs

@@ -19,8 +19,10 @@ pub(crate) struct FirewallApi {
     pub(crate) ifname: String,
     #[cfg(any(target_os = "freebsd", target_os = "macos", target_os = "netbsd"))]
     pub(crate) file: File,
-    #[cfg(any(target_os = "freebsd", target_os = "macos", target_os = "netbsd"))]
     pub(crate) default_policy: Policy,
+    /// pflog unit to log to; `None` disables logging.
+    #[cfg(any(target_os = "freebsd", target_os = "macos", target_os = "netbsd"))]
+    pub(crate) log_interface: Option<u8>,
     #[cfg(target_os = "linux")]
     pub(crate) socket: mnl::Socket,
 }
@@ -34,8 +36,9 @@ impl FirewallApi {
             ifname: ifname.into(),
             #[cfg(any(target_os = "freebsd", target_os = "macos", target_os = "netbsd"))]
             file: OpenOptions::new().read(true).write(true).open(DEV_PF)?,
-            #[cfg(any(target_os = "freebsd", target_os = "macos", target_os = "netbsd"))]
             default_policy: Policy::Deny,
+            #[cfg(any(target_os = "freebsd", target_os = "macos", target_os = "netbsd"))]
+            log_interface: None,
             #[cfg(target_os = "linux")]
             socket: mnl::Socket::new(mnl::Bus::Netfilter).map_err(|err| {
                 FirewallError::NetlinkError(format!("Failed to create socket: {err:?}"))

src/enterprise/firewall/flow.rs

@@ -0,0 +1,143 @@
+//! Identification of connections from raw packets, shared by firewall connection monitors.
+
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
+
+// IANA protocol numbers not exposed by `libc`.
+const IPPROTO_DCCP: u8 = 33;
+const IPPROTO_SCTP: u8 = 132;
+const IPPROTO_UDPLITE: u8 = 136;
+
+/// Uniquely identifies a connection, as seen in its original direction.
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub(crate) struct FlowKey {
+    pub(crate) protocol: u8,
+    pub(crate) source: IpAddr,
+    pub(crate) destination: IpAddr,
+    /// Source port, or ICMP identifier.
+    pub(crate) source_port: u16,
+    pub(crate) destination_port: u16,
+}
+
+pub(crate) fn protocol_has_ports(protocol: u8) -> bool {
+    matches!(i32::from(protocol), libc::IPPROTO_TCP | libc::IPPROTO_UDP)
+        || matches!(protocol, IPPROTO_DCCP | IPPROTO_SCTP | IPPROTO_UDPLITE)
+}
+
+pub(crate) fn be_u16(data: &[u8]) -> Option<u16> {
+    Some(u16::from_be_bytes(data.get(..2)?.try_into().ok()?))
+}
+
+pub(crate) fn ipv4(data: &[u8]) -> Option<IpAddr> {
+    Some(IpAddr::V4(Ipv4Addr::from(
+        <[u8; 4]>::try_from(data.get(..4)?).ok()?,
+    )))
+}
+
+pub(crate) fn ipv6(data: &[u8]) -> Option<IpAddr> {
+    Some(IpAddr::V6(Ipv6Addr::from(
+        <[u8; 16]>::try_from(data.get(..16)?).ok()?,
+    )))
+}
+
+/// Extract connection details from a packet, starting at the network header.
+pub(crate) fn parse_packet(packet: &[u8]) -> Option<FlowKey> {
+    let (protocol, source, destination, transport) = match packet.first()? >> 4 {
+        4 => {
+            let header_len = usize::from(packet[0] & 0x0f) * 4;
+            (
+                *packet.get(9)?,
+                ipv4(packet.get(12..)?)?,
+                ipv4(packet.get(16..)?)?,
+                packet.get(header_len..)?,
+            )
+        }
+        6 => {
+            let mut next_header = *packet.get(6)?;
+            let mut offset = 40;
+            loop {
+                match i32::from(next_header) {
+                    libc::IPPROTO_HOPOPTS | libc::IPPROTO_ROUTING | libc::IPPROTO_DSTOPTS => {
+                        let header = packet.get(offset..offset + 2)?;
+                        next_header = header[0];
+                        offset += (usize::from(header[1]) + 1) * 8;
+                    }
+                    libc::IPPROTO_FRAGMENT => {
+                        next_header = *packet.get(offset)?;
+                        offset += 8;
+                    }
+                    _ => break,
+                }
+            }
+            (
+                next_header,
+                ipv6(packet.get(8..)?)?,
+                ipv6(packet.get(24..)?)?,
+                packet.get(offset..)?,
+            )
+        }
+        _ => return None,
+    };
+
+    // Match the way conntrack identifies connections.
+    let (source_port, destination_port) = match i32::from(protocol) {
+        libc::IPPROTO_ICMP | libc::IPPROTO_ICMPV6 => (be_u16(transport.get(4..)?)?, 0),
+        _ if protocol_has_ports(protocol) => (be_u16(transport)?, be_u16(transport.get(2..)?)?),
+        _ => (0, 0),
+    };
+
+    Some(FlowKey {
+        protocol,
+        source,
+        destination,
+        source_port,
+        destination_port,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn tcp_packet() -> Vec<u8> {
+        let mut packet = vec![0u8; 40];
+        packet[0] = 0x45;
+        packet[9] = libc::IPPROTO_TCP as u8;
+        packet[12..16].copy_from_slice(&[10, 0, 0, 2]);
+        packet[16..20].copy_from_slice(&[192, 168, 1, 1]);
+        packet[20..22].copy_from_slice(&51000u16.to_be_bytes());
+        packet[22..24].copy_from_slice(&443u16.to_be_bytes());
+        packet
+    }
+
+    #[test]
+    fn test_parse_ipv4_tcp_packet() {
+        assert_eq!(
+            parse_packet(&tcp_packet()),
+            Some(FlowKey {
+                protocol: libc::IPPROTO_TCP as u8,
+                source: IpAddr::V4(Ipv4Addr::new(10, 0, 0, 2)),
+                destination: IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)),
+                source_port: 51000,
+                destination_port: 443,
+            })
+        );
+    }
+
+    #[test]
+    fn test_parse_ipv6_udp_packet_with_extension_header() {
+        let mut packet = vec![0u8; 56];
+        packet[0] = 0x60;
+        packet[6] = libc::IPPROTO_DSTOPTS as u8;
+        packet[8..24].copy_from_slice(&Ipv6Addr::LOCALHOST.octets());
+        packet[24..40].copy_from_slice(&Ipv6Addr::LOCALHOST.octets());
+        // Destination options header, 8 bytes long.
+        packet[40] = libc::IPPROTO_UDP as u8;
+        packet[48..50].copy_from_slice(&5353u16.to_be_bytes());
+        packet[50..52].copy_from_slice(&53u16.to_be_bytes());
+
+        let key = parse_packet(&packet).unwrap();
+        assert_eq!(key.protocol, libc::IPPROTO_UDP as u8);
+        assert_eq!(key.source_port, 5353);
+        assert_eq!(key.destination_port, 53);
+    }
+}

src/enterprise/firewall/mod.rs

@@ -1,5 +1,8 @@
 pub mod api;
+#[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
+mod flow;
 mod iprange;
+pub mod monitor;
 #[cfg(target_os = "linux")]
 mod nftables;
 #[cfg(any(target_os = "freebsd", target_os = "macos", target_os = "netbsd"))]
@@ -18,6 +21,11 @@ use tracing::error;
 
 use crate::proto;
 
+#[cfg(target_os = "linux")]
+pub use nftables::ConnectionMonitor;
+#[cfg(any(target_os = "freebsd", target_os = "macos"))]
+pub use packetfilter::ConnectionMonitor;
+
 #[derive(Clone, Debug, PartialEq)]
 pub(crate) enum Address {
     Network(IpNetwork),
@@ -189,6 +197,16 @@ pub(crate) enum Policy {
     Deny,
 }
 
+#[cfg(target_os = "linux")]
+impl Policy {
+    pub(crate) fn as_str(&self) -> &str {
+        match self {
+            Self::Allow => "a",
+            Self::Deny => "d",
+        }
+    }
+}
+
 impl From<bool> for Policy {
     fn from(allow: bool) -> Self {
         if allow { Self::Allow } else { Self::Deny }