diff --git a/proxy_agent_shared/src/misc_helpers.rs b/proxy_agent_shared/src/misc_helpers.rs index 673d4591..b1a81241 100644 --- a/proxy_agent_shared/src/misc_helpers.rs +++ b/proxy_agent_shared/src/misc_helpers.rs @@ -680,6 +680,24 @@ pub fn xml_escape(s: String) -> String { .replace('>', ">") } +/// Truncate the given string to the specified maximum number of bytes, ensuring that it does not cut off in the middle of a character. +/// # Arguments +/// * `text` - The string to be truncated +/// * `max_bytes_size` - The maximum number of bytes the string should occupy +/// # Notes +/// This function ensures that the resulting string is valid UTF-8 by truncating at character boundaries. +pub fn truncate_to_char_boundary(text: &mut String, max_bytes_size: usize) { + if text.len() <= max_bytes_size { + return; + } + + let mut end = max_bytes_size; + while !text.is_char_boundary(end) { + end -= 1; + } + text.truncate(end); +} + #[cfg(test)] mod tests { use regex::Regex; diff --git a/proxy_agent_shared/src/telemetry/event_logger.rs b/proxy_agent_shared/src/telemetry/event_logger.rs index b8f438ef..26bc9629 100644 --- a/proxy_agent_shared/src/telemetry/event_logger.rs +++ b/proxy_agent_shared/src/telemetry/event_logger.rs @@ -13,7 +13,6 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Duration; -const MAX_MESSAGE_LENGTH: usize = 1024 * 4; // 4KB static EVENT_QUEUE: Lazy> = Lazy::new(|| ConcurrentQueue::::bounded(1000)); static SHUT_DOWN: Lazy> = Lazy::new(|| Arc::new(AtomicBool::new(false))); @@ -300,11 +299,13 @@ pub fn write_event( /// Write event only without logging to file /// This event will send out as `TelemetryGenericLogsEvent` pub fn write_event_only(level: Level, message: String, method_name: &str, module_name: &str) { - let event_message = if message.len() > MAX_MESSAGE_LENGTH { - message[..MAX_MESSAGE_LENGTH].to_string() - } else { - message.to_string() - }; + // Truncate the event message to the maximum allowed telemetry message length before sending it to the memory queue. + let mut event_message = message; + misc_helpers::truncate_to_char_boundary( + &mut event_message, + super::telemetry_event::MAX_TELEMETRY_MESSAGE_LENGTH, + ); + match EVENT_QUEUE.push(Event::new( level.to_string(), event_message, @@ -327,14 +328,12 @@ pub fn push_windows_event(windows_event: crate::windows_events::models::WindowsE if let (Some(provider_name), Some(task_name)) = (&windows_event.provider_name, &windows_event.task_name) { - let event_message = { - let message = windows_event.get_message(); - if message.len() > MAX_MESSAGE_LENGTH { - message[..MAX_MESSAGE_LENGTH].to_string() - } else { - message - } - }; + // Truncate the event message to the maximum allowed telemetry message length before sending it to the memory queue. + let mut event_message = windows_event.get_message(); + misc_helpers::truncate_to_char_boundary( + &mut event_message, + super::telemetry_event::MAX_TELEMETRY_MESSAGE_LENGTH, + ); match EVENT_QUEUE.push(Event { EventLevel: windows_event.get_level_string(), diff --git a/proxy_agent_shared/src/telemetry/event_sender.rs b/proxy_agent_shared/src/telemetry/event_sender.rs index cbaf713e..e8b1c11d 100644 --- a/proxy_agent_shared/src/telemetry/event_sender.rs +++ b/proxy_agent_shared/src/telemetry/event_sender.rs @@ -108,7 +108,11 @@ impl EventSender { let mut add_more_events = true; while !TELEMETRY_EVENT_QUEUE.is_empty() && add_more_events { match TELEMETRY_EVENT_QUEUE.pop() { - Ok(event) => { + Ok(mut event) => { + // Redact only in this sequential background consumer. + // Producers stay off the blocking regex path. + event.redact_secrets(); + telemetry_data.add_event(event.clone()); if telemetry_data.get_size() >= MAX_MESSAGE_SIZE { diff --git a/proxy_agent_shared/src/telemetry/telemetry_event.rs b/proxy_agent_shared/src/telemetry/telemetry_event.rs index 0b25b484..38d9d0de 100644 --- a/proxy_agent_shared/src/telemetry/telemetry_event.rs +++ b/proxy_agent_shared/src/telemetry/telemetry_event.rs @@ -8,6 +8,10 @@ use crate::{current_info, misc_helpers}; use once_cell::sync::Lazy; use serde_derive::{Deserialize, Serialize}; +// Keep telemetry messages bounded before secret redaction. Besides limiting the wire payload, this +// prevents externally supplied extension status from causing disproportionate regex work or memory use. +pub const MAX_TELEMETRY_MESSAGE_LENGTH: usize = 4 * 1024; + const METRICS_PROVIDER_ID: &str = "FFF0196F-EE4C-4EAF-9AA5-776F622DEB4F"; const STATUS_PROVIDER_ID: &str = "69B669B9-4AF8-4C50-BDC4-6006FA76E975"; @@ -279,6 +283,22 @@ impl TelemetryEvent { TelemetryEvent::ExtensionEvent(event) => event.to_xml_event(vm_data), } } + + /// Redact an event in the background telemetry-consumer path. + pub(crate) fn redact_secrets(&mut self) { + match self { + TelemetryEvent::GenericLogsEvent(event) => { + event.context1 = crate::secrets_redactor::redact_secrets_string(std::mem::take( + &mut event.context1, + )); + } + TelemetryEvent::ExtensionEvent(event) => { + event.message = crate::secrets_redactor::redact_secrets_string(std::mem::take( + &mut event.message, + )); + } + } + } } /// Struct to hold Generic Logs telemetry event data without VM metadata. @@ -311,9 +331,10 @@ impl TelemetryGenericLogsEvent { Some(version) => (version, format!("{}-{}", event_name, event_log.Version)), None => (event_log.Version.clone(), event_name), }; - // redact secrets in the message before sending to telemetry - let message = event_log.Message.clone(); - let message = crate::secrets_redactor::redact_secrets_string(message); + // Bound producer work and queue memory here; redaction runs later in the background sender. + let mut message = event_log.Message.clone(); + misc_helpers::truncate_to_char_boundary(&mut message, MAX_TELEMETRY_MESSAGE_LENGTH); + TelemetryGenericLogsEvent { event_name, ga_version, @@ -416,9 +437,10 @@ impl TelemetryExtensionEventsEvent { execution_mode: String, ga_version: String, ) -> Self { - // redact secrets in the message before sending to telemetry - let message = event.operation_status.message.clone(); - let message = crate::secrets_redactor::redact_secrets_string(message); + // Bound producer work and queue memory here; redaction runs later in the background sender. + let mut message = event.operation_status.message.clone(); + misc_helpers::truncate_to_char_boundary(&mut message, MAX_TELEMETRY_MESSAGE_LENGTH); + TelemetryExtensionEventsEvent { ga_version, execution_mode, @@ -832,6 +854,47 @@ mod tests { assert!(xml.contains("500")); } + #[test] + fn test_extension_event_bounds_message_before_redaction() { + let mut extension_status_event = create_test_extension_status_event(); + extension_status_event.operation_status.message = format!( + "Authorization: Bearer secret\n{}", + "x".repeat(MAX_TELEMETRY_MESSAGE_LENGTH * 16) + ); + + let telemetry_event = TelemetryExtensionEventsEvent::from_extension_status_event( + &extension_status_event, + "production".to_string(), + "1.0.0".to_string(), + ); + let mut telemetry_event = TelemetryEvent::ExtensionEvent(telemetry_event); + telemetry_event.redact_secrets(); + + let TelemetryEvent::ExtensionEvent(telemetry_event) = telemetry_event else { + unreachable!(); + }; + assert!(telemetry_event.message.len() <= MAX_TELEMETRY_MESSAGE_LENGTH); + assert!(telemetry_event.message.starts_with("[REDACTED]\n")); + assert!(!telemetry_event.message.contains("secret")); + } + + #[test] + fn test_extension_event_truncates_at_utf8_boundary() { + let mut extension_status_event = create_test_extension_status_event(); + extension_status_event.operation_status.message = "é".repeat(MAX_TELEMETRY_MESSAGE_LENGTH); + + let telemetry_event = TelemetryExtensionEventsEvent::from_extension_status_event( + &extension_status_event, + "production".to_string(), + "1.0.0".to_string(), + ); + + assert_eq!(telemetry_event.message.len(), MAX_TELEMETRY_MESSAGE_LENGTH); + assert!(telemetry_event + .message + .is_char_boundary(telemetry_event.message.len())); + } + /// Tests TelemetryExtensionEventsEvent with operation failure #[test] fn test_telemetry_extension_events_event_failure() {