datasecurity: add dd-sds scanning module with config parsing and scan tests

Author: aimenebelfodil

pkg/collector/sharedlibrary/rustchecks/Cargo.lock

@@ -6,9 +6,13 @@ edition = "2024"
 [dependencies]
 anyhow = "1.0.100"
 core = { path = "../../core" }
+dd_sds = { package = "dd-sensitive-data-scanner", version = "=0.1.0-20260715-fcd8f9716b61", default-features = false, features = ["dd-sds"] }
 libc = "0.2.182"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 
+[dev-dependencies]
+serde_yaml = "0.9.34"
+
 [lib]
 crate-type = ["cdylib"]

pkg/collector/sharedlibrary/rustchecks/checks/datasecurity/Cargo.toml

@@ -3,35 +3,43 @@ use core::*;
 use serde_json::Value;
 
 use crate::config::{CheckConfig, SubTask};
-use crate::payload::{Match, ScanEventPayload};
+use crate::payload::ScanEventPayload;
+use crate::scanning::Scanner;
 
 /// Check implementation (scaffolding).
 pub fn check(check: &AgentCheck) -> Result<()> {
     let config = CheckConfig::from_instance(check)?;
     println!(
-        "datasecurity: check started (task_id={}, {} sub task(s))",
+        "datasecurity: check started (task_id={}, {} rule(s), {} sub task(s))",
         config.task_id,
+        config.scanning_rules.len(),
         config.scan_data.len()
     );
 
+    let scanner = Scanner::new(&config.scanning_rules).context("creating sds scanner")?;
+
     for sub_task in &config.scan_data {
-        run_sub_task(check, &config, sub_task)?;
+        run_sub_task(check, &config, &scanner, sub_task)?;
     }
 
     println!("datasecurity: check completed");
     Ok(())
 }
 
-fn run_sub_task(check: &AgentCheck, config: &CheckConfig, sub_task: &SubTask) -> Result<()> {
+fn run_sub_task(
+    check: &AgentCheck,
+    config: &CheckConfig,
+    scanner: &Scanner,
+    sub_task: &SubTask,
+) -> Result<()> {
     println!(
         "datasecurity: running sub task (sub_task_id={})",
         sub_task.sub_task_id
     );
 
     // TODO(DSEC-139): fetch the rows from postgres.
     let data = fetch_data(sub_task);
-    // TODO(DSEC-138): scan the rows with the SDS scanner.
-    let matches = scan(&data);
+    let matches = scanner.scan(&data).context("scanning sub task data")?;
 
     let payload = ScanEventPayload {
         task_id: config.task_id.clone(),
@@ -68,40 +76,3 @@ fn run_sub_task(check: &AgentCheck, config: &CheckConfig, sub_task: &SubTask) ->
 fn fetch_data(sub_task: &SubTask) -> Value {
     sub_task.dummy_response.clone()
 }
-
-/// Placeholder scan over the returned columns.
-// TODO(DSEC-138): replace with the SDS scanner.
-fn scan(scan_result: &Value) -> Vec<Match> {
-    let mut matches = Vec::new();
-    if let Some(emails) = scan_result.get("email").and_then(Value::as_array) {
-        let count = emails
-            .iter()
-            .filter(|value| value.as_str().is_some_and(|email| email.contains('@')))
-            .count() as i64;
-        if count > 0 {
-            matches.push(Match {
-                rule_id: "email-scanner".to_string(),
-                column_name: "email".to_string(),
-                count_matched_rows: count,
-            });
-        }
-    }
-    matches
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use serde_json::json;
-
-    #[test]
-    fn scan_dummy_response_returns_email_matches() {
-        let scan_result = json!({
-            "email": ["alice@example.com", "bob@test.com", "charlie@corp.com"],
-        });
-        let matches = scan(&scan_result);
-        assert_eq!(matches.len(), 1);
-        assert_eq!(matches[0].column_name, "email");
-        assert_eq!(matches[0].count_matched_rows, 3);
-    }
-}

pkg/collector/sharedlibrary/rustchecks/checks/datasecurity/src/check.rs

@@ -3,11 +3,19 @@ use core::AgentCheck;
 use serde::Deserialize;
 use serde_json::Value;
 
+use crate::scanning::ScanningRule;
+
 impl CheckConfig {
     /// Reads the check instance config into a `CheckConfig`.
     pub fn from_instance(check: &AgentCheck) -> Result<Self> {
         Ok(Self {
             task_id: check.instance.get("task_id").unwrap_or_default(),
+            // `scanning_rules` is common to every sub task; `scan_data` is the
+            // list of sub tasks to run against it.
+            scanning_rules: check
+                .instance
+                .get("scanning_rules")
+                .context("reading scanning_rules from instance config")?,
             scan_data: check
                 .instance
                 .get("scan_data")
@@ -26,6 +34,8 @@ pub struct CheckConfig {
     #[serde(default)]
     pub task_id: String,
     #[serde(default)]
+    pub scanning_rules: Vec<ScanningRule>,
+    #[serde(default)]
     pub scan_data: Vec<SubTask>,
 }
 

pkg/collector/sharedlibrary/rustchecks/checks/datasecurity/src/config.rs

@@ -5,6 +5,7 @@ use check::check;
 
 mod config;
 mod payload;
+mod scanning;
 mod version;
 use version::VERSION;
 

pkg/collector/sharedlibrary/rustchecks/checks/datasecurity/src/lib.rs

@@ -0,0 +1,110 @@
+//! Self-contained SDS scanning: rule config parsing (`rule`) plus building and
+//! running the dd-sds scanner over column-oriented query results. Everything
+//! scanning-related lives in this module so the rest of the check stays lean.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+
+use anyhow::{Context, Result};
+use dd_sds::{RootRuleConfig, RuleConfig, RuleMatch, Scanner as SdsScanner, ScannerBuilder};
+use serde_json::{Map, Value};
+
+use crate::payload::Match;
+
+mod rule;
+pub use rule::ScanningRule;
+
+#[cfg(test)]
+mod tests;
+
+/// A built dd-sds scanner plus the rule ids in builder order. dd-sds reports
+/// matches by rule index, so we keep the ids to map matches back to our rules.
+pub struct Scanner {
+    scanner: SdsScanner,
+    rule_ids: Vec<String>,
+}
+
+impl Scanner {
+    /// Builds a scanner from the check's scanning rules. The full dd-sds rule
+    /// surface (pattern, proximity keywords, suppressions, precedence, secondary
+    /// validators such as Luhn/JWT, ...) is supported because each rule carries
+    /// the flattened `RootRuleConfig` verbatim.
+    pub fn new(rules: &[ScanningRule]) -> Result<Self> {
+        let mut rule_ids = Vec::with_capacity(rules.len());
+        let compiled: Vec<RootRuleConfig<Arc<dyn RuleConfig>>> = rules
+            .iter()
+            .map(|rule| {
+                rule_ids.push(rule.id.clone());
+                rule.config.clone().into_dyn()
+            })
+            .collect();
+
+        let scanner = ScannerBuilder::new(&compiled)
+            .build()
+            .context("building sds scanner")?;
+        Ok(Self { scanner, rule_ids })
+    }
+
+    /// Scans a column-oriented result (`{ column: [values...] }`) and returns
+    /// one `Match` per (column, rule) pair with the count of matched rows.
+    pub fn scan(&self, data: &Value) -> Result<Vec<Match>> {
+        let mut event = Map::new();
+        if let Some(columns) = data.as_object() {
+            for (name, values) in columns {
+                let str_values: Vec<Value> = values
+                    .as_array()
+                    .map(|vs| {
+                        vs.iter()
+                            .filter_map(|v| v.as_str().map(|s| Value::String(s.to_string())))
+                            .collect()
+                    })
+                    .unwrap_or_default();
+                event.insert(name.clone(), Value::Array(str_values));
+            }
+        }
+
+        let mut event_value = Value::Object(event);
+        let hits = self
+            .scanner
+            .scan(&mut event_value)
+            .context("scanning query result")?;
+
+        Ok(aggregate_matches(&self.rule_ids, &hits))
+    }
+}
+
+/// Groups raw dd-sds hits into `(column, rule)` pairs, counting distinct matched
+/// rows (paths) per pair. Output is sorted so emitted events are deterministic.
+fn aggregate_matches(rule_ids: &[String], hits: &[RuleMatch]) -> Vec<Match> {
+    let mut rows: HashMap<(String, String), HashSet<String>> = HashMap::new();
+    for hit in hits {
+        let path = hit.path.to_string();
+        let column = column_name_from_path(&path);
+        let rule_id = rule_ids.get(hit.rule_index).cloned().unwrap_or_default();
+        rows.entry((column, rule_id)).or_default().insert(path);
+    }
+
+    let mut matches: Vec<Match> = rows
+        .into_iter()
+        .map(|((column, rule_id), paths)| Match {
+            rule_id,
+            column_name: column,
+            count_matched_rows: paths.len() as i64,
+        })
+        .collect();
+
+    matches.sort_by(|a, b| {
+        a.column_name
+            .cmp(&b.column_name)
+            .then_with(|| a.rule_id.cmp(&b.rule_id))
+    });
+    matches
+}
+
+/// `foo[0]` / `foo.bar` -> `foo`.
+fn column_name_from_path(path: &str) -> String {
+    match path.find(['[', '.']) {
+        Some(i) => path[..i].to_string(),
+        None => path.to_string(),
+    }
+}

pkg/collector/sharedlibrary/rustchecks/checks/datasecurity/src/scanning/mod.rs

@@ -0,0 +1,17 @@
+use dd_sds::{RegexRuleConfig, RootRuleConfig};
+use serde::Deserialize;
+
+/// A scanning rule, shared by every sub task. `id` (used to map matches back to
+/// the rule) and `name` are ours; everything else is the flattened dd-sds
+/// `RootRuleConfig<RegexRuleConfig>`, so the full rule schema — pattern,
+/// proximity keywords, suppressions, precedence and secondary validators (Luhn,
+/// JWT, ...) — comes straight from dd-sds with no duplication.
+#[derive(Debug, Deserialize)]
+pub struct ScanningRule {
+    pub id: String,
+    #[serde(default)]
+    #[allow(dead_code)]
+    pub name: String,
+    #[serde(flatten)]
+    pub config: RootRuleConfig<RegexRuleConfig>,
+}