From 514d46d7f360a09cb8ba780c5a5728ef8e84b7b5 Mon Sep 17 00:00:00 2001 From: Anton Liashkevich Date: Fri, 17 Jul 2026 17:24:05 -0400 Subject: [PATCH 1/6] Vendor nullnet-libresmon into wallguard as an internal module Replaces the external crate dependency with local code to avoid maintaining a separately published package for such a small, single-consumer library. Co-Authored-By: Claude Sonnet 5 --- Cargo.lock | 29 +----- wallguard/Cargo.toml | 3 +- .../src/data_transmission/resources/mod.rs | 1 + .../data_transmission/resources/monitor.rs | 93 +++++++++++++++++++ .../resources/transmitter.rs | 3 +- .../data_transmission/transmission_manager.rs | 4 +- 6 files changed, 101 insertions(+), 32 deletions(-) create mode 100644 wallguard/src/data_transmission/resources/monitor.rs diff --git a/Cargo.lock b/Cargo.lock index ce8fbecd..b717b562 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3544,16 +3544,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "nullnet-libresmon" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "11eb36074a3ff5aaf8d47507d0f241fd3dfaf977594232753bef57f33ea21dc6" -dependencies = [ - "async-channel", - "sysinfo 0.35.2", -] - [[package]] name = "nullnet-traffic-monitor" version = "0.1.6" @@ -6074,20 +6064,6 @@ dependencies = [ "syn 2.0.118", ] -[[package]] -name = "sysinfo" -version = "0.35.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c3ffa3e4ff2b324a57f7aeb3c349656c7b127c3c189520251a648102a92496e" -dependencies = [ - "libc", - "memchr", - "ntapi", - "objc2-core-foundation", - "objc2-io-kit", - "windows 0.61.3", -] - [[package]] name = "sysinfo" version = "0.37.2" @@ -6694,7 +6670,6 @@ dependencies = [ "nftables", "nix 0.31.3", "nullnet-liberror", - "nullnet-libresmon", "nullnet-traffic-monitor", "once_cell", "openh264", @@ -6705,7 +6680,7 @@ dependencies = [ "serde_json", "sha2 0.10.9", "smbios-lib", - "sysinfo 0.37.2", + "sysinfo", "tokio", "tokio-rustls", "tonic", @@ -6733,7 +6708,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.10.9", - "sysinfo 0.37.2", + "sysinfo", "tar", "tokio", "tonic", diff --git a/wallguard/Cargo.toml b/wallguard/Cargo.toml index 27a3b1b9..75106009 100644 --- a/wallguard/Cargo.toml +++ b/wallguard/Cargo.toml @@ -18,9 +18,8 @@ chrono = "0.4.41" once_cell = "1.21.4" nullnet-traffic-monitor = "0.1.6" etherparse = "0.19.0" -sysinfo = { version = "0.37.2", default-features = false, features = ["disk"] } +sysinfo = { version = "0.37.2", default-features = false, features = ["disk", "system", "component"] } async-channel = "2.3.1" -nullnet-libresmon = "0.1.2" wallguard-common = { path = "../wallguard-common" } xmltree = "0.12.0" md5 = "0.8.0" diff --git a/wallguard/src/data_transmission/resources/mod.rs b/wallguard/src/data_transmission/resources/mod.rs index 0def8407..38189b84 100644 --- a/wallguard/src/data_transmission/resources/mod.rs +++ b/wallguard/src/data_transmission/resources/mod.rs @@ -1 +1,2 @@ +pub(crate) mod monitor; pub(crate) mod transmitter; diff --git a/wallguard/src/data_transmission/resources/monitor.rs b/wallguard/src/data_transmission/resources/monitor.rs new file mode 100644 index 00000000..0f92ec08 --- /dev/null +++ b/wallguard/src/data_transmission/resources/monitor.rs @@ -0,0 +1,93 @@ +use async_channel::Receiver; +use std::collections::HashMap; +use std::path::Path; +use sysinfo::{ + Components, CpuRefreshKind, DiskRefreshKind, Disks, MemoryRefreshKind, RefreshKind, System, +}; + +static SYSTEM_REFRESH_KIND: std::sync::LazyLock = std::sync::LazyLock::new(|| { + RefreshKind::nothing() + .with_cpu(CpuRefreshKind::nothing().with_cpu_usage()) + .with_memory(MemoryRefreshKind::nothing().with_ram()) +}); + +static DISK_REFRESH_KIND: std::sync::LazyLock = + std::sync::LazyLock::new(|| DiskRefreshKind::nothing().with_io_usage().with_storage()); + +#[derive(Default)] +pub(crate) struct SystemResources { + pub num_cpus: usize, + pub global_cpu_usage: f32, + pub cpu_usages: HashMap, + pub total_memory: u64, + pub used_memory: u64, + pub total_disk_space: u64, + pub available_disk_space: u64, + pub read_bytes: u64, + pub written_bytes: u64, + pub temperatures: HashMap>, +} + +#[must_use] +pub(crate) fn poll_system_resources(interval_msec: u64) -> Receiver { + let (tx, rx) = async_channel::bounded(60); + + std::thread::spawn(move || { + let mut sys = System::new_with_specifics(*SYSTEM_REFRESH_KIND); + let mut disks = Disks::new_with_refreshed_list_specifics(*DISK_REFRESH_KIND); + let mut components = Components::new_with_refreshed_list(); + loop { + std::thread::sleep(std::time::Duration::from_millis(interval_msec)); + + sys.refresh_specifics(*SYSTEM_REFRESH_KIND); + disks.refresh_specifics(true, *DISK_REFRESH_KIND); + components.refresh(true); + + let mut cpu_usages = HashMap::new(); + for cpu in sys.cpus() { + let usage = cpu.cpu_usage(); + cpu_usages.insert(cpu.name().to_string(), usage); + } + + let mut total_disk_space = 0; + let mut available_disk_space = 0; + let mut read_bytes = 0; + let mut written_bytes = 0; + for disk in &disks { + if disk.mount_point() == Path::new("/") { + total_disk_space = disk.total_space(); + available_disk_space = disk.available_space(); + let disk_usage = disk.usage(); + read_bytes = disk_usage.read_bytes; + written_bytes = disk_usage.written_bytes; + } + } + + let mut temperatures = HashMap::new(); + for component in &components { + let temperature = component.temperature(); + temperatures.insert(component.label().to_string(), temperature); + } + + let resources = SystemResources { + num_cpus: sys.cpus().len(), + global_cpu_usage: sys.global_cpu_usage(), + cpu_usages, + total_memory: sys.total_memory(), + used_memory: sys.used_memory(), + total_disk_space, + available_disk_space, + read_bytes, + written_bytes, + temperatures, + }; + + // send resources to caller, or exit if channel is closed + let Ok(()) = tx.send_blocking(resources) else { + return; + }; + } + }); + + rx +} diff --git a/wallguard/src/data_transmission/resources/transmitter.rs b/wallguard/src/data_transmission/resources/transmitter.rs index 09b1fa22..4c475b47 100644 --- a/wallguard/src/data_transmission/resources/transmitter.rs +++ b/wallguard/src/data_transmission/resources/transmitter.rs @@ -1,6 +1,7 @@ use crate::constants::QUEUE_SIZE_RESOURCES; use crate::data_transmission::dump_dir::{DumpDir, DumpItem}; use crate::data_transmission::item_buffer::ItemBuffer; +use crate::data_transmission::resources::monitor::SystemResources; use crate::token_provider::TokenProvider; use crate::wg_server::WGServer; use async_channel::Receiver; @@ -8,7 +9,7 @@ use chrono::Utc; use wallguard_common::protobuf::wallguard_service::{SystemResource, SystemResourcesData}; pub(crate) async fn transmit_system_resources( - rx: Receiver, + rx: Receiver, token_provider: TokenProvider, dump_dir: DumpDir, client: WGServer, diff --git a/wallguard/src/data_transmission/transmission_manager.rs b/wallguard/src/data_transmission/transmission_manager.rs index a381bcf8..b497c921 100644 --- a/wallguard/src/data_transmission/transmission_manager.rs +++ b/wallguard/src/data_transmission/transmission_manager.rs @@ -2,13 +2,13 @@ use crate::client_data::Platform; use crate::constants::SNAPLEN; use crate::data_transmission::grpc_handler::handle_connection_and_retransmission; use crate::data_transmission::packets::transmitter::transmit_packets; +use crate::data_transmission::resources::monitor::SystemResources; use crate::data_transmission::resources::transmitter::transmit_system_resources; use crate::data_transmission::sysconfig; use crate::netinfo::monitor_services; use crate::wg_server::WGServer; use crate::{data_transmission::dump_dir::DumpDir, token_provider::TokenProvider}; use async_channel::Receiver; -use nullnet_libresmon::SystemResources; use nullnet_traffic_monitor::PacketInfo; use tokio::sync::broadcast; @@ -115,7 +115,7 @@ impl TransmissionManager { } log::info!("Starting resource monitoring"); - let rx = nullnet_libresmon::poll_system_resources(1000); + let rx = crate::data_transmission::resources::monitor::poll_system_resources(1000); self.resource_monitoring = Some(rx.clone()); let token_provider = self.token_provider.clone(); let dump_dir = self.dump_dir.clone(); From 8ea495867ef0be7529ffcd0c37faa1f26f2d53cf Mon Sep 17 00:00:00 2001 From: Anton Liashkevich Date: Fri, 17 Jul 2026 23:06:22 -0400 Subject: [PATCH 2/6] Fix empty FreeBSD resource-monitoring records Temperatures were always empty because pfSense/OPNsense don't load the coretemp/amdtemp kernel modules by default, so the dev.cpu.N.temperature sysctls sysinfo reads never existed; now best-effort kldload them before polling starts. Disk read/written bytes were always zero on ZFS-rooted boxes (the modern pfSense/OPNsense default) because sysinfo attributes I/O by matching a /dev/... device path, but a ZFS root reports its source as a dataset name instead. Added a FreeBSD-only fallback that resolves the pool backing "/" via zpool status and reads GEOM's raw devstat counters for just its backing disks, avoiding double-counting from the partition/label layers stacked on top of each disk. Co-Authored-By: Claude Sonnet 5 --- .../resources/freebsd_disk_io.rs | 279 ++++++++++++++++++ .../src/data_transmission/resources/mod.rs | 2 + .../data_transmission/resources/monitor.rs | 25 ++ 3 files changed, 306 insertions(+) create mode 100644 wallguard/src/data_transmission/resources/freebsd_disk_io.rs diff --git a/wallguard/src/data_transmission/resources/freebsd_disk_io.rs b/wallguard/src/data_transmission/resources/freebsd_disk_io.rs new file mode 100644 index 00000000..f946c03f --- /dev/null +++ b/wallguard/src/data_transmission/resources/freebsd_disk_io.rs @@ -0,0 +1,279 @@ +#![cfg(target_os = "freebsd")] + +//! `sysinfo`'s FreeBSD disk-I/O accounting attributes bytes to a mount point by +//! matching its underlying device against a `/dev/...` path resolved from +//! `kern.geom.conftxt`. That works for UFS, but a ZFS-backed mount reports its +//! source as a dataset name (e.g. "zroot/ROOT/default") which never matches a +//! `/dev/...` path, so `read_bytes`/`written_bytes` stay 0 forever on any +//! ZFS-rooted box — the default on modern pfSense/OPNsense installs. +//! +//! This module fills that gap for the root filesystem specifically: it resolves +//! the ZFS pool backing "/" to its physical leaf disks via `zpool status`, then +//! reads GEOM's raw devstat counters for exactly those disks (never summing +//! every GEOM layer, which would double-count the same I/O once per partition +//! and label provider stacked on top of each disk). + +use libc::{c_char, c_int, c_void, devstat, devstat_getversion, size_t}; +use std::ffi::CString; +use std::process::Command; +use std::ptr::null_mut; +use std::sync::OnceLock; + +const DEVSTAT_READ: usize = 0x01; +const DEVSTAT_WRITE: usize = 0x02; + +#[link(name = "geom")] +unsafe extern "C" { + fn geom_stats_open() -> c_int; + fn geom_stats_snapshot_get() -> *mut c_void; + fn geom_stats_snapshot_next(arg: *mut c_void) -> *mut devstat; + fn geom_stats_snapshot_free(arg: *mut c_void); +} + +fn geom_ready() -> bool { + static READY: OnceLock = OnceLock::new(); + *READY.get_or_init(|| unsafe { devstat_getversion(null_mut()) == 6 && geom_stats_open() == 0 }) +} + +fn c_buf_to_string(buf: &[c_char]) -> Option { + let bytes: &[u8] = unsafe { std::slice::from_raw_parts(buf.as_ptr().cast(), buf.len()) }; + let len = bytes.iter().position(|&b| b == 0)?; + std::str::from_utf8(&bytes[..len]).ok().map(str::to_owned) +} + +/// Reads a string-valued sysctl by name (e.g. "kern.geom.conftxt"). +fn sysctl_string(name: &str) -> Option { + let c_name = CString::new(name).ok()?; + let mut len: size_t = 0; + unsafe { + if libc::sysctlbyname( + c_name.as_ptr(), + null_mut(), + &mut len, + null_mut(), + 0, + ) != 0 + { + return None; + } + let mut buf = vec![0u8; len]; + if libc::sysctlbyname( + c_name.as_ptr(), + buf.as_mut_ptr().cast(), + &mut len, + null_mut(), + 0, + ) != 0 + { + return None; + } + buf.truncate(len); + // The sysctl is a NUL-terminated C string; trailing NULs would otherwise + // survive into the String and break later exact-match comparisons. + while buf.last() == Some(&0) { + buf.pop(); + } + String::from_utf8(buf).ok() + } +} + +/// Maps every alternate `/dev/...` path GEOM knows for a disk (partitions, GPT +/// labels, gptid, diskid, ...) back to that disk's base name (e.g. "ada0"). +/// Mirrors the mapping `sysinfo` itself builds from the same sysctl internally. +fn disk_label_mapping() -> std::collections::HashMap { + let mut mapping = std::collections::HashMap::new(); + let Some(conftxt) = sysctl_string("kern.geom.conftxt") else { + return mapping; + }; + + let mut last_id = String::new(); + for line in conftxt.lines() { + let mut parts = line.split_whitespace(); + let Some(kind) = parts.next() else { continue }; + + if kind == "0" { + if let Some("DISK") = parts.next() + && let Some(id) = parts.next() + { + last_id.clear(); + last_id.push_str(id); + } + } else if kind == "2" && !last_id.is_empty() { + if let Some("LABEL") = parts.next() + && let Some(path) = parts.next() + { + mapping.insert(format!("/dev/{path}"), last_id.clone()); + } + } + } + mapping +} + +/// "ada0p3" -> "ada0", "nvd0p2" -> "nvd0", "da1" (whole-disk vdev) -> "da1". +fn strip_partition_suffix(device: &str) -> String { + match device.rfind('p') { + Some(idx) + if idx + 1 < device.len() && device[idx + 1..].bytes().all(|b| b.is_ascii_digit()) => + { + device[..idx].to_string() + } + _ => device.to_string(), + } +} + +/// Resolves a `zpool status` leaf device entry (a raw name like "ada0p3", or a +/// label like "gpt/zfs0" / "gptid/") to the base GEOM disk name that +/// devstat actually tracks. +fn resolve_base_disk(leaf: &str, label_mapping: &std::collections::HashMap) -> Option { + if leaf.contains('/') { + label_mapping.get(&format!("/dev/{leaf}")).cloned() + } else { + Some(strip_partition_suffix(leaf)) + } +} + +/// Runs `zpool status ` and returns the base disk names backing every +/// leaf vdev (mirrors/raidz members included; spares and cache/log devices are +/// included too since they're real devices attached to the pool). +fn zpool_backing_disks(pool: &str) -> Vec { + let Ok(output) = Command::new("/sbin/zpool").arg("status").arg(pool).output() else { + return Vec::new(); + }; + if !output.status.success() { + return Vec::new(); + } + let text = String::from_utf8_lossy(&output.stdout); + let label_mapping = disk_label_mapping(); + + let mut in_config = false; + let mut seen_header = false; + let mut disks = Vec::new(); + + for line in text.lines() { + let trimmed = line.trim(); + if trimmed == "config:" { + in_config = true; + continue; + } + if !in_config { + continue; + } + if trimmed.is_empty() { + if seen_header { + break; + } + continue; + } + let Some(name) = trimmed.split_whitespace().next() else { + continue; + }; + if name == "NAME" { + seen_header = true; + continue; + } + if name == pool + || name.starts_with("mirror-") + || name.starts_with("raidz") + || name == "spares" + || name == "logs" + || name == "cache" + { + continue; + } + if let Some(base) = resolve_base_disk(name, &label_mapping) { + disks.push(base); + } + } + + disks.sort(); + disks.dedup(); + disks +} + +/// Sums read/write bytes across exactly the named base-disk devstat entries, +/// never across every GEOM layer (which would multiply-count the same I/O). +fn read_named_disk_totals(names: &[String]) -> Option<(u64, u64)> { + if !geom_ready() || names.is_empty() { + return None; + } + let snap = unsafe { geom_stats_snapshot_get() }; + if snap.is_null() { + return None; + } + + let mut read_bytes = 0u64; + let mut written_bytes = 0u64; + loop { + let device = unsafe { geom_stats_snapshot_next(snap) }; + if device.is_null() { + break; + } + let device = unsafe { &*device }; + let Some(device_name) = c_buf_to_string(&device.device_name) else { + continue; + }; + let full_name = format!("{device_name}{}", device.unit_number); + if names.iter().any(|n| *n == full_name) { + read_bytes = read_bytes.saturating_add(device.bytes[DEVSTAT_READ]); + written_bytes = written_bytes.saturating_add(device.bytes[DEVSTAT_WRITE]); + } + } + unsafe { geom_stats_snapshot_free(snap) }; + Some((read_bytes, written_bytes)) +} + +/// Tracks cumulative-to-delta conversion for the root filesystem's backing +/// disk(s), across polling intervals. +pub(crate) struct RootDiskIo { + disk_names: Option>, + prev: Option<(u64, u64)>, +} + +impl RootDiskIo { + pub(crate) fn new() -> Self { + let disk_names = mount_source("/").and_then(|(fstype, source)| { + if fstype != "zfs" { + // Not ZFS: sysinfo's own dev_id-based matching already works here. + return None; + } + let pool = source.split('/').next().unwrap_or(&source); + let disks = zpool_backing_disks(pool); + if disks.is_empty() { None } else { Some(disks) } + }); + Self { + disk_names, + prev: None, + } + } + + /// Returns `(read_bytes, written_bytes)` deltas since the last call, or + /// `None` when "/" isn't ZFS or pool resolution failed — callers should + /// keep whatever `sysinfo` already computed in that case. + pub(crate) fn refresh(&mut self) -> Option<(u64, u64)> { + let names = self.disk_names.as_ref()?; + let (total_read, total_written) = read_named_disk_totals(names)?; + let delta = match self.prev { + Some((prev_read, prev_written)) => ( + total_read.saturating_sub(prev_read), + total_written.saturating_sub(prev_written), + ), + None => (0, 0), + }; + self.prev = Some((total_read, total_written)); + Some(delta) + } +} + +/// Returns the raw fstype + mount-source ("f_mntfromname") for a path, as +/// reported by `statfs(2)`. For ZFS this is the dataset name; for UFS/other +/// it's a device path. +fn mount_source(path: &str) -> Option<(String, String)> { + let c_path = CString::new(path).ok()?; + let mut buf: libc::statfs = unsafe { std::mem::zeroed() }; + if unsafe { libc::statfs(c_path.as_ptr(), &mut buf) } != 0 { + return None; + } + let fstype = c_buf_to_string(&buf.f_fstypename)?; + let source = c_buf_to_string(&buf.f_mntfromname)?; + Some((fstype, source)) +} diff --git a/wallguard/src/data_transmission/resources/mod.rs b/wallguard/src/data_transmission/resources/mod.rs index 38189b84..1d926431 100644 --- a/wallguard/src/data_transmission/resources/mod.rs +++ b/wallguard/src/data_transmission/resources/mod.rs @@ -1,2 +1,4 @@ +#[cfg(target_os = "freebsd")] +mod freebsd_disk_io; pub(crate) mod monitor; pub(crate) mod transmitter; diff --git a/wallguard/src/data_transmission/resources/monitor.rs b/wallguard/src/data_transmission/resources/monitor.rs index 0f92ec08..84c5b182 100644 --- a/wallguard/src/data_transmission/resources/monitor.rs +++ b/wallguard/src/data_transmission/resources/monitor.rs @@ -14,6 +14,17 @@ static SYSTEM_REFRESH_KIND: std::sync::LazyLock = std::sync::LazyLo static DISK_REFRESH_KIND: std::sync::LazyLock = std::sync::LazyLock::new(|| DiskRefreshKind::nothing().with_io_usage().with_storage()); +/// FreeBSD only exposes `dev.cpu.N.temperature` once the vendor-specific sensor +/// driver is attached; pfSense/OPNsense don't load it by default, so temperature +/// readings are silently empty until we do this ourselves. Loading the wrong +/// vendor's module is harmless: it simply fails to attach to the hardware. +#[cfg(target_os = "freebsd")] +fn load_temperature_sensors() { + for module in ["coretemp", "amdtemp"] { + let _ = std::process::Command::new("kldload").arg(module).status(); + } +} + #[derive(Default)] pub(crate) struct SystemResources { pub num_cpus: usize, @@ -33,9 +44,14 @@ pub(crate) fn poll_system_resources(interval_msec: u64) -> Receiver Receiver Date: Mon, 27 Jul 2026 21:21:20 -0400 Subject: [PATCH 3/6] Fix CPU-spike risks in wallguard: socket scan, queue drains, blocking work Replaces the hand-rolled per-platform socket enumeration (netinfo/sock) with the listeners crate, removing unshielded blocking Win32 syscalls on the async executor and a sequential /proc walk. Switches ItemBuffer to VecDeque and restructures the dump-file retransmission loop to avoid O(n^2) front-drains on large backlogs. Moves JSON (de)serialization and fireparse config parsing onto spawn_blocking so large payloads can't stall the shared tokio runtime. Also caches the config digest instead of re-hashing unchanged content each tick, parallelizes per-socket HTTP/SSH service probing, fixes a short-circuit bug in sysconfig's update_all, and trims a couple of small per-tick allocations in FreeBSD disk-IO and resource monitoring. Co-Authored-By: Claude Sonnet 5 --- Cargo.lock | 14 +++ wallguard/Cargo.toml | 3 +- wallguard/src/data_transmission/dump_dir.rs | 27 +++-- .../src/data_transmission/grpc_handler.rs | 47 +++++--- .../src/data_transmission/item_buffer.rs | 16 ++- .../data_transmission/packets/transmitter.rs | 7 +- .../resources/freebsd_disk_io.rs | 11 +- .../data_transmission/resources/monitor.rs | 4 +- .../sysconfig/data/config_xml.rs | 13 ++- .../src/data_transmission/sysconfig/mod.rs | 15 ++- wallguard/src/netinfo/service/http.rs | 31 +++-- wallguard/src/netinfo/service/ssh.rs | 27 +++-- wallguard/src/netinfo/sock/freebsd.rs | 83 ------------- wallguard/src/netinfo/sock/linux/inode_pid.rs | 58 ---------- .../src/netinfo/sock/linux/inode_sock.rs | 106 ----------------- wallguard/src/netinfo/sock/linux/mod.rs | 23 ---- wallguard/src/netinfo/sock/mod.rs | 58 ++++------ wallguard/src/netinfo/sock/windows/mod.rs | 33 ------ .../src/netinfo/sock/windows/proc_snap.rs | 50 -------- wallguard/src/netinfo/sock/windows/tcp.rs | 109 ------------------ wallguard/src/netinfo/sock/windows/udp.rs | 99 ---------------- 21 files changed, 182 insertions(+), 652 deletions(-) delete mode 100644 wallguard/src/netinfo/sock/freebsd.rs delete mode 100644 wallguard/src/netinfo/sock/linux/inode_pid.rs delete mode 100644 wallguard/src/netinfo/sock/linux/inode_sock.rs delete mode 100644 wallguard/src/netinfo/sock/linux/mod.rs delete mode 100644 wallguard/src/netinfo/sock/windows/mod.rs delete mode 100644 wallguard/src/netinfo/sock/windows/proc_snap.rs delete mode 100644 wallguard/src/netinfo/sock/windows/tcp.rs delete mode 100644 wallguard/src/netinfo/sock/windows/udp.rs diff --git a/Cargo.lock b/Cargo.lock index b717b562..4791e01c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3137,6 +3137,19 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" +[[package]] +name = "listeners" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e06a2deee6dfb82d91999635c5cdd263563c62a0b85c71b4afd43dbb939382e" +dependencies = [ + "byteorder", + "cc", + "libc", + "rustix", + "windows 0.62.2", +] + [[package]] name = "litemap" version = "0.8.2" @@ -6665,6 +6678,7 @@ dependencies = [ "flexi_logger", "is_elevated", "libc", + "listeners", "log", "md5", "nftables", diff --git a/wallguard/Cargo.toml b/wallguard/Cargo.toml index 75106009..02fc788f 100644 --- a/wallguard/Cargo.toml +++ b/wallguard/Cargo.toml @@ -36,6 +36,7 @@ nullnet-liberror.workspace = true rustls.workspace = true tokio-rustls.workspace = true whoami = "2.0.2" +listeners = "0.6" [target.'cfg(any(target_os = "linux", target_os = "freebsd"))'.dependencies] x11rb = "0.13" @@ -51,7 +52,7 @@ evdev = "0.12" [target.'cfg(windows)'.dependencies] is_elevated = "0.1.2" -winapi = {version = "0.3.9", features = ["winerror", "iphlpapi", "handleapi", "tlhelp32", "wingdi", "winuser", "windef", "minwindef"]} +winapi = {version = "0.3.9", features = ["wingdi", "winuser", "windef", "minwindef"]} [target.'cfg(unix)'.dependencies] nix = { version = "0.31.3", features = ["user", "mman", "fs"] } diff --git a/wallguard/src/data_transmission/dump_dir.rs b/wallguard/src/data_transmission/dump_dir.rs index 53cb0e0c..0e5c222c 100644 --- a/wallguard/src/data_transmission/dump_dir.rs +++ b/wallguard/src/data_transmission/dump_dir.rs @@ -55,22 +55,29 @@ impl DumpDir { pub(crate) async fn dump_item_to_file(&self, dump_item: DumpItem) { let now = chrono::Utc::now().to_rfc3339(); let file_path = self.get_file_path(&now, &dump_item); - tokio::fs::write( - file_path, - serde_json::to_string(&dump_item).expect("Failed to serialize item"), - ) + // Serializing a dump item (up to the full queue, e.g. 1M records) is + // CPU-bound work; run it on the blocking pool so it can't stall the + // tokio runtime that also drives gRPC/heartbeat traffic. + let json = tokio::task::spawn_blocking(move || { + serde_json::to_string(&dump_item).expect("Failed to serialize item") + }) .await - .expect("Failed to write dump file"); + .expect("Serialization task panicked"); + tokio::fs::write(file_path, json) + .await + .expect("Failed to write dump file"); } pub(crate) async fn update_items_dump_file(&self, file_path: PathBuf, mut dump: DumpItem) { dump.set_token(String::new()); - tokio::fs::write( - file_path, - serde_json::to_string(&dump).expect("Failed to serialize items"), - ) + let json = tokio::task::spawn_blocking(move || { + serde_json::to_string(&dump).expect("Failed to serialize items") + }) .await - .expect("Failed to write dump file"); + .expect("Serialization task panicked"); + tokio::fs::write(file_path, json) + .await + .expect("Failed to write dump file"); } } diff --git a/wallguard/src/data_transmission/grpc_handler.rs b/wallguard/src/data_transmission/grpc_handler.rs index 5b5a5945..0b61a29d 100644 --- a/wallguard/src/data_transmission/grpc_handler.rs +++ b/wallguard/src/data_transmission/grpc_handler.rs @@ -37,30 +37,42 @@ pub(crate) async fn handle_connection_and_retransmission( let Ok(string) = fs::read_to_string(file.path()).await else { continue; }; - let Ok(mut dump) = serde_json::from_str::(&string) else { + // Deserializing a dump file (up to the full queue, e.g. 1M + // records) is CPU-bound; keep it off the tokio runtime so it + // can't stall gRPC/heartbeat traffic sharing the same executor. + let Ok(Ok(mut dump)) = + tokio::task::spawn_blocking(move || serde_json::from_str::(&string)) + .await + else { continue; }; // update auth token of items retrieved from disk dump.set_token(token.clone()); - while dump.size() != 0 { - let range = ..min(dump.size(), BATCH_SIZE); - // `dump.set_token` above already updated the token field in - // place, so only the (cheap) token string needs cloning here - // — cloning the whole item via `..c.clone()` used to clone - // the entire, not-yet-drained items vector on every batch. + // Batches are sliced by a `sent` offset rather than drained from + // the front on every iteration: draining a Vec's front repeatedly + // shifts the remaining tail down each time (O(remaining) per + // batch), which turns replaying a large backlog file into an + // O(n^2) sequence of memmoves. Slicing leaves the vector + // untouched until a single drain(..sent) at the end. + let total = dump.size(); + let mut sent = 0; + let mut failed = false; + + while sent < total { + let range = ..min(total - sent, BATCH_SIZE); let send_res = match &dump { DumpItem::Connections(c) => { let msg = ConnectionsData { token: c.token.clone(), - connections: c.connections.get(range).unwrap_or_default().to_vec(), + connections: c.connections[sent..][range].to_vec(), }; interface.handle_connections_data(msg).await } DumpItem::Resources(r) => { let msg = SystemResourcesData { token: r.token.clone(), - resources: r.resources.get(range).unwrap_or_default().to_vec(), + resources: r.resources[sent..][range].to_vec(), }; interface.handle_system_resources_data(msg).await } @@ -75,13 +87,18 @@ pub(crate) async fn handle_connection_and_retransmission( // back off before retrying instead of immediately // re-reading and re-sending the same file in a tight loop. log::warn!("Failed to send dump. Reconnecting...",); - // update dump file with unsent items - dump_dir.update_items_dump_file(file.path(), dump).await; - tokio::time::sleep(Duration::from_secs(10)).await; - break 'file_loop; + failed = true; + break; } - // remove sent items from dump - dump.drain(range); + sent += range.end; + } + + if failed { + // remove the items that did get sent, in one shot, and persist the rest + dump.drain(..sent); + dump_dir.update_items_dump_file(file.path(), dump).await; + tokio::time::sleep(Duration::from_secs(10)).await; + break 'file_loop; } log::info!("Dump file '{:?}' sent successfully", file.file_name()); diff --git a/wallguard/src/data_transmission/item_buffer.rs b/wallguard/src/data_transmission/item_buffer.rs index 86b3fd53..608df989 100644 --- a/wallguard/src/data_transmission/item_buffer.rs +++ b/wallguard/src/data_transmission/item_buffer.rs @@ -1,28 +1,34 @@ +use std::collections::VecDeque; use std::ops::RangeTo; +// Backed by a VecDeque (not a Vec) so that repeatedly draining a batch off +// the front — the access pattern every caller uses — costs O(batch), not +// O(remaining length): a Vec::drain(..batch) has to shift the whole +// remaining tail down on every call, which turns catching up a large backlog +// into an O(n^2) sequence of memmoves. pub(crate) struct ItemBuffer { - buffer: Vec, + buffer: VecDeque, size: usize, } impl ItemBuffer { pub(crate) fn new(size: usize) -> Self { Self { - buffer: Vec::with_capacity(size), + buffer: VecDeque::with_capacity(size), size, } } pub(crate) fn push(&mut self, item: T) { - self.buffer.push(item); + self.buffer.push_back(item); } pub(crate) fn take(&mut self) -> Vec { - std::mem::take(&mut self.buffer) + Vec::from(std::mem::take(&mut self.buffer)) } pub(crate) fn get(&mut self, range: RangeTo) -> Vec { - self.buffer.get(range).unwrap_or_default().to_vec() + self.buffer.iter().take(range.end).cloned().collect() } pub(crate) fn extend(&mut self, items: Vec) { diff --git a/wallguard/src/data_transmission/packets/transmitter.rs b/wallguard/src/data_transmission/packets/transmitter.rs index 850e5ad6..47e9da3d 100644 --- a/wallguard/src/data_transmission/packets/transmitter.rs +++ b/wallguard/src/data_transmission/packets/transmitter.rs @@ -27,7 +27,12 @@ pub(crate) async fn transmit_packets( if raw_batch.len() >= batch_size || timer.is_expired() { timer.reset(); - let connections = parse_packets(std::mem::take(&mut raw_batch)); + // Swap in a fresh, pre-sized buffer rather than `mem::take`ing + // (which would leave a 0-capacity Vec behind) so the next + // accumulation cycle doesn't have to regrow from scratch back up + // to `batch_size` on every flush. + let batch = std::mem::replace(&mut raw_batch, Vec::with_capacity(batch_size)); + let connections = parse_packets(batch); connection_queue.extend(connections); send_connections(&client, &mut connection_queue, &token_provider, batch_size).await; diff --git a/wallguard/src/data_transmission/resources/freebsd_disk_io.rs b/wallguard/src/data_transmission/resources/freebsd_disk_io.rs index f946c03f..33d13167 100644 --- a/wallguard/src/data_transmission/resources/freebsd_disk_io.rs +++ b/wallguard/src/data_transmission/resources/freebsd_disk_io.rs @@ -212,8 +212,15 @@ fn read_named_disk_totals(names: &[String]) -> Option<(u64, u64)> { let Some(device_name) = c_buf_to_string(&device.device_name) else { continue; }; - let full_name = format!("{device_name}{}", device.unit_number); - if names.iter().any(|n| *n == full_name) { + // Avoid allocating a fresh "{device_name}{unit_number}" String for + // every GEOM provider on the system (there can be dozens once + // partitions/labels are counted) just to compare it against the + // handful of names we actually care about. + let is_named_disk = names.iter().any(|n| { + n.strip_prefix(device_name.as_str()) + .is_some_and(|suffix| suffix.parse::() == Ok(device.unit_number)) + }); + if is_named_disk { read_bytes = read_bytes.saturating_add(device.bytes[DEVSTAT_READ]); written_bytes = written_bytes.saturating_add(device.bytes[DEVSTAT_WRITE]); } diff --git a/wallguard/src/data_transmission/resources/monitor.rs b/wallguard/src/data_transmission/resources/monitor.rs index 84c5b182..d3b29be1 100644 --- a/wallguard/src/data_transmission/resources/monitor.rs +++ b/wallguard/src/data_transmission/resources/monitor.rs @@ -59,7 +59,7 @@ pub(crate) fn poll_system_resources(interval_msec: u64) -> Receiver Receiver Result { - let prev = utilities::hash::sha256_digest_bytes(&self.content); let content = tokio::fs::read(FILE_PATH).await.handle_err(location!())?; - self.content = String::from_utf8_lossy(content.as_slice()).into(); - let curr = utilities::hash::sha256_digest_bytes(&self.content); - Ok(prev != curr) + let digest = utilities::hash::sha256_digest_bytes(&self.content); + let changed = digest != self.digest; + self.digest = digest; + + Ok(changed) } } diff --git a/wallguard/src/data_transmission/sysconfig/mod.rs b/wallguard/src/data_transmission/sysconfig/mod.rs index 9efbf08d..5915abb2 100644 --- a/wallguard/src/data_transmission/sysconfig/mod.rs +++ b/wallguard/src/data_transmission/sysconfig/mod.rs @@ -124,8 +124,15 @@ async fn upload_all( snapshot.push(file.take_snapshot()); } + // Parsing a firewall config (nftables ruleset / pfSense-OPNsense XML) is + // CPU-bound and can be sizeable; run it on the blocking pool so a slow + // parse can't stall the tokio runtime shared with gRPC/heartbeat traffic. + let configuration = tokio::task::spawn_blocking(move || Fireparse::parse(snapshot, platform)) + .await + .handle_err(location!())??; + let data = ConfigSnapshot { - configuration: Some(Fireparse::parse(snapshot, platform)?), + configuration: Some(configuration), token: token_provider .get() .await @@ -142,8 +149,12 @@ async fn upload_all( async fn update_all(files: &mut [SystemConfigurationFile]) -> Result { let mut retval = false; + // `retval ||= ...` would short-circuit and skip update() on the + // remaining files once one returns true — every file must always be + // refreshed so its cached content/digest stays current. for file in files.iter_mut() { - retval = retval || file.update().await?; + let changed = file.update().await?; + retval = retval || changed; } Ok(retval) diff --git a/wallguard/src/netinfo/service/http.rs b/wallguard/src/netinfo/service/http.rs index 2b84500b..a4bfef01 100644 --- a/wallguard/src/netinfo/service/http.rs +++ b/wallguard/src/netinfo/service/http.rs @@ -110,22 +110,37 @@ pub(super) async fn filter(sockets: &mut Vec) -> Vec { let mut services = Vec::new(); let mut remaining = Vec::with_capacity(sockets.len()); + // Probe every candidate socket concurrently instead of awaiting each TLS + // handshake + HTTP probe one at a time: sequential probing made this scale + // linearly (up to TIMEOUT_VALUE per socket) with the number of open + // listening ports every scan cycle. + let mut set = tokio::task::JoinSet::new(); for socket in sockets.drain(..) { - if matches!(socket.protocol, crate::netinfo::sock::Protocol::Tcp) - && let Some((protocol, code)) = detect_protocol(socket.sockaddr).await - { - if (200..300).contains(&code) { + set.spawn(async move { + let detected = if matches!(socket.protocol, crate::netinfo::sock::Protocol::Tcp) { + detect_protocol(socket.sockaddr).await + } else { + None + }; + (socket, detected) + }); + } + + while let Some(joined) = set.join_next().await { + let Ok((socket, detected)) = joined else { + continue; + }; + + match detected { + Some((protocol, code)) if (200..300).contains(&code) => { services.push(ServiceInfo { addr: socket.sockaddr, protocol, program: socket.process_name.clone(), }); } - - continue; + _ => remaining.push(socket), } - - remaining.push(socket); } *sockets = remaining; diff --git a/wallguard/src/netinfo/service/ssh.rs b/wallguard/src/netinfo/service/ssh.rs index e6bc175b..96e9eeae 100644 --- a/wallguard/src/netinfo/service/ssh.rs +++ b/wallguard/src/netinfo/service/ssh.rs @@ -28,20 +28,33 @@ pub(super) async fn filter(sockets: &mut Vec) -> Vec { let mut services = Vec::new(); let mut remaining = Vec::with_capacity(sockets.len()); + // Probe every candidate socket concurrently rather than awaiting each one + // sequentially — otherwise cost scales linearly (up to SSH_TIMEOUT per + // socket) with the number of open listening ports every scan cycle. + let mut set = tokio::task::JoinSet::new(); for socket in sockets.drain(..) { - if matches!(socket.protocol, crate::netinfo::sock::Protocol::Tcp) - && is_ssh(socket.sockaddr).await - { + set.spawn(async move { + let matched = + matches!(socket.protocol, crate::netinfo::sock::Protocol::Tcp) + && is_ssh(socket.sockaddr).await; + (socket, matched) + }); + } + + while let Some(joined) = set.join_next().await { + let Ok((socket, matched)) = joined else { + continue; + }; + + if matched { services.push(ServiceInfo { addr: socket.sockaddr, protocol: Protocol::Ssh, program: socket.process_name.clone(), }); - - continue; + } else { + remaining.push(socket); } - - remaining.push(socket); } *sockets = remaining; diff --git a/wallguard/src/netinfo/sock/freebsd.rs b/wallguard/src/netinfo/sock/freebsd.rs deleted file mode 100644 index 3e092286..00000000 --- a/wallguard/src/netinfo/sock/freebsd.rs +++ /dev/null @@ -1,83 +0,0 @@ -use super::{Protocol, SocketInfo}; -use std::net::SocketAddr; -use tokio::process::Command; - -fn parse_sockstat_addr(addr_str: &str, is_ipv6: bool) -> Option { - use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6}; - - let (host, port_str) = addr_str.rsplit_once(':')?; - let port = port_str.parse::().ok()?; - - if is_ipv6 { - let host = if host.starts_with('[') && host.ends_with(']') { - &host[1..host.len() - 1] - } else { - host - }; - - if host == "*" { - Some(SocketAddr::V6(SocketAddrV6::new( - Ipv6Addr::UNSPECIFIED, - port, - 0, - 0, - ))) - } else { - host.parse::() - .ok() - .map(|addr| SocketAddr::V6(SocketAddrV6::new(addr, port, 0, 0))) - } - } else if host == "*" { - Some(SocketAddr::V4(SocketAddrV4::new( - Ipv4Addr::UNSPECIFIED, - port, - ))) - } else { - host.parse::() - .ok() - .map(|addr| SocketAddr::V4(SocketAddrV4::new(addr, port))) - } -} - -pub(super) async fn get_sockets_info() -> Vec { - let Ok(output) = Command::new("sockstat").args(["-l"]).output().await else { - return vec![]; - }; - - if !output.status.success() { - return vec![]; - } - - let stdout = String::from_utf8_lossy(&output.stdout); - let mut sockets = Vec::new(); - - for line in stdout.lines().skip(1) { - let parts: Vec<&str> = line.split_whitespace().collect(); - if parts.len() < 6 { - continue; - } - - // Format: USER COMMAND PID FD PROTO LOCAL REMOTE - let command = parts[1]; - let proto_str = parts[4]; - let addr_str = parts[5]; - - let (protocol, is_ipv6) = match proto_str { - "tcp4" | "tcp46" => (Protocol::Tcp, false), - "tcp6" => (Protocol::Tcp, true), - "udp4" | "udp46" => (Protocol::Udp, false), - "udp6" => (Protocol::Udp, true), - _ => continue, - }; - - if let Some(sockaddr) = parse_sockstat_addr(addr_str, is_ipv6) { - sockets.push(SocketInfo { - process_name: command.to_string(), - protocol, - sockaddr, - }); - } - } - - sockets -} diff --git a/wallguard/src/netinfo/sock/linux/inode_pid.rs b/wallguard/src/netinfo/sock/linux/inode_pid.rs deleted file mode 100644 index 9d4170e7..00000000 --- a/wallguard/src/netinfo/sock/linux/inode_pid.rs +++ /dev/null @@ -1,58 +0,0 @@ -use std::{collections::HashMap, path::PathBuf}; -use tokio::fs; - -async fn read_proc_comm(pid: u32) -> Option { - let mut path = PathBuf::from("/proc"); - path.push(pid.to_string()); - path.push("comm"); - - match fs::read_to_string(&path).await { - Ok(mut name) => { - if let Some('\n') = name.chars().last() { - name.pop(); - } - Some(name) - } - Err(_) => None, - } -} - -pub(super) async fn build_inode_pid_map() -> HashMap { - let mut map = HashMap::new(); - - let Ok(mut rd) = fs::read_dir("/proc").await else { - return map; - }; - - while let Ok(Some(entry)) = rd.next_entry().await { - let pid_str = entry.file_name().into_string().unwrap_or_default(); - - if pid_str.chars().any(|c| !c.is_ascii_digit()) { - continue; - }; - - let Ok(pid) = pid_str.parse::() else { - continue; - }; - - let Ok(mut fd) = fs::read_dir(&format!("/proc/{pid}/fd")).await else { - continue; - }; - - while let Ok(Some(fds)) = fd.next_entry().await { - let Ok(link) = fs::read_link(fds.path()).await else { - continue; - }; - - if let Some(inner) = link.to_str().and_then(|s| s.strip_prefix("socket:[")) - && let Some(num) = inner.strip_suffix(']') - && let Ok(inode) = num.parse::() - && let Some(process_name) = read_proc_comm(pid).await - { - map.insert(inode, process_name); - } - } - } - - map -} diff --git a/wallguard/src/netinfo/sock/linux/inode_sock.rs b/wallguard/src/netinfo/sock/linux/inode_sock.rs deleted file mode 100644 index 6769ad04..00000000 --- a/wallguard/src/netinfo/sock/linux/inode_sock.rs +++ /dev/null @@ -1,106 +0,0 @@ -use std::{ - collections::HashMap, - net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6}, -}; - -use tokio::{ - fs, - io::{self, AsyncBufReadExt}, -}; - -use crate::netinfo::sock::{IpVersion, Protocol}; - -async fn parse_proc_net( - path: &str, - proto: Protocol, - version: IpVersion, -) -> io::Result> { - let file = fs::File::open(path).await?; - let reader = io::BufReader::new(file); - let mut map = HashMap::new(); - - let mut lines = reader.lines(); - - // Skip the header line - lines.next_line().await?; - - while let Some(line) = lines.next_line().await? { - let cols: Vec<&str> = line.split_whitespace().collect(); - if cols.len() < 10 { - continue; - } - - if let Protocol::Tcp = proto { - // Only include TCP if state == "0A" (LISTEN) - let state_hex = cols[3]; - if state_hex != "0A" { - continue; - } - } - - let local = cols[1]; - let inode: u64 = cols[9].parse().unwrap_or(0); - - if let Some(colon) = local.find(':') { - let (addr_hex, port_hex) = local.split_at(colon); - let addr_hex = &addr_hex[..addr_hex.len()]; - let port_hex = &port_hex[1..]; - - if let Ok(port) = u16::from_str_radix(port_hex, 16) { - let sockaddr = match version { - IpVersion::V4 => { - if addr_hex.len() != 8 { - continue; - }; - - let ip = u32::from_str_radix(addr_hex, 16).unwrap(); - let a = (ip & 0xff) as u8; - let b = ((ip >> 8) & 0xff) as u8; - let c = ((ip >> 16) & 0xff) as u8; - let d = ((ip >> 24) & 0xff) as u8; - - SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(a, b, c, d), port)) - } - IpVersion::V6 => { - if addr_hex.len() != 32 { - continue; - }; - - let mut bytes = [0u8; 16]; - for i in 0..16 { - let byte = u8::from_str_radix(&addr_hex[i * 2..i * 2 + 2], 16).unwrap(); - bytes[15 - i] = byte; - } - - SocketAddr::V6(SocketAddrV6::new(Ipv6Addr::from_octets(bytes), port, 0, 0)) - } - }; - - map.insert(inode, (sockaddr, proto)); - } - } - } - Ok(map) -} - -pub(super) async fn build_inode_sock_map() -> HashMap { - let mut inode_map = HashMap::new(); - - if let Ok(tcp_map) = parse_proc_net("/proc/net/tcp", Protocol::Tcp, IpVersion::V4).await { - inode_map.extend(tcp_map); - } - - if let Ok(tcp6_map) = parse_proc_net("/proc/net/tcp6", Protocol::Tcp, IpVersion::V6).await { - inode_map.extend(tcp6_map); - } - - if let Ok(udp_map) = parse_proc_net("/proc/net/udp", Protocol::Udp, IpVersion::V4).await { - inode_map.extend(udp_map); - } - - if let Ok(udp6_map) = parse_proc_net("/proc/net/udp6", Protocol::Udp, IpVersion::V6).await { - inode_map.extend(udp6_map); - } - - inode_map -} diff --git a/wallguard/src/netinfo/sock/linux/mod.rs b/wallguard/src/netinfo/sock/linux/mod.rs deleted file mode 100644 index fd189afc..00000000 --- a/wallguard/src/netinfo/sock/linux/mod.rs +++ /dev/null @@ -1,23 +0,0 @@ -use crate::netinfo::sock::SocketInfo; - -mod inode_pid; -mod inode_sock; - -pub(super) async fn get_sockets_info() -> Vec { - let sock_map = inode_sock::build_inode_sock_map().await; - let pid_map = inode_pid::build_inode_pid_map().await; - - let mut results = vec![]; - - for (inode, (sockaddr, protocol)) in sock_map { - if let Some(proc_name) = pid_map.get(&inode) { - results.push(SocketInfo { - process_name: proc_name.into(), - sockaddr, - protocol, - }); - } - } - - results -} diff --git a/wallguard/src/netinfo/sock/mod.rs b/wallguard/src/netinfo/sock/mod.rs index 7c4f6873..ef722dca 100644 --- a/wallguard/src/netinfo/sock/mod.rs +++ b/wallguard/src/netinfo/sock/mod.rs @@ -1,13 +1,6 @@ use std::net::SocketAddr; -#[cfg(target_os = "linux")] -mod linux; - -#[cfg(target_os = "freebsd")] -mod freebsd; - -#[cfg(target_os = "windows")] -mod windows; +use listeners::{Listener, Protocol as ListenersProtocol, SocketState}; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum Protocol { @@ -15,13 +8,6 @@ pub enum Protocol { Udp, } -#[allow(dead_code)] -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub(crate) enum IpVersion { - V4, - V6, -} - #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct SocketInfo { pub process_name: String, @@ -29,27 +15,31 @@ pub struct SocketInfo { pub sockaddr: SocketAddr, } -#[cfg(target_os = "linux")] -async fn get_sockets_info_impl() -> Vec { - linux::get_sockets_info().await -} - -#[cfg(target_os = "windows")] -async fn get_sockets_info_impl() -> Vec { - windows::get_sockets_info() -} - -#[cfg(target_os = "freebsd")] -async fn get_sockets_info_impl() -> Vec { - freebsd::get_sockets_info().await -} +fn into_socket_info(listener: Listener) -> Option { + let protocol = match listener.protocol { + // UDP has no listening state of its own, so every UDP socket is kept + // (matches the previous per-platform implementations' behavior). + ListenersProtocol::UDP => Protocol::Udp, + // Only TCP sockets actively accepting connections are candidate services. + ListenersProtocol::TCP if listener.state == SocketState::Listen => Protocol::Tcp, + ListenersProtocol::TCP => return None, + }; -// macOS socket enumeration is not yet implemented. -#[cfg(target_os = "macos")] -async fn get_sockets_info_impl() -> Vec { - vec![] + Some(SocketInfo { + process_name: listener.process.name, + protocol, + sockaddr: listener.socket, + }) } pub async fn get_sockets_info() -> Vec { - get_sockets_info_impl().await + tokio::task::spawn_blocking(|| { + listeners::get_all() + .unwrap_or_default() + .into_iter() + .filter_map(into_socket_info) + .collect() + }) + .await + .unwrap_or_default() } diff --git a/wallguard/src/netinfo/sock/windows/mod.rs b/wallguard/src/netinfo/sock/windows/mod.rs deleted file mode 100644 index d8a5fdad..00000000 --- a/wallguard/src/netinfo/sock/windows/mod.rs +++ /dev/null @@ -1,33 +0,0 @@ -mod proc_snap; -mod tcp; -mod udp; - -use proc_snap::snapshot_processes; -use tcp::{tcp_sockets, tcp6_sockets}; -use udp::{udp_sockets, udp6_sockets}; - -use super::{Protocol, SocketInfo}; - -pub(super) fn get_sockets_info() -> Vec { - let sockets = [ - (tcp_sockets().unwrap_or_default(), Protocol::Tcp), - (tcp6_sockets().unwrap_or_default(), Protocol::Tcp), - (udp_sockets().unwrap_or_default(), Protocol::Udp), - (udp6_sockets().unwrap_or_default(), Protocol::Udp), - ]; - - let snapshot = snapshot_processes().unwrap_or_default(); - - sockets - .iter() - .flat_map(|(socks, protocol)| { - socks.iter().filter_map(|(sockaddr, pid)| { - snapshot.get(pid).map(|proc_name| SocketInfo { - process_name: proc_name.into(), - protocol: *protocol, - sockaddr: *sockaddr, - }) - }) - }) - .collect() -} diff --git a/wallguard/src/netinfo/sock/windows/proc_snap.rs b/wallguard/src/netinfo/sock/windows/proc_snap.rs deleted file mode 100644 index 0329a13a..00000000 --- a/wallguard/src/netinfo/sock/windows/proc_snap.rs +++ /dev/null @@ -1,50 +0,0 @@ -use std::collections::HashMap; -use std::ffi::OsString; -use std::os::windows::ffi::OsStringExt; - -use winapi::um::handleapi::CloseHandle; -use winapi::um::tlhelp32::{ - CreateToolhelp32Snapshot, PROCESSENTRY32W, Process32FirstW, Process32NextW, TH32CS_SNAPPROCESS, -}; -use winapi::um::winnt::HANDLE; - -pub fn snapshot_processes() -> std::io::Result> { - let mut map = HashMap::::new(); - - unsafe { - let snapshot: HANDLE = CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0); - - if snapshot.is_null() { - return Err(std::io::Error::last_os_error()); - } - - let mut entry: PROCESSENTRY32W = std::mem::zeroed(); - entry.dwSize = std::mem::size_of::() as u32; - - if Process32FirstW(snapshot, &mut entry as *mut _) == 0 { - CloseHandle(snapshot); - return Err(std::io::Error::last_os_error()); - } - - loop { - let len = entry - .szExeFile - .iter() - .position(|&c| c == 0) - .unwrap_or(entry.szExeFile.len()); - let name = OsString::from_wide(&entry.szExeFile[..len]) - .to_string_lossy() - .into_owned(); - - map.insert(entry.th32ProcessID, name); - - if Process32NextW(snapshot, &mut entry as *mut _) == 0 { - break; - } - } - - CloseHandle(snapshot); - } - - Ok(map) -} diff --git a/wallguard/src/netinfo/sock/windows/tcp.rs b/wallguard/src/netinfo/sock/windows/tcp.rs deleted file mode 100644 index 917023b1..00000000 --- a/wallguard/src/netinfo/sock/windows/tcp.rs +++ /dev/null @@ -1,109 +0,0 @@ -use std::io; -use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6}; -use std::ptr; - -use winapi::shared::iprtrmib::TCP_TABLE_OWNER_PID_ALL; -use winapi::shared::minwindef::FALSE; -use winapi::shared::ntdef::ULONG; -use winapi::shared::tcpmib::{ - MIB_TCP_STATE_LISTEN, MIB_TCP6ROW_OWNER_PID, MIB_TCP6TABLE_OWNER_PID, MIB_TCPROW_OWNER_PID, - MIB_TCPTABLE_OWNER_PID, -}; -use winapi::shared::winerror::NO_ERROR; -use winapi::shared::ws2def::{AF_INET, AF_INET6}; -use winapi::um::iphlpapi::GetExtendedTcpTable; - -fn get_tcp_table(af: ULONG) -> io::Result> { - let mut size: u32 = 0; - - unsafe { - GetExtendedTcpTable( - ptr::null_mut(), - &mut size, - FALSE, - af, - TCP_TABLE_OWNER_PID_ALL, - 0, - ); - - let mut buffer = vec![0u8; size as usize]; - - let ret = GetExtendedTcpTable( - buffer.as_mut_ptr() as _, - &mut size, - FALSE, - af, - TCP_TABLE_OWNER_PID_ALL, - 0, - ); - - if ret != NO_ERROR { - return Err(io::Error::last_os_error()); - } - - Ok(buffer) - } -} - -pub(crate) fn tcp_sockets() -> io::Result> { - unsafe { - let buffer = get_tcp_table(AF_INET as ULONG)?; - let table_ptr = buffer.as_ptr() as *const MIB_TCPTABLE_OWNER_PID; - let num_entries = (*table_ptr).dwNumEntries as usize; - - let rows = std::slice::from_raw_parts( - &(*table_ptr).table as *const MIB_TCPROW_OWNER_PID, - num_entries, - ); - - let values = rows - .iter() - .filter(|row| row.dwState == MIB_TCP_STATE_LISTEN) - .map(|row| { - let addr = Ipv4Addr::new( - (row.dwLocalAddr & 0xff) as u8, - ((row.dwLocalAddr >> 8) & 0xff) as u8, - ((row.dwLocalAddr >> 16) & 0xff) as u8, - ((row.dwLocalAddr >> 24) & 0xff) as u8, - ); - let port = u16::from_be((row.dwLocalPort & 0xFFFF) as u16); - - ( - SocketAddr::V4(SocketAddrV4::new(addr, port)), - row.dwOwningPid, - ) - }) - .collect::>(); - - Ok(values) - } -} - -pub(crate) fn tcp6_sockets() -> io::Result> { - unsafe { - let buffer = get_tcp_table(AF_INET6 as ULONG)?; - let table_ptr = buffer.as_ptr() as *const MIB_TCP6TABLE_OWNER_PID; - let num_entries = (*table_ptr).dwNumEntries as usize; - - let rows = std::slice::from_raw_parts( - &(*table_ptr).table as *const MIB_TCP6ROW_OWNER_PID, - num_entries, - ); - - let values = rows - .iter() - .filter(|row| row.dwState == MIB_TCP_STATE_LISTEN) - .map(|row| { - let addr = Ipv6Addr::from(row.ucLocalAddr); - let port = u16::from_be((row.dwLocalPort & 0xFFFF) as u16); - - ( - SocketAddr::V6(SocketAddrV6::new(addr, port, 0, row.dwLocalScopeId)), - row.dwOwningPid, - ) - }) - .collect::>(); - - Ok(values) - } -} diff --git a/wallguard/src/netinfo/sock/windows/udp.rs b/wallguard/src/netinfo/sock/windows/udp.rs deleted file mode 100644 index c23fb1d9..00000000 --- a/wallguard/src/netinfo/sock/windows/udp.rs +++ /dev/null @@ -1,99 +0,0 @@ -use std::io; -use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6}; -use std::ptr; - -use winapi::shared::iprtrmib::UDP_TABLE_OWNER_PID; -use winapi::shared::minwindef::FALSE; -use winapi::shared::udpmib::{ - MIB_UDP6ROW_OWNER_PID, MIB_UDP6TABLE_OWNER_PID, MIB_UDPROW_OWNER_PID, MIB_UDPTABLE_OWNER_PID, -}; -use winapi::shared::ws2def::{AF_INET, AF_INET6}; -use winapi::um::iphlpapi::GetExtendedUdpTable; - -fn get_udp_table(af: u32) -> io::Result> { - let mut size: u32 = 0; - unsafe { - GetExtendedUdpTable( - ptr::null_mut(), - &mut size, - FALSE, - af, - UDP_TABLE_OWNER_PID, - 0, - ); - - let mut buffer = vec![0u8; size as usize]; - - let ret = GetExtendedUdpTable( - buffer.as_mut_ptr() as _, - &mut size, - FALSE, - af, - UDP_TABLE_OWNER_PID, - 0, - ); - - if ret != 0 { - return Err(io::Error::last_os_error()); - } - - Ok(buffer) - } -} - -pub(crate) fn udp_sockets() -> io::Result> { - unsafe { - let buffer = get_udp_table(AF_INET as u32)?; - let table_ptr = buffer.as_ptr() as *const MIB_UDPTABLE_OWNER_PID; - let count = (*table_ptr).dwNumEntries as usize; - - let rows = - std::slice::from_raw_parts(&(*table_ptr).table as *const MIB_UDPROW_OWNER_PID, count); - - let values = rows - .iter() - .map(|row| { - let addr = Ipv4Addr::new( - (row.dwLocalAddr & 0xff) as u8, - ((row.dwLocalAddr >> 8) & 0xff) as u8, - ((row.dwLocalAddr >> 16) & 0xff) as u8, - ((row.dwLocalAddr >> 24) & 0xff) as u8, - ); - let port = u16::from_be((row.dwLocalPort & 0xFFFF) as u16); - - ( - SocketAddr::V4(SocketAddrV4::new(addr, port)), - row.dwOwningPid, - ) - }) - .collect::>(); - - Ok(values) - } -} - -pub(crate) fn udp6_sockets() -> io::Result> { - unsafe { - let buffer = get_udp_table(AF_INET6 as u32)?; - let table_ptr = buffer.as_ptr() as *const MIB_UDP6TABLE_OWNER_PID; - let count = (*table_ptr).dwNumEntries as usize; - - let rows = - std::slice::from_raw_parts(&(*table_ptr).table as *const MIB_UDP6ROW_OWNER_PID, count); - - let values = rows - .iter() - .map(|row| { - let addr = Ipv6Addr::from(row.ucLocalAddr); - let port = u16::from_be((row.dwLocalPort & 0xFFFF) as u16); - - ( - SocketAddr::V6(SocketAddrV6::new(addr, port, 0, row.dwLocalScopeId)), - row.dwOwningPid, - ) - }) - .collect::>(); - - Ok(values) - } -} From cc9ba9c8244cf86a96af1221fb2847c1b9ea0fe3 Mon Sep 17 00:00:00 2001 From: Anton Liashkevich Date: Mon, 27 Jul 2026 21:27:29 -0400 Subject: [PATCH 4/6] Run cargo fmt Co-Authored-By: Claude Sonnet 5 --- .../data_transmission/resources/freebsd_disk_io.rs | 14 +++++--------- wallguard/src/netinfo/service/ssh.rs | 5 ++--- 2 files changed, 7 insertions(+), 12 deletions(-) diff --git a/wallguard/src/data_transmission/resources/freebsd_disk_io.rs b/wallguard/src/data_transmission/resources/freebsd_disk_io.rs index 33d13167..f38f364f 100644 --- a/wallguard/src/data_transmission/resources/freebsd_disk_io.rs +++ b/wallguard/src/data_transmission/resources/freebsd_disk_io.rs @@ -46,14 +46,7 @@ fn sysctl_string(name: &str) -> Option { let c_name = CString::new(name).ok()?; let mut len: size_t = 0; unsafe { - if libc::sysctlbyname( - c_name.as_ptr(), - null_mut(), - &mut len, - null_mut(), - 0, - ) != 0 - { + if libc::sysctlbyname(c_name.as_ptr(), null_mut(), &mut len, null_mut(), 0) != 0 { return None; } let mut buf = vec![0u8; len]; @@ -124,7 +117,10 @@ fn strip_partition_suffix(device: &str) -> String { /// Resolves a `zpool status` leaf device entry (a raw name like "ada0p3", or a /// label like "gpt/zfs0" / "gptid/") to the base GEOM disk name that /// devstat actually tracks. -fn resolve_base_disk(leaf: &str, label_mapping: &std::collections::HashMap) -> Option { +fn resolve_base_disk( + leaf: &str, + label_mapping: &std::collections::HashMap, +) -> Option { if leaf.contains('/') { label_mapping.get(&format!("/dev/{leaf}")).cloned() } else { diff --git a/wallguard/src/netinfo/service/ssh.rs b/wallguard/src/netinfo/service/ssh.rs index 96e9eeae..71e548f9 100644 --- a/wallguard/src/netinfo/service/ssh.rs +++ b/wallguard/src/netinfo/service/ssh.rs @@ -34,9 +34,8 @@ pub(super) async fn filter(sockets: &mut Vec) -> Vec { let mut set = tokio::task::JoinSet::new(); for socket in sockets.drain(..) { set.spawn(async move { - let matched = - matches!(socket.protocol, crate::netinfo::sock::Protocol::Tcp) - && is_ssh(socket.sockaddr).await; + let matched = matches!(socket.protocol, crate::netinfo::sock::Protocol::Tcp) + && is_ssh(socket.sockaddr).await; (socket, matched) }); } From 842906d33900e482918c6a1cbe8c1fa2c953016d Mon Sep 17 00:00:00 2001 From: Anton Liashkevich Date: Mon, 27 Jul 2026 21:33:48 -0400 Subject: [PATCH 5/6] Run cargo fmt --all across the workspace Co-Authored-By: Claude Sonnet 5 --- wallguard-cli/src/update.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/wallguard-cli/src/update.rs b/wallguard-cli/src/update.rs index f558cd0f..aacb0fa6 100644 --- a/wallguard-cli/src/update.rs +++ b/wallguard-cli/src/update.rs @@ -205,7 +205,10 @@ async fn apply_update(version: &str) -> AnyResult<()> { // for the lock at all. let new_version = poll_agent_version(20, Duration::from_millis(500)).await; - if new_version.as_deref().is_some_and(|v| versions_match(v, version)) { + if new_version + .as_deref() + .is_some_and(|v| versions_match(v, version)) + { let _ = std::fs::remove_file(&backup_path); println!("WallGuard successfully updated to v{version}."); return Ok(()); From 95955ee26bb95eb54dccaa1a872ff09407fa9d5e Mon Sep 17 00:00:00 2001 From: Anton Liashkevich Date: Mon, 27 Jul 2026 21:52:21 -0400 Subject: [PATCH 6/6] Fix clippy -D warnings failures in wallguard-server and freebsd disk-io for_kv_map (iterate lock.values() instead of (_, v) in lock.iter()) and needless_borrow in wallguard-server, plus collapsible_if in the FreeBSD-only GEOM label-mapping parser (invisible to clippy on non-FreeBSD hosts). Co-Authored-By: Claude Sonnet 5 --- wallguard-server/src/token.rs | 2 +- wallguard-server/src/tunneling/timeout_controller.rs | 2 +- .../data_transmission/resources/freebsd_disk_io.rs | 12 ++++++------ 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/wallguard-server/src/token.rs b/wallguard-server/src/token.rs index 8c82007f..d9f5a3ba 100644 --- a/wallguard-server/src/token.rs +++ b/wallguard-server/src/token.rs @@ -90,7 +90,7 @@ mod tests { fn test_token() { let jwt = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJhY2NvdW50Ijp7ImFjY291bnRfaWQiOiJxd2JxNDZqcWNsZXYiLCJhY2NvdW50X29yZ2FuaXphdGlvbl9pZCI6bnVsbCwiYWNjb3VudF9zdGF0dXMiOiJBY3RpdmUiLCJjb250YWN0Ijp7fSwiZGV2aWNlIjp7fSwiaWQiOiIwMUtSUE41RUpLUEhUSldUM1Y2WUM3NjhXQSIsIm9yZ2FuaXphdGlvbiI6eyJjYXRlZ29yaWVzIjpbIlBlcnNvbmFsIl0sImNvZGUiOiJPMDAwMDIxIiwiaWQiOiIwMUtSUE41RVA3QTdOWjJWQTU2SjBBOVNZSyIsIm5hbWUiOiJQZXJzb25hbCBPcmdhbml6YXRpb24iLCJvcmdhbml6YXRpb25faWQiOiIwMUtSUE41RVA3QTdOWjJWQTU2SjBBOVNZSyIsInBhcmVudF9vcmdhbml6YXRpb25faWQiOm51bGwsInN0YXR1cyI6IkFjdGl2ZSJ9LCJvcmdhbml6YXRpb25faWQiOiIwMUtSUE41RVA3QTdOWjJWQTU2SjBBOVNZSyIsInByb2ZpbGUiOnsiYWNjb3VudF9pZCI6IjAxS1JQTjVFSktQSFRKV1QzVjZZQzc2OFdBIiwiY2F0ZWdvcmllcyI6W10sImNvZGUiOm51bGwsImVtYWlsIjoicXdicTQ2anFjbGV2IiwiZmlyc3RfbmFtZSI6IiIsImlkIjoiMDFLUlBONUZCQ1cwRU5GWDNDNTBGOE01TVYiLCJsYXN0X25hbWUiOiIiLCJvcmdhbml6YXRpb25faWQiOiIwMUtSUE41RVA3QTdOWjJWQTU2SjBBOVNZSyIsInN0YXR1cyI6IkFjdGl2ZSJ9LCJyb2xlX2lkIjpudWxsLCJzZXNzaW9uSUQiOiIifSwiZXhwIjoxNzc4OTYzOTM4LCJpYXQiOjE3Nzg4Nzc1MzgsInJvbGVfbmFtZSI6IiIsInNlbnNpdGl2aXR5X2xldmVsIjoxMDAwLCJzZXNzaW9uSUQiOiIiLCJzaWduZWRfaW5fYWNjb3VudCI6eyJhY2NvdW50X2lkIjoicXdicTQ2anFjbGV2IiwiYWNjb3VudF9vcmdhbml6YXRpb25faWQiOm51bGwsImFjY291bnRfc3RhdHVzIjoiQWN0aXZlIiwiY29udGFjdCI6e30sImRldmljZSI6e30sImlkIjoiMDFLUlBONUVKS1BIVEpXVDNWNllDNzY4V0EiLCJvcmdhbml6YXRpb24iOnsiY2F0ZWdvcmllcyI6WyJQZXJzb25hbCJdLCJjb2RlIjoiTzAwMDAyMSIsImlkIjoiMDFLUlBONUVQN0E3TloyVkE1NkowQTlTWUsiLCJuYW1lIjoiUGVyc29uYWwgT3JnYW5pemF0aW9uIiwib3JnYW5pemF0aW9uX2lkIjoiMDFLUlBONUVQN0E3TloyVkE1NkowQTlTWUsiLCJwYXJlbnRfb3JnYW5pemF0aW9uX2lkIjpudWxsLCJzdGF0dXMiOiJBY3RpdmUifSwib3JnYW5pemF0aW9uX2lkIjoiMDFLUlBONUVQN0E3TloyVkE1NkowQTlTWUsiLCJwcm9maWxlIjp7ImFjY291bnRfaWQiOiIwMUtSUE41RUpLUEhUSldUM1Y2WUM3NjhXQSIsImNhdGVnb3JpZXMiOltdLCJjb2RlIjpudWxsLCJlbWFpbCI6InF3YnE0NmpxY2xldiIsImZpcnN0X25hbWUiOiIiLCJpZCI6IjAxS1JQTjVGQkNXMEVORlgzQzUwRjhNNU1WIiwibGFzdF9uYW1lIjoiIiwib3JnYW5pemF0aW9uX2lkIjoiMDFLUlBONUVQN0E3TloyVkE1NkowQTlTWUsiLCJzdGF0dXMiOiJBY3RpdmUifSwicm9sZV9pZCI6bnVsbCwic2Vzc2lvbklEIjoiIn19.uCwpxbfDo6-3v-2hkbgPisbEo0GzMaQUv9SxKXIhWfo"; - let result = Token::from_jwt(&jwt); + let result = Token::from_jwt(jwt); assert!(result.is_ok()); } diff --git a/wallguard-server/src/tunneling/timeout_controller.rs b/wallguard-server/src/tunneling/timeout_controller.rs index bd329dcd..3830ab8e 100644 --- a/wallguard-server/src/tunneling/timeout_controller.rs +++ b/wallguard-server/src/tunneling/timeout_controller.rs @@ -47,7 +47,7 @@ impl TimeoutController { let lock = self.tunnels.lock().await; - for (_, tunnel) in lock.iter() { + for tunnel in lock.values() { match tunnel { WallguardTunnel::Http(http_tunnel) => { let tun = http_tunnel.lock().await; diff --git a/wallguard/src/data_transmission/resources/freebsd_disk_io.rs b/wallguard/src/data_transmission/resources/freebsd_disk_io.rs index f38f364f..070ecfd0 100644 --- a/wallguard/src/data_transmission/resources/freebsd_disk_io.rs +++ b/wallguard/src/data_transmission/resources/freebsd_disk_io.rs @@ -91,12 +91,12 @@ fn disk_label_mapping() -> std::collections::HashMap { last_id.clear(); last_id.push_str(id); } - } else if kind == "2" && !last_id.is_empty() { - if let Some("LABEL") = parts.next() - && let Some(path) = parts.next() - { - mapping.insert(format!("/dev/{path}"), last_id.clone()); - } + } else if kind == "2" + && !last_id.is_empty() + && let Some("LABEL") = parts.next() + && let Some(path) = parts.next() + { + mapping.insert(format!("/dev/{path}"), last_id.clone()); } } mapping