fix(sccm): harden client intake and CCM ambiguity (#319)

crates/cmtraceopen-parser/src/parser/ccm.rs

@@ -14,6 +14,8 @@ use super::severity::detect_severity_from_text;
 use crate::models::log_entry::{LogEntry, LogFormat, ParserSpecialization, Severity};
 use std::sync::OnceLock;
 
+const CCM_RECORD_OPENER: &str = "<![LOG[";
+
 /// Compiled regex matching a complete CCM log line.
 ///
 /// Based on the binary's scanf pattern:
@@ -377,12 +379,40 @@ fn scan_ccm_content(content: &str, file_path: &str, mode: CcmScanMode) -> CcmSca
     let mut errors = 0u32;
     let mut id_counter = 0u64;
     let mut cursor = 0usize;
+    let mut search_cursor = 0usize;
     let mut matched_any = false;
 
-    for caps in ccm_re().captures_iter(content) {
-        let Some(full_match) = caps.get(0) else {
+    while let Some(caps) = ccm_re().captures_at(content, search_cursor) {
+        let full_match = caps
+            .get(0)
+            .expect("CCM regex captures always include the full match");
+
+        let message_end = caps
+            .name("msg")
+            .expect("complete CCM matches always include the message capture")
+            .end();
+        if newest_nested_opener(content, full_match.start(), message_end).is_some()
+            && mode == CcmScanMode::SccmEvidence
+        {
+            // A line-start opener inside a complete-looking multiline record
+            // is byte-for-byte ambiguous: it can be a literal message token
+            // or a recovery boundary after partial input. SCCM evidence must
+            // not promote either interpretation. The public projection keeps
+            // the legacy full-match result to preserve LogEntry compatibility.
+            push_unmatched_plain(
+                &content[cursor..full_match.end()],
+                cursor,
+                &line_starts,
+                file_path,
+                &mut records,
+                &mut id_counter,
+                &mut errors,
+            );
+            cursor = full_match.end();
+            search_cursor = full_match.end();
+            matched_any = true;
             continue;
-        };
+        }
 
         // Emit unmatched text between the previous match and this one
         push_unmatched_plain(
@@ -426,6 +456,7 @@ fn scan_ccm_content(content: &str, file_path: &str, mode: CcmScanMode) -> CcmSca
         }
 
         cursor = full_match.end();
+        search_cursor = full_match.end();
         matched_any = true;
     }
 
@@ -462,6 +493,30 @@ fn scan_ccm_content(content: &str, file_path: &str, mode: CcmScanMode) -> CcmSca
     CcmScan { records, errors }
 }
 
+fn newest_nested_opener(
+    content: &str,
+    full_match_start: usize,
+    message_end: usize,
+) -> Option<usize> {
+    // Only physical-line openers inside the message can delimit recovery
+    // records. Attribute values are outside the raw CCM payload and may
+    // contain literal opener text. Scan newest first so an adversarial run of
+    // partial message openers is discarded in one pass.
+    let nested_search_start = full_match_start.checked_add(CCM_RECORD_OPENER.len())?;
+    content
+        .get(nested_search_start..message_end)?
+        .rmatch_indices(CCM_RECORD_OPENER)
+        .find_map(|(offset, _)| {
+            let absolute = nested_search_start + offset;
+            let starts_logical_line = absolute == 0
+                || content
+                    .as_bytes()
+                    .get(absolute - 1)
+                    .is_some_and(|previous| matches!(previous, b'\n' | b'\r'));
+            starts_logical_line.then_some(absolute)
+        })
+}
+
 /// Parse named captures from a CCM regex match into a CcmParsed struct.
 fn parse_captures(caps: &regex::Captures<'_>) -> Option<CcmParsed> {
     let msg = caps.name("msg").map(|m| m.as_str().to_string())?;
@@ -1011,6 +1066,115 @@ mod tests {
         assert!(records[0].timestamp.utc_millis.is_some());
     }
 
+    #[test]
+    fn logical_scanner_treats_line_start_opener_inside_multiline_message_as_ambiguous() {
+        for newline in ["\n", "\r\n"] {
+            let text = format!(
+                concat!(
+                    "<![LOG[first line{newline}",
+                    "<![LOG[literal at continuation start]LOG]!>",
+                    "<time=\"10:00:00.000-240\" date=\"07-30-2026\" ",
+                    "component=\"PolicyAgent\" context=\"\" type=\"1\" thread=\"42\">"
+                ),
+                newline = newline,
+            );
+
+            assert!(
+                scan_logical_records(&text, "PolicyAgent.log").is_empty(),
+                "{newline:?} physical-line ambiguity must remain coverage-only"
+            );
+        }
+    }
+
+    #[test]
+    fn public_projection_keeps_multiline_line_start_literal_opener_as_one_log_entry() {
+        for newline in ["\n", "\r\n"] {
+            let text = format!(
+                concat!(
+                    "<![LOG[first line{newline}",
+                    "<![LOG[literal at continuation start]LOG]!>",
+                    "<time=\"10:00:00.000-240\" date=\"07-30-2026\" ",
+                    "component=\"PolicyAgent\" context=\"\" type=\"1\" thread=\"42\">"
+                ),
+                newline = newline,
+            );
+
+            let (entries, errors) = parse_content(&text, "PolicyAgent.log", None);
+
+            assert_eq!(errors, 0, "{newline:?} public parse must remain compatible");
+            assert_eq!(entries.len(), 1, "{newline:?} must remain one LogEntry");
+            assert_eq!(entries[0].format, LogFormat::Ccm);
+            assert_eq!(
+                entries[0].message,
+                format!("first line{newline}<![LOG[literal at continuation start")
+            );
+        }
+    }
+
+    #[test]
+    fn logical_scanner_keeps_same_line_literal_opener_in_one_complete_record() {
+        let text = concat!(
+            "<![LOG[Diagnostic text retained a literal <![LOG[ token]LOG]!>",
+            "<time=\"10:00:00.000-240\" date=\"07-30-2026\" ",
+            "component=\"PolicyAgent\" context=\"\" type=\"1\" thread=\"42\">"
+        );
+
+        let records = scan_logical_records(text, "PolicyAgent.log");
+
+        assert_eq!(records.len(), 1);
+        assert_eq!(
+            records[0].entry.message,
+            "Diagnostic text retained a literal <![LOG[ token"
+        );
+        assert_eq!(records[0].line_start, 1);
+        assert_eq!(records[0].line_end, 1);
+    }
+
+    #[test]
+    fn logical_scanner_ignores_line_start_opener_inside_attribute_value() {
+        let text = concat!(
+            "<![LOG[Policy request completed]LOG]!>",
+            "<time=\"10:00:00.000-240\" date=\"07-30-2026\" ",
+            "component=\"PolicyAgent\" context=\"diagnostic continuation\n",
+            "<![LOG[literal attribute token\" type=\"1\" thread=\"42\">"
+        );
+
+        let records = scan_logical_records(text, "PolicyAgent.log");
+
+        assert_eq!(records.len(), 1);
+        assert_eq!(records[0].entry.message, "Policy request completed");
+        assert_eq!(
+            records[0].context.as_deref(),
+            Some("diagnostic continuation\n<![LOG[literal attribute token")
+        );
+    }
+
+    #[test]
+    fn ambiguity_recovery_selects_the_newest_physical_line_opener() {
+        let partial_prefix = "<![LOG[partial\n".repeat(4_096);
+        let text = format!(
+            concat!(
+                "{partial_prefix}<![LOG[complete record]LOG]!><time=\"10:00:00.000-240\" ",
+                "date=\"07-30-2026\" component=\"PolicyAgent\" context=\"\" ",
+                "type=\"1\" thread=\"42\">"
+            ),
+            partial_prefix = partial_prefix,
+        );
+        let full_match = ccm_re()
+            .find(&text)
+            .expect("the partial prefix and terminal close form one complete-looking match");
+
+        assert_eq!(
+            newest_nested_opener(&text, full_match.start(), full_match.end()),
+            text.rfind("<![LOG["),
+            "recovery must jump to the newest nested opener in one scan"
+        );
+        assert!(
+            scan_logical_records(&text, "PolicyAgent.log").is_empty(),
+            "the adversarial ambiguous segment must remain coverage-only"
+        );
+    }
+
     fn logical_record_for_time_tail(tail: &str) -> CcmLogicalRecord {
         let text = format!(
             concat!(

crates/cmtraceopen-parser/src/sccm/client/intake.rs

@@ -1,8 +1,13 @@
 use std::cmp::Ordering;
 use std::collections::{BTreeMap, BTreeSet};
+use std::fmt;
 
 use chrono::{DateTime, SecondsFormat, Utc};
-use serde::{de::Error as _, ser::Error as _, Deserialize, Deserializer, Serialize, Serializer};
+use serde::{
+    de::{Error as _, IgnoredAny, SeqAccess, Visitor},
+    ser::Error as _,
+    Deserialize, Deserializer, Serialize, Serializer,
+};
 use thiserror::Error;
 
 use crate::sccm::catalog::{
@@ -12,6 +17,16 @@ use crate::sccm::{
     SccmArtifact, SccmCoverageState, SccmRole, SccmRotation, SCCM_DIAGNOSTICS_SCHEMA_VERSION,
 };
 
+/// Maximum artifact declarations admitted by one SCCM client intake bundle.
+///
+/// This is the authoritative v1 ceiling for both pure client intake and the
+/// native SCCM client manifest/capture boundary. Native readers, writers, and
+/// collectors must import and reuse this constant rather than define a wider
+/// or otherwise parallel limit. Pure intake validates it before allocating
+/// per-artifact indexes so a malformed bundle cannot turn validation into an
+/// unbounded allocation path.
+pub const MAX_SCCM_CLIENT_INTAKE_ARTIFACTS: usize = 4096;
+
 const MAX_ARTIFACT_ID_CHARS: usize = 160;
 const MAX_BASENAME_CHARS: usize = 160;
 const MAX_COLLECTED_AT_CHARS: usize = 64;
@@ -172,12 +187,87 @@ pub struct SccmClientIntakeArtifact {
     pub fragment_complete: Option<bool>,
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct SccmClientIntakeBundle {
     pub artifacts: Vec<SccmClientIntakeArtifact>,
 }
 
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+struct SccmClientIntakeBundleWire {
+    #[serde(deserialize_with = "deserialize_bounded_client_artifacts")]
+    artifacts: Vec<SccmClientIntakeArtifact>,
+}
+
+impl<'de> Deserialize<'de> for SccmClientIntakeBundle {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let wire = SccmClientIntakeBundleWire::deserialize(deserializer)?;
+        Ok(Self {
+            artifacts: wire.artifacts,
+        })
+    }
+}
+
+fn deserialize_bounded_client_artifacts<'de, D>(
+    deserializer: D,
+) -> Result<Vec<SccmClientIntakeArtifact>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    struct BoundedArtifactVisitor;
+
+    impl<'de> Visitor<'de> for BoundedArtifactVisitor {
+        type Value = Vec<SccmClientIntakeArtifact>;
+
+        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+            write!(
+                formatter,
+                "at most {MAX_SCCM_CLIENT_INTAKE_ARTIFACTS} SCCM client artifacts"
+            )
+        }
+
+        fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
+        where
+            A: SeqAccess<'de>,
+        {
+            if sequence
+                .size_hint()
+                .is_some_and(|size| size > MAX_SCCM_CLIENT_INTAKE_ARTIFACTS)
+            {
+                return Err(A::Error::custom(
+                    SccmClientIntakeError::ArtifactLimitExceeded,
+                ));
+            }
+
+            let initial_capacity = sequence
+                .size_hint()
+                .unwrap_or_default()
+                .min(MAX_SCCM_CLIENT_INTAKE_ARTIFACTS);
+            let mut artifacts = Vec::with_capacity(initial_capacity);
+            while artifacts.len() < MAX_SCCM_CLIENT_INTAKE_ARTIFACTS {
+                let Some(artifact) = sequence.next_element()? else {
+                    return Ok(artifacts);
+                };
+                artifacts.push(artifact);
+            }
+
+            if sequence.next_element::<IgnoredAny>()?.is_some() {
+                return Err(A::Error::custom(
+                    SccmClientIntakeError::ArtifactLimitExceeded,
+                ));
+            }
+
+            Ok(artifacts)
+        }
+    }
+
+    deserializer.deserialize_seq(BoundedArtifactVisitor)
+}
+
 #[derive(Debug, Clone, PartialEq)]
 pub struct SccmClientIntakeFragment {
     pub artifact_id: String,
@@ -489,6 +579,8 @@ impl SccmClientIntakeAssessment {
 
 #[derive(Debug, Clone, PartialEq, Eq, Error)]
 pub enum SccmClientIntakeError {
+    #[error("client intake artifact count exceeds the supported limit")]
+    ArtifactLimitExceeded,
     #[error("client intake artifact identity is empty, unsafe, or too long")]
     InvalidArtifactId,
     #[error("client intake artifact basename is empty, unsafe, or too long")]
@@ -804,6 +896,10 @@ fn unsupported_as_intake_artifact(
 }
 
 fn validate_bundle(bundle: &SccmClientIntakeBundle) -> Result<(), SccmClientIntakeError> {
+    if bundle.artifacts.len() > MAX_SCCM_CLIENT_INTAKE_ARTIFACTS {
+        return Err(SccmClientIntakeError::ArtifactLimitExceeded);
+    }
+
     let mut artifact_ids = BTreeSet::new();
     let mut path_fingerprint_bindings: BTreeMap<String, (Option<String>, String)> = BTreeMap::new();
     let mut rotation_lineage_bindings = BTreeMap::new();