diff --git a/frontend/e2e/alerts.spec.ts b/frontend/e2e/alerts.spec.ts index 475547b68..4b7fb5516 100644 --- a/frontend/e2e/alerts.spec.ts +++ b/frontend/e2e/alerts.spec.ts @@ -560,6 +560,34 @@ test('Slack and Alertmanager targets retain their distinct fields', async ({ pag kind === 'slack' ? 'Slack' : 'Alertmanager', ); } + // Stored credentials come back masked: neither is prefilled, and both must be re-entered. + await rowAction(page, 'Test alertManager', 'Edit'); + const sheet = page.getByRole('dialog', { name: 'Edit target', exact: true }); + const username = sheet.getByLabel('Username', { exact: true }); + const password = sheet.getByLabel('Password', { exact: true }); + await expect(sheet.getByLabel('Target type')).toHaveValue('alertManager'); + await expect(username).toHaveValue(''); + await expect(password).toHaveValue(''); + await expect( + sheet.getByText('Re-enter both username and password to keep basic authentication', { + exact: false, + }), + ).toBeVisible(); + await sheet.getByLabel('Endpoint URL').fill('https://example.com/api/v2/alerts'); + await username.fill('********'); + await password.fill('secret'); + await sheet.getByRole('button', { name: 'Save target' }).click(); + await expect(username).toHaveAccessibleDescription('Enter the real username, not the mask.'); + await username.fill('svc'); + await password.fill(''); + await password.blur(); + await expect(password).toHaveAccessibleDescription( + /Enter both username and password, or leave both empty\./, + ); + await password.fill('new-secret'); + await sheet.getByRole('button', { name: 'Save target' }).click(); + await expect(sheet).toHaveCount(0); + await expect(namedRow(page, 'Test alertManager')).toContainText('Alertmanager'); }); test('validation errors describe their controls and block invalid thresholds, frequency and hold', async ({ page, diff --git a/frontend/src/features/alerts/TargetSheet.tsx b/frontend/src/features/alerts/TargetSheet.tsx index c944209e3..516e9a3dd 100644 --- a/frontend/src/features/alerts/TargetSheet.tsx +++ b/frontend/src/features/alerts/TargetSheet.tsx @@ -218,11 +218,19 @@ export function TargetSheet({ )} {draft.type === 'alertManager' && ( <> + {target && ( +

+ Stored credentials are masked. Re-enter both username and password to keep basic + authentication; leaving both empty removes it. +

+ )} shown.touch('username')} onChange={(event) => update('username', event.target.value)} /> shown.touch('password')} onChange={(event) => update('password', event.target.value)} diff --git a/frontend/src/features/alerts/helpers.test.ts b/frontend/src/features/alerts/helpers.test.ts index a5dde151f..8492428f8 100644 --- a/frontend/src/features/alerts/helpers.test.ts +++ b/frontend/src/features/alerts/helpers.test.ts @@ -443,28 +443,65 @@ describe('list filtering, sorting and bounded type discovery', () => { }); }); describe('target forms', () => { + const maskedAlertmanager = { + id: 'am', + name: 'AM', + type: 'webhook' as const, + endpoint: 'https://********', + username: '********', + password: '********', + }; it('recognizes masked Alertmanager, requires re-entry and builds top-level credentials', () => { - const target = { - id: 'am', - name: 'AM', - type: 'webhook' as const, - endpoint: 'https://********', - username: 'svc', - password: '********', - }; - expect(targetType(target)).toBe('alertManager'); - const form = targetDraft(target); - expect(form.endpoint).toBe(''); - expect(form.password).toBe(''); + expect(targetType(maskedAlertmanager)).toBe('alertManager'); + expect(targetType({ ...maskedAlertmanager, username: null, password: null })).toBe( + 'alertManager', + ); + const form = targetDraft(maskedAlertmanager); + // Neither mask is prefilled, so it can never be resubmitted as a real credential. + expect([form.endpoint, form.username, form.password]).toEqual(['', '', '']); expect(validateTarget(form)).toHaveProperty('endpoint'); form.endpoint = 'https://example.com/api/v2/alerts'; + form.username = 'svc'; form.password = 'secret'; - expect(buildTargetPayload(form)).toMatchObject({ + expect(validateTarget(form)).toEqual({}); + expect(buildTargetPayload(form)).toEqual({ + name: 'AM', type: 'alertManager', + endpoint: 'https://example.com/api/v2/alerts', + skipTlsCheck: false, username: 'svc', password: 'secret', }); - expect(buildTargetPayload(form)).not.toHaveProperty('auth'); + }); + it('removes Alertmanager auth when both credentials are left empty on edit', () => { + const form = { + ...targetDraft(maskedAlertmanager), + endpoint: 'https://example.com/api/v2/alerts', + }; + expect(validateTarget(form)).toEqual({}); + const payload = buildTargetPayload(form); + expect(payload).not.toHaveProperty('username'); + expect(payload).not.toHaveProperty('password'); + expect(JSON.stringify(payload)).not.toContain('********'); + }); + it('requires the Alertmanager username and password as a pair and rejects masks', () => { + const form = { + ...targetDraft(maskedAlertmanager), + endpoint: 'https://example.com/api/v2/alerts', + }; + const pair = 'Enter both username and password, or leave both empty.'; + expect(validateTarget({ ...form, username: 'svc' })).toEqual({ password: pair }); + expect(validateTarget({ ...form, password: 'secret' })).toEqual({ password: pair }); + expect(validateTarget({ ...form, username: '********', password: 'secret' })).toEqual({ + username: 'Enter the real username, not the mask.', + }); + expect(validateTarget({ ...form, username: 'svc', password: '********' })).toEqual({ + password: 'Enter the real password, not the mask.', + }); + expect(validateTarget({ ...form, username: '********', password: '********' })).toEqual({ + username: 'Enter the real username, not the mask.', + password: 'Enter the real password, not the mask.', + }); }); it('omits TLS and auth fields for Slack', () => { const form = { @@ -487,6 +524,7 @@ describe('target forms', () => { for (const headers of [ [{ id: 'a', key: 'Cookie', value: 'a' }], [{ id: 'a', key: 'X-Key', value: '' }], + [{ id: 'a', key: 'X-Key', value: '********' }], [ { id: 'a', key: 'X-Key', value: 'a' }, { id: 'b', key: 'x-key', value: 'b' }, diff --git a/frontend/src/features/alerts/targetHelpers.ts b/frontend/src/features/alerts/targetHelpers.ts index 130c4febb..63f1e6ca7 100644 --- a/frontend/src/features/alerts/targetHelpers.ts +++ b/frontend/src/features/alerts/targetHelpers.ts @@ -1,6 +1,9 @@ import type { AlertTarget, AlertTargetRequest, AlertTargetType } from '../../lib/types'; import { createId } from '../../lib/ids'; +/** The placeholder the server returns in place of stored target secrets. */ +export const MASKED = '********'; + export function targetType(target: AlertTarget): AlertTargetType { return target.type === 'webhook' && !target.headers && @@ -25,7 +28,8 @@ export function targetDraft(target?: AlertTarget): TargetDraft { type: target ? targetType(target) : 'webhook', endpoint: '', skipTls: target?.skipTlsCheck ?? false, - username: target?.username ?? '', + // Credentials come back masked, so both start empty and must be re-entered together. + username: '', password: '', headers: Object.keys(target?.headers ?? {}).map((key) => ({ id: createId(), key, value: '' })), }; @@ -57,6 +61,8 @@ export function invalidHeader( fields, message: 'Each header needs a valid name and a non-empty value without line breaks.', }; + if (header.value === MASKED) + return { index, fields: ['value'], message: 'Enter the real header value, not the mask.' }; if (deniedHeaders.includes(name)) return { index, @@ -75,7 +81,7 @@ export function validateTarget(draft: TargetDraft): Partial { @@ -61,25 +61,68 @@ describe('demo alerts', () => { }); it('masks every target secret and recognizes the backend Alertmanager shape', async () => { const client = createDemoClient(); + const secrets = ['svc-user', 'secret-pass', 'secret-path', 'header-secret']; + const leaks = (value: unknown) => + secrets.filter((secret) => JSON.stringify(value).includes(secret)); const target = await client.createAlertTarget({ name: 'AM', type: 'alertManager', - endpoint: 'https://example.com/api/v2/alerts', - username: 'svc', - password: 'secret', + endpoint: 'https://example.com/secret-path/api/v2/alerts', + username: 'svc-user', + password: 'secret-pass', skipTlsCheck: false, }); + expect(target).toMatchObject({ + type: 'webhook', + endpoint: 'https://********', + username: '********', + password: '********', + }); expect(targetType(target)).toBe('alertManager'); - expect(target.password).toBe('********'); - expect(JSON.stringify(await client.listAlertTargets())).not.toContain('secret'); + const fetched = (await client.getAlertTarget(target.id)).target; + expect(fetched).toEqual(target); + expect(alertTarget(fetched)).toBe(true); + expect(leaks([target, fetched, await client.listAlertTargets()])).toEqual([]); + // The edit form never carries masks forward; the demo uses the same form validation. + const form = targetDraft(fetched); + expect([form.username, form.password]).toEqual(['', '']); + for (const credentials of [ + { username: '********', password: 'secret-pass' }, + { username: 'svc-user', password: '********' }, + ]) + await expect( + client.updateAlertTarget(target.id, { + name: 'AM', + type: 'alertManager', + endpoint: 'https://example.com/api/v2/alerts', + ...credentials, + skipTlsCheck: false, + }), + ).rejects.toMatchObject({ status: 400 }); + const unauthenticated = await client.updateAlertTarget(target.id, { + name: 'AM', + type: 'alertManager', + endpoint: 'https://example.com/api/v2/alerts', + skipTlsCheck: false, + }); + expect(unauthenticated).toMatchObject({ username: null, password: null }); + expect(targetType(unauthenticated)).toBe('alertManager'); const updated = await client.updateAlertTarget(target.id, { name: 'AM', type: 'webhook', - endpoint: 'https://example.com/new-secret', - headers: { 'X-Key': 'secret' }, + endpoint: 'https://example.com/secret-path', + headers: { 'X-Key': 'header-secret' }, skipTlsCheck: false, }); - expect(updated.headers).toEqual({ 'X-Key': '********' }); + expect(updated).toMatchObject({ + endpoint: 'https://********', + headers: { 'X-Key': '********' }, + }); + expect(updated).not.toHaveProperty('username'); + expect(targetType(updated)).toBe('webhook'); + expect( + leaks([updated, await client.getAlertTarget(target.id), await client.listAlertTargets()]), + ).toEqual([]); await client.deleteAlertTarget(target.id); await expect(client.getAlertTarget(target.id)).rejects.toMatchObject({ status: 400 }); }); diff --git a/frontend/src/lib/demoAlerts.ts b/frontend/src/lib/demoAlerts.ts index 5d0546ea4..ff72e3769 100644 --- a/frontend/src/lib/demoAlerts.ts +++ b/frontend/src/lib/demoAlerts.ts @@ -188,7 +188,8 @@ export function createDemoAlerts(now = Date.now()): Pick>, } +/// Extension fields holding runtime state whose diagnostics may carry target URLs. +const RUNTIME_FIELDS: [&str; 2] = ["promqlRuntime", "sqlRuntime"]; + +/// Sanitize runtime diagnostics in extension fields, leaving every other field untouched. +pub(super) fn sanitize_runtime_fields(fields: &mut serde_json::Map) { + for name in RUNTIME_FIELDS { + if let Some(runtime) = fields.get_mut(name) { + if name == "promqlRuntime" { + sanitize_promql_runtime_errors(runtime); + } else { + sanitize_runtime_errors(runtime); + } + } + } +} + +// PromQL labels are series identity, not diagnostics. Preserve only the label +// payloads at the known instance/delivery positions; a diagnostic container +// named `labels` elsewhere must still be scrubbed. +fn sanitize_promql_runtime_errors(value: &mut Value) { + let Value::Object(fields) = value else { + sanitize_runtime_errors(value); + return; + }; + for (key, value) in fields { + match (key.as_str(), value) { + ("instances", Value::Object(instances)) => { + instances + .values_mut() + .for_each(sanitize_promql_entry_errors); + } + ("deliveries", Value::Array(deliveries)) => { + deliveries.iter_mut().for_each(sanitize_promql_entry_errors); + } + (_, value) => sanitize_runtime_error_field(key, value), + } + } +} + +fn sanitize_promql_entry_errors(value: &mut Value) { + if let Value::Object(fields) = value { + for (key, value) in fields { + if key != "labels" { + sanitize_runtime_error_field(key, value); + } + } + } else { + sanitize_runtime_errors(value); + } +} + +fn sanitize_runtime_error_field(key: &str, value: &mut Value) { + match value { + Value::String(error) + if key == "error" && super::target::runtime_error_needs_redaction(error.as_str()) => + { + *error = super::target::safe_runtime_error(error.as_str()); + } + _ => sanitize_runtime_errors(value), + } +} + +/// Recursively sanitize string `error` fields, preserving the surrounding shape. +/// Containers are always descended into, including ones stored under `error`. +pub(super) fn sanitize_runtime_errors(value: &mut Value) { + match value { + Value::Object(fields) => { + for (key, value) in fields { + sanitize_runtime_error_field(key, value); + } + } + Value::Array(values) => values.iter_mut().for_each(sanitize_runtime_errors), + _ => {} + } +} + impl AlertConfig { + /// Sanitize stored runtime diagnostics in place. Used when configs enter memory; + /// the stored object keeps its contents until the next normal save. + pub fn sanitize_runtime_diagnostics(&mut self) { + if let Some(fields) = self.other_fields.as_mut() { + sanitize_runtime_fields(fields); + } + } + /// Filters out reserved field names from other_fields /// This prevents conflicts when flattening other_fields during serialization pub fn sanitize_other_fields(&mut self) { @@ -489,7 +574,8 @@ impl AlertConfig { } } - pub fn to_response(self) -> AlertConfigResponse { + pub fn to_response(mut self) -> AlertConfigResponse { + self.sanitize_runtime_diagnostics(); AlertConfigResponse { version: self.version, id: self.id, @@ -942,10 +1028,155 @@ impl MetastoreObject for MTTRHistory { } #[cfg(test)] -mod tests { - use super::AlertRequest; +pub(crate) mod tests { + use serde_json::{Map, Value, json}; + + use super::{AlertConfig, AlertRequest, sanitize_runtime_fields}; use crate::alerts::AlertError; + pub(crate) const REDACTED: &str = "Request failed (URL details redacted)"; + + /// Extension fields as stored by older builds, with raw delivery URLs in errors. + pub(crate) fn legacy_runtime_fields() -> Map { + let labels = json!({ + "host":"a", + "error":format!("https://metric.example/failure/{}", "é".repeat(1500)), + "runbook":"https://runbook.example/keep" + }); + let delivery = |error: &str| { + json!({"labels":labels.clone(),"value":2.0,"firing":true, + "target":"01ARZ3NDEKTSV4RRFFQ69G5FAV","attempts":3,"error":error}) + }; + let Value::Object(fields) = json!({ + "promqlConfig":{"holdDuration":"5m"}, + "executionIdentity":{"userId":"owner","tenantId":"DEFAULT_TENANT"}, + "promqlRuntime":{ + "health":"error", + "error":"error sending request for url (https://user:SECRET-PASS@am.example/api?token=SECRET-TOKEN#frag)", + "lastEvaluatedAt":"2026-01-01T00:00:00Z", + "instances":{ + "a":{"labels":labels,"state":"firing","pendingSince":null, + "lastSeen":"2026-01-01T00:00:00Z","value":2.0}, + "b":{"labels":{"host":"b","error":"é".repeat(3000)},"state":"resolved", + "pendingSince":null,"lastSeen":"2026-01-01T00:00:00Z","value":0.0} + }, + "diagnostics":{"error":{"labels":{"error":"https://hook.example/SECRET-DIAGNOSTIC-LABEL"}}}, + "deliveries":[ + delivery("HTTP status server error (500) for url (HTTP://hook.example/SECRET-PATH)"), + delivery("Notification delivery timed out"), + ] + }, + "sqlRuntime":{ + "runs":[ + {"error":"hTtPs://hook.example/SECRET-SQL"}, + {"nested":{"error":"é".repeat(3000)}}, + {"error":42} + ], + // Containers stored directly under `error` are still descended into. + "error":{"error":"https://user:SECRET-NESTED@host.example/token","code":7}, + "history":{"error":[ + {"error":"https://hook.example/SECRET-ARRAY"}, + [{"error":{"error":"http://hook.example/SECRET-DEEP"}}], + "plain array entry", + null + ]} + }, + "customNote":{"error":"https://docs.example/keep-this"} + }) else { + unreachable!() + }; + fields + } + + /// `legacy_runtime_fields` with only the runtime `error` strings sanitized. + pub(crate) fn sanitized_runtime_fields() -> Map { + let mut fields = legacy_runtime_fields(); + fields["promqlRuntime"]["error"] = REDACTED.into(); + fields["promqlRuntime"]["deliveries"][0]["error"] = REDACTED.into(); + fields["promqlRuntime"]["diagnostics"]["error"]["labels"]["error"] = REDACTED.into(); + fields["sqlRuntime"]["runs"][0]["error"] = REDACTED.into(); + fields["sqlRuntime"]["runs"][1]["nested"]["error"] = "é".repeat(1024).into(); + fields["sqlRuntime"]["error"]["error"] = REDACTED.into(); + fields["sqlRuntime"]["history"]["error"][0]["error"] = REDACTED.into(); + fields["sqlRuntime"]["history"]["error"][1][0]["error"]["error"] = REDACTED.into(); + fields + } + + pub(crate) fn legacy_alert_json(state: &str) -> Value { + let mut alert = json!({ + "version":"v2","id":ulid::Ulid::new(),"severity":"high","title":"Legacy", + "queryType":"promql","query":"up","datasets":["metrics"],"alertType":"threshold", + "thresholdConfig":{"operator":">","value":1.0}, + "evalConfig":{"rollingWindow":{"evalStart":"5m","evalEnd":"now","evalFrequency":1}}, + "targets":[],"state":state,"notificationState":"notify","notificationConfig":{"interval":1}, + "created":"2026-01-01T00:00:00Z","tags":["prod"] + }); + alert + .as_object_mut() + .unwrap() + .extend(legacy_runtime_fields()); + alert + } + + #[test] + fn runtime_sanitization_only_touches_runtime_error_strings() { + let mut fields = legacy_runtime_fields(); + sanitize_runtime_fields(&mut fields); + assert_eq!(fields, sanitized_runtime_fields()); + // Idempotent: a second pass leaves the result unchanged. + sanitize_runtime_fields(&mut fields); + assert_eq!(fields, sanitized_runtime_fields()); + } + + #[test] + fn detail_response_sanitizes_legacy_runtime_and_keeps_shape() { + let raw = legacy_alert_json("triggered"); + let config: AlertConfig = serde_json::from_value(raw.clone()).unwrap(); + let response = serde_json::to_value(config.to_response()).unwrap(); + + let text = response.to_string(); + for secret in [ + "SECRET-PASS", + "SECRET-TOKEN", + "SECRET-PATH", + "SECRET-SQL", + "SECRET-NESTED", + "SECRET-ARRAY", + "SECRET-DEEP", + "SECRET-DIAGNOSTIC-LABEL", + ] { + assert!(!text.contains(secret), "{secret} leaked in: {text}"); + } + for (key, value) in sanitized_runtime_fields() { + assert_eq!(response[&key], value, "{key}"); + } + for key in [ + "id", + "title", + "query", + "queryType", + "datasets", + "state", + "tags", + ] { + assert_eq!(response[key], raw[key], "{key}"); + } + } + + #[tokio::test] + async fn alert_request_sanitizes_client_runtime_extras() { + let mut request = alert_request("code", "SELECT COUNT(*) FROM \"logs-a\"", &[]); + let mut extras = legacy_runtime_fields(); + extras.remove("executionIdentity"); + request.other_fields = Some(extras); + + let config = request.into(None).await.unwrap(); + + let mut expected = sanitized_runtime_fields(); + expected.remove("executionIdentity"); + assert_eq!(config.other_fields, Some(expected)); + } + fn alert_request(query_type: &str, query: &str, datasets: &[&str]) -> AlertRequest { serde_json::from_value(serde_json::json!({ "severity": "high", @@ -1006,4 +1237,140 @@ mod tests { if message == "Alert should include only one dataset. Found: [\"metrics-a\", \"metrics-b\"]" )); } + + /// Ignored microbenchmark for the response path: the existing clone + serialize + /// of runtime extension fields versus clone + sanitize + serialize. It measures + /// per-response/runtime overhead only, not whole-server throughput. Run with + /// `cargo test --release --lib bench_runtime_sanitization -- --ignored --nocapture`. + #[test] + #[ignore = "microbenchmark; run explicitly with --ignored --nocapture"] + fn bench_runtime_sanitization_overhead() { + use std::hint::black_box; + use std::time::Instant; + + const WARMUP: usize = 500; + const ITERATIONS: usize = 2_000; + const ROUNDS: usize = 6; + + fn fixture(errors: &[String]) -> Map { + let instances: Map = (0..50) + .map(|i| { + let state = if i % 3 == 0 { "firing" } else { "resolved" }; + ( + format!("host={i}"), + json!({"labels":{"host":format!("host-{i}"),"region":"us-east-1"}, + "state":state,"pendingSince":null, + "lastSeen":"2026-01-01T00:00:00Z","value":i as f64}), + ) + }) + .collect(); + let deliveries: Vec = (0..20) + .map(|i| { + json!({"labels":{"host":format!("host-{i}")},"value":i as f64,"firing":true, + "target":"01ARZ3NDEKTSV4RRFFQ69G5FAV","attempts":1, + "error":errors[i % errors.len()]}) + }) + .collect(); + let Value::Object(fields) = json!({ + "promqlConfig":{"holdDuration":"5m"}, + "executionIdentity":{"userId":"owner","tenantId":"DEFAULT_TENANT"}, + "promqlRuntime":{"health":"error","error":errors[0], + "lastEvaluatedAt":"2026-01-01T00:00:00Z", + "instances":instances,"deliveries":deliveries} + }) else { + unreachable!() + }; + fields + } + + fn serialize(fields: &Map, sanitize: bool) -> Vec { + let mut fields = black_box(fields).clone(); + if sanitize { + sanitize_runtime_fields(&mut fields); + } + black_box(serde_json::to_vec(black_box(&fields)).unwrap()) + } + + /// Nanoseconds per operation for one round of `iterations` calls. + fn run(fields: &Map, sanitize: bool, iterations: usize) -> f64 { + let start = Instant::now(); + for _ in 0..iterations { + black_box(serialize(fields, sanitize)); + } + start.elapsed().as_nanos() as f64 / iterations as f64 + } + + fn median(mut samples: Vec) -> f64 { + samples.sort_by(f64::total_cmp); + let mid = samples.len() / 2; + if samples.len().is_multiple_of(2) { + (samples[mid - 1] + samples[mid]) / 2.0 + } else { + samples[mid] + } + } + + let clean = fixture(&[ + "Notification delivery timed out".into(), + "HTTP status server error (503 Service Unavailable)".into(), + "error sending request".into(), + ]); + let legacy = fixture(&[ + "error sending request for url (https://user:secret@am.example/api/v2/alerts?token=abc)" + .into(), + format!( + "HTTP status client error (401 Unauthorized) for url (http://hook.example/{}?key=secret)", + "p".repeat(4000) + ), + "x".repeat(5000), + ]); + + // Correctness checks run outside the timed section. + assert_eq!(serialize(&clean, false), serialize(&clean, true)); + let sanitized_legacy = String::from_utf8(serialize(&legacy, true)).unwrap(); + for needle in ["://", "secret", "token="] { + assert!( + !sanitized_legacy.contains(needle), + "{needle} survived sanitization" + ); + } + + println!( + "Response-path microbenchmark: clone+serialize vs clone+sanitize+serialize; \ + not whole-server throughput. Medians of {ROUNDS} rounds x {ITERATIONS} iterations." + ); + for (name, fields) in [ + ("clean typical", &clean), + ("legacy credential URL", &legacy), + ] { + run(fields, false, WARMUP); + run(fields, true, WARMUP); + let (mut baseline, mut sanitized) = (Vec::new(), Vec::new()); + for round in 0..ROUNDS { + // Alternate order so warm caches and frequency scaling favour neither side. + let order = if round % 2 == 0 { + [false, true] + } else { + [true, false] + }; + for sanitize in order { + let ns = run(fields, sanitize, ITERATIONS); + if sanitize { + sanitized.push(ns); + } else { + baseline.push(ns); + } + } + } + let baseline_ns = median(baseline); + let sanitized_ns = median(sanitized); + let baseline_bytes = serialize(fields, false).len(); + let sanitized_bytes = serialize(fields, true).len(); + println!( + "{name}: baseline {baseline_ns:.0} ns/op, sanitized {sanitized_ns:.0} ns/op, \ + ratio {:.3}, payload {baseline_bytes} -> {sanitized_bytes} bytes", + sanitized_ns / baseline_ns + ); + } + } } diff --git a/src/alerts/mod.rs b/src/alerts/mod.rs index 36323f6f0..0a93c1540 100644 --- a/src/alerts/mod.rs +++ b/src/alerts/mod.rs @@ -102,7 +102,18 @@ fn ensure_schedulable_in_oss(alert: &dyn AlertTrait) -> Result<(), AlertError> { Ok(()) } +/// Parse or migrate a stored alert, scrubbing historic runtime diagnostics in memory +/// before the alert is scheduled or inserted. Storage is not rewritten here. async fn parse_alert_config(alert_bytes: &[u8], tenant: &Option) -> Option { + let mut alert = parse_stored_alert_config(alert_bytes, tenant).await?; + alert.sanitize_runtime_diagnostics(); + Some(alert) +} + +async fn parse_stored_alert_config( + alert_bytes: &[u8], + tenant: &Option, +) -> Option { let json_value: JsonValue = match serde_json::from_slice(alert_bytes) { Ok(val) => val, Err(e) => { @@ -829,9 +840,9 @@ impl AlertConfig { ); if let Some(other_fields) = &self.other_fields { - for (key, value) in other_fields { - map.insert(key.clone(), value.clone()); - } + let mut other_fields = other_fields.clone(); + alert_structs::sanitize_runtime_fields(&mut other_fields); + map.extend(other_fields); } map @@ -1187,6 +1198,17 @@ pub fn get_number_of_agg_exprs_from_plan(plan: &LogicalPlan) -> Result &'static str { + self.0 + } +} + #[derive(Debug, thiserror::Error)] pub enum AlertError { #[error("Storage Error: {0}")] @@ -1197,8 +1219,12 @@ pub enum AlertError { Metadata(&'static str), #[error("User is not authorized to run this query")] Unauthorized, - #[error("Alert target outbound policy rejected request:{0}")] - OutboundPolicy(#[from] outbound_http_policy::OutboundPolicyError), + // Only the sanitized category is retained, including in Debug/error chains. + #[error( + "Alert target outbound policy rejected request: {}", + .0.sanitized_message() + )] + OutboundPolicy(SanitizedOutboundPolicyError), #[error("ActixError: {0}")] Error(#[from] actix_web::Error), #[error("DataFusion Error: {0}")] @@ -1239,6 +1265,12 @@ pub enum AlertError { MetastoreError(#[from] MetastoreError), } +impl From for AlertError { + fn from(error: outbound_http_policy::OutboundPolicyError) -> Self { + Self::OutboundPolicy(SanitizedOutboundPolicyError(error.sanitized_message())) + } +} + impl actix_web::ResponseError for AlertError { fn status_code(&self) -> StatusCode { match self { @@ -1847,4 +1879,96 @@ mod tests { Some("tenant-a".to_owned()) ); } + + #[test] + fn summary_sanitizes_legacy_runtime_and_keeps_extension_fields() { + use crate::alerts::alert_structs::tests::{legacy_alert_json, sanitized_runtime_fields}; + let raw = legacy_alert_json("triggered"); + let config: super::AlertConfig = serde_json::from_value(raw.clone()).unwrap(); + let summary = config.to_summary(); + for (key, value) in sanitized_runtime_fields() { + assert_eq!(summary[&key], value, "{key}"); + } + for key in ["title", "id", "state", "queryType", "datasets", "tags"] { + assert_eq!(summary[key], raw[key], "{key}"); + } + // The config itself is not modified by building a summary. + assert_eq!( + config.other_fields.unwrap()["promqlRuntime"], + raw["promqlRuntime"] + ); + } + + #[tokio::test] + async fn startup_parse_sanitizes_historic_runtime_errors_including_disabled() { + use crate::alerts::alert_structs::tests::{legacy_alert_json, sanitized_runtime_fields}; + for state in ["disabled", "triggered", "not-triggered"] { + let raw = legacy_alert_json(state); + let bytes = serde_json::to_vec(&raw).unwrap(); + let config = super::parse_alert_config(&bytes, &None).await.unwrap(); + assert_eq!(serde_json::to_value(config.state).unwrap(), state); + assert_eq!( + config.other_fields, + Some(sanitized_runtime_fields()), + "{state}" + ); + assert_eq!(config.title, "Legacy"); + assert_eq!(config.tags, Some(vec!["prod".to_string()])); + } + } + + #[test] + fn outbound_policy_errors_stay_sanitized_in_actix_logs_and_responses() { + use super::outbound_http_policy::OutboundPolicyError; + use actix_web::ResponseError; + use std::error::Error; + + let errors = [ + OutboundPolicyError::DeniedDomain("secret-host.internal".into()), + OutboundPolicyError::PrivateAddressNotAllowed("10.1.2.3".parse().unwrap()), + OutboundPolicyError::UnsupportedScheme("secret-scheme".into()), + OutboundPolicyError::ResolveFailed { + host: "secret-host.internal".into(), + source: std::io::Error::other("lookup failed for secret-host.internal"), + }, + ]; + for error in errors { + let expected = error.sanitized_message(); + let error = super::AlertError::from(error); + assert_eq!( + error.to_string(), + format!("Alert target outbound policy rejected request: {expected}") + ); + assert!(error.source().is_none()); + + let response = error.error_response(); + assert_eq!(response.status(), actix_web::http::StatusCode::BAD_REQUEST); + let body = actix_web::body::to_bytes(response.into_body()); + let body = futures::executor::block_on(body).unwrap(); + let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!( + body["error"], + "Alert target blocked by outbound security policy" + ); + assert_eq!(body["message"], error.to_string()); + for secret in ["secret-host", "10.1.2.3", "secret-scheme"] { + assert!(!body.to_string().contains(secret), "{secret} leaked"); + } + + // Actix stores this error on the response and its Logger formats + // it with Debug, rather than the sanitized Display implementation. + let response = actix_web::HttpResponse::from_error(error); + assert_eq!(response.status(), actix_web::http::StatusCode::BAD_REQUEST); + let stored_error = response.error().unwrap(); + for formatted in [ + format!("Error in response: {stored_error:?}"), + format!("{stored_error:#?}"), + ] { + assert!(formatted.contains(expected), "{formatted}"); + for secret in ["secret-host", "10.1.2.3", "secret-scheme"] { + assert!(!formatted.contains(secret), "{secret} leaked: {formatted}"); + } + } + } + } } diff --git a/src/alerts/promql_alerts.rs b/src/alerts/promql_alerts.rs index b449b2647..f3326499b 100644 --- a/src/alerts/promql_alerts.rs +++ b/src/alerts/promql_alerts.rs @@ -2,8 +2,10 @@ use crate::{ alerts::{ AlertConfig, AlertError, AlertQueryType, AlertState, NotificationState, - alert_structs::DeploymentInfo, alert_types::ThresholdAlert, - alerts_utils::evaluate_condition, target::TARGETS, + alert_structs::DeploymentInfo, + alert_types::ThresholdAlert, + alerts_utils::evaluate_condition, + target::{TARGETS, safe_runtime_error}, }, handlers::http::promql::AlertQueryError, parseable::{DEFAULT_TENANT, PARSEABLE}, @@ -177,6 +179,16 @@ pub(super) fn authorize_execution(config: &AlertConfig) -> Result<(), String> { } impl Runtime { + /// Scrub diagnostics restored from storage, which may predate URL redaction. + pub fn sanitize_errors(&mut self) { + let errors = self + .deliveries + .iter_mut() + .map(|delivery| &mut delivery.error); + for error in std::iter::once(&mut self.error).chain(errors).flatten() { + *error = safe_runtime_error(error); + } + } pub fn discontinuity(&mut self, health: &str, error: Option, now: DateTime) { self.health = health.into(); self.error = error; @@ -469,6 +481,7 @@ pub async fn evaluate(id: Ulid, tenant_id: &Option) -> Result<(), AlertE .map(|v| serde_json::from_value(v.clone())) .transpose()? .unwrap_or_default(); + runtime.sanitize_errors(); let authorized = authorize_execution(&config); let may_deliver = authorized.is_ok(); let result = match authorized { @@ -485,7 +498,7 @@ pub async fn evaluate(id: Ulid, tenant_id: &Option) -> Result<(), AlertE let result = match result { Ok(samples) => runtime.observe_transactionally(samples, &config, now), Err(AlertQueryError::Busy(error)) => { - runtime.interrupted(error, &config, now); + runtime.interrupted(safe_runtime_error(&error), &config, now); Ok(vec![]) } Err(AlertQueryError::Failed(error)) => Err(error), @@ -493,7 +506,7 @@ pub async fn evaluate(id: Ulid, tenant_id: &Option) -> Result<(), AlertE let transitions = match result { Ok(transitions) => transitions, Err(error) => { - runtime.discontinuity("error", Some(error), now); + runtime.discontinuity("error", Some(safe_runtime_error(&error)), now); vec![] } }; @@ -575,7 +588,7 @@ pub async fn evaluate(id: Ulid, tenant_id: &Option) -> Result<(), AlertE let target_result = TARGETS .get_target_by_id(&delivery.target, &updated.tenant_id) .await - .map_err(|error| error.to_string()); + .map_err(|error| safe_runtime_error(&error.to_string())); let result = match target_result { Ok(target) => match tokio::time::timeout_at( deadline.min(tokio::time::Instant::now() + std::time::Duration::from_secs(5)), @@ -593,7 +606,7 @@ pub async fn evaluate(id: Ulid, tenant_id: &Option) -> Result<(), AlertE delivery.attempts = u32::MAX; delivery.error = None; } - Err(error) => delivery.error = Some(error), + Err(error) => delivery.error = Some(safe_runtime_error(&error)), } } // Alertmanager expects active alerts to be re-sent, so every @@ -636,8 +649,9 @@ pub async fn evaluate(id: Ulid, tenant_id: &Option) -> Result<(), AlertE .unwrap_or_else(|_| Err("Notification delivery timed out".into())); if let Err(error) = result { tracing::warn!( - "PromQL alert {} could not re-send firing alerts to Alertmanager target {target_id}: {error}", - updated.id + "PromQL alert {} could not re-send firing alerts to Alertmanager target {target_id}: {}", + updated.id, + safe_runtime_error(&error) ); } } @@ -1183,4 +1197,118 @@ mod tests { now.to_rfc3339_opts(chrono::SecondsFormat::Millis, true) ); } + + #[test] + fn restored_runtime_preserves_error_labels_and_prunes_stale_firing_delivery() { + let mut config = config(); + config.other_fields.as_mut().unwrap().insert( + "promqlConfig".into(), + serde_json::json!({"holdDuration":"0s"}), + ); + config.targets = vec![Ulid::new()]; + let labels = BTreeMap::from([ + ("host".into(), "a".into()), + ( + "error".into(), + format!("https://metric.example/failure/{}", "é".repeat(1500)), + ), + ]); + let key = serde_json::to_string(&labels).unwrap(); + let now = Utc::now(); + let mut runtime = Runtime::default(); + let firing = runtime + .observe_transactionally(vec![(labels.clone(), 9.0)], &config, now) + .unwrap(); + assert_eq!(firing, vec![(labels.clone(), 9.0, true)]); + runtime.enqueue(&firing, &config.targets); + runtime.deliveries[0].error = + Some("request failed for https://hook.example/TARGET-SECRET".into()); + + // Exercise the JSON boundary used at startup and for API responses, + // then restore the typed outbox as the next evaluation would. + config.other_fields.as_mut().unwrap().insert( + "promqlRuntime".into(), + serde_json::to_value(runtime).unwrap(), + ); + config.sanitize_runtime_diagnostics(); + let mut restored: Runtime = + serde_json::from_value(config.other_fields.as_ref().unwrap()["promqlRuntime"].clone()) + .unwrap(); + restored.sanitize_errors(); + assert_eq!(restored.instances[&key].labels, labels); + assert_eq!(restored.deliveries[0].labels, labels); + assert_eq!( + serde_json::to_string(&restored.instances[&key].labels).unwrap(), + key + ); + assert!( + !restored.deliveries[0] + .error + .as_ref() + .unwrap() + .contains("TARGET-SECRET") + ); + + let recovery = restored + .observe_transactionally( + vec![(labels.clone(), 0.0)], + &config, + now + chrono::Duration::minutes(1), + ) + .unwrap(); + assert_eq!(recovery, vec![(labels.clone(), 0.0, false)]); + assert!(restored.deliveries.is_empty()); + restored.enqueue(&recovery, &config.targets); + assert_eq!(restored.deliveries.len(), 1); + assert_eq!(restored.deliveries[0].labels, labels); + assert!(!restored.deliveries[0].firing); + } + + #[test] + fn restored_runtime_errors_are_sanitized_without_changing_state() { + let target = ulid::Ulid::new(); + let delivery = |error: Option<&str>| Delivery { + labels: BTreeMap::from([("host".into(), "a".into())]), + value: 9.0, + firing: true, + target, + attempts: 2, + error: error.map(Into::into), + }; + let stored = serde_json::json!({ + "health":"error", + "error":"error sending request for url (https://am.example/api?token=SECRET)", + "lastEvaluatedAt":Utc::now(), + "instances":{"a":{"labels":{"host":"a"},"state":"firing","pendingSince":null, + "lastSeen":Utc::now(),"value":9.0}}, + "deliveries":[ + delivery(Some("HTTP status server error (500) for url (HTTP://u:p@hook.example/SECRET)")), + delivery(Some("Notification delivery timed out")), + delivery(None), + ] + }); + let mut runtime: Runtime = serde_json::from_value(stored).unwrap(); + runtime.sanitize_errors(); + assert_eq!( + runtime.error.as_deref(), + Some("Request failed (URL details redacted)") + ); + let errors: Vec<_> = runtime + .deliveries + .iter() + .map(|delivery| delivery.error.as_deref()) + .collect(); + assert_eq!( + errors, + [ + Some("Request failed (URL details redacted)"), + Some("Notification delivery timed out"), + None + ] + ); + assert_eq!(runtime.health, "error"); + assert_eq!(runtime.instances["a"].state, "firing"); + assert!(runtime.deliveries.iter().all(|d| d.attempts == 2)); + assert!(!serde_json::to_string(&runtime).unwrap().contains("SECRET")); + } } diff --git a/src/alerts/target.rs b/src/alerts/target.rs index 3a42af234..3cc13286d 100644 --- a/src/alerts/target.rs +++ b/src/alerts/target.rs @@ -45,6 +45,41 @@ use crate::{ storage::object_storage::target_json_path, }; +const MAX_RUNTIME_ERROR_CHARS: usize = 1024; +const REDACTED_URL_ERROR: &str = "Request failed (URL details redacted)"; +const INVALID_TARGET_CONFIGURATION: &str = "invalid alert target configuration"; + +/// Bound diagnostics and remove URL-bearing details before storage or API output. +/// Historic reqwest messages can include credentials in paths, query strings or userinfo. +pub fn safe_runtime_error(error: &str) -> String { + if contains_http_url(error) { + return REDACTED_URL_ERROR.into(); + } + error.chars().take(MAX_RUNTIME_ERROR_CHARS).collect() +} + +/// Whether `safe_runtime_error` would change `error`, so clean diagnostics are not reallocated. +pub(super) fn runtime_error_needs_redaction(error: &str) -> bool { + contains_http_url(error) + || (error.len() > MAX_RUNTIME_ERROR_CHARS + && error.chars().nth(MAX_RUNTIME_ERROR_CHARS).is_some()) +} + +// Case-insensitive `http://` or `https://` marker, without allocating a lowercase copy. +fn contains_http_url(error: &str) -> bool { + let bytes = error.as_bytes(); + error.match_indices("://").any(|(index, _)| { + let ends_with = |scheme: &[u8]| { + index >= scheme.len() && bytes[index - scheme.len()..index].eq_ignore_ascii_case(scheme) + }; + ends_with(b"http") || ends_with(b"https") + }) +} + +fn safe_delivery_error(error: reqwest::Error) -> String { + safe_runtime_error(&error.without_url().to_string()) +} + pub static TARGETS: Lazy = Lazy::new(|| TargetConfigs { target_configs: RwLock::new(HashMap::new()), }); @@ -163,9 +198,10 @@ impl Default for Retry { } } -#[derive(Debug, serde::Serialize, serde::Deserialize, Clone)] +/// Serialized in full so stored targets keep their credentials. Deserialization goes +/// through `TargetVerifier`, see the `Deserialize` impl below. +#[derive(Debug, serde::Serialize, Clone)] #[serde(rename_all = "camelCase")] -#[serde(try_from = "TargetVerifier")] pub struct Target { pub name: String, #[serde(flatten)] @@ -246,14 +282,13 @@ impl Target { } TargetType::AlertManager(alert_manager) => { let endpoint = format!("{}://********", alert_manager.endpoint.scheme()); - if let Some(auth) = alert_manager.auth { - let password = "********"; + if alert_manager.auth.is_some() { json!({ "name":self.name, "type":"webhook", "endpoint":endpoint, - "username":auth.username, - "password":password, + "username":"********", + "password":"********", "skipTlsCheck":alert_manager.skip_tls_check, "id":self.id }) @@ -298,7 +333,7 @@ impl Target { None, ) .await - .map_err(|e| e.to_string())?; + .map_err(|e| e.sanitized_message().to_string())?; prepared .client .post(target.endpoint.clone()) @@ -313,7 +348,7 @@ impl Target { Some(&target.headers), ) .await - .map_err(|e| e.to_string())?; + .map_err(|e| e.sanitized_message().to_string())?; prepared .client .post(target.endpoint.clone()) @@ -325,9 +360,9 @@ impl Target { request .send() .await - .map_err(|e| e.to_string())? + .map_err(safe_delivery_error)? .error_for_status() - .map_err(|e| e.to_string())?; + .map_err(safe_delivery_error)?; Ok(()) } @@ -523,6 +558,20 @@ impl TryFrom for Target { } } +// Serde errors echo rejected values (for example an endpoint string or a header value), +// which may be secrets. Report one generic error to HTTP, tool and stored-target callers. +impl<'de> serde::Deserialize<'de> for Target { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + let verifier = ::deserialize(deserializer) + .map_err(|_| ::custom(INVALID_TARGET_CONFIGURATION))?; + Target::try_from(verifier) + .map_err(|_| ::custom(INVALID_TARGET_CONFIGURATION)) + } +} + #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "camelCase")] #[serde(tag = "type")] @@ -577,7 +626,10 @@ impl CallableTarget for SlackWebHook { { Ok(prepared) => prepared, Err(err) => { - error!("Slack alert target rejected by outbound policy:{err}"); + error!( + "Slack alert target rejected by outbound policy:{}", + err.sanitized_message() + ); return; } }; @@ -588,7 +640,10 @@ impl CallableTarget for SlackWebHook { .send() .await { - error!("Couldn't make call to webhook, error:{}", e) + error!( + "Couldn't make call to webhook, error:{}", + safe_delivery_error(e) + ) } } } @@ -621,7 +676,10 @@ impl CallableTarget for OtherWebHook { { Ok(prepared) => prepared, Err(err) => { - error!("Webhook alert target rejected by outbound policy:{err}"); + error!( + "Webhook alert target rejected by outbound policy:{}", + err.sanitized_message() + ); return; } }; @@ -630,7 +688,10 @@ impl CallableTarget for OtherWebHook { .post(self.endpoint.clone()) .headers(prepared.headers); if let Err(e) = request.body(alert).send().await { - error!("Couldn't make call to webhook, error:{}", e) + error!( + "Couldn't make call to webhook, error:{}", + safe_delivery_error(e) + ) } } } @@ -660,7 +721,7 @@ impl AlertManager { None, ) .await - .map_err(|e| e.to_string())?; + .map_err(|e| e.sanitized_message().to_string())?; let mut request = prepared.client.post(self.endpoint.clone()).json(alerts); if let Some(auth) = &self.auth { if !prepared.authorization_allowed { @@ -671,9 +732,9 @@ impl AlertManager { request .send() .await - .map_err(|e| e.to_string())? + .map_err(safe_delivery_error)? .error_for_status() - .map_err(|e| e.to_string())?; + .map_err(safe_delivery_error)?; Ok(()) } } @@ -722,7 +783,10 @@ impl CallableTarget for AlertManager { { Ok(prepared) => prepared, Err(err) => { - error!("Alertmanager target rejected by outbound policy:{err}"); + error!( + "Alertmanager target rejected by outbound policy:{}", + err.sanitized_message() + ); return; } }; @@ -732,8 +796,7 @@ impl CallableTarget for AlertManager { // Credentials are only safe when the destination itself is explicitly allowlisted. if !prepared.authorization_allowed { error!( - "Alertmanager credentials blocked by outbound policy for destination:{}", - self.endpoint + "Alertmanager credentials blocked by outbound policy: destination is not explicitly allowlisted" ); return; } @@ -742,7 +805,10 @@ impl CallableTarget for AlertManager { request = request.header(AUTHORIZATION, basic_auth_value); } if let Err(e) = request.send().await { - error!("Couldn't make call to alertmanager, error:{}", e) + error!( + "Couldn't make call to alertmanager, error:{}", + safe_delivery_error(e) + ) } } } @@ -781,7 +847,556 @@ pub struct Auth { #[cfg(test)] mod tests { + use std::{ + io::{Read, Write}, + net::{TcpListener, TcpStream}, + thread::JoinHandle, + }; + use super::*; + use crate::alerts::{ + AlertInfo, DeploymentInfo, NotificationState, + outbound_http_policy::{ALERT_TARGET_POLICY, AlertTargetPolicyConfig}, + }; + + const SECRETS: [&str; 7] = [ + "SECRET-USERINFO", + "SECRET-PATH-TOKEN", + "SECRET-QUERY-TOKEN", + "SECRET-FRAGMENT", + "SECRET-HEADER-VALUE", + "SECRET-AM-USER", + "SECRET-AM-PASSWORD", + ]; + + fn secret_url(host: &str) -> Url { + format!( + "http://url-user:SECRET-USERINFO@{host}/hooks/SECRET-PATH-TOKEN?token=SECRET-QUERY-TOKEN#SECRET-FRAGMENT" + ) + .parse() + .unwrap() + } + + fn named(target: TargetType, tenant: &str) -> Target { + Target { + name: "redaction-target".into(), + target, + id: Ulid::new(), + tenant: Some(tenant.to_owned()), + } + } + + fn webhook(endpoint: Url) -> TargetType { + TargetType::Other(OtherWebHook { + endpoint, + headers: HashMap::from([("X-Api-Key".into(), "SECRET-HEADER-VALUE".into())]), + skip_tls_check: false, + }) + } + + fn alertmanager(endpoint: Url, with_auth: bool) -> TargetType { + TargetType::AlertManager(AlertManager { + endpoint, + skip_tls_check: false, + auth: with_auth.then(|| Auth { + username: "SECRET-AM-USER".into(), + password: "SECRET-AM-PASSWORD".into(), + }), + }) + } + + fn context() -> Context { + Context::new( + AlertInfo::new( + Ulid::new(), + "Redaction alert".into(), + AlertState::Triggered, + NotificationState::Notify, + "high".into(), + ), + DeploymentInfo::new("instance".into(), Ulid::new(), "standalone".into()), + NotificationConfig::default(), + "alert message".into(), + ) + } + + fn unique_tenant() -> String { + format!("target-redaction-{}", Ulid::new()) + } + + /// Allow loopback delivery for one tenant only, as an operator allowlist would. + async fn allow_loopback(tenant: &str) { + ALERT_TARGET_POLICY.write().await.insert( + tenant.to_owned(), + AlertTargetPolicyConfig { + allow_private: true, + allowed_cidrs: vec!["127.0.0.1/32".into()], + ..Default::default() + }, + ); + } + + async fn remove_policy(tenant: &str) { + ALERT_TARGET_POLICY.write().await.remove(tenant); + } + + fn assert_redacted(text: &str, port: Option) { + for secret in SECRETS { + assert!(!text.contains(secret), "{secret} leaked in: {text}"); + } + assert!(!text.contains("127.0.0.1"), "host leaked in: {text}"); + assert!(!text.contains("203.0.113.10"), "host leaked in: {text}"); + assert!(!text.contains("url-user"), "userinfo leaked in: {text}"); + if let Some(port) = port { + assert!( + !text.contains(&format!(":{port}")), + "port leaked in: {text}" + ); + } + } + + /// Serve one request with `status` from a local loopback listener and + /// return the raw request so callers can check what was delivered. + fn serve_once(status: &'static str) -> (u16, JoinHandle) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let handle = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let request = read_request(&mut stream); + write!( + stream, + "HTTP/1.1 {status}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + ) + .unwrap(); + request + }); + (port, handle) + } + + fn read_request(stream: &mut TcpStream) -> String { + let mut data = Vec::new(); + let mut buf = [0u8; 4096]; + loop { + let read = stream.read(&mut buf).unwrap_or(0); + if read == 0 { + break; + } + data.extend_from_slice(&buf[..read]); + let Some(end) = data.windows(4).position(|w| w == b"\r\n\r\n") else { + continue; + }; + let head = String::from_utf8_lossy(&data[..end]).to_ascii_lowercase(); + let body_len = head + .lines() + .find_map(|line| line.strip_prefix("content-length:")) + .and_then(|len| len.trim().parse::().ok()) + .unwrap_or(0); + if data.len() >= end + 4 + body_len { + break; + } + } + String::from_utf8_lossy(&data).into_owned() + } + + /// A loopback port with nothing listening, so connecting fails locally. + fn refused_port() -> u16 { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.local_addr().unwrap().port() + } + + #[derive(Clone, Default)] + struct CapturedLogs(Arc>>); + + impl Write for CapturedLogs { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + /// Run `future` with a thread-local subscriber that records this crate's + /// alert logs, so the actual formatted log lines can be checked. + async fn capture_alert_logs(future: impl std::future::Future) -> String { + let logs = CapturedLogs::default(); + let writer = logs.clone(); + let subscriber = tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::new( + "parseable::alerts=trace", + )) + .with_writer(move || writer.clone()) + .with_ansi(false) + .finish(); + let _guard = tracing::subscriber::set_default(subscriber); + future.await; + String::from_utf8(logs.0.lock().unwrap().clone()).unwrap() + } + + #[test] + fn diagnostics_redact_url_secrets_and_bound_non_url_errors() { + for error in [ + "request failed for https://user:secret@host/path", + "HTTP://host/private-token", + "hTtPs://host/path?token=secret#fragment", + "error sending request for url (http://10.0.0.1:9093/api/v2/alerts)", + "échec: https://hôte.example/路径?jeton=秘密", + "first http://a.example/one then HTTPS://b.example/two", + "prefixhttps://glued.example/token", + ] { + assert_eq!(safe_runtime_error(error), REDACTED_URL_ERROR, "{error}"); + } + // A URL beyond the length cap is still redacted rather than truncated away. + let late_url = format!("{}https://host/late-token", "x".repeat(3000)); + assert_eq!(safe_runtime_error(&late_url), REDACTED_URL_ERROR); + + for clean in [ + "Notification delivery timed out", + "HTTP status server error (500 Internal Server Error)", + "error sending request", + "ftp://not-an-http-target", + "Alert owner's dataset permission has been revoked", + ] { + assert_eq!(safe_runtime_error(clean), clean); + } + + assert_eq!(safe_runtime_error(&"x".repeat(5000)).len(), 1024); + let unicode = safe_runtime_error(&"é".repeat(5000)); + assert_eq!(unicode.chars().count(), 1024); + assert_eq!(unicode.len(), 2048); + assert_eq!(safe_runtime_error(&"x".repeat(1024)).len(), 1024); + } + + #[test] + fn diagnostics_redaction_is_idempotent_and_detects_needed_changes() { + let long = "é".repeat(1500); + let at_cap = "x".repeat(1024); + let multibyte_at_cap = "é".repeat(1024); + for error in [ + "https://user:secret@host/path", + "Notification delivery timed out", + long.as_str(), + at_cap.as_str(), + multibyte_at_cap.as_str(), + "", + ] { + let once = safe_runtime_error(error); + assert_eq!(safe_runtime_error(&once), once); + assert!(!runtime_error_needs_redaction(&once)); + assert_eq!( + runtime_error_needs_redaction(error), + once != error, + "{error}" + ); + } + } + + #[test] + fn masked_targets_hide_urls_headers_and_credentials() { + let tenant = unique_tenant(); + let slack = named( + TargetType::Slack(SlackWebHook { + endpoint: "https://hooks.slack.com/services/SECRET-PATH-TOKEN" + .parse() + .unwrap(), + }), + &tenant, + ); + let id = slack.id; + assert_eq!( + slack.mask(), + json!({"name":"redaction-target","type":"slack","endpoint":"https://********","id":id}) + ); + + let other = named(webhook(secret_url("127.0.0.1:8080")), &tenant); + let id = other.id; + assert_eq!( + other.mask(), + json!({ + "name":"redaction-target","type":"webhook","endpoint":"http://********", + "headers":{"X-Api-Key":"********"},"skipTlsCheck":false,"id":id + }) + ); + + let with_auth = named(alertmanager(secret_url("127.0.0.1:9093"), true), &tenant); + let id = with_auth.id; + assert_eq!( + with_auth.mask(), + json!({ + "name":"redaction-target","type":"webhook","endpoint":"http://********", + "username":"********","password":"********","skipTlsCheck":false,"id":id + }) + ); + + let without_auth = named(alertmanager(secret_url("127.0.0.1:9093"), false), &tenant); + let id = without_auth.id; + assert_eq!( + without_auth.mask(), + json!({ + "name":"redaction-target","type":"webhook","endpoint":"http://********", + "username":null,"password":null,"skipTlsCheck":false,"id":id + }) + ); + } + + #[test] + fn storage_serialization_keeps_credentials_and_round_trips() { + let tenant = unique_tenant(); + for target in [ + named( + TargetType::Slack(SlackWebHook { + endpoint: "https://hooks.slack.com/services/SECRET-PATH-TOKEN" + .parse() + .unwrap(), + }), + &tenant, + ), + named(webhook(secret_url("127.0.0.1:8080")), &tenant), + named(alertmanager(secret_url("127.0.0.1:9093"), true), &tenant), + named(alertmanager(secret_url("127.0.0.1:9093"), false), &tenant), + ] { + let stored = serde_json::to_string(&target).unwrap(); + assert!(stored.contains("SECRET-PATH-TOKEN")); + if matches!(target.target, TargetType::Other(_)) { + assert!(stored.contains("SECRET-HEADER-VALUE")); + } + if target.alertmanager().is_some_and(|am| am.auth.is_some()) { + assert!(stored.contains("SECRET-AM-USER")); + assert!(stored.contains("SECRET-AM-PASSWORD")); + } + let restored: Target = serde_json::from_str(&stored).unwrap(); + assert_eq!(restored.name, target.name); + assert_eq!(restored.id, target.id); + assert_eq!(restored.target, target.target); + } + } + + #[test] + fn valid_target_requests_still_deserialize_through_the_verifier() { + let id = Ulid::new(); + let target: Target = serde_json::from_value(json!({ + "name":"am","type":"alertManager","endpoint":"https://am.example/api/v2/alerts", + "username":"user","password":"pass","skipTlsCheck":true, + "notificationConfig":{"interval":5,"times":2},"id":id,"tenantId":"tenant-a" + })) + .unwrap(); + assert_eq!(target.id, id); + assert_eq!(target.tenant.as_deref(), Some("tenant-a")); + let am = target.alertmanager().unwrap(); + assert!(am.skip_tls_check); + assert_eq!( + am.auth, + Some(Auth { + username: "user".into(), + password: "pass".into() + }) + ); + + let generated: Target = serde_json::from_value(json!({ + "name":"hook","type":"webhook","endpoint":"https://hook.example/x" + })) + .unwrap(); + assert!(matches!(generated.target, TargetType::Other(ref hook) if hook.headers.is_empty())); + } + + #[test] + fn invalid_secret_bearing_targets_report_generic_errors() { + let cases = [ + json!({"name":"t","type":"webhook","endpoint":"not a url SECRET-ENDPOINT-VALUE"}), + json!({"name":"t","type":"webhook","endpoint":"https://h.example","headers":{"X-Api-Key":424242}}), + json!({"name":"t","type":"webhook","endpoint":"https://h.example","skipTlsCheck":"SECRET-BOOL-VALUE"}), + json!({"name":"t","type":"alertManager","endpoint":"https://h.example","username":"SECRET-AUTH-USER","password":"SECRET-AUTH-PASSWORD","notificationConfig":{"interval":"SECRET-AUTH-INTERVAL"}}), + json!({"name":"t","type":"SECRET-TYPE-VALUE","endpoint":"https://h.example"}), + json!({"name":"t","type":"slack","endpoint":"https://hooks.slack.com/x","id":"SECRET-ID-VALUE"}), + ]; + for case in cases { + for error in [ + serde_json::from_value::(case.clone()) + .unwrap_err() + .to_string(), + serde_json::from_str::(&case.to_string()) + .unwrap_err() + .to_string(), + ] { + assert!(error.starts_with(INVALID_TARGET_CONFIGURATION), "{error}"); + for secret in [ + "SECRET-ENDPOINT-VALUE", + "424242", + "SECRET-BOOL-VALUE", + "SECRET-AUTH-USER", + "SECRET-AUTH-PASSWORD", + "SECRET-AUTH-INTERVAL", + "SECRET-TYPE-VALUE", + "SECRET-ID-VALUE", + ] { + assert!(!error.contains(secret), "{secret} leaked in: {error}"); + } + } + } + } + + #[tokio::test] + async fn webhook_delivery_status_and_transport_errors_hide_urls() { + let tenant = unique_tenant(); + allow_loopback(&tenant).await; + + let (port, server) = serve_once("500 Internal Server Error"); + let target = named(webhook(secret_url(&format!("127.0.0.1:{port}"))), &tenant); + let status = target + .deliver_promql(&target.tenant, "message", &[]) + .await + .unwrap_err(); + let request = server.join().unwrap().to_ascii_lowercase(); + assert!(request.contains("x-api-key: secret-header-value")); + assert!(status.contains("500"), "{status}"); + assert_redacted(&status, Some(port)); + + let port = refused_port(); + let target = named(webhook(secret_url(&format!("127.0.0.1:{port}"))), &tenant); + let transport = target + .deliver_promql(&target.tenant, "message", &[]) + .await + .unwrap_err(); + assert!(transport.contains("error sending request"), "{transport}"); + assert_redacted(&transport, Some(port)); + + remove_policy(&tenant).await; + } + + #[tokio::test] + async fn alertmanager_delivery_status_and_transport_errors_hide_urls() { + let tenant = unique_tenant(); + allow_loopback(&tenant).await; + let alerts = [json!({"labels":{"alertname":"x"}})]; + + let (port, server) = serve_once("503 Service Unavailable"); + let target = named( + alertmanager(secret_url(&format!("127.0.0.1:{port}")), true), + &tenant, + ); + let status = target + .deliver_promql(&target.tenant, "message", &alerts) + .await + .unwrap_err(); + let request = server.join().unwrap().to_ascii_lowercase(); + let credentials = + base64::prelude::BASE64_STANDARD.encode("SECRET-AM-USER:SECRET-AM-PASSWORD"); + assert!( + request.contains(&format!("authorization: basic {credentials}").to_ascii_lowercase()) + ); + assert!(status.contains("503"), "{status}"); + assert_redacted(&status, Some(port)); + + let port = refused_port(); + let target = named( + alertmanager(secret_url(&format!("127.0.0.1:{port}")), true), + &tenant, + ); + let transport = target + .alertmanager() + .unwrap() + .post_promql(&target.tenant, &alerts) + .await + .unwrap_err(); + assert!(transport.contains("error sending request"), "{transport}"); + assert_redacted(&transport, Some(port)); + + remove_policy(&tenant).await; + } + + #[tokio::test] + async fn delivery_policy_errors_use_sanitized_messages() { + let tenant = unique_tenant(); + ALERT_TARGET_POLICY.write().await.insert( + tenant.clone(), + AlertTargetPolicyConfig { + denied_domains: vec!["secret-denied.example".into()], + ..Default::default() + }, + ); + let denied = named(webhook(secret_url("secret-denied.example")), &tenant); + let error = denied + .deliver_promql(&denied.tenant, "message", &[]) + .await + .unwrap_err(); + assert_eq!(error, "Target domain is denied by outbound policy"); + + // Slack hosts are pinned before any DNS lookup, so this makes no request. + let slack = named( + TargetType::Slack(SlackWebHook { + endpoint: "https://secret-slack-host.example/services/SECRET-PATH-TOKEN" + .parse() + .unwrap(), + }), + &tenant, + ); + let error = slack + .deliver_promql(&slack.tenant, "message", &[]) + .await + .unwrap_err(); + assert_eq!(error, "Slack target host is invalid"); + + // A public, non-allowlisted address may receive alerts but not credentials. + let blocked = named(alertmanager(secret_url("203.0.113.10"), true), &tenant); + let error = blocked + .deliver_promql(&blocked.tenant, "message", &[]) + .await + .unwrap_err(); + assert_eq!(error, "Target credentials blocked by outbound policy"); + + remove_policy(&tenant).await; + } + + #[tokio::test] + async fn legacy_delivery_logs_hide_urls_and_credentials() { + let tenant = unique_tenant(); + allow_loopback(&tenant).await; + let tenant_id = Some(tenant.clone()); + let webhook_port = refused_port(); + let alertmanager_port = refused_port(); + + let logs = capture_alert_logs(async { + webhook(secret_url(&format!("127.0.0.1:{webhook_port}"))) + .call(&tenant_id, &context()) + .await; + alertmanager(secret_url(&format!("127.0.0.1:{alertmanager_port}")), true) + .call(&tenant_id, &context()) + .await; + // No request is sent: credentials are blocked before delivery. + alertmanager(secret_url("203.0.113.10"), true) + .call(&tenant_id, &context()) + .await; + TargetType::Slack(SlackWebHook { + endpoint: "https://secret-slack-host.example/services/SECRET-PATH-TOKEN" + .parse() + .unwrap(), + }) + .call(&tenant_id, &context()) + .await; + }) + .await; + + assert!(logs.contains("Couldn't make call to webhook"), "{logs}"); + assert!( + logs.contains("Couldn't make call to alertmanager"), + "{logs}" + ); + assert!( + logs.contains("Alertmanager credentials blocked by outbound policy"), + "{logs}" + ); + assert!(logs.contains("Slack target host is invalid"), "{logs}"); + assert!(!logs.contains("secret-slack-host"), "{logs}"); + assert_redacted(&logs, Some(webhook_port)); + assert_redacted(&logs, Some(alertmanager_port)); + + remove_policy(&tenant).await; + } #[test] fn test_masked_target_hides_secrets() { diff --git a/src/handlers/http/targets.rs b/src/handlers/http/targets.rs index ae9d312ea..db48c07fc 100644 --- a/src/handlers/http/targets.rs +++ b/src/handlers/http/targets.rs @@ -224,4 +224,128 @@ mod tests { ); assert!(!response.to_string().contains(denied_domain)); } + + #[actix_web::test] + async fn post_rejects_malformed_targets_without_echoing_values() { + let app = test::init_service(App::new().route("/targets", web::post().to(post))).await; + for body in [ + json!({"name":"t","type":"webhook","endpoint":"SECRET-ENDPOINT https://u:p@h/x"}), + json!({"name":"t","type":"webhook","endpoint":"https://h.example","headers":{"X-Api-Key":31337}}), + json!({"name":"t","type":"alertManager","endpoint":"https://h.example","username":"SECRET-AUTH-USER","password":"SECRET-AUTH-PASSWORD","notificationConfig":{"interval":"SECRET-AUTH-INTERVAL"}}), + ] { + let request = test::TestRequest::post() + .uri("/targets") + .insert_header(( + "x-p-tenant", + format!("targets-handler-test-{}", Ulid::new()), + )) + .set_json(&body) + .to_request(); + let response = test::call_service(&app, request).await; + assert_eq!(response.status(), actix_web::http::StatusCode::BAD_REQUEST); + let body = String::from_utf8(test::read_body(response).await.to_vec()).unwrap(); + assert!( + body.contains("invalid alert target configuration"), + "{body}" + ); + for secret in [ + "SECRET-ENDPOINT", + "u:p@h", + "31337", + "SECRET-AUTH-USER", + "SECRET-AUTH-PASSWORD", + "SECRET-AUTH-INTERVAL", + ] { + assert!(!body.contains(secret), "{secret} leaked in: {body}"); + } + } + } + + #[actix_web::test] + async fn post_policy_rejection_hides_target_host() { + let tenant = format!("targets-handler-test-{}", Ulid::new()); + let denied_domain = "secret-denied.internal.example"; + ALERT_TARGET_POLICY.write().await.insert( + tenant.clone(), + AlertTargetPolicyConfig { + denied_domains: vec![denied_domain.to_string()], + ..Default::default() + }, + ); + + let app = test::init_service(App::new().route("/targets", web::post().to(post))).await; + let request = test::TestRequest::post() + .uri("/targets") + .insert_header(("x-p-tenant", tenant.clone())) + .set_json(json!({ + "name": "denied", + "type": "webhook", + "endpoint": format!("https://{denied_domain}/hook/SECRET-PATH?token=SECRET-TOKEN"), + "headers": {"X-Api-Key": "SECRET-HEADER"} + })) + .to_request(); + let response = test::call_service(&app, request).await; + let status = response.status(); + let body: Value = test::read_body_json(response).await; + + ALERT_TARGET_POLICY.write().await.remove(&tenant); + + assert_eq!(status, actix_web::http::StatusCode::BAD_REQUEST); + assert_eq!( + body["message"], + "Alert target outbound policy rejected request: Target domain is denied by outbound policy" + ); + let text = body.to_string(); + for secret in [ + denied_domain, + "SECRET-PATH", + "SECRET-TOKEN", + "SECRET-HEADER", + ] { + assert!(!text.contains(secret), "{secret} leaked in: {text}"); + } + assert!(TARGETS.list(&Some(tenant)).await.unwrap().is_empty()); + } + + #[actix_web::test] + async fn get_masks_alertmanager_username_and_password() { + let tenant = format!("targets-handler-test-{}", Ulid::new()); + let target: Target = serde_json::from_value(json!({ + "name": "am", + "type": "alertManager", + "endpoint": "https://203.0.113.10/api/SECRET-PATH", + "username": "SECRET-USER", + "password": "SECRET-PASSWORD", + "tenantId": tenant + })) + .unwrap(); + let id = target.id; + TARGETS + .target_configs + .write() + .await + .entry(tenant.clone()) + .or_default() + .insert(id, target); + + // An IP literal avoids DNS; the public, non-allowlisted address blocks credentials. + let response = get_internal(id, &Some(tenant.clone())).await.unwrap(); + TARGETS.target_configs.write().await.remove(&tenant); + + assert_eq!(response["target"]["username"], "********"); + assert_eq!(response["target"]["password"], "********"); + assert_eq!(response["target"]["endpoint"], "https://********"); + assert_eq!(response["target"]["type"], "webhook"); + assert_eq!(response["enabled"], false); + assert_eq!(response["error"], "Target contains a denied header"); + let text = response.to_string(); + for secret in [ + "SECRET-USER", + "SECRET-PASSWORD", + "SECRET-PATH", + "203.0.113.10", + ] { + assert!(!text.contains(secret), "{secret} leaked in: {text}"); + } + } }