feat: Add support for marking steps as sensitive

This leads to logs and alerts being redacted so that the sensitive
response bodies aren't included.

Author: stevensdavid

Cargo.lock

@@ -2,7 +2,7 @@
 
 [package]
 name = "prodzilla"
-version = "0.0.4"
+version = "0.0.3-rc.2"
 edition = "2021"
 
 [dependencies]

Cargo.toml

@@ -76,6 +76,7 @@ A complete Probe config looks as follows:
   - name: Your Post Url
     url: https://your.site/some/path
     http_method: POST
+    sensitive: false
     with:
       headers:
         x-client-id: ClientId
@@ -162,14 +163,34 @@ The webhook looks as such:
   "probe_name": "Your Probe",
   "failure_timestamp": "2024-01-26T02:41:02.983025Z",
   "trace_id": "123456789abcdef",
-  "error_message": "Failed to meet expectation for field 'StatusCode' with operation Equals \"200\". Received: status '500', body '\"Internal Server Error\"' (truncatated to 100 characters)."
+  "error_message": "Failed to meet expectation for field 'StatusCode' with operation Equals \"200\".",
+  "status_code": 500,
+  "body": "Internal Server Error"
 }
 
 ```
 
+Response bodies are truncated to 500 characters. If a step or probe is marked as sensitive, the request body will be redacted from logs and alerts.
+
 Prodzilla will also recognize the Slack webhook domain `hooks.slack.com` and produce messages like:
 
-> Probe Your Probe failed at 2024-06-10 08:16:33.935659994 UTC. Trace ID: 123456789abcdef. Error: Failed to meet expectation for field 'StatusCode' with operation Equals "200". Received: status '500', body '"Internal Server Error"' (truncatated to 100 characters).
+> **"Your Probe" failed.**
+>
+> Error message:
+> 
+> > Failed to meet expectation for field 'StatusCode' with operation Equals "429".
+> 
+> Received status code **500**
+> 
+> Received body:
+>
+> ```
+> Internal Server Error
+> ```
+> 
+> Time: **2024-06-26 14:36:30.094126 UTC**
+> 
+> Trace ID: **e03cc9b03185db8004400049264331de**
 
 OpsGenie, and PagerDuty notification integrations are planned.
 

README.md

@@ -31,20 +31,26 @@ pub async fn alert_if_failure(
         return Ok(());
     }
     let error_message = error.unwrap_or("No error message");
-    let truncated_body = probe_response.map(|r| r.truncated_body(500));
+    let status_code = probe_response.map(|r| r.status_code);
+    let truncated_body = match probe_response {
+        Some(r) if !r.sensitive => Some(r.truncated_body(500)),
+        Some(_) => Some("Redacted".to_owned()),
+        None => None,
+    };
     warn!(
         "Probe {probe_name} failed at {failure_timestamp} with trace ID {}. Status code: {}. Error: {error_message}. Body: {}",
         trace_id.as_ref().unwrap_or(&"N/A".to_owned()),
-        probe_response.map_or("N/A".to_owned(), |r| r.status_code.to_string()),
-        truncated_body.unwrap_or("N/A".to_owned()),
+        status_code.map_or("N/A".to_owned(), |code| code.to_string()),
+        truncated_body.as_ref().unwrap_or(&"N/A".to_owned()),
     );
     let mut errors = Vec::new();
     if let Some(alerts_vec) = alerts {
         for alert in alerts_vec {
             if let Err(e) = send_alert(
                 alert,
                 probe_name.to_owned(),
-                probe_response,
+                status_code,
+                truncated_body.as_deref(),
                 error_message,
                 failure_timestamp,
                 trace_id.clone(),
@@ -86,7 +92,8 @@ pub async fn send_generic_webhook(
 pub async fn send_webhook_alert(
     url: &String,
     probe_name: String,
-    probe_response: Option<&ProbeResponse>,
+    status_code: Option<u32>,
+    body: Option<&str>,
     error_message: &str,
     failure_timestamp: DateTime<Utc>,
     trace_id: Option<String>,
@@ -97,8 +104,8 @@ pub async fn send_webhook_alert(
         error_message: error_message.to_owned(),
         failure_timestamp,
         trace_id,
-        body: probe_response.map(|r| r.truncated_body(500)),
-        status_code: probe_response.map(|r| r.status_code),
+        body: body.map(|s| s.to_owned()),
+        status_code,
     };
 
     let json = serde_json::to_string(&request_body).map_to_send_err()?;
@@ -108,7 +115,8 @@ pub async fn send_webhook_alert(
 pub async fn send_slack_alert(
     webhook_url: &String,
     probe_name: String,
-    probe_response: Option<&ProbeResponse>,
+    status_code: Option<u32>,
+    body: Option<&str>,
     error_message: &str,
     failure_timestamp: DateTime<Utc>,
     trace_id: Option<String>,
@@ -133,26 +141,26 @@ pub async fn send_slack_alert(
         },
     ];
 
-    if let Some(response) = probe_response {
-        blocks.extend([
-            SlackBlock {
-                r#type: "divider".to_owned(),
-                elements: None,
-                text: None,
-            },
-            SlackBlock {
-                r#type: "section".to_owned(),
-                elements: None,
-                text: Some(SlackTextBlock {
-                    r#type: "mrkdwn".to_owned(),
-                    text: format!(
-                        "Received status code *{}* and (truncated) body:\n```\n{}\n```",
-                        response.status_code,
-                        response.body.chars().take(500).collect::<String>()
-                    ),
-                }),
-            },
-        ])
+    if let Some(code) = status_code {
+        blocks.push(SlackBlock {
+            r#type: "section".to_owned(),
+            elements: None,
+            text: Some(SlackTextBlock {
+                r#type: "mrkdwn".to_owned(),
+                text: format!("Received status code *{}*", code,),
+            }),
+        })
+    }
+
+    if let Some(s) = body {
+        blocks.push(SlackBlock {
+            r#type: "section".to_owned(),
+            elements: None,
+            text: Some(SlackTextBlock {
+                r#type: "mrkdwn".to_owned(),
+                text: format!("Received body:\n```\n{}\n```", s,),
+            }),
+        })
     }
 
     blocks.push(SlackBlock {
@@ -178,7 +186,8 @@ pub async fn send_slack_alert(
 pub async fn send_alert(
     alert: &ProbeAlert,
     probe_name: String,
-    probe_response: Option<&ProbeResponse>,
+    status_code: Option<u32>,
+    body: Option<&str>,
     error_message: &str,
     failure_timestamp: DateTime<Utc>,
     trace_id: Option<String>,
@@ -189,7 +198,8 @@ pub async fn send_alert(
             send_slack_alert(
                 &alert.url,
                 probe_name.clone(),
-                probe_response,
+                status_code,
+                body,
                 error_message,
                 failure_timestamp,
                 trace_id.clone(),
@@ -200,7 +210,8 @@ pub async fn send_alert(
             send_webhook_alert(
                 &alert.url,
                 probe_name.clone(),
-                probe_response,
+                status_code,
+                body,
                 error_message,
                 failure_timestamp,
                 trace_id.clone(),

src/alerts/outbound_webhook.rs

@@ -4,9 +4,11 @@ use std::time::Duration;
 use crate::errors::MapToSendError;
 use chrono::Utc;
 use lazy_static::lazy_static;
-use opentelemetry::global::ObjectSafeSpan;
+use opentelemetry::KeyValue;
+use opentelemetry_semantic_conventions::trace as semconv;
 
 use opentelemetry::trace::FutureExt;
+use opentelemetry::trace::Span;
 use opentelemetry::trace::SpanId;
 use opentelemetry::trace::TraceId;
 
@@ -32,6 +34,7 @@ pub async fn call_endpoint(
     http_method: &str,
     url: &String,
     input_parameters: &Option<ProbeInputParameters>,
+    sensitive: bool,
 ) -> Result<EndpointResult, Box<dyn std::error::Error + Send>> {
     let timestamp_start = Utc::now();
     let (otel_headers, cx, span_id, trace_id) =
@@ -41,20 +44,41 @@ pub async fn call_endpoint(
     let response = request
         .timeout(Duration::from_secs(REQUEST_TIMEOUT_SECS))
         .send()
-        .with_context(cx)
+        .with_context(cx.clone())
         .await
         .map_to_send_err()?;
 
     let timestamp_response = Utc::now();
 
-    Ok(EndpointResult {
+    let result = EndpointResult {
         timestamp_request_started: timestamp_start,
         timestamp_response_received: timestamp_response,
         status_code: response.status().as_u16() as u32,
         body: response.text().await.map_to_send_err()?,
+        sensitive,
         trace_id: trace_id.to_string(),
         span_id: span_id.to_string(),
-    })
+    };
+    let span = cx.span();
+    span.set_attributes(vec![
+        KeyValue::new(semconv::HTTP_METHOD, http_method.to_owned()),
+        KeyValue::new(semconv::HTTP_URL, url.clone()),
+    ]);
+    span.set_attribute(KeyValue::new(
+        semconv::HTTP_STATUS_CODE,
+        result.status_code.to_string(),
+    ));
+    if !sensitive {
+        span.add_event(
+            "response",
+            vec![KeyValue::new(
+                "body",
+                result.body.chars().take(500).collect::<String>(),
+            )],
+        )
+    }
+
+    Ok(result)
 }
 
 fn get_otel_headers(span_name: String) -> (HeaderMap, Context, SpanId, TraceId) {
@@ -130,7 +154,7 @@ mod http_tests {
             format!("{}/test", mock_server.uri()),
             "".to_owned(),
         );
-        let endpoint_result = call_endpoint(&probe.http_method, &probe.url, &probe.with)
+        let endpoint_result = call_endpoint(&probe.http_method, &probe.url, &probe.with, false)
             .await
             .unwrap();
         let check_expectations_result = validate_response(
@@ -160,7 +184,8 @@ mod http_tests {
             format!("{}/test", mock_server.uri()),
             body.to_string(),
         );
-        let endpoint_result = call_endpoint(&probe.http_method, &probe.url, &probe.with).await;
+        let endpoint_result =
+            call_endpoint(&probe.http_method, &probe.url, &probe.with, false).await;
 
         assert!(endpoint_result.is_err());
     }
@@ -183,7 +208,7 @@ mod http_tests {
             format!("{}/test", mock_server.uri()),
             body.to_string(),
         );
-        let endpoint_result = call_endpoint(&probe.http_method, &probe.url, &probe.with)
+        let endpoint_result = call_endpoint(&probe.http_method, &probe.url, &probe.with, false)
             .await
             .unwrap();
         let check_expectations_result = validate_response(
@@ -220,7 +245,7 @@ mod http_tests {
             format!("{}/test", mock_server.uri()),
             request_body.to_owned(),
         );
-        let endpoint_result = call_endpoint(&probe.http_method, &probe.url, &probe.with)
+        let endpoint_result = call_endpoint(&probe.http_method, &probe.url, &probe.with, false)
             .await
             .unwrap();
         let check_expectations_result = validate_response(

src/probe/http_probe.rs

@@ -12,6 +12,8 @@ pub struct Probe {
     pub expectations: Option<Vec<ProbeExpectation>>,
     pub schedule: ProbeScheduleParameters,
     pub alerts: Option<Vec<ProbeAlert>>,
+    #[serde(default)] // default to false
+    pub sensitive: bool,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -73,6 +75,7 @@ pub struct ProbeResponse {
     pub timestamp_received: DateTime<Utc>,
     pub status_code: u32,
     pub body: String,
+    pub sensitive: bool,
 }
 
 impl ProbeResponse {
@@ -96,6 +99,8 @@ pub struct Step {
     pub http_method: String,
     pub with: Option<ProbeInputParameters>,
     pub expectations: Option<Vec<ProbeExpectation>>,
+    #[serde(default)] // default to false
+    pub sensitive: bool,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -128,6 +133,7 @@ pub struct EndpointResult {
     pub body: String,
     pub trace_id: String,
     pub span_id: String,
+    pub sensitive: bool,
 }
 
 impl EndpointResult {
@@ -136,6 +142,7 @@ impl EndpointResult {
             timestamp_received: self.timestamp_response_received,
             status_code: self.status_code,
             body: self.body.clone(),
+            sensitive: self.sensitive,
         }
     }
 }