rsigma-parser 0.18.0

//! ADS (Alerting and Detection Strategy) section vocabulary and reading helpers.
//!
//! The Palantir ADS framework describes nine sections every production
//! detection should carry: a goal, an ATT&CK categorization, a strategy
//! abstract, technical context, stated blind spots and assumptions,
//! false-positive notes, a true-positive validation recipe, a priority, and a
//! response plan. RSigma already homes four of them on standard Sigma fields
//! (`description`, `tags`, `falsepositives`, `level`) and carries the rest as
//! plain documentation under a `rsigma.ads.*` custom-attribute namespace.
//!
//! [`ads_catalogue`] is the single source of truth for that vocabulary: one
//! [`AdsSectionInfo`] per section (its stable snake_case id, the field that
//! carries it, whether it is required by default, and a one-line description).
//! The linter, the `rsigma rule doc` command, the MCP `rsigma://ads/schema`
//! resource, and the docs all ground on this list. The list is generated by one
//! macro so the same source drives both the catalogue and an *exhaustive*
//! `match`: adding an [`AdsSection`] variant without a catalogue entry is a
//! compile error.
//!
//! These values are pure documentation. The engine never interprets them, so
//! they carry zero runtime cost.
//!
//! # Example
//!
//! ```rust
//! use rsigma_parser::ads::{ads_catalogue, AdsSection};
//!
//! let sections = ads_catalogue();
//! assert_eq!(sections.len(), 9);
//!
//! let goal = sections.iter().find(|s| s.id == "goal").unwrap();
//! assert!(goal.default_required);
//! assert_eq!(AdsSection::Goal.carrier_field(), "description");
//! ```

use serde::Serialize;

use crate::ast::SigmaRule;

/// The `rsigma.ads.*` custom-attribute key that opts a rule out of ADS
/// enforcement (`rsigma.ads.exempt: true`).
pub const EXEMPT_KEY: &str = "rsigma.ads.exempt";

/// The shared prefix of every `rsigma.ads.*` custom-attribute key.
pub const ADS_PREFIX: &str = "rsigma.ads.";

/// One ADS section.
///
/// Reference: Palantir Alerting and Detection Strategy framework.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum AdsSection {
    /// What the detection is trying to catch (carried by `description`).
    Goal,
    /// The ATT&CK categorization (carried by `attack.*` `tags`).
    Categorization,
    /// A one-paragraph abstract of the detection approach.
    Strategy,
    /// The data source, fields, and environment knowledge the detection needs.
    TechnicalContext,
    /// How an attacker could evade the detection, and what it assumes.
    BlindSpots,
    /// Known benign triggers (carried by `falsepositives`).
    FalsePositives,
    /// A recipe that produces a true-positive event the detection fires on.
    Validation,
    /// Why the detection's `level` is what it is (the priority rationale).
    Priority,
    /// What an analyst should do when the detection fires.
    Response,
}

/// Where an ADS section's content lives on a rule.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case", tag = "kind", content = "field")]
pub enum AdsCarrier {
    /// A standard Sigma field reused as-is (`description`, `tags`,
    /// `falsepositives`, `level`).
    StandardField(&'static str),
    /// A new `rsigma.ads.*` custom-attribute key.
    CustomAttribute(&'static str),
}

impl AdsCarrier {
    /// The field name or attribute key, regardless of carrier kind.
    pub fn name(&self) -> &'static str {
        match self {
            AdsCarrier::StandardField(name) | AdsCarrier::CustomAttribute(name) => name,
        }
    }
}

/// Metadata describing one ADS section.
#[derive(Debug, Clone, Copy, Serialize)]
pub struct AdsSectionInfo {
    /// The section variant.
    pub section: AdsSection,
    /// Stable snake_case identifier (matches `AdsSection`'s serde rename and
    /// the `ads.<id>` config key).
    pub id: &'static str,
    /// Where the section's content is carried on a rule.
    pub carrier: AdsCarrier,
    /// Whether the section is required by default (before config overrides).
    pub default_required: bool,
    /// One-line, human-readable description of the section.
    pub description: &'static str,
}

/// Build the catalogue plus the exhaustive metadata lookup from one list.
///
/// Every `AdsSection` variant must appear exactly once. The generated
/// `describe` match has no wildcard arm, so a new variant fails to compile
/// until it is added here.
macro_rules! ads_catalogue {
    ($($variant:ident => ($id:expr, $carrier:expr, $required:expr, $desc:expr)),+ $(,)?) => {
        /// All ADS sections, in canonical (framework) order.
        const ALL_ADS_SECTIONS: &[AdsSection] = &[$(AdsSection::$variant),+];

        fn describe(section: AdsSection) -> AdsSectionInfo {
            match section {
                $(AdsSection::$variant => AdsSectionInfo {
                    section: AdsSection::$variant,
                    id: $id,
                    carrier: $carrier,
                    default_required: $required,
                    description: $desc,
                }),+
            }
        }
    };
}

use AdsCarrier::{CustomAttribute, StandardField};

ads_catalogue! {
    Goal => ("goal", StandardField("description"), true,
        "What the detection is trying to catch."),
    Categorization => ("categorization", StandardField("tags"), true,
        "The ATT&CK categorization, carried by attack.* tags."),
    Strategy => ("strategy", CustomAttribute("rsigma.ads.strategy"), true,
        "A one-paragraph abstract of the detection approach."),
    TechnicalContext => ("technical_context", CustomAttribute("rsigma.ads.technical_context"), true,
        "The data source, fields, and environment knowledge the detection needs."),
    BlindSpots => ("blind_spots", CustomAttribute("rsigma.ads.blind_spots"), true,
        "How an attacker could evade the detection, and what it assumes."),
    FalsePositives => ("false_positives", StandardField("falsepositives"), true,
        "Known benign triggers, carried by falsepositives."),
    Validation => ("validation", CustomAttribute("rsigma.ads.validation"), true,
        "A recipe that produces a true-positive event the detection fires on."),
    Priority => ("priority", CustomAttribute("rsigma.ads.priority"), true,
        "Why the detection's level is what it is (the priority rationale)."),
    Response => ("response", CustomAttribute("rsigma.ads.response"), true,
        "What an analyst should do when the detection fires."),
}

/// Return metadata for every [`AdsSection`], in canonical order.
pub fn ads_catalogue() -> Vec<AdsSectionInfo> {
    ALL_ADS_SECTIONS.iter().map(|&s| describe(s)).collect()
}

impl AdsSection {
    /// All sections, in canonical order.
    pub fn all() -> &'static [AdsSection] {
        ALL_ADS_SECTIONS
    }

    /// Look up a section by its stable snake_case id.
    pub fn from_id(id: &str) -> Option<AdsSection> {
        ALL_ADS_SECTIONS.iter().copied().find(|s| s.info().id == id)
    }

    /// This section's catalogue metadata.
    pub fn info(&self) -> AdsSectionInfo {
        describe(*self)
    }

    /// The stable snake_case id (e.g. `blind_spots`).
    pub fn id(&self) -> &'static str {
        self.info().id
    }

    /// The carrier of this section's content.
    pub fn carrier(&self) -> AdsCarrier {
        self.info().carrier
    }

    /// The field name or attribute key that carries this section.
    pub fn carrier_field(&self) -> &'static str {
        self.info().carrier.name()
    }

    /// Whether this section is required by default.
    pub fn default_required(&self) -> bool {
        self.info().default_required
    }

    /// Extract this section's content from a rule, or `None` when the section
    /// is absent or blank.
    pub fn content(&self, rule: &SigmaRule) -> Option<AdsContent> {
        match self {
            AdsSection::Goal => rule
                .description
                .as_deref()
                .and_then(non_blank)
                .map(|s| AdsContent::Text(s.to_string())),
            AdsSection::Categorization => {
                let tags: Vec<String> = attack_tags(rule).map(str::to_string).collect();
                if tags.is_empty() {
                    None
                } else {
                    Some(AdsContent::List(tags))
                }
            }
            AdsSection::FalsePositives => {
                let items: Vec<String> = rule
                    .falsepositives
                    .iter()
                    .filter_map(|s| non_blank(s).map(str::to_string))
                    .collect();
                if items.is_empty() {
                    None
                } else {
                    Some(AdsContent::List(items))
                }
            }
            other => {
                let key = other.carrier_field();
                rule.custom_attributes.get(key).and_then(content_from_value)
            }
        }
    }

    /// Whether this section's content is present and non-blank on the rule.
    pub fn is_present(&self, rule: &SigmaRule) -> bool {
        self.content(rule).is_some()
    }
}

/// Rendered content of an ADS section.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(untagged)]
pub enum AdsContent {
    /// A single prose value (block scalar).
    Text(String),
    /// A list of values (blind spots, response steps, tags).
    List(Vec<String>),
}

impl AdsContent {
    /// Render as plain text, one list item per line.
    pub fn as_text(&self) -> String {
        match self {
            AdsContent::Text(s) => s.clone(),
            AdsContent::List(items) => items.join("\n"),
        }
    }

    /// The list items, treating a single text value as a one-element list.
    pub fn items(&self) -> Vec<String> {
        match self {
            AdsContent::Text(s) => vec![s.clone()],
            AdsContent::List(items) => items.clone(),
        }
    }
}

/// Whether a rule is exempt from ADS enforcement (`rsigma.ads.exempt: true`).
pub fn is_exempt(rule: &SigmaRule) -> bool {
    rule.custom_attributes
        .get(EXEMPT_KEY)
        .and_then(|v| v.as_bool())
        .unwrap_or(false)
}

/// The `attack.*` tags on a rule (the ATT&CK categorization carrier).
pub fn attack_tags(rule: &SigmaRule) -> impl Iterator<Item = &str> {
    rule.tags
        .iter()
        .map(String::as_str)
        .filter(|t| t.starts_with("attack."))
}

/// Whether the rule carries an ATT&CK categorization: an `attack.*` tag, or a
/// tag in any of the `extra_namespaces` (a private ATT&CK-adjacent taxonomy a
/// team recognises via the linter's `tag_namespaces` setting).
///
/// [`AdsSection::Categorization`]'s own [`content`](AdsSection::content) and
/// [`is_present`](AdsSection::is_present) consider only `attack.*`; this is the
/// config-aware variant the linter, `rule doc`, and the `author_ads` tool use so
/// the three agree on whether a rule is categorized.
pub fn has_categorization(rule: &SigmaRule, extra_namespaces: &[String]) -> bool {
    rule.tags
        .iter()
        .filter_map(|t| t.split('.').next())
        .any(|ns| ns == "attack" || extra_namespaces.iter().any(|e| e == ns))
}

/// The status of one ADS section on a rule: which section, whether it is
/// present, and its content when present.
#[derive(Debug, Clone, Serialize)]
pub struct AdsSectionStatus {
    /// The section id (e.g. `validation`).
    pub id: &'static str,
    /// Whether the section is required (by default; callers may override).
    pub required: bool,
    /// Whether the section's content is present on the rule.
    pub present: bool,
    /// The carrier field or attribute key.
    pub carrier: &'static str,
    /// The rendered content when present.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub content: Option<AdsContent>,
}

/// The assembled ADS document for one rule: every section, its presence, and
/// its content.
#[derive(Debug, Clone, Serialize)]
pub struct AdsDocument {
    /// One entry per ADS section, in canonical order.
    pub sections: Vec<AdsSectionStatus>,
}

impl AdsDocument {
    /// Assemble the ADS document for a rule from its reused fields and
    /// `rsigma.ads.*` sections.
    pub fn from_rule(rule: &SigmaRule) -> Self {
        let sections = AdsSection::all()
            .iter()
            .map(|s| {
                let content = s.content(rule);
                AdsSectionStatus {
                    id: s.id(),
                    required: s.default_required(),
                    present: content.is_some(),
                    carrier: s.carrier_field(),
                    content,
                }
            })
            .collect();
        AdsDocument { sections }
    }

    /// The ids of required sections missing from the rule.
    pub fn missing_required(&self) -> Vec<&'static str> {
        self.sections
            .iter()
            .filter(|s| s.required && !s.present)
            .map(|s| s.id)
            .collect()
    }
}

/// One entry of a generated ADS scaffold: a `rsigma.ads.*` key and a
/// placeholder value for an author or agent to complete.
#[derive(Debug, Clone, Serialize)]
pub struct AdsScaffoldEntry {
    /// The `rsigma.ads.*` custom-attribute key.
    pub key: &'static str,
    /// The placeholder content.
    pub placeholder: AdsContent,
}

/// Build placeholder `rsigma.ads.*` entries for the sections a rule is missing.
///
/// Only the custom-attribute sections are scaffolded; the reused fields
/// (`description`, `tags`, `falsepositives`) already live on the rule, so the
/// scaffold leaves them in place and fills the gaps under `rsigma.ads.*`.
pub fn scaffold_missing(rule: &SigmaRule) -> Vec<AdsScaffoldEntry> {
    AdsSection::all()
        .iter()
        .filter(|s| matches!(s.carrier(), AdsCarrier::CustomAttribute(_)))
        .filter(|s| !s.is_present(rule))
        .map(|s| AdsScaffoldEntry {
            key: s.carrier_field(),
            placeholder: placeholder_for(*s),
        })
        .collect()
}

fn placeholder_for(section: AdsSection) -> AdsContent {
    match section {
        AdsSection::Strategy => AdsContent::Text(
            "TODO: a one-paragraph abstract of what this detection does and the approach it takes."
                .to_string(),
        ),
        AdsSection::TechnicalContext => AdsContent::Text(
            "TODO: the data source, fields, and environment knowledge needed to understand this \
             detection."
                .to_string(),
        ),
        AdsSection::BlindSpots => AdsContent::List(vec![
            "TODO: a way an attacker could evade this detection.".to_string(),
            "TODO: an assumption this detection relies on.".to_string(),
        ]),
        AdsSection::Validation => AdsContent::Text(
            "TODO: the steps to generate a true-positive event that triggers this detection."
                .to_string(),
        ),
        AdsSection::Priority => AdsContent::Text(
            "TODO: why this detection's level is set as it is, and what it implies for response \
             urgency."
                .to_string(),
        ),
        AdsSection::Response => AdsContent::List(vec![
            "TODO: the first triage step when this detection fires.".to_string(),
            "TODO: the escalation or containment action.".to_string(),
        ]),
        // The reused-field sections are never scaffolded under rsigma.ads.*.
        AdsSection::Goal | AdsSection::Categorization | AdsSection::FalsePositives => {
            AdsContent::Text(String::new())
        }
    }
}

fn non_blank(s: &str) -> Option<&str> {
    let t = s.trim();
    if t.is_empty() { None } else { Some(t) }
}

fn content_from_value(v: &yaml_serde::Value) -> Option<AdsContent> {
    use yaml_serde::Value;
    match v {
        Value::Sequence(seq) => {
            let items: Vec<String> = seq.iter().filter_map(scalar_text).collect();
            if items.is_empty() {
                None
            } else {
                Some(AdsContent::List(items))
            }
        }
        other => scalar_text(other).map(AdsContent::Text),
    }
}

fn scalar_text(v: &yaml_serde::Value) -> Option<String> {
    use yaml_serde::Value;
    match v {
        Value::String(s) => non_blank(s).map(str::to_string),
        Value::Bool(b) => Some(b.to_string()),
        Value::Number(n) => Some(n.to_string()),
        _ => None,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::parse_sigma_yaml;

    fn rule(yaml: &str) -> SigmaRule {
        parse_sigma_yaml(yaml).unwrap().rules.pop().unwrap()
    }

    const FULL_RULE: &str = r#"
title: Whoami execution
description: Detects whoami execution, a common discovery step.
status: stable
logsource:
    category: process_creation
    product: windows
detection:
    selection:
        CommandLine|contains: whoami
    condition: selection
level: medium
falsepositives:
    - Legitimate administrators enumerating their own privileges
tags:
    - attack.execution
    - attack.t1059
custom_attributes:
    rsigma.ads.strategy: Watch for the whoami binary in process creation events.
    rsigma.ads.technical_context: Requires process_creation telemetry with CommandLine.
    rsigma.ads.blind_spots:
        - Renamed whoami binaries evade the image match.
        - Assumes CommandLine logging is enabled.
    rsigma.ads.validation: Run `whoami` in a lab and confirm the rule fires.
    rsigma.ads.priority: Medium because discovery is mid-kill-chain.
    rsigma.ads.response:
        - Confirm the user and host.
        - Correlate with other discovery activity.
"#;

    #[test]
    fn catalogue_has_nine_sections() {
        let cat = ads_catalogue();
        assert_eq!(cat.len(), 9);
        assert_eq!(ALL_ADS_SECTIONS.len(), 9);
    }

    #[test]
    fn ids_are_unique_and_round_trip() {
        use std::collections::HashSet;
        let mut seen = HashSet::new();
        for &s in AdsSection::all() {
            let id = s.id();
            assert!(seen.insert(id), "duplicate ADS section id: {id}");
            assert_eq!(AdsSection::from_id(id), Some(s));
        }
        assert_eq!(AdsSection::from_id("nope"), None);
    }

    #[test]
    fn carriers_match_the_schema() {
        assert_eq!(AdsSection::Goal.carrier_field(), "description");
        assert_eq!(AdsSection::Categorization.carrier_field(), "tags");
        assert_eq!(AdsSection::FalsePositives.carrier_field(), "falsepositives");
        assert_eq!(AdsSection::Strategy.carrier_field(), "rsigma.ads.strategy");
        assert!(matches!(
            AdsSection::Goal.carrier(),
            AdsCarrier::StandardField(_)
        ));
        assert!(matches!(
            AdsSection::Response.carrier(),
            AdsCarrier::CustomAttribute(_)
        ));
    }

    #[test]
    fn full_rule_has_every_section_present() {
        let rule = rule(FULL_RULE);
        let doc = AdsDocument::from_rule(&rule);
        assert!(doc.missing_required().is_empty(), "{doc:?}");
        for s in AdsSection::all() {
            assert!(s.is_present(&rule), "{} should be present", s.id());
        }
    }

    #[test]
    fn reused_fields_satisfy_their_sections() {
        // description satisfies goal, attack.* tags satisfy categorization,
        // falsepositives satisfies false_positives.
        let rule = rule(FULL_RULE);
        assert!(AdsSection::Goal.is_present(&rule));
        assert!(AdsSection::Categorization.is_present(&rule));
        assert!(AdsSection::FalsePositives.is_present(&rule));
    }

    #[test]
    fn list_content_preserves_items() {
        let rule = rule(FULL_RULE);
        match AdsSection::BlindSpots.content(&rule).unwrap() {
            AdsContent::List(items) => assert_eq!(items.len(), 2),
            other => panic!("expected list, got {other:?}"),
        }
    }

    #[test]
    fn bare_rule_is_missing_custom_sections() {
        let rule = rule(
            r#"
title: Bare
status: stable
logsource:
    category: test
detection:
    selection:
        field: value
    condition: selection
"#,
        );
        let doc = AdsDocument::from_rule(&rule);
        let missing = doc.missing_required();
        // Every section is missing: no description, no tags, no falsepositives,
        // no rsigma.ads.* keys.
        assert_eq!(missing.len(), 9);
    }

    #[test]
    fn scaffold_fills_only_missing_custom_sections() {
        let rule = rule(
            r#"
title: Partly documented
description: Has a goal already.
status: stable
logsource:
    category: test
detection:
    selection:
        field: value
    condition: selection
custom_attributes:
    rsigma.ads.strategy: Already written.
"#,
        );
        let entries = scaffold_missing(&rule);
        let keys: Vec<&str> = entries.iter().map(|e| e.key).collect();
        // strategy is present, so it is not scaffolded; the other five custom
        // sections are.
        assert!(!keys.contains(&"rsigma.ads.strategy"));
        assert!(keys.contains(&"rsigma.ads.validation"));
        assert!(keys.contains(&"rsigma.ads.response"));
        assert_eq!(entries.len(), 5);
    }

    #[test]
    fn categorization_honours_extra_namespaces() {
        let rule = rule(
            r#"
title: Private taxonomy
status: stable
logsource:
    category: test
detection:
    selection:
        field: value
    condition: selection
tags:
    - myorg.technique
"#,
        );
        // attack.* alone does not satisfy it.
        assert!(!AdsSection::Categorization.is_present(&rule));
        assert!(!has_categorization(&rule, &[]));
        // A configured namespace does.
        assert!(has_categorization(&rule, &["myorg".to_string()]));
    }

    #[test]
    fn exempt_flag_is_read() {
        let rule = rule(
            r#"
title: Vendor import
status: stable
logsource:
    category: test
detection:
    selection:
        field: value
    condition: selection
custom_attributes:
    rsigma.ads.exempt: true
"#,
        );
        assert!(is_exempt(&rule));
    }
}