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

dd_sds/scanner/
mod.rs
