diff --git a/fact/src/bpf/mod.rs b/fact/src/bpf/mod.rs index 1eb58add..c8a8beb4 100644 --- a/fact/src/bpf/mod.rs +++ b/fact/src/bpf/mod.rs @@ -1,4 +1,4 @@ -use std::{io, path::PathBuf}; +use std::io; use anyhow::{Context, bail}; use aya::{ @@ -7,7 +7,6 @@ use aya::{ programs::{Program, lsm::LsmLink}, }; use checks::Checks; -use globset::{Glob, GlobSet, GlobSetBuilder}; use libc::c_char; use log::{error, info, warn}; use tokio::{ @@ -16,7 +15,12 @@ use tokio::{ task::JoinSet, }; -use crate::{config::BpfConfig, event::Event, host_info, metrics::EventCounter}; +use crate::{ + config::{BpfConfig, PathsConfig}, + event::Event, + host_info, + metrics::EventCounter, +}; use fact_ebpf::{LPM_SIZE_MAX, event_t, inode_key_t, inode_value_t, metrics_t, path_prefix_t}; @@ -30,11 +34,9 @@ pub struct Bpf { tx: mpsc::Sender, - paths_config: watch::Receiver>, + paths_config: watch::Receiver, paths_lpm_map: LpmTrie, - paths_globset: GlobSet, - links: Vec, running: watch::Receiver, @@ -43,7 +45,7 @@ pub struct Bpf { impl Bpf { pub fn new( - paths_config: watch::Receiver>, + paths_config: watch::Receiver, bpf_config: &BpfConfig, running: watch::Receiver, metrics: EventCounter, @@ -66,7 +68,6 @@ impl Bpf { checks, tx, paths_config, - paths_globset: GlobSet::empty(), paths_lpm_map, links: Vec::new(), running, @@ -176,10 +177,9 @@ impl Bpf { } fn load_paths(&mut self) -> anyhow::Result<()> { - if self.paths_config.borrow().is_empty() { + if self.paths_config.borrow().patterns().is_empty() { self.detach_progs(); self.cleanup_lpm_map(&[])?; - self.paths_globset = GlobSet::empty(); return Ok(()); } @@ -190,22 +190,13 @@ impl Bpf { // Add the new prefixes let new_paths = { let paths_config = self.paths_config.borrow(); - let mut new_paths = Vec::with_capacity(paths_config.len()); - let mut builder = GlobSetBuilder::new(); - for p in paths_config.iter() { - let Some(glob_str) = p.to_str() else { - bail!("failed to convert path {} to string", p.display()); - }; - - builder.add( - Glob::new(glob_str).with_context(|| format!("invalid glob {}", glob_str))?, - ); - + let patterns = paths_config.patterns(); + let mut new_paths = Vec::with_capacity(patterns.len()); + for p in patterns { let prefix = path_prefix_t::try_from(p)?; self.paths_lpm_map.insert(&prefix.into(), 0, 0)?; new_paths.push(prefix); } - self.paths_globset = builder.build()?; new_paths }; @@ -318,7 +309,7 @@ impl Bpf { // so we let the event go into HostScanner and make the // decision there. if !event.is_monitored_by_parent() && - event.is_ignored(&self.paths_globset) { + event.is_ignored(&self.paths_config.borrow().globset) { self.metrics.dropped(); continue; } @@ -395,7 +386,7 @@ mod bpf_tests { let mut config = FactConfig::default(); config.set_paths(paths); let bpf_config = config.bpf.clone(); - let reloader = Reloader::from(config); + let reloader = Reloader::try_from(config).unwrap(); let metrics = Metrics::new(); let (run_tx, run_rx) = watch::channel(true); let (bpf, mut rx) = Bpf::new( diff --git a/fact/src/config/mod.rs b/fact/src/config/mod.rs index 1542f86a..3a062ef6 100644 --- a/fact/src/config/mod.rs +++ b/fact/src/config/mod.rs @@ -10,6 +10,7 @@ use std::{ use anyhow::{Context, bail}; use clap::Parser; +use globset::{Glob, GlobSet}; use log::info; use yaml_rust2::{Yaml, YamlLoader, yaml}; @@ -33,7 +34,7 @@ fn yaml_to_duration_secs(v: &Yaml) -> Option { #[derive(Debug, Default, PartialEq, Clone)] pub struct FactConfig { - paths: Option>, + paths: PathsConfig, pub grpc: GrpcConfig, pub otel: OTelConfig, pub endpoint: EndpointConfig, @@ -82,10 +83,7 @@ impl FactConfig { } pub fn update(&mut self, from: &FactConfig) { - if let Some(paths) = from.paths.as_deref() { - self.paths = Some(paths.to_owned()); - } - + self.paths.update(&from.paths); self.grpc.update(&from.grpc); self.otel.update(&from.otel); self.endpoint.update(&from.endpoint); @@ -116,10 +114,6 @@ impl FactConfig { } } - pub fn paths(&self) -> &[PathBuf] { - self.paths.as_ref().map(|v| v.as_ref()).unwrap_or(&[]) - } - pub fn skip_pre_flight(&self) -> bool { self.skip_pre_flight.unwrap_or(false) } @@ -146,7 +140,7 @@ impl FactConfig { #[cfg(test)] pub fn set_paths(&mut self, paths: Vec) { - self.paths = Some(paths); + self.paths = paths.try_into().expect("Invalid paths"); } } @@ -188,21 +182,10 @@ impl TryFrom> for FactConfig { match k { "paths" if v.is_array() => { - let paths = v - .as_vec() - .unwrap() - .iter() - .map(|p| { - let Some(p) = p.as_str() else { - bail!("Path has invalid type: {p:?}"); - }; - Ok(PathBuf::from(p)) - }) - .collect::>()?; - config.paths = Some(paths); + config.paths = v.as_vec().unwrap().try_into()?; } "paths" if v.is_null() => { - config.paths = Some(Vec::new()); + config.paths = PathsConfig::empty(); } "grpc" if v.is_hash() => { let grpc = v.as_hash().unwrap(); @@ -271,6 +254,83 @@ impl TryFrom> for FactConfig { } } +#[derive(Debug, Default, Clone)] +pub struct PathsConfig { + patterns: Option>, + pub globset: GlobSet, +} + +impl PathsConfig { + const fn empty() -> Self { + PathsConfig { + patterns: Some(Vec::new()), + globset: GlobSet::empty(), + } + } + + fn globset_build<'a>(patterns: impl Iterator) -> anyhow::Result { + let mut builder = GlobSet::builder(); + for p in patterns { + let Some(p) = p.to_str() else { + bail!("paths item has invalid UTF-8: {}", p.display()); + }; + let p = Glob::new(p).with_context(|| format!("invalid glob {}", p))?; + builder.add(p); + } + + builder.build().map_err(anyhow::Error::from) + } + + fn update(&mut self, other: &Self) { + if other.patterns.is_some() { + self.patterns = other.patterns.clone(); + self.globset = other.globset.clone(); + } + } + + pub fn patterns(&self) -> &[PathBuf] { + self.patterns.as_deref().unwrap_or(&[]) + } +} + +impl PartialEq for PathsConfig { + fn eq(&self, other: &Self) -> bool { + self.patterns() == other.patterns() + } +} + +impl TryFrom<&yaml::Array> for PathsConfig { + type Error = anyhow::Error; + + fn try_from(value: &yaml::Array) -> Result { + let paths = value + .iter() + .map(|p| match p.as_str() { + Some(p) => Ok(p.into()), + None => bail!("paths field has invalid type: {p:?}"), + }) + .collect::, _>>()?; + let globset = PathsConfig::globset_build(paths.iter())?; + + Ok(PathsConfig { + patterns: Some(paths), + globset, + }) + } +} + +impl TryFrom> for PathsConfig { + type Error = anyhow::Error; + + fn try_from(paths: Vec) -> Result { + let globset = PathsConfig::globset_build(paths.iter())?; + Ok(PathsConfig { + patterns: Some(paths), + globset, + }) + } +} + #[derive(Debug, Default, PartialEq, Eq, Clone)] pub struct EndpointConfig { address: Option, @@ -915,7 +975,10 @@ pub struct FactCli { impl FactCli { fn into_config(self) -> FactConfig { FactConfig { - paths: self.paths, + paths: self + .paths + .map(|patterns| patterns.try_into().expect("Invalid paths configuration")) + .unwrap_or_default(), grpc: GrpcConfig { url: self.url, certs: self.certs, diff --git a/fact/src/config/reloader/mod.rs b/fact/src/config/reloader/mod.rs index cb99b898..78e60742 100644 --- a/fact/src/config/reloader/mod.rs +++ b/fact/src/config/reloader/mod.rs @@ -8,7 +8,7 @@ use tokio::{ time::interval, }; -use crate::config::OTelConfig; +use crate::config::{OTelConfig, PathsConfig}; use super::{CONFIG_FILES, EndpointConfig, FactConfig, GrpcConfig}; @@ -18,7 +18,7 @@ pub struct Reloader { endpoint: watch::Sender, grpc: watch::Sender, otel: watch::Sender, - paths: watch::Sender>, + paths: watch::Sender, files: HashMap<&'static str, (i64, i64)>, scan_interval: watch::Sender, rate_limit: watch::Sender, @@ -80,7 +80,7 @@ impl Reloader { /// Subscribe to get notifications when paths configuration is /// changed. - pub fn paths(&self) -> watch::Receiver> { + pub fn paths(&self) -> watch::Receiver { self.paths.subscribe() } @@ -147,17 +147,6 @@ impl Reloader { /// Propagate configuration changes to all subscribers that need it fn send_updates(&self, new: FactConfig) { - self.paths.send_if_modified(|old| { - let new = new.paths(); - if *old != new { - debug!("Sending new paths configuration..."); - *old = new.to_vec(); - true - } else { - false - } - }); - self.scan_interval.send_if_modified(|old| { let new = new.scan_interval(); if *old != new { @@ -188,9 +177,20 @@ impl Reloader { endpoint, grpc, otel, + paths, .. } = new; + self.paths.send_if_modified(|old| { + if *old != paths { + debug!("Sending new paths configuration..."); + *old = paths; + true + } else { + false + } + }); + self.endpoint.send_if_modified(|old| { if *old != endpoint { debug!("Sending new endpoint configuration..."); @@ -242,8 +242,10 @@ impl Reloader { } } -impl From for Reloader { - fn from(config: FactConfig) -> Self { +impl TryFrom for Reloader { + type Error = anyhow::Error; + + fn try_from(config: FactConfig) -> Result { let files = CONFIG_FILES .iter() .filter_map(|path| { @@ -265,7 +267,6 @@ impl From for Reloader { .collect(); let enabled = config.hotreload(); - let (paths, _) = watch::channel(config.paths().to_vec()); let (scan_interval, _) = watch::channel(config.scan_interval()); let (rate_limit, _) = watch::channel(config.rate_limit()); @@ -273,15 +274,17 @@ impl From for Reloader { endpoint, grpc, otel, + paths, .. } = config; let (endpoint, _) = watch::channel(endpoint); let (grpc, _) = watch::channel(grpc); let (otel, _) = watch::channel(otel); + let (paths, _) = watch::channel(paths); let trigger = Arc::new(Notify::new()); - Reloader { + Ok(Reloader { enabled, endpoint, grpc, @@ -291,6 +294,6 @@ impl From for Reloader { rate_limit, files, trigger, - } + }) } } diff --git a/fact/src/config/reloader/tests.rs b/fact/src/config/reloader/tests.rs index a5e128b5..ab2d673e 100644 --- a/fact/src/config/reloader/tests.rs +++ b/fact/src/config/reloader/tests.rs @@ -32,7 +32,7 @@ macro_rules! generate_test { ($testname:ident, $channel:ident, $old:expr, $new:expr, $expected:expr) => { #[test] fn $testname() { - let reloader = Reloader::from($old); + let reloader = Reloader::try_from($old).unwrap(); let channel = reloader.$channel(); reloader.send_updates($new); @@ -57,7 +57,7 @@ generate_paths_test! { test_reloader_paths_from_default_to_empty, FactConfig::default(), FactConfig { - paths: Some(vec![]), + paths: PathsConfig::default(), ..Default::default() }, None @@ -65,23 +65,23 @@ generate_paths_test! { generate_paths_test! { test_reloader_paths_config_change, FactConfig { - paths: Some(vec!["/home".into()]), + paths: vec!["/home".into()].try_into().unwrap(), ..Default::default() }, FactConfig { - paths: Some(vec!["/etc".into()]), + paths: vec!["/etc".into()].try_into().unwrap(), ..Default::default() }, - Some(vec![PathBuf::from("/etc")]) + Some(vec![PathBuf::from("/etc")].try_into().unwrap()) } generate_paths_test! { test_reloader_paths_no_config_change, FactConfig { - paths: Some(vec!["/home".into()]), + paths: vec!["/home".into()].try_into().unwrap(), ..Default::default() }, FactConfig { - paths: Some(vec!["/home".into()]), + paths: vec!["/home".into()].try_into().unwrap(), scan_interval: Some(Duration::from_secs(10)), ..Default::default() }, @@ -156,7 +156,7 @@ generate_scan_interval_test! { }, FactConfig { scan_interval: Some(Duration::from_secs(60)), - paths: Some(vec!["/etc".into()]), + paths: vec!["/etc".into()].try_into().unwrap(), ..Default::default() }, None @@ -230,7 +230,7 @@ generate_rate_limit_test! { }, FactConfig { rate_limit: Some(1000), - paths: Some(vec!["/etc".into()]), + paths: vec!["/etc".into()].try_into().unwrap(), ..Default::default() }, None @@ -266,7 +266,7 @@ generate_endpoint_test! { health_check: Some(true), introspection: Some(true), }, - paths: Some(vec!["/etc".into()]), + paths: vec!["/etc".into()].try_into().unwrap(), ..Default::default() }, None @@ -546,7 +546,7 @@ generate_grpc_test! { retries_max: Some(GRPC_BACKOFF_RETRIES_NEW), } }, - paths: Some(vec!["/etc".into()]), + paths: vec!["/etc".into()].try_into().unwrap(), ..Default::default() }, None @@ -1412,7 +1412,7 @@ generate_otel_test! { otel: OTelConfig { endpoint: Some(OTEL_ENDPOINT_NEW.into()), }, - paths: Some(vec!["/etc".into()]), + paths: vec!["/etc".into()].try_into().unwrap(), ..Default::default() }, None diff --git a/fact/src/config/tests.rs b/fact/src/config/tests.rs index 596ca711..b7d3031e 100644 --- a/fact/src/config/tests.rs +++ b/fact/src/config/tests.rs @@ -12,14 +12,16 @@ fn parsing() { ( "paths:", FactConfig { - paths: Some(Vec::new()), + paths: PathsConfig::default(), ..Default::default() }, ), ( "paths: [/etc, /bin]", FactConfig { - paths: Some(vec![PathBuf::from("/etc"), PathBuf::from("/bin")]), + paths: vec![PathBuf::from("/etc"), PathBuf::from("/bin")] + .try_into() + .unwrap(), ..Default::default() }, ), @@ -509,7 +511,7 @@ fn parsing() { replay: /some/path.jsonl "#, FactConfig { - paths: Some(vec![PathBuf::from("/etc")]), + paths: vec![PathBuf::from("/etc")].try_into().unwrap(), grpc: GrpcConfig { url: Some(String::from("https://svc.sensor.stackrox:9090")), certs: Some(PathBuf::from("/etc/stackrox/certs")), @@ -594,7 +596,7 @@ paths: ("- something", "Wrong configuration type"), ("true: something", "key is not string: Boolean(true)"), ("4: something", "key is not string: Integer(4)"), - ("paths: [4]", "Path has invalid type: Integer(4)"), + ("paths: [4]", "paths field has invalid type: Integer(4)"), ( "grpc: true", "Invalid field 'grpc' with value: Boolean(true)", @@ -985,7 +987,7 @@ fn update() { "paths:", FactConfig::default(), FactConfig { - paths: Some(Vec::new()), + paths: PathsConfig::default(), ..Default::default() }, ), @@ -993,40 +995,61 @@ fn update() { "paths: [/etc, /bin]", FactConfig::default(), FactConfig { - paths: Some(vec![PathBuf::from("/etc"), PathBuf::from("/bin")]), + paths: vec![PathBuf::from("/etc"), PathBuf::from("/bin")] + .try_into() + .unwrap(), ..Default::default() }, ), ( "paths: [/bin]", FactConfig { - paths: Some(vec![PathBuf::from("/etc")]), + paths: vec![PathBuf::from("/etc")].try_into().unwrap(), ..Default::default() }, FactConfig { - paths: Some(vec![PathBuf::from("/bin")]), + paths: vec![PathBuf::from("/bin")].try_into().unwrap(), ..Default::default() }, ), ( "paths:", FactConfig { - paths: Some(vec![PathBuf::from("/etc")]), + paths: vec![PathBuf::from("/etc")].try_into().unwrap(), ..Default::default() }, FactConfig { - paths: Some(Vec::new()), + paths: PathsConfig::default(), ..Default::default() }, ), ( "paths: [/etc, /bin]", FactConfig { - paths: Some(vec![PathBuf::from("/etc"), PathBuf::from("/bin")]), + paths: vec![PathBuf::from("/etc"), PathBuf::from("/bin")] + .try_into() + .unwrap(), ..Default::default() }, FactConfig { - paths: Some(vec![PathBuf::from("/etc"), PathBuf::from("/bin")]), + paths: vec![PathBuf::from("/etc"), PathBuf::from("/bin")] + .try_into() + .unwrap(), + ..Default::default() + }, + ), + ( + "", + FactConfig { + paths: vec![PathBuf::from("/etc"), PathBuf::from("/bin")] + .try_into() + .unwrap(), + ..Default::default() + }, + FactConfig { + paths: vec![PathBuf::from("/etc"), PathBuf::from("/bin")] + .try_into() + .unwrap(), ..Default::default() }, ), @@ -1936,7 +1959,9 @@ fn update() { rate_limit: 1000 "#, FactConfig { - paths: Some(vec![PathBuf::from("/etc"), PathBuf::from("/bin")]), + paths: vec![PathBuf::from("/etc"), PathBuf::from("/bin")] + .try_into() + .unwrap(), grpc: GrpcConfig { url: Some(String::from("http://localhost")), certs: Some(PathBuf::from("/etc/certs")), @@ -1976,7 +2001,7 @@ fn update() { replay: None, }, FactConfig { - paths: Some(vec![PathBuf::from("/etc")]), + paths: vec![PathBuf::from("/etc")].try_into().unwrap(), grpc: GrpcConfig { url: Some(String::from("https://svc.sensor.stackrox:9090")), certs: Some(PathBuf::from("/etc/stackrox/certs")), @@ -2038,8 +2063,8 @@ fn update() { #[test] fn defaults() { let config = FactConfig::default(); - let default_paths: &[PathBuf] = &[]; - assert_eq!(config.paths(), default_paths); + assert!(config.paths.patterns().is_empty()); + assert!(config.paths.globset.is_empty()); assert_eq!(config.grpc.url(), None); assert_eq!(config.grpc.certs(), None); assert_eq!( @@ -2223,7 +2248,9 @@ fn env_vars() { value: "/etc:/var/log", }, FactConfig { - paths: Some(vec![PathBuf::from("/etc"), PathBuf::from("/var/log")]), + paths: vec![PathBuf::from("/etc"), PathBuf::from("/var/log")] + .try_into() + .unwrap(), ..Default::default() }, ), @@ -2580,7 +2607,7 @@ fn env_vars_override_yaml() { }, "paths:\n- /etc", FactConfig { - paths: Some(vec![PathBuf::from("/var/log")]), + paths: vec![PathBuf::from("/var/log")].try_into().unwrap(), ..Default::default() }, ), diff --git a/fact/src/host_scanner.rs b/fact/src/host_scanner.rs index 1ea00a5d..83a87103 100644 --- a/fact/src/host_scanner.rs +++ b/fact/src/host_scanner.rs @@ -36,7 +36,6 @@ use aya::{ sys::SyscallError, }; use fact_ebpf::{inode_key_t, inode_value_t, monitored_t}; -use globset::{Glob, GlobSet, GlobSetBuilder}; use log::{debug, info, warn}; use serde::{Serialize, ser::SerializeMap}; use tokio::{ @@ -47,6 +46,7 @@ use tokio::{ use crate::{ bpf::Bpf, + config::PathsConfig, event::Event, host_info, metrics::host_scanner::{HostScannerMetrics, ScanLabels}, @@ -95,7 +95,7 @@ pub struct HostScanner { kernel_inode_map: RefCell>, inode_map: RefCell, - paths: watch::Receiver>, + paths: watch::Receiver, scan_interval: watch::Receiver, rx: mpsc::Receiver, @@ -103,15 +103,13 @@ pub struct HostScanner { introspection: mpsc::Receiver>>, metrics: HostScannerMetrics, - - paths_globset: GlobSet, } impl HostScanner { pub fn new( bpf: &mut Bpf, rx: mpsc::Receiver, - paths: watch::Receiver>, + paths: watch::Receiver, scan_interval: watch::Receiver, metrics: HostScannerMetrics, introspection: mpsc::Receiver>>, @@ -119,7 +117,6 @@ impl HostScanner { let kernel_inode_map = RefCell::new(bpf.take_inode_map()?); let inode_map = RefCell::new(InodeMap::new()); let (tx, output) = mpsc::channel(100); - let paths_globset = HostScanner::build_globset(paths.borrow().as_slice())?; let host_scanner = HostScanner { kernel_inode_map, @@ -130,7 +127,6 @@ impl HostScanner { tx, introspection, metrics, - paths_globset, }; // Run an initial scan to fill in the inode map @@ -139,36 +135,18 @@ impl HostScanner { Ok((host_scanner, output)) } - fn build_globset(paths: &[PathBuf]) -> anyhow::Result { - let mut builder = GlobSetBuilder::new(); - for p in paths.iter() { - let Some(glob_str) = p.to_str() else { - bail!("failed to convert path {} to string", p.display()); - }; - - builder.add( - Glob::new(glob_str) - .with_context(|| format!("invalid glob {}", glob_str)) - .unwrap(), - ); - } - Ok(builder.build()?) - } - fn scan(&self) -> anyhow::Result<()> { info!("Host scan started"); let start = Instant::now(); self.metrics.scan_inc(ScanLabels::Scans); - let config = self.paths.borrow(); + let paths = self.paths.borrow(); // Cleanup any items that are either: // * Not configured to be monitored anymore. // * Are configured to be monitored but no longer are found in // the file system. self.inode_map.borrow_mut().retain(|inode, path| { - if config.iter().any(|prefix| path.starts_with(prefix)) - && host_info::prepend_host_mount(path).exists() - { + if paths.globset.is_match(&path) && host_info::prepend_host_mount(path).exists() { true } else { let _ = self.kernel_inode_map.borrow_mut().remove(inode); @@ -177,7 +155,7 @@ impl HostScanner { } }); - for pattern in self.paths.borrow().iter() { + for pattern in paths.patterns() { let path = host_info::prepend_host_mount(pattern); self.scan_inner(&path)?; } @@ -437,7 +415,7 @@ You can increase this limit with: unreachable!("Rename event did not have an old host path"); }; - if self.paths_globset.is_match(&new_host_path) { + if self.paths.borrow().globset.is_match(&new_host_path) { // New path needs to be tracked. // Move all entries for the old host path to the new one for path in inode_map.values_mut() { @@ -540,11 +518,12 @@ You can increase this limit with: /// the host paths for matches in events that are monitored by /// parent. fn event_is_ignored(&self, event: &Event) -> bool { - event.is_ignored(&self.paths_globset) - && !self.paths_globset.is_match(event.get_host_path()) + let paths = self.paths.borrow(); + event.is_ignored(&paths.globset) + && !paths.globset.is_match(event.get_host_path()) && event .get_old_host_path() - .is_none_or(|path| !self.paths_globset.is_match(path)) + .is_none_or(|path| !paths.globset.is_match(path)) } pub fn start(mut self, task_set: &mut JoinSet>) { @@ -640,7 +619,6 @@ You can increase this limit with: } _ = scan_trigger.notified() => self.scan()?, _ = self.paths.changed() => { - self.paths_globset = HostScanner::build_globset(self.paths.borrow().as_slice())?; self.scan()?; } } diff --git a/fact/src/lib.rs b/fact/src/lib.rs index 0969615b..4dc99450 100644 --- a/fact/src/lib.rs +++ b/fact/src/lib.rs @@ -120,7 +120,7 @@ pub async fn run(config: FactConfig) -> anyhow::Result<()> { let metrics_userspace = Metrics::new(); let mut task_set = JoinSet::new(); - let reloader = config::reloader::Reloader::from(config); + let reloader = config::reloader::Reloader::try_from(config)?; let config_trigger = reloader.get_trigger(); let setup_args = SetupArgs {