dd_sds/scanner/
mod.rs

1use crate::encoding::Encoding;
2use crate::event::Event;
3use std::future::Future;
4
5use crate::match_validation::{
6    config::InternalMatchValidationType, config::MatchValidationType, match_status::MatchStatus,
7    match_validator::MatchValidator,
8};
9
10#[cfg(feature = "third-party-active-checkers")]
11use error::MatchValidatorCreationError;
12
13use self::metrics::ScannerMetrics;
14#[cfg(feature = "third-party-active-checkers")]
15use crate::match_validation::match_validator::RAYON_THREAD_POOL;
16use crate::observability::labels::Labels;
17use crate::rule_match::{InternalRuleMatch, RuleMatch};
18use crate::scanner::config::RuleConfig;
19use crate::scanner::internal_rule_match_set::InternalRuleMatchSet;
20use crate::scanner::regex_rule::compiled::RegexCompiledRule;
21use crate::scanner::regex_rule::{RegexCaches, access_regex_caches};
22use crate::scanner::scope::Scope;
23pub use crate::scanner::shared_data::SharedData;
24use crate::scanner::suppression::{CompiledSuppressions, SuppressionValidationError, Suppressions};
25use crate::scoped_ruleset::{ContentVisitor, ExclusionCheck, ScopedRuleSet};
26pub use crate::secondary_validation::Validator;
27use crate::stats::GLOBAL_STATS;
28use crate::tokio::TOKIO_RUNTIME;
29use crate::{CreateScannerError, EncodeIndices, MatchAction, Path, ScannerError};
30use ahash::AHashMap;
31use futures::executor::block_on;
32use serde::{Deserialize, Serialize};
33use serde_with::serde_as;
34use std::ops::Deref;
35use std::pin::Pin;
36use std::sync::Arc;
37use std::time::{Duration, Instant};
38use tokio::task::JoinHandle;
39use tokio::time::timeout;
40
41pub mod config;
42pub mod debug_scan;
43pub mod error;
44pub mod metrics;
45pub mod regex_rule;
46pub mod scope;
47pub mod shared_data;
48pub mod shared_pool;
49pub mod suppression;
50
51mod internal_rule_match_set;
52#[cfg(test)]
53mod test;
54
55#[derive(Clone)]
56pub struct StringMatch {
57    pub start: usize,
58    pub end: usize,
59    // The keyword that was used to match this rule. Optional, only some rules may set this value.
60    pub keyword: Option<String>,
61}
62
63pub trait MatchEmitter<T = ()> {
64    fn emit(&mut self, string_match: StringMatch) -> T;
65}
66
67// This implements MatchEmitter for mutable closures (so you can use a closure instead of a custom
68// struct that implements MatchEmitter)
69impl<F, T> MatchEmitter<T> for F
70where
71    F: FnMut(StringMatch) -> T,
72{
73    fn emit(&mut self, string_match: StringMatch) -> T {
74        // This just calls the closure (itself)
75        (self)(string_match)
76    }
77}
78
79/// The precedence of a rule. Catchall is the lowest precedence, Specific is the highest precedence.
80/// The default precedence is Specific.
81/// For rules that:
82/// - Have the same mutation priority
83/// - Match at the same index
84/// - Match the same number of characters
85///
86/// Then the rule with the highest precedence will be used.
87#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Copy, Default)]
88pub enum Precedence {
89    Catchall,
90    Generic,
91    #[default]
92    Specific,
93}
94
95#[serde_as]
96#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
97pub struct RootRuleConfig<T> {
98    #[serde(default)]
99    pub match_action: MatchAction,
100    #[serde(default)]
101    pub scope: Scope,
102    #[deprecated(note = "Use `third_party_active_checker` instead")]
103    match_validation_type: Option<MatchValidationType>,
104    third_party_active_checker: Option<MatchValidationType>,
105    suppressions: Option<Suppressions>,
106    #[serde(default)]
107    precedence: Precedence,
108    #[serde(default)]
109    pub is_supporting_rule: bool,
110    #[serde(flatten)]
111    pub inner: T,
112}
113
114impl<T> RootRuleConfig<T>
115where
116    T: RuleConfig + 'static,
117{
118    pub fn new_dyn(inner: T) -> RootRuleConfig<Arc<dyn RuleConfig>> {
119        RootRuleConfig::new(Arc::new(inner) as Arc<dyn RuleConfig>)
120    }
121
122    pub fn into_dyn(self) -> RootRuleConfig<Arc<dyn RuleConfig>> {
123        self.map_inner(|x| Arc::new(x) as Arc<dyn RuleConfig>)
124    }
125}
126
127impl<T> RootRuleConfig<T> {
128    pub fn new(inner: T) -> Self {
129        #[allow(deprecated)]
130        Self {
131            match_action: MatchAction::None,
132            scope: Scope::all(),
133            match_validation_type: None,
134            third_party_active_checker: None,
135            suppressions: None,
136            precedence: Precedence::default(),
137            is_supporting_rule: false,
138            inner,
139        }
140    }
141
142    pub fn map_inner<U>(self, func: impl FnOnce(T) -> U) -> RootRuleConfig<U> {
143        #[allow(deprecated)]
144        RootRuleConfig {
145            match_action: self.match_action,
146            scope: self.scope,
147            match_validation_type: self.match_validation_type,
148            third_party_active_checker: self.third_party_active_checker,
149            suppressions: self.suppressions,
150            precedence: self.precedence,
151            is_supporting_rule: self.is_supporting_rule,
152            inner: func(self.inner),
153        }
154    }
155
156    pub fn match_action(mut self, action: MatchAction) -> Self {
157        self.match_action = action;
158        self
159    }
160
161    pub fn precedence(mut self, precedence: Precedence) -> Self {
162        self.precedence = precedence;
163        self
164    }
165
166    pub fn scope(mut self, scope: Scope) -> Self {
167        self.scope = scope;
168        self
169    }
170
171    pub fn third_party_active_checker(
172        mut self,
173        match_validation_type: MatchValidationType,
174    ) -> Self {
175        self.third_party_active_checker = Some(match_validation_type);
176        self
177    }
178
179    pub fn suppressions(mut self, suppressions: Suppressions) -> Self {
180        self.suppressions = Some(suppressions);
181        self
182    }
183
184    pub fn is_supporting_rule(mut self, value: bool) -> Self {
185        self.is_supporting_rule = value;
186        self
187    }
188
189    fn get_third_party_active_checker(&self) -> Option<&MatchValidationType> {
190        #[allow(deprecated)]
191        self.third_party_active_checker
192            .as_ref()
193            .or(self.match_validation_type.as_ref())
194    }
195}
196
197impl<T> Deref for RootRuleConfig<T> {
198    type Target = T;
199
200    fn deref(&self) -> &Self::Target {
201        &self.inner
202    }
203}
204pub struct RootCompiledRule {
205    pub inner: Box<dyn CompiledRule>,
206    pub scope: Scope,
207    pub match_action: MatchAction,
208    pub match_validation_type: Option<MatchValidationType>,
209    pub suppressions: Option<CompiledSuppressions>,
210    pub precedence: Precedence,
211    pub is_supporting_rule: bool,
212}
213
214impl RootCompiledRule {
215    pub fn internal_match_validation_type(&self) -> Option<InternalMatchValidationType> {
216        self.match_validation_type
217            .as_ref()
218            .map(|x| x.get_internal_match_validation_type())
219    }
220}
221
222impl Deref for RootCompiledRule {
223    type Target = dyn CompiledRule;
224
225    fn deref(&self) -> &Self::Target {
226        self.inner.as_ref()
227    }
228}
229
230pub struct StringMatchesCtx<'a> {
231    rule_index: usize,
232    pub regex_caches: &'a mut RegexCaches,
233    pub exclusion_check: &'a ExclusionCheck<'a>,
234    pub excluded_matches: &'a mut AHashMap<String, String>,
235    pub match_emitter: &'a mut dyn MatchEmitter,
236    pub wildcard_indices: Option<&'a Vec<(usize, usize)>>,
237    pub enable_debug_observability: bool,
238
239    // Shared Data
240    pub per_string_data: &'a mut SharedData,
241    pub per_scanner_data: &'a SharedData,
242    pub per_event_data: &'a mut SharedData,
243    pub event_id: Option<&'a str>,
244
245    // Per-scan metadata supplied by the caller via `ScanOptions::scan_metadata`.
246    pub scan_metadata: &'a AHashMap<String, String>,
247}
248
249impl StringMatchesCtx<'_> {
250    /// If a `get_string_matches` implementation needs to do any async processing (e.g. I/O),
251    /// this function can be used to return an "async job" to find matches. The return value
252    /// of `process_async` should be returned from the `get_string_matches` function. The future
253    /// passed into this function will be spawned and executed immediately without blocking
254    /// other `get_string_matches` calls. This means all the async jobs will run concurrently.
255    ///
256    /// The `ctx` available to async jobs is more restrictive than the normal `ctx` available in
257    /// `get_string_matches`. The only thing you can do is return matches. If other data is needed,
258    /// it should be accessed before `process_async` is called.
259    pub fn process_async(
260        &self,
261        func: impl for<'a> FnOnce(
262            &'a mut AsyncStringMatchesCtx,
263        )
264            -> Pin<Box<dyn Future<Output = Result<(), ScannerError>> + Send + 'a>>
265        + Send
266        + 'static,
267    ) -> RuleResult {
268        let rule_index = self.rule_index;
269
270        // The future is spawned onto the tokio runtime immediately so it starts running
271        // in the background
272        let fut = TOKIO_RUNTIME.spawn(async move {
273            let start = Instant::now();
274            let mut ctx = AsyncStringMatchesCtx {
275                rule_matches: vec![],
276            };
277            (func)(&mut ctx).await?;
278            let io_duration = start.elapsed();
279
280            Ok(AsyncRuleInfo {
281                rule_index,
282                rule_matches: ctx.rule_matches,
283                io_duration,
284            })
285        });
286
287        Ok(RuleStatus::Pending(fut))
288    }
289}
290
291pub struct AsyncStringMatchesCtx {
292    rule_matches: Vec<StringMatch>,
293}
294
295impl AsyncStringMatchesCtx {
296    pub fn emit_match(&mut self, string_match: StringMatch) {
297        self.rule_matches.push(string_match);
298    }
299}
300
301#[must_use]
302pub enum RuleStatus {
303    Done,
304    Pending(PendingRuleResult),
305}
306
307// pub type PendingRuleResult = BoxFuture<'static, Result<AsyncRuleInfo, ScannerError>>;
308pub type PendingRuleResult = JoinHandle<Result<AsyncRuleInfo, ScannerError>>;
309
310pub struct PendingRuleJob {
311    fut: PendingRuleResult,
312    path: Path<'static>,
313}
314
315pub struct AsyncRuleInfo {
316    rule_index: usize,
317    rule_matches: Vec<StringMatch>,
318    io_duration: Duration,
319}
320
321/// A rule result that cannot be async
322pub type RuleResult = Result<RuleStatus, ScannerError>;
323
324// This is the public trait that is used to define the behavior of a compiled rule.
325pub trait CompiledRule: Send + Sync {
326    fn init_per_scanner_data(&self, _per_scanner_data: &mut SharedData) {
327        // by default, no per-scanner data is initialized
328    }
329
330    fn init_per_string_data(&self, _labels: &Labels, _per_string_data: &mut SharedData) {
331        // by default, no per-string data is initialized
332    }
333
334    fn init_per_event_data(&self, _per_event_data: &mut SharedData) {
335        // by default, no per-event data is initialized
336    }
337
338    fn get_string_matches(
339        &self,
340        content: &str,
341        path: &Path,
342        ctx: &mut StringMatchesCtx<'_>,
343    ) -> RuleResult;
344
345    // Whether a match from this rule should be excluded (marked as a false-positive)
346    // if the content of this match was found in a match from an excluded scope
347    fn should_exclude_multipass_v0(&self) -> bool {
348        // default is to NOT use Multi-pass V0
349        false
350    }
351
352    fn on_excluded_match_multipass_v0(
353        &self,
354        _path: &Path,
355        _excluded_path: &str,
356        _enable_debug_observability: bool,
357    ) {
358        // default is to do nothing
359    }
360
361    fn as_regex_rule(&self) -> Option<&RegexCompiledRule> {
362        None
363    }
364
365    fn as_regex_rule_mut(&mut self) -> Option<&mut RegexCompiledRule> {
366        None
367    }
368
369    fn allow_scanner_to_exclude_namespace(&self) -> bool {
370        true
371    }
372}
373
374impl<T> RuleConfig for Box<T>
375where
376    T: RuleConfig + ?Sized,
377{
378    fn convert_to_compiled_rule(
379        &self,
380        rule_index: usize,
381        labels: Labels,
382    ) -> Result<Box<dyn CompiledRule>, CreateScannerError> {
383        self.as_ref().convert_to_compiled_rule(rule_index, labels)
384    }
385}
386
387#[derive(Debug, PartialEq, Clone)]
388struct ScannerFeatures {
389    pub add_implicit_index_wildcards: bool,
390    pub multipass_v0_enabled: bool,
391    pub return_matches: bool,
392    pub enable_debug_observability: bool,
393}
394
395impl Default for ScannerFeatures {
396    fn default() -> Self {
397        Self {
398            add_implicit_index_wildcards: false,
399            multipass_v0_enabled: true,
400            return_matches: false,
401            enable_debug_observability: false,
402        }
403    }
404}
405
406pub struct ScanOptions {
407    // The blocked_rules_idx parameter is a list of rule indices that should be skipped for this scan.
408    // this list shall be small (<10), so a linear search is acceptable otherwise performance will be impacted.
409    pub blocked_rules_idx: Vec<usize>,
410    // The wildcarded_indices parameter is a map containing a list of tuples of (start, end) indices that should be treated as wildcards (for the message key only) per path.
411    pub wildcarded_indices: AHashMap<Path<'static>, Vec<(usize, usize)>>,
412    // Whether to validate matches using third-party validators (e.g., checksum validation for credit cards).
413    // When enabled, the scanner automatically collects match content needed for validation.
414    pub validate_matches: bool,
415    // Arbitrary key-value metadata passed through to each rule's `get_string_matches` call via
416    // `StringMatchesCtx::scan_metadata`. Rules may use this to receive per-scan context (e.g.
417    // an org identifier) without requiring changes to the `CompiledRule` trait signature.
418    pub scan_metadata: AHashMap<String, String>,
419}
420
421impl Default for ScanOptions {
422    fn default() -> Self {
423        Self {
424            blocked_rules_idx: vec![],
425            wildcarded_indices: AHashMap::new(),
426            validate_matches: false,
427            scan_metadata: AHashMap::new(),
428        }
429    }
430}
431
432pub struct ScanOptionBuilder {
433    blocked_rules_idx: Vec<usize>,
434    wildcarded_indices: AHashMap<Path<'static>, Vec<(usize, usize)>>,
435    validate_matches: bool,
436    scan_metadata: AHashMap<String, String>,
437}
438
439impl ScanOptionBuilder {
440    pub fn new() -> Self {
441        Self {
442            blocked_rules_idx: vec![],
443            wildcarded_indices: AHashMap::new(),
444            validate_matches: false,
445            scan_metadata: AHashMap::new(),
446        }
447    }
448
449    pub fn with_blocked_rules_idx(mut self, blocked_rules_idx: Vec<usize>) -> Self {
450        self.blocked_rules_idx = blocked_rules_idx;
451        self
452    }
453
454    pub fn with_wildcarded_indices(
455        mut self,
456        wildcarded_indices: AHashMap<Path<'static>, Vec<(usize, usize)>>,
457    ) -> Self {
458        self.wildcarded_indices = wildcarded_indices;
459        self
460    }
461
462    pub fn with_validate_matching(mut self, validate_matches: bool) -> Self {
463        self.validate_matches = validate_matches;
464        self
465    }
466
467    pub fn with_scan_metadata(mut self, scan_metadata: AHashMap<String, String>) -> Self {
468        self.scan_metadata = scan_metadata;
469        self
470    }
471
472    pub fn build(self) -> ScanOptions {
473        ScanOptions {
474            blocked_rules_idx: self.blocked_rules_idx,
475            wildcarded_indices: self.wildcarded_indices,
476            validate_matches: self.validate_matches,
477            scan_metadata: self.scan_metadata,
478        }
479    }
480}
481
482pub struct Scanner {
483    rules: Vec<RootCompiledRule>,
484    scoped_ruleset: ScopedRuleSet,
485    scanner_features: ScannerFeatures,
486    metrics: ScannerMetrics,
487    labels: Labels,
488    // Without third-party active checkers this map is always empty and never read, but it is kept
489    // in the struct so the type stays identical across build configurations.
490    #[cfg_attr(not(feature = "third-party-active-checkers"), allow(dead_code))]
491    match_validators_per_type: AHashMap<InternalMatchValidationType, Box<dyn MatchValidator>>,
492    per_scanner_data: SharedData,
493    async_scan_timeout: Duration,
494}
495
496impl Scanner {
497    pub fn builder(rules: &[RootRuleConfig<Arc<dyn RuleConfig>>]) -> ScannerBuilder<'_> {
498        ScannerBuilder::new(rules)
499    }
500
501    // This function scans the given event with the rules configured in the scanner.
502    // The event parameter is a mutable reference to the event that should be scanned (implemented the Event trait).
503    // The return value is a list of RuleMatch objects, which contain information about the matches that were found.
504    // This version uses default scan options (no validation, no blocked rules, no wildcarded indices).
505    pub fn scan<E: Event>(&self, event: &mut E) -> Result<Vec<RuleMatch>, ScannerError> {
506        self.scan_with_options(event, ScanOptions::default())
507    }
508
509    // This function scans the given event with the rules configured in the scanner.
510    // The event parameter is a mutable reference to the event that should be scanned (implemented the Event trait).
511    // The options parameter allows customizing the scan behavior (validation, blocked rules, etc.).
512    // The return value is a list of RuleMatch objects, which contain information about the matches that were found.
513    pub fn scan_with_options<E: Event>(
514        &self,
515        event: &mut E,
516        options: ScanOptions,
517    ) -> Result<Vec<RuleMatch>, ScannerError> {
518        let start = Instant::now();
519        let validate = options.validate_matches;
520        // Collect matches inside block_on, then run finalize_matches (which uses rayon) outside of
521        // it to avoid re-entrancy between the futures LocalPool executor and RAYON_THREAD_POOL.
522        let result = block_on(self.internal_scan_collect(event, options));
523        match result {
524            Ok((mut rule_matches, io_duration)) => {
525                self.finalize_matches(&mut rule_matches, validate);
526                self.record_metrics(&rule_matches, start, Some(io_duration));
527                Ok(rule_matches)
528            }
529            Err(e) => {
530                self.record_metrics(&[], start, None);
531                Err(e)
532            }
533        }
534    }
535
536    // This function scans the given event with the rules configured in the scanner.
537    // The event parameter is a mutable reference to the event that should be scanned (implemented the Event trait).
538    // The return value is a list of RuleMatch objects, which contain information about the matches that were found.
539    pub async fn scan_async<E: Event>(
540        &self,
541        event: &mut E,
542    ) -> Result<Vec<RuleMatch>, ScannerError> {
543        self.scan_async_with_options(event, ScanOptions::default())
544            .await
545    }
546
547    pub async fn scan_async_with_options<E: Event>(
548        &self,
549        event: &mut E,
550        options: ScanOptions,
551    ) -> Result<Vec<RuleMatch>, ScannerError> {
552        let start = Instant::now();
553        let validate = options.validate_matches;
554        let fut = self.internal_scan_collect(event, options);
555
556        // The sleep from the timeout requires being in a tokio context
557        // The guard needs to be dropped before await since the guard is !Send
558        let timeout_result = {
559            let _tokio_guard = TOKIO_RUNTIME.enter();
560            timeout(self.async_scan_timeout, fut)
561        };
562
563        let result = timeout_result.await.unwrap_or(Err(ScannerError::Transient(
564            "Async scan timeout".to_string(),
565        )));
566
567        match result {
568            Ok((mut rule_matches, io_duration)) => {
569                self.finalize_matches(&mut rule_matches, validate);
570                self.record_metrics(&rule_matches, start, Some(io_duration));
571                Ok(rule_matches)
572            }
573            Err(e) => {
574                self.record_metrics(&[], start, None);
575                Err(e)
576            }
577        }
578    }
579
580    fn record_metrics(
581        &self,
582        output_rule_matches: &[RuleMatch],
583        start: Instant,
584        io_duration: Option<Duration>,
585    ) {
586        // Add number of scanned events
587        self.metrics.num_scanned_events.increment(1);
588        // Add number of matches
589        self.metrics
590            .match_count
591            .increment(output_rule_matches.len() as u64);
592
593        if let Some(io_duration) = io_duration {
594            let total_duration = start.elapsed();
595            let cpu_duration = total_duration.saturating_sub(io_duration);
596            self.metrics
597                .cpu_duration
598                .increment(cpu_duration.as_nanos() as u64);
599        }
600    }
601
602    fn process_rule_matches<E: Event>(
603        &self,
604        event: &mut E,
605        rule_matches: InternalRuleMatchSet<E::Encoding>,
606        excluded_matches: AHashMap<String, String>,
607        output_rule_matches: &mut Vec<RuleMatch>,
608        need_match_content: bool,
609    ) {
610        if rule_matches.is_empty() {
611            return;
612        }
613        access_regex_caches(|regex_caches| {
614            for (path, mut rule_matches) in rule_matches.into_iter() {
615                // All rule matches in each inner list are for a single path, so they can be processed independently.
616                event.visit_string_mut(&path, |content| {
617                    // calculate_indices requires that matches are sorted by start index
618                    rule_matches.sort_unstable_by_key(|rule_match| rule_match.utf8_start);
619
620                    <<E as Event>::Encoding>::calculate_indices(
621                        content,
622                        rule_matches.iter_mut().map(
623                            |rule_match: &mut InternalRuleMatch<E::Encoding>| EncodeIndices {
624                                utf8_start: rule_match.utf8_start,
625                                utf8_end: rule_match.utf8_end,
626                                custom_start: &mut rule_match.custom_start,
627                                custom_end: &mut rule_match.custom_end,
628                            },
629                        ),
630                    );
631
632                    if self.scanner_features.multipass_v0_enabled {
633                        // Now that the `excluded_matches` set is fully populated, filter out any matches
634                        // that are the same as excluded matches (also known as "Multi-pass V0")
635                        rule_matches.retain(|rule_match| {
636                            if self.rules[rule_match.rule_index]
637                                .inner
638                                .should_exclude_multipass_v0()
639                            {
640                                let match_content =
641                                    &content[rule_match.utf8_start..rule_match.utf8_end];
642                                let excluded_path = excluded_matches.get(match_content);
643                                if let Some(excluded_path) = excluded_path {
644                                    self.rules[rule_match.rule_index]
645                                        .on_excluded_match_multipass_v0(
646                                            &path,
647                                            excluded_path,
648                                            self.scanner_features.enable_debug_observability,
649                                        );
650                                }
651                                excluded_path.is_none()
652                            } else {
653                                true
654                            }
655                        });
656                    }
657
658                    self.suppress_matches::<E::Encoding>(&mut rule_matches, content, regex_caches);
659
660                    self.sort_and_remove_overlapping_rules::<E::Encoding>(&mut rule_matches);
661
662                    let will_mutate = rule_matches.iter().any(|rule_match| {
663                        self.rules[rule_match.rule_index].match_action.is_mutating()
664                    });
665
666                    self.apply_match_actions(
667                        content,
668                        &path,
669                        rule_matches,
670                        output_rule_matches,
671                        need_match_content,
672                    );
673
674                    will_mutate
675                });
676            }
677        });
678    }
679
680    async fn internal_scan_collect<E: Event>(
681        &self,
682        event: &mut E,
683        options: ScanOptions,
684    ) -> Result<(Vec<RuleMatch>, Duration), ScannerError> {
685        // If validation is requested, we need to collect match content even if the scanner
686        // wasn't originally configured to return matches
687        let need_match_content = self.scanner_features.return_matches || options.validate_matches;
688        // All matches, after some (but not all) false-positives have been removed.
689        let mut rule_matches = InternalRuleMatchSet::new();
690        let mut excluded_matches = AHashMap::new();
691        let mut async_jobs = vec![];
692
693        access_regex_caches(|regex_caches| {
694            self.scoped_ruleset.visit_string_rule_combinations(
695                event,
696                ScannerContentVisitor {
697                    scanner: self,
698                    regex_caches,
699                    rule_matches: &mut rule_matches,
700                    blocked_rules: &options.blocked_rules_idx,
701                    excluded_matches: &mut excluded_matches,
702                    per_event_data: SharedData::new(),
703                    wildcarded_indexes: &options.wildcarded_indices,
704                    async_jobs: &mut async_jobs,
705                    event_id: event.get_id().map(|s| s.to_string()),
706                    scan_metadata: &options.scan_metadata,
707                },
708            )
709        })?;
710
711        // The async jobs were already spawned on the tokio runtime, so the
712        // results just need to be collected
713        let mut total_io_duration = Duration::ZERO;
714        for job in async_jobs {
715            let rule_info = job.fut.await.unwrap()?;
716            total_io_duration += rule_info.io_duration;
717            rule_matches.push_async_matches(
718                &job.path,
719                rule_info
720                    .rule_matches
721                    .into_iter()
722                    .map(|x| InternalRuleMatch::new(rule_info.rule_index, x)),
723            );
724        }
725
726        let mut output_rule_matches = vec![];
727
728        self.process_rule_matches(
729            event,
730            rule_matches,
731            excluded_matches,
732            &mut output_rule_matches,
733            need_match_content,
734        );
735
736        Ok((output_rule_matches, total_io_duration))
737    }
738
739    pub fn suppress_matches<E: Encoding>(
740        &self,
741        rule_matches: &mut Vec<InternalRuleMatch<E>>,
742        content: &str,
743        regex_caches: &mut RegexCaches,
744    ) {
745        rule_matches.retain(|rule_match| {
746            if let Some(suppressions) = &self.rules[rule_match.rule_index].suppressions {
747                let match_should_be_suppressed = suppressions.should_match_be_suppressed(
748                    &content[rule_match.utf8_start..rule_match.utf8_end],
749                    regex_caches,
750                );
751
752                if match_should_be_suppressed {
753                    self.metrics.suppressed_match_count.increment(1);
754                }
755                !match_should_be_suppressed
756            } else {
757                true
758            }
759        });
760    }
761
762    #[cfg(feature = "third-party-active-checkers")]
763    pub fn validate_matches(&self, rule_matches: &mut Vec<RuleMatch>) {
764        // Create MatchValidatorRuleMatch per match_validator_type to pass it to each match_validator
765        let mut match_validator_rule_match_per_type = AHashMap::new();
766
767        let mut validated_rule_matches = vec![];
768
769        for mut rule_match in rule_matches.drain(..) {
770            let rule = &self.rules[rule_match.rule_index];
771            if let Some(match_validation_type) = rule.internal_match_validation_type() {
772                match_validator_rule_match_per_type
773                    .entry(match_validation_type)
774                    .or_insert_with(Vec::new)
775                    .push(rule_match)
776            } else {
777                // There is no match validator for this rule, so mark it as not available.
778                rule_match.match_status.merge(MatchStatus::NotAvailable);
779                validated_rule_matches.push(rule_match);
780            }
781        }
782
783        // Skip the pool hop when there's nothing to validate.
784        if !match_validator_rule_match_per_type.is_empty() {
785            let run_validation = || {
786                use rayon::prelude::*;
787
788                match_validator_rule_match_per_type.par_iter_mut().for_each(
789                    |(match_validation_type, matches_per_type)| {
790                        let match_validator =
791                            self.match_validators_per_type.get(match_validation_type);
792                        if let Some(match_validator) = match_validator {
793                            match_validator
794                                .as_ref()
795                                .validate(matches_per_type, &self.rules)
796                        }
797                    },
798                );
799            };
800
801            // TODO(SDSP-450): move validation onto the async TOKIO_RUNTIME. It is I/O-bound
802            // (blocking HTTP to third-party checkers), so a Rayon (CPU-bound) pool is a poor
803            // fit and forces the workaround below. Async validation would remove it entirely.
804            //
805            // If we're already on a Rayon worker (e.g. the caller scans from its own parallel
806            // iterator), calling `install` here blocks the worker, and Rayon keeps it busy by
807            // stealing more of the caller's jobs — which scan and validate and steal again,
808            // recursing until the stack overflows. Running `install` on a separate OS thread
809            // parks this worker instead, breaking the recursion; validation still runs in
810            // parallel on RAYON_THREAD_POOL.
811            if rayon::current_thread_index().is_some() {
812                std::thread::scope(|s| {
813                    s.spawn(|| RAYON_THREAD_POOL.install(run_validation));
814                });
815            } else {
816                RAYON_THREAD_POOL.install(run_validation);
817            }
818        }
819
820        // Refill the rule_matches with the validated matches
821        for (_, mut matches) in match_validator_rule_match_per_type {
822            validated_rule_matches.append(&mut matches);
823        }
824
825        // Sort rule_matches by start index
826        validated_rule_matches.sort_by_key(|rule_match| rule_match.start_index);
827        *rule_matches = validated_rule_matches;
828    }
829
830    // Runs optional validation and drops supporting-rule matches from the output.
831    // Must be called OUTSIDE of any futures executor (e.g. block_on) because validate_matches
832    // uses RAYON_THREAD_POOL internally; running rayon inside block_on causes an EnterError panic
833    // when the calling thread re-enters the LocalPool executor context.
834    fn finalize_matches(&self, rule_matches: &mut Vec<RuleMatch>, validate: bool) {
835        #[cfg(feature = "third-party-active-checkers")]
836        if validate {
837            self.validate_matches(rule_matches);
838        }
839        #[cfg(not(feature = "third-party-active-checkers"))]
840        let _ = validate;
841        // Supporting rules exist only to provide template variables to CustomHttpV2 validators of
842        // other rules. Their matches must not appear in the final output. They are retained until
843        // after validate_matches so that match pairing can reference their match values.
844        rule_matches.retain(|rule_match| !self.rules[rule_match.rule_index].is_supporting_rule);
845    }
846
847    /// Apply mutations from actions, and shift indices to match the mutated values.
848    /// This assumes the matches are all from the content given, and are sorted by start index.
849    fn apply_match_actions<E: Encoding>(
850        &self,
851        content: &mut String,
852        path: &Path<'static>,
853        rule_matches: Vec<InternalRuleMatch<E>>,
854        output_rule_matches: &mut Vec<RuleMatch>,
855        need_match_content: bool,
856    ) {
857        let mut utf8_byte_delta: isize = 0;
858        let mut custom_index_delta: <E>::IndexShift = <E>::zero_shift();
859
860        for rule_match in rule_matches {
861            output_rule_matches.push(self.apply_match_actions_for_string::<E>(
862                content,
863                path.clone(),
864                rule_match,
865                &mut utf8_byte_delta,
866                &mut custom_index_delta,
867                need_match_content,
868            ));
869        }
870    }
871
872    /// This will be called once for each match of a single string. The rules must be passed in in order of the start index. Mutating rules must not overlap.
873    fn apply_match_actions_for_string<E: Encoding>(
874        &self,
875        content: &mut String,
876        path: Path<'static>,
877        rule_match: InternalRuleMatch<E>,
878        // The current difference in length between the original and mutated string
879        utf8_byte_delta: &mut isize,
880
881        // The difference between the custom index on the original string and the mutated string
882        custom_index_delta: &mut <E>::IndexShift,
883        need_match_content: bool,
884    ) -> RuleMatch {
885        let rule = &self.rules[rule_match.rule_index];
886
887        let custom_start =
888            (<E>::get_index(&rule_match.custom_start, rule_match.utf8_start) as isize
889                + <E>::get_shift(custom_index_delta, *utf8_byte_delta)) as usize;
890
891        let mut matched_content_copy = None;
892
893        if need_match_content {
894            // This copies part of the is_mutating block but is seperate since can't mix compilation condition and code condition
895            let mutated_utf8_match_start =
896                (rule_match.utf8_start as isize + *utf8_byte_delta) as usize;
897            let mutated_utf8_match_end = (rule_match.utf8_end as isize + *utf8_byte_delta) as usize;
898
899            // Matches for mutating rules must have valid indices
900            debug_assert!(content.is_char_boundary(mutated_utf8_match_start));
901            debug_assert!(content.is_char_boundary(mutated_utf8_match_end));
902
903            let matched_content = &content[mutated_utf8_match_start..mutated_utf8_match_end];
904            matched_content_copy = Some(matched_content.to_string());
905        }
906
907        if rule.match_action.is_mutating() {
908            let mutated_utf8_match_start =
909                (rule_match.utf8_start as isize + *utf8_byte_delta) as usize;
910            let mutated_utf8_match_end = (rule_match.utf8_end as isize + *utf8_byte_delta) as usize;
911
912            // Matches for mutating rules must have valid indices
913            debug_assert!(content.is_char_boundary(mutated_utf8_match_start));
914            debug_assert!(content.is_char_boundary(mutated_utf8_match_end));
915
916            let matched_content = &content[mutated_utf8_match_start..mutated_utf8_match_end];
917            if let Some(replacement) = rule.match_action.get_replacement(matched_content) {
918                let before_replacement = &matched_content[replacement.start..replacement.end];
919
920                // update indices to match the new mutated content
921                <E>::adjust_shift(
922                    custom_index_delta,
923                    before_replacement,
924                    &replacement.replacement,
925                );
926                *utf8_byte_delta +=
927                    replacement.replacement.len() as isize - before_replacement.len() as isize;
928
929                let replacement_start = mutated_utf8_match_start + replacement.start;
930                let replacement_end = mutated_utf8_match_start + replacement.end;
931                content.replace_range(replacement_start..replacement_end, &replacement.replacement);
932            }
933        }
934
935        let shift_offset = <E>::get_shift(custom_index_delta, *utf8_byte_delta);
936        let custom_end = (<E>::get_index(&rule_match.custom_end, rule_match.utf8_end) as isize
937            + shift_offset) as usize;
938
939        let rule = &self.rules[rule_match.rule_index];
940
941        let match_status: MatchStatus = if rule.match_validation_type.is_some() {
942            MatchStatus::NotChecked
943        } else {
944            MatchStatus::NotAvailable
945        };
946
947        RuleMatch {
948            rule_index: rule_match.rule_index,
949            path,
950            replacement_type: rule.match_action.replacement_type(),
951            start_index: custom_start,
952            end_index_exclusive: custom_end,
953            shift_offset,
954            match_value: matched_content_copy,
955            match_status,
956            keyword: rule_match.keyword,
957        }
958    }
959
960    fn sort_and_remove_overlapping_rules<E: Encoding>(
961        &self,
962        rule_matches: &mut Vec<InternalRuleMatch<E>>,
963    ) {
964        // Some of the scanner code relies on the behavior here, such as the sort order and removal of overlapping mutating rules.
965        // Be very careful if this function is modified.
966
967        rule_matches.sort_unstable_by(|a, b| {
968            // Mutating rules are a higher priority (earlier in the list)
969            let ord = self.rules[a.rule_index]
970                .match_action
971                .is_mutating()
972                .cmp(&self.rules[b.rule_index].match_action.is_mutating())
973                .reverse();
974
975            // Earlier start offset
976            let ord = ord.then(a.utf8_start.cmp(&b.utf8_start));
977
978            // Longer matches
979            let ord = ord.then(a.len().cmp(&b.len()).reverse());
980
981            // Matches with higher precedence come first
982            let ord = ord.then(
983                self.rules[a.rule_index]
984                    .precedence
985                    .cmp(&self.rules[b.rule_index].precedence)
986                    .reverse(),
987            );
988
989            // Matches from earlier rules
990            let ord = ord.then(a.rule_index.cmp(&b.rule_index));
991
992            // swap the order of everything so matches can be efficiently popped off the back as they are processed
993            ord.reverse()
994        });
995
996        let mut retained_rules: Vec<InternalRuleMatch<E>> = vec![];
997
998        'rule_matches: while let Some(rule_match) = rule_matches.pop() {
999            if self.rules[rule_match.rule_index].match_action.is_mutating() {
1000                // Mutating rules are kept only if they don't overlap with a previous rule.
1001                if let Some(last) = retained_rules.last()
1002                    && last.utf8_end > rule_match.utf8_start
1003                {
1004                    continue;
1005                }
1006            } else {
1007                // Only retain if it doesn't overlap with any other rule. Since mutating matches are sorted before non-mutated matches
1008                // this needs to check all retained matches (instead of just the last one)
1009                for retained_rule in &retained_rules {
1010                    if retained_rule.utf8_start < rule_match.utf8_end
1011                        && retained_rule.utf8_end > rule_match.utf8_start
1012                    {
1013                        continue 'rule_matches;
1014                    }
1015                }
1016            };
1017            retained_rules.push(rule_match);
1018        }
1019
1020        // ensure rules are sorted by start index (other parts of the library required this to function correctly)
1021        retained_rules.sort_unstable_by_key(|rule_match| rule_match.utf8_start);
1022
1023        *rule_matches = retained_rules;
1024    }
1025}
1026
1027impl Drop for Scanner {
1028    fn drop(&mut self) {
1029        let stats = &*GLOBAL_STATS;
1030        stats.scanner_deletions.increment(1);
1031        stats.decrement_total_scanners();
1032    }
1033}
1034
1035#[derive(Default)]
1036pub struct ScannerBuilder<'a> {
1037    rules: &'a [RootRuleConfig<Arc<dyn RuleConfig>>],
1038    labels: Labels,
1039    scanner_features: ScannerFeatures,
1040    async_scan_timeout: Duration,
1041}
1042
1043impl ScannerBuilder<'_> {
1044    pub fn new(rules: &[RootRuleConfig<Arc<dyn RuleConfig>>]) -> ScannerBuilder<'_> {
1045        ScannerBuilder {
1046            rules,
1047            labels: Labels::empty(),
1048            scanner_features: ScannerFeatures::default(),
1049            async_scan_timeout: Duration::from_secs(60 * 5),
1050        }
1051    }
1052
1053    pub fn labels(mut self, labels: Labels) -> Self {
1054        self.labels = labels;
1055        self
1056    }
1057
1058    pub fn with_async_scan_timeout(mut self, duration: Duration) -> Self {
1059        self.async_scan_timeout = duration;
1060        self
1061    }
1062
1063    pub fn with_implicit_wildcard_indexes_for_scopes(mut self, value: bool) -> Self {
1064        self.scanner_features.add_implicit_index_wildcards = value;
1065        self
1066    }
1067
1068    pub fn with_return_matches(mut self, value: bool) -> Self {
1069        self.scanner_features.return_matches = value;
1070        self
1071    }
1072
1073    /// Enables/Disables the Multipass V0 feature. This defaults to TRUE.
1074    /// Multipass V0 saves matches from excluded scopes, and marks any identical
1075    /// matches in included scopes as a false positive.
1076    pub fn with_multipass_v0(mut self, value: bool) -> Self {
1077        self.scanner_features.multipass_v0_enabled = value;
1078        self
1079    }
1080
1081    /// Enables/Disables debug observability features. This defaults to FALSE.
1082    /// When enabled, metrics will include additional tags (such as `sds_namespace`)
1083    /// to help debug the source of matches.
1084    pub fn with_debug_observability(mut self, value: bool) -> Self {
1085        self.scanner_features.enable_debug_observability = value;
1086        self
1087    }
1088
1089    pub fn build(self) -> Result<Scanner, CreateScannerError> {
1090        // When third-party active checkers are compiled out, no validators are ever
1091        // created; the map stays empty and `mut` would be unused.
1092        #[cfg_attr(not(feature = "third-party-active-checkers"), allow(unused_mut))]
1093        let mut match_validators_per_type = AHashMap::new();
1094
1095        #[cfg(feature = "third-party-active-checkers")]
1096        for rule in self.rules.iter() {
1097            if let Some(match_validation_type) = &rule.get_third_party_active_checker()
1098                && match_validation_type.can_create_match_validator()
1099            {
1100                let internal_type = match_validation_type.get_internal_match_validation_type();
1101                let match_validator = match_validation_type.into_match_validator();
1102                if let Ok(match_validator) = match_validator {
1103                    if !match_validators_per_type.contains_key(&internal_type) {
1104                        match_validators_per_type.insert(internal_type, match_validator);
1105                    }
1106                } else {
1107                    return Err(CreateScannerError::InvalidMatchValidator(
1108                        MatchValidatorCreationError::InternalError,
1109                    ));
1110                }
1111            }
1112        }
1113
1114        let compiled_rules = self
1115            .rules
1116            .iter()
1117            .enumerate()
1118            .map(|(rule_index, config)| {
1119                if config.is_supporting_rule && config.match_action != MatchAction::None {
1120                    return Err(CreateScannerError::SupportingRuleHasMatchAction);
1121                }
1122                let inner = config.convert_to_compiled_rule(rule_index, self.labels.clone())?;
1123                config.match_action.validate()?;
1124                let compiled_suppressions = match &config.suppressions {
1125                    Some(s) => s.compile()?,
1126                    None => None,
1127                };
1128                Ok(RootCompiledRule {
1129                    inner,
1130                    scope: config.scope.clone(),
1131                    match_action: config.match_action.clone(),
1132                    match_validation_type: config.get_third_party_active_checker().cloned(),
1133                    suppressions: compiled_suppressions,
1134                    precedence: config.precedence,
1135                    is_supporting_rule: config.is_supporting_rule,
1136                })
1137            })
1138            .collect::<Result<Vec<RootCompiledRule>, CreateScannerError>>()?;
1139
1140        let mut per_scanner_data = SharedData::new();
1141
1142        compiled_rules.iter().for_each(|rule| {
1143            rule.init_per_scanner_data(&mut per_scanner_data);
1144        });
1145
1146        let scoped_ruleset = ScopedRuleSet::new(
1147            &compiled_rules
1148                .iter()
1149                .map(|rule| rule.scope.clone())
1150                .collect::<Vec<_>>(),
1151        )
1152        .with_implicit_index_wildcards(self.scanner_features.add_implicit_index_wildcards);
1153
1154        {
1155            let stats = &*GLOBAL_STATS;
1156            stats.scanner_creations.increment(1);
1157            stats.increment_total_scanners();
1158        }
1159
1160        Ok(Scanner {
1161            rules: compiled_rules,
1162            scoped_ruleset,
1163            scanner_features: self.scanner_features,
1164            metrics: ScannerMetrics::new(&self.labels),
1165            match_validators_per_type,
1166            labels: self.labels,
1167            per_scanner_data,
1168            async_scan_timeout: self.async_scan_timeout,
1169        })
1170    }
1171}
1172
1173struct ScannerContentVisitor<'a, E: Encoding> {
1174    scanner: &'a Scanner,
1175    regex_caches: &'a mut RegexCaches,
1176    rule_matches: &'a mut InternalRuleMatchSet<E>,
1177    // Rules that shall be skipped for this scan
1178    // This list shall be small (<10), so a linear search is acceptable
1179    blocked_rules: &'a Vec<usize>,
1180    excluded_matches: &'a mut AHashMap<String, String>,
1181    per_event_data: SharedData,
1182    wildcarded_indexes: &'a AHashMap<Path<'static>, Vec<(usize, usize)>>,
1183    async_jobs: &'a mut Vec<PendingRuleJob>,
1184    event_id: Option<String>,
1185    scan_metadata: &'a AHashMap<String, String>,
1186}
1187
1188impl<'a, E: Encoding> ContentVisitor<'a> for ScannerContentVisitor<'a, E> {
1189    fn visit_content<'b>(
1190        &'b mut self,
1191        path: &Path<'a>,
1192        content: &str,
1193        mut rule_visitor: crate::scoped_ruleset::RuleIndexVisitor,
1194        exclusion_check: ExclusionCheck<'b>,
1195    ) -> Result<bool, ScannerError> {
1196        // matches for a single path
1197        let mut path_rules_matches = vec![];
1198
1199        // Create a map of per rule type data that can be shared between rules of the same type
1200        let mut per_string_data = SharedData::new();
1201        let wildcard_indices_per_path = self.wildcarded_indexes.get(path);
1202
1203        rule_visitor.visit_rule_indices(|rule_index| {
1204            if self.blocked_rules.contains(&rule_index) {
1205                return Ok(());
1206            }
1207            let rule = &self.scanner.rules[rule_index];
1208            {
1209                if rule.inner.allow_scanner_to_exclude_namespace() {
1210                    // check if the path is excluded
1211                    if exclusion_check.is_excluded(rule_index) {
1212                        return Ok(());
1213                    }
1214                }
1215                // creating the emitter is basically free, it will get mostly optimized away
1216                let mut emitter = |rule_match: StringMatch| {
1217                    // This should never happen, but to ensure no empty match is ever generated
1218                    // (which may cause an infinite loop), this will panic instead.
1219                    assert_ne!(
1220                        rule_match.start, rule_match.end,
1221                        "empty match detected on rule with index {rule_index}"
1222                    );
1223                    path_rules_matches.push(InternalRuleMatch::new(rule_index, rule_match));
1224                };
1225
1226                rule.init_per_string_data(&self.scanner.labels, &mut per_string_data);
1227
1228                // TODO: move this somewhere higher?
1229                rule.init_per_event_data(&mut self.per_event_data);
1230
1231                let mut ctx = StringMatchesCtx {
1232                    rule_index,
1233                    regex_caches: self.regex_caches,
1234                    exclusion_check: &exclusion_check,
1235                    excluded_matches: self.excluded_matches,
1236                    match_emitter: &mut emitter,
1237                    wildcard_indices: wildcard_indices_per_path,
1238                    enable_debug_observability: self
1239                        .scanner
1240                        .scanner_features
1241                        .enable_debug_observability,
1242                    per_string_data: &mut per_string_data,
1243                    per_scanner_data: &self.scanner.per_scanner_data,
1244                    per_event_data: &mut self.per_event_data,
1245                    event_id: self.event_id.as_deref(),
1246                    scan_metadata: self.scan_metadata,
1247                };
1248
1249                let async_status = rule.get_string_matches(content, path, &mut ctx)?;
1250
1251                match async_status {
1252                    RuleStatus::Done => {
1253                        // nothing to do
1254                    }
1255                    RuleStatus::Pending(fut) => {
1256                        self.async_jobs.push(PendingRuleJob {
1257                            fut,
1258                            path: path.into_static(),
1259                        });
1260                    }
1261                }
1262            }
1263            Ok(())
1264        })?;
1265
1266        // If there are any matches, the string will need to be accessed to check for false positives from
1267        // excluded matches, any to potentially mutate the string.
1268        // If there are any async jobs, this is also true since it's not known yet whether there
1269        // will be a match
1270        let needs_to_access_content = !path_rules_matches.is_empty() || !self.async_jobs.is_empty();
1271
1272        self.rule_matches
1273            .push_sync_matches(path, path_rules_matches);
1274
1275        Ok(needs_to_access_content)
1276    }
1277}
1278
1279// Calculates the next starting position for a regex match if a the previous match is a false positive
1280fn get_next_regex_start(content: &str, regex_match: (usize, usize)) -> Option<usize> {
1281    // The next valid UTF8 char after the start of the regex match is used
1282    if let Some((i, _)) = content[regex_match.0..].char_indices().nth(1) {
1283        Some(regex_match.0 + i)
1284    } else {
1285        // There are no more chars left in the string to scan
1286        None
1287    }
1288}
1289
1290fn is_false_positive_match(
1291    regex_match_range: (usize, usize),
1292    rule: &RegexCompiledRule,
1293    content: &str,
1294    check_excluded_keywords: bool,
1295) -> bool {
1296    if check_excluded_keywords
1297        && let Some(excluded_keywords) = &rule.excluded_keywords
1298        && excluded_keywords.is_false_positive_match(content, regex_match_range.0)
1299    {
1300        return true;
1301    }
1302
1303    if let Some(validator) = rule.validator.as_ref()
1304        && !validator.is_valid_match(&content[regex_match_range.0..regex_match_range.1])
1305    {
1306        return true;
1307    }
1308    false
1309}
dd_sds/scanner/mod.rs

dd_sds/scanner/
mod.rs
