diff --git a/devops/docker/compose/.env.example b/devops/docker/compose/.env.example index f2cf2b858..08b8e5515 100644 --- a/devops/docker/compose/.env.example +++ b/devops/docker/compose/.env.example @@ -38,6 +38,9 @@ TROGON_GATEWAY_LOCAL_GITHUB_WEBHOOK_SECRET=local-dev-secret # --- Sentry Source --- # SENTRY_PRIMARY_CLIENT_SECRET= +# --- Datadog Source --- +# DATADOG_PRIMARY_WEBHOOK_TOKEN= + # --- Discord Source --- # DISCORD_BOT_TOKEN= diff --git a/devops/docker/compose/services/trogon-gateway/README.md b/devops/docker/compose/services/trogon-gateway/README.md index 6aafa19a9..1e12a7d79 100644 --- a/devops/docker/compose/services/trogon-gateway/README.md +++ b/devops/docker/compose/services/trogon-gateway/README.md @@ -19,6 +19,7 @@ named TOML integrations, each with its own literal or env-backed secret. | Microsoft Graph change notifications | `/sources/microsoft-graph/{integration}/webhook` | `client_state` | | Notion | `/sources/notion/{integration}/webhook` | `verification_token` | | Sentry | `/sources/sentry/{integration}/webhook` | `client_secret` | +| Datadog | `/sources/datadog/{integration}/webhook` | `webhook_token` | The gateway port is configured via `TROGON_GATEWAY_PORT` (default `8080`). Liveness and readiness probes are available at `GET /-/liveness` and `GET /-/readiness`. diff --git a/devops/docker/compose/services/trogon-gateway/gateway.toml b/devops/docker/compose/services/trogon-gateway/gateway.toml index 06b10e561..3956d80a5 100644 --- a/devops/docker/compose/services/trogon-gateway/gateway.toml +++ b/devops/docker/compose/services/trogon-gateway/gateway.toml @@ -81,6 +81,17 @@ webhook_secret = { env = "TROGON_GATEWAY_LOCAL_GITHUB_WEBHOOK_SECRET" } # [sources.sentry.integrations.primary.webhook] # client_secret = { env = "SENTRY_PRIMARY_CLIENT_SECRET" } # +# [sources.datadog.integrations.primary] +# subject_prefix = "datadog-primary" +# stream_name = "DATADOG_PRIMARY" +# stream_max_age_secs = 604800 +# nats_ack_timeout_secs = 10 +# +# [sources.datadog.integrations.primary.webhook] +# webhook_token = { env = "DATADOG_PRIMARY_WEBHOOK_TOKEN" } +# webhook_token_header = "X-Datadog-Webhook-Token" # optional; override the custom header name +# timestamp_tolerance_secs = 300 # optional anti-replay window; requires "timestamp": "$DATE_POSIX" in the payload +# # [sources.discord] # bot_token = { env = "DISCORD_BOT_TOKEN" } # gateway_intents = "guilds,guild_members,guild_messages,guild_message_reactions,direct_messages,message_content,guild_voice_states" diff --git a/otel/semconv/registry/gateway.yaml b/otel/semconv/registry/gateway.yaml index 7952425b7..3e8cdf943 100644 --- a/otel/semconv/registry/gateway.yaml +++ b/otel/semconv/registry/gateway.yaml @@ -21,7 +21,7 @@ groups: - id: event_id type: string stability: development - brief: Webhook event identifier (Slack, Notion). + brief: Webhook event identifier (Slack, Notion, Datadog). - id: event type: string stability: development @@ -157,3 +157,12 @@ groups: attributes: - ref: notification_count - ref: subject + - id: span.datadog.webhook + type: span + span_kind: server + stability: development + brief: Datadog webhook ingestion. + attributes: + - ref: event_type + - ref: event_id + - ref: subject diff --git a/rsworkspace/crates/trogon-gateway/README.md b/rsworkspace/crates/trogon-gateway/README.md index bfd84830c..508db1ade 100644 --- a/rsworkspace/crates/trogon-gateway/README.md +++ b/rsworkspace/crates/trogon-gateway/README.md @@ -44,6 +44,7 @@ configured as a named integration and mounted at its integration path. | Microsoft Graph change notifications | `/sources/microsoft-graph/{integration}/webhook` | | Notion | `/sources/notion/{integration}/webhook` | | Sentry | `/sources/sentry/{integration}/webhook` | +| Datadog | `/sources/datadog/{integration}/webhook` | Integration IDs must contain only ASCII letters, numbers, `_`, or `-`. They cannot be empty, contain path separators, or exceed 64 characters. @@ -70,6 +71,7 @@ or explicit environment reference shape. | Microsoft Graph change notifications | `client_state` | | Notion | `verification_token` | | Sentry | `client_secret` | +| Datadog | `webhook_token` | ## Core configuration @@ -103,6 +105,8 @@ Source-specific extras: - Telegram `webhook_registration_mode = "startup"` attempts registration on startup and requires `bot_token` plus `public_webhook_url` - Linear `timestamp_tolerance_secs` (default: `60`, `0` disables tolerance) - Twitter/X `consumer_secret` is used for both CRC responses and `x-twitter-webhooks-signature` validation +- Datadog webhooks are not signed. `webhook_token` is a shared secret the operator sends as a custom header (configured in the Datadog webhook's *Custom Headers*), verified in constant time. The header name defaults to `X-Datadog-Webhook-Token` and can be overridden per integration with `webhook_token_header`. Routing keys off an `event_type` field in the payload template, falling back to `.unroutable` when it is absent or invalid; an optional `id` field is used as the NATS message ID for deduplication +- Datadog `timestamp_tolerance_secs` (optional, disabled by default) enables a freshness/anti-replay check. Datadog sends no timestamp header, so the operator must template a `timestamp` field carrying `$DATE_POSIX` (POSIX seconds) into the payload. When enabled, requests whose `timestamp` is outside the window are rejected (`401`), and requests missing/with an unparseable `timestamp` are rejected (`400`). `0` or unset disables the check ## Config file shape @@ -163,8 +167,34 @@ verification_token = "notion-verification-token-example" [sources.sentry.integrations.primary.webhook] client_secret = "sentry-client-secret" + +[sources.datadog.integrations.primary.webhook] +webhook_token = { env = "DATADOG_PRIMARY_WEBHOOK_TOKEN" } +# webhook_token_header = "X-Datadog-Webhook-Token" # optional; override the custom header name +# timestamp_tolerance_secs = 300 # optional anti-replay window; requires "timestamp": "$DATE_POSIX" in the payload +``` + +Configure the matching Datadog webhook (Integrations > Webhooks) with a custom header and a payload template, for example: + +```json +{ + "X-Datadog-Webhook-Token": "$WEBHOOK_TOKEN" +} ``` +```json +{ + "id": "$ID", + "event_type": "$EVENT_TYPE", + "timestamp": "$DATE_POSIX", + "title": "$EVENT_TITLE", + "body": "$EVENT_MSG", + "alert_transition": "$ALERT_TRANSITION" +} +``` + +`timestamp` is only required when `timestamp_tolerance_secs` is set; `$DATE_POSIX` is POSIX epoch seconds (`$DATE` is epoch milliseconds, which this source does not accept). + ## Microsoft Graph change notifications This source receives Microsoft Graph change notifications. It does not implement diff --git a/rsworkspace/crates/trogon-gateway/src/config.rs b/rsworkspace/crates/trogon-gateway/src/config.rs index f00694862..0db0e3c42 100644 --- a/rsworkspace/crates/trogon-gateway/src/config.rs +++ b/rsworkspace/crates/trogon-gateway/src/config.rs @@ -1,6 +1,7 @@ use std::collections::BTreeMap; use std::path::Path; +use crate::source::datadog::DatadogWebhookToken; use crate::source::discord::config::DiscordBotToken; use crate::source::github::config::GitHubWebhookSecret; use crate::source::gitlab::GitLabSigningToken; @@ -19,6 +20,7 @@ use crate::source::telegram::config::{ }; use crate::source::twitter::config::TwitterConsumerSecret; use crate::source_integration_id::{SourceIntegrationId, SourceIntegrationIdError}; +use axum::http::HeaderName; use confique::Config; #[cfg(test)] use trogon_nats::NatsAuth; @@ -350,6 +352,8 @@ struct SourcesConfig { notion: NotionConfig, #[config(nested)] sentry: SentryConfig, + #[config(nested)] + datadog: DatadogConfig, } #[derive(Config)] @@ -448,6 +452,14 @@ struct SentryConfig { integrations: BTreeMap>, } +#[derive(Config)] +#[config(layer_attr(serde(deny_unknown_fields)))] +struct DatadogConfig { + status: Option, + #[config(default = {})] + integrations: BTreeMap>, +} + #[derive(serde::Deserialize)] #[serde(deny_unknown_fields)] struct SourceIntegrationInput { @@ -544,6 +556,14 @@ struct SentryWebhookConfig { client_secret: Option, } +#[derive(serde::Deserialize)] +#[serde(deny_unknown_fields)] +struct DatadogWebhookConfig { + webhook_token: Option, + webhook_token_header: Option, + timestamp_tolerance_secs: Option, +} + pub struct ResolvedHttpServerConfig { pub port: u16, } @@ -573,6 +593,7 @@ pub struct ResolvedConfig { pub microsoft_graph: Vec>, pub notion: Vec>, pub sentry: Vec>, + pub datadog: Vec>, } impl ResolvedConfig { @@ -588,6 +609,7 @@ impl ResolvedConfig { || !self.microsoft_graph.is_empty() || !self.notion.is_empty() || !self.sentry.is_empty() + || !self.datadog.is_empty() } } @@ -620,6 +642,7 @@ fn resolve(cfg: GatewayConfig, nats_overrides: &NatsArgs) -> Result Result, +) -> Vec> { + let DatadogConfig { + status, + integrations: configured_integrations, + } = section; + let mut integrations = Vec::new(); + if !resolve_source_status("datadog", status.as_deref(), errors) { + return integrations; + } + for (raw_id, integration) in configured_integrations { + let Some(id) = resolve_integration_id("datadog", raw_id, errors) else { + continue; + }; + if !resolve_integration_source_status("datadog", &id, integration.status.as_deref(), errors) { + continue; + } + let Some(webhook) = integration.webhook else { + errors.push(ConfigValidationError::missing_integration("datadog", &id, "webhook")); + continue; + }; + let Some(token) = + require_integration_value("datadog", &id, "webhook_token", webhook.webhook_token, env, errors) + else { + continue; + }; + let webhook_token = match DatadogWebhookToken::new(token) { + Ok(token) => token, + Err(error) => { + errors.push(ConfigValidationError::invalid_integration( + "datadog", + &id, + "webhook_token", + error, + )); + continue; + } + }; + let Some((subject_prefix, stream_name, stream_max_age, nats_ack_timeout)) = resolve_common_integration_fields( + CommonIntegrationFieldsInput { + source: "datadog", + id: &id, + subject_source_prefix: "datadog", + stream_source_prefix: "DATADOG", + subject_prefix: integration.subject_prefix, + stream_name: integration.stream_name, + stream_max_age_secs: integration.stream_max_age_secs, + nats_ack_timeout_secs: integration.nats_ack_timeout_secs, + default_nats_ack_timeout_secs: DEFAULT_NATS_ACK_TIMEOUT_SECS, + }, + errors, + ) else { + continue; + }; + let webhook_token_header = match webhook.webhook_token_header { + Some(raw) => match HeaderName::try_from(raw.trim()) { + Ok(header) => header, + Err(error) => { + errors.push(ConfigValidationError::invalid_integration( + "datadog", + &id, + "webhook_token_header", + error, + )); + continue; + } + }, + None => HeaderName::from_static(crate::source::datadog::constants::DEFAULT_HEADER_WEBHOOK_TOKEN), + }; + let timestamp_tolerance = webhook + .timestamp_tolerance_secs + .and_then(|secs| NonZeroDuration::from_secs(secs).ok()); + integrations.push(SourceIntegration::new( + id, + crate::source::datadog::DatadogConfig { + webhook_token, + webhook_token_header, + subject_prefix, + stream_name, + stream_max_age, + nats_ack_timeout, + timestamp_tolerance, + }, + )); + } + integrations +} + fn resolve_integration_id( source: &'static str, raw_id: String, diff --git a/rsworkspace/crates/trogon-gateway/src/config/tests.rs b/rsworkspace/crates/trogon-gateway/src/config/tests.rs index 9ab95011f..b7c6cd72a 100644 --- a/rsworkspace/crates/trogon-gateway/src/config/tests.rs +++ b/rsworkspace/crates/trogon-gateway/src/config/tests.rs @@ -128,6 +128,15 @@ client_secret = "{secret}" ) } +fn datadog_toml(token: &str) -> String { + format!( + r#" +[sources.datadog.integrations.primary.webhook] +webhook_token = "{token}" +"# + ) +} + fn incidentio_valid_test_secret() -> String { ["whsec_", "dGVzdC1zZWNyZXQ="].concat() } @@ -879,6 +888,179 @@ client_secret = "sentry-client-secret" assert!(cfg.sentry.is_empty()); } +#[test] +fn datadog_resolves_with_valid_token() { + let f = write_toml(&datadog_toml("datadog-webhook-token")); + let cfg = load(Some(f.path())).expect("load failed"); + assert!(!cfg.datadog.is_empty()); + assert!(cfg.has_any_source()); +} + +#[test] +fn datadog_disabled_is_empty() { + let toml = r#" +[sources.datadog] +status = "disabled" + +[sources.datadog.integrations.primary.webhook] +webhook_token = "datadog-webhook-token" +"#; + let f = write_toml(toml); + let cfg = load(Some(f.path())).expect("load failed"); + assert!(cfg.datadog.is_empty()); +} + +#[test] +fn datadog_enabled_without_webhook_token_is_invalid() { + let toml = r#" +[sources.datadog.integrations.primary] +status = "enabled" + +[sources.datadog.integrations.primary.webhook] + +"#; + let f = write_toml(toml); + let result = load(Some(f.path())); + assert!( + matches!(result, Err(ConfigError::Validation(ref errs)) if errs.iter().any(|e| e.contains("datadog/primary: missing webhook_token"))) + ); +} + +#[test] +fn datadog_empty_webhook_token_is_invalid() { + let f = write_toml(&datadog_toml("")); + let result = load(Some(f.path())); + assert!( + matches!(result, Err(ConfigError::Validation(ref errs)) if errs.iter().any(|e| e.contains("datadog/primary: invalid webhook_token"))) + ); +} + +#[test] +fn datadog_enabled_integration_without_webhook_block_is_invalid() { + let toml = r#" +[sources.datadog.integrations.primary] +subject_prefix = "datadog-primary" +"#; + let f = write_toml(toml); + let result = load(Some(f.path())); + assert!( + matches!(result, Err(ConfigError::Validation(ref errs)) if errs.iter().any(|e| e.contains("datadog/primary: missing webhook"))) + ); +} + +#[test] +fn datadog_invalid_integration_id_is_invalid() { + let toml = r#" +[sources.datadog.integrations."bad/id".webhook] +webhook_token = "datadog-webhook-token" +"#; + let f = write_toml(toml); + let result = load(Some(f.path())); + assert!( + matches!(result, Err(ConfigError::Validation(ref errs)) if errs.iter().any(|e| e.contains("datadog/bad/id: invalid integration id"))) + ); +} + +#[test] +fn datadog_integration_disabled_is_skipped() { + let toml = r#" +[sources.datadog.integrations.primary] +status = "disabled" + +[sources.datadog.integrations.primary.webhook] +webhook_token = "datadog-webhook-token" +"#; + let f = write_toml(toml); + let cfg = load(Some(f.path())).expect("load failed"); + assert!(cfg.datadog.is_empty()); +} + +#[test] +fn datadog_zero_stream_max_age_is_error() { + let toml = r#" +[sources.datadog.integrations.primary] +stream_max_age_secs = 0 + +[sources.datadog.integrations.primary.webhook] +webhook_token = "datadog-webhook-token" +"#; + let f = write_toml(toml); + let result = load(Some(f.path())); + assert!( + matches!(result, Err(ConfigError::Validation(ref errs)) if errs.iter().any(|e| e.contains("datadog/primary: stream_max_age_secs must not be zero"))) + ); +} + +#[test] +fn datadog_timestamp_tolerance_defaults_to_disabled() { + let f = write_toml(&datadog_toml("datadog-webhook-token")); + let cfg = load(Some(f.path())).expect("load failed"); + assert!(cfg.datadog[0].config.timestamp_tolerance.is_none()); +} + +#[test] +fn datadog_timestamp_tolerance_resolves_when_set() { + let toml = r#" +[sources.datadog.integrations.primary.webhook] +webhook_token = "datadog-webhook-token" +timestamp_tolerance_secs = 300 +"#; + let f = write_toml(toml); + let cfg = load(Some(f.path())).expect("load failed"); + assert_eq!( + cfg.datadog[0].config.timestamp_tolerance.map(std::time::Duration::from), + Some(std::time::Duration::from_secs(300)), + ); +} + +#[test] +fn datadog_zero_timestamp_tolerance_disables_check() { + let toml = r#" +[sources.datadog.integrations.primary.webhook] +webhook_token = "datadog-webhook-token" +timestamp_tolerance_secs = 0 +"#; + let f = write_toml(toml); + let cfg = load(Some(f.path())).expect("load failed"); + assert!(cfg.datadog[0].config.timestamp_tolerance.is_none()); +} + +#[test] +fn datadog_webhook_token_header_defaults() { + let f = write_toml(&datadog_toml("datadog-webhook-token")); + let cfg = load(Some(f.path())).expect("load failed"); + assert_eq!( + cfg.datadog[0].config.webhook_token_header.as_str(), + "x-datadog-webhook-token", + ); +} + +#[test] +fn datadog_custom_webhook_token_header_resolves() { + let toml = r#" +[sources.datadog.integrations.primary.webhook] +webhook_token = "datadog-webhook-token" +webhook_token_header = "X-Acme-Token" +"#; + let f = write_toml(toml); + let cfg = load(Some(f.path())).expect("load failed"); + assert_eq!(cfg.datadog[0].config.webhook_token_header.as_str(), "x-acme-token"); +} + +#[test] +fn datadog_invalid_webhook_token_header_is_invalid() { + let toml = r#" +[sources.datadog.integrations.primary.webhook] +webhook_token = "datadog-webhook-token" +webhook_token_header = "bad header" +"#; + let f = write_toml(toml); + let result = load(Some(f.path())); + assert!( + matches!(result, Err(ConfigError::Validation(ref errs)) if errs.iter().any(|e| e.contains("datadog/primary: invalid webhook_token_header"))) + ); +} + #[test] fn sentry_missing_client_secret_returns_none_when_status_unspecified() { let toml = r#" diff --git a/rsworkspace/crates/trogon-gateway/src/http/tests.rs b/rsworkspace/crates/trogon-gateway/src/http/tests.rs index a364d365f..d739a867f 100644 --- a/rsworkspace/crates/trogon-gateway/src/http/tests.rs +++ b/rsworkspace/crates/trogon-gateway/src/http/tests.rs @@ -63,6 +63,9 @@ verification_token = "notion-verification-token-example" [sources.sentry.integrations.primary.webhook] client_secret = "sentry-client-secret" + +[sources.datadog.integrations.primary.webhook] +webhook_token = "datadog-webhook-token" "# .to_string() } diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/config.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/config.rs new file mode 100644 index 000000000..358bcdce5 --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/config.rs @@ -0,0 +1,20 @@ +use super::datadog_webhook_token::DatadogWebhookToken; +use axum::http::HeaderName; +use trogon_nats::NatsToken; +use trogon_nats::jetstream::StreamMaxAge; +use trogon_std::NonZeroDuration; + +pub struct DatadogConfig { + pub webhook_token: DatadogWebhookToken, + /// HTTP header the shared secret is delivered in. Defaults to + /// `X-Datadog-Webhook-Token`; operators may override the name. + pub webhook_token_header: HeaderName, + pub subject_prefix: NatsToken, + pub stream_name: NatsToken, + pub stream_max_age: StreamMaxAge, + pub nats_ack_timeout: NonZeroDuration, + /// Optional freshness window for the payload `timestamp` field. `None` + /// disables the check. Datadog does not send a timestamp header, so the + /// operator must template `$DATE_POSIX` (POSIX seconds) into the payload. + pub timestamp_tolerance: Option, +} diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/constants.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/constants.rs new file mode 100644 index 000000000..7798f4674 --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/constants.rs @@ -0,0 +1,13 @@ +use trogon_std::{ByteSize, HttpBodySizeMax}; + +pub const HTTP_BODY_SIZE_MAX: HttpBodySizeMax = HttpBodySizeMax::new(ByteSize::mib(2)).unwrap(); + +/// Default HTTP header carrying the shared secret token the operator configures +/// as a custom header on the Datadog webhook. Datadog does not sign webhook +/// requests, so this header is the only authentication signal. Operators may +/// override the header name via `webhook_token_header`. +pub const DEFAULT_HEADER_WEBHOOK_TOKEN: &str = "x-datadog-webhook-token"; + +pub const NATS_HEADER_EVENT_TYPE: &str = "X-Datadog-Event-Type"; +pub const NATS_HEADER_EVENT_ID: &str = "X-Datadog-Event-Id"; +pub const NATS_HEADER_REJECT_REASON: &str = "X-Datadog-Reject-Reason"; diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_event_type.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_event_type.rs new file mode 100644 index 000000000..b4614538e --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_event_type.rs @@ -0,0 +1,56 @@ +use std::fmt; + +use trogon_nats::DottedNatsToken; +use trogon_nats::SubjectTokenViolation; + +#[derive(Debug, Clone, PartialEq, thiserror::Error)] +pub enum DatadogEventTypeError { + #[error("event_type must not be empty")] + Empty, + #[error("event_type contains invalid character: {0:?}")] + InvalidCharacter(char), + #[error("event_type is too long: {0} bytes (max 128)")] + TooLong(usize), +} + +impl From for DatadogEventTypeError { + fn from(violation: SubjectTokenViolation) -> Self { + match violation { + SubjectTokenViolation::Empty => Self::Empty, + SubjectTokenViolation::InvalidCharacter(ch) => Self::InvalidCharacter(ch), + SubjectTokenViolation::TooLong(len) => Self::TooLong(len), + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct DatadogEventType(DottedNatsToken); + +impl DatadogEventType { + pub fn new(value: impl AsRef) -> Result { + DottedNatsToken::new(value) + .map(Self) + .map_err(DatadogEventTypeError::from) + } + + pub fn as_str(&self) -> &str { + self.0.as_str() + } +} + +impl fmt::Display for DatadogEventType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +impl std::ops::Deref for DatadogEventType { + type Target = str; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +#[cfg(test)] +mod tests; diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_event_type/tests.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_event_type/tests.rs new file mode 100644 index 000000000..95fcfab5e --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_event_type/tests.rs @@ -0,0 +1,68 @@ +use super::*; + +#[test] +fn empty_error_display_is_specific() { + let err = DatadogEventType::new("").unwrap_err(); + assert_eq!(err.to_string(), "event_type must not be empty"); +} + +#[test] +fn invalid_character_error_display_is_specific() { + let err = DatadogEventType::new("metric alert").unwrap_err(); + assert_eq!(err.to_string(), "event_type contains invalid character: ' '"); +} + +#[test] +fn too_long_error_display_is_specific() { + let long_event_type = "a".repeat(129); + let err = DatadogEventType::new(&long_event_type).unwrap_err(); + assert_eq!(err.to_string(), "event_type is too long: 129 bytes (max 128)"); +} + +#[test] +fn accepts_datadog_event_types() { + let event_type = DatadogEventType::new("metric_alert_monitor").unwrap(); + assert_eq!(event_type.as_str(), "metric_alert_monitor"); +} + +#[test] +fn accepts_dotted_event_types() { + let event_type = DatadogEventType::new("monitor.triggered").unwrap(); + assert_eq!(event_type.as_str(), "monitor.triggered"); +} + +#[test] +fn rejects_empty() { + assert!(matches!(DatadogEventType::new(""), Err(DatadogEventTypeError::Empty))); +} + +#[test] +fn rejects_wildcards() { + assert!(DatadogEventType::new("monitor.*").is_err()); + assert!(DatadogEventType::new("monitor.>").is_err()); +} + +#[test] +fn rejects_whitespace() { + assert!(DatadogEventType::new("metric alert").is_err()); +} + +#[test] +fn rejects_malformed_dots() { + assert!(DatadogEventType::new(".monitor").is_err()); + assert!(DatadogEventType::new("monitor.").is_err()); + assert!(DatadogEventType::new("monitor..triggered").is_err()); +} + +#[test] +fn display_roundtrips() { + let event_type = DatadogEventType::new("event_alert").unwrap(); + assert_eq!(event_type.to_string(), "event_alert"); +} + +#[test] +fn deref_roundtrips_to_str() { + let event_type = DatadogEventType::new("event_alert").unwrap(); + let event_type_str: &str = &event_type; + assert_eq!(event_type_str, "event_alert"); +} diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_webhook_token.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_webhook_token.rs new file mode 100644 index 000000000..381a256b4 --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_webhook_token.rs @@ -0,0 +1,25 @@ +use std::fmt; + +use trogon_std::{EmptySecret, SecretString}; + +#[derive(Clone)] +pub struct DatadogWebhookToken(SecretString); + +impl DatadogWebhookToken { + pub fn new(value: impl AsRef) -> Result { + SecretString::new(value).map(Self) + } + + pub fn as_str(&self) -> &str { + self.0.as_str() + } +} + +impl fmt::Debug for DatadogWebhookToken { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("DatadogWebhookToken(****)") + } +} + +#[cfg(test)] +mod tests; diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_webhook_token/tests.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_webhook_token/tests.rs new file mode 100644 index 000000000..e9e27dcd1 --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_webhook_token/tests.rs @@ -0,0 +1,18 @@ +use super::*; + +#[test] +fn datadog_webhook_token_roundtrips() { + let token = DatadogWebhookToken::new("super-secret").unwrap(); + assert_eq!(token.as_str(), "super-secret"); +} + +#[test] +fn datadog_webhook_token_debug_redacts() { + let token = DatadogWebhookToken::new("super-secret").unwrap(); + assert_eq!(format!("{token:?}"), "DatadogWebhookToken(****)"); +} + +#[test] +fn datadog_webhook_token_rejects_empty() { + assert!(DatadogWebhookToken::new("").is_err()); +} diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/mod.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/mod.rs new file mode 100644 index 000000000..f2bbfd79b --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/mod.rs @@ -0,0 +1,19 @@ +//! Datadog webhook receiver that publishes verified events to NATS JetStream. +//! +//! Datadog webhooks are not signed. Requests are authenticated with a shared +//! secret token the operator configures as a custom header on the Datadog +//! webhook, and the payload is an operator-defined JSON template. Routing keys +//! off an `event_type` field in the body, falling back to `.unroutable` when it +//! is absent or invalid. + +pub mod config; +pub mod constants; +pub mod datadog_event_type; +pub mod datadog_webhook_token; +pub mod server; +pub mod signature; + +pub use config::DatadogConfig; +pub use datadog_event_type::DatadogEventType; +pub use datadog_webhook_token::DatadogWebhookToken; +pub use server::{provision, router}; diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs new file mode 100644 index 000000000..e65f61f76 --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs @@ -0,0 +1,279 @@ +use std::fmt; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use axum::{ + Router, body::Bytes, extract::DefaultBodyLimit, extract::State, http::HeaderMap, http::HeaderName, + http::StatusCode, http::header::AsHeaderName, routing::post, +}; +use tracing::{info, instrument, warn}; +use trogon_nats::NatsToken; +use trogon_nats::jetstream::{ + ClaimCheckPublisher, JetStreamContext, JetStreamPublisher, ObjectStorePut, PublishOutcome, +}; +use trogon_semconv::span::DATADOG_WEBHOOK; +use trogon_std::NonZeroDuration; + +use super::DatadogConfig; +use super::DatadogEventType; +use super::DatadogWebhookToken; +use super::constants::{HTTP_BODY_SIZE_MAX, NATS_HEADER_EVENT_ID, NATS_HEADER_EVENT_TYPE, NATS_HEADER_REJECT_REASON}; +use super::signature; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RejectReason { + InvalidJson, + MissingEventType, + InvalidEventType, +} + +impl RejectReason { + fn as_str(self) -> &'static str { + match self { + Self::InvalidJson => "invalid_json", + Self::MissingEventType => "missing_event_type", + Self::InvalidEventType => "invalid_event_type", + } + } +} + +fn header_value(headers: &HeaderMap, name: impl AsHeaderName) -> Option<&str> { + headers + .get(name) + .and_then(|value| value.to_str().ok()) + .map(str::trim) + .filter(|value| !value.is_empty()) +} + +fn body_event_id(payload: &serde_json::Value) -> Option<&str> { + payload + .get("id") + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) +} + +/// Reads the `timestamp` field (POSIX epoch seconds, e.g. Datadog's +/// `$DATE_POSIX`) from the payload. Datadog does not send a timestamp header, so +/// the operator templates it into the body. Accepts an integer JSON number or a +/// numeric string; fractional values are rejected so truncation cannot shift the +/// replay-window decision at the tolerance boundary. +fn body_timestamp_secs(payload: &serde_json::Value) -> Option { + let value = payload.get("timestamp")?; + if let Some(secs) = value.as_u64() { + return Some(secs); + } + value.as_str().and_then(|secs| secs.trim().parse::().ok()) +} + +fn timestamp_is_fresh(timestamp_secs: u64, tolerance: NonZeroDuration) -> bool { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + now.abs_diff(timestamp_secs) <= Duration::from(tolerance).as_secs() +} + +/// Boundary view of an untrusted Datadog webhook payload. All the fields the +/// handler routes on are extracted from the raw JSON exactly once here, so +/// downstream logic works with typed values instead of poking at +/// `serde_json::Value` throughout the request path. The raw `body` is still +/// forwarded verbatim to NATS; this type only drives routing decisions. +struct DatadogWebhookWire<'a> { + event_type: Option<&'a str>, + event_id: Option<&'a str>, + timestamp_secs: Option, +} + +impl<'a> DatadogWebhookWire<'a> { + fn from_payload(payload: &'a serde_json::Value) -> Self { + Self { + event_type: payload.get("event_type").and_then(serde_json::Value::as_str), + event_id: body_event_id(payload), + timestamp_secs: body_timestamp_secs(payload), + } + } +} + +fn outcome_to_status(outcome: PublishOutcome) -> StatusCode { + if outcome.is_ok() { + info!("Published Datadog event to NATS"); + StatusCode::OK + } else { + outcome.log_on_error("datadog"); + StatusCode::INTERNAL_SERVER_ERROR + } +} + +async fn publish_unroutable( + publisher: &ClaimCheckPublisher, + subject_prefix: &NatsToken, + event_id: Option<&str>, + reason: RejectReason, + body: Bytes, + ack_timeout: NonZeroDuration, +) -> StatusCode { + let subject = format!("{subject_prefix}.unroutable"); + let mut headers = async_nats::HeaderMap::new(); + headers.insert(NATS_HEADER_REJECT_REASON, reason.as_str()); + if let Some(event_id) = event_id { + headers.insert(async_nats::header::NATS_MESSAGE_ID, event_id); + headers.insert(NATS_HEADER_EVENT_ID, event_id); + } + + let outcome = publisher + .publish_event(subject, headers, body, ack_timeout.into()) + .await; + if outcome.is_ok() { + StatusCode::OK + } else { + outcome.log_on_error("datadog.unroutable"); + StatusCode::INTERNAL_SERVER_ERROR + } +} + +#[derive(Clone)] +struct AppState { + publisher: ClaimCheckPublisher, + webhook_token: DatadogWebhookToken, + webhook_token_header: HeaderName, + subject_prefix: NatsToken, + timestamp_tolerance: Option, + nats_ack_timeout: NonZeroDuration, +} + +pub async fn provision(js: &C, config: &DatadogConfig) -> Result<(), C::Error> { + js.get_or_create_stream(async_nats::jetstream::stream::Config { + name: config.stream_name.as_str().to_owned(), + subjects: vec![format!("{}.>", config.subject_prefix)], + max_age: config.stream_max_age.into(), + ..Default::default() + }) + .await?; + + let stream = config.stream_name.as_str(); + let max_age_secs = Duration::from(config.stream_max_age).as_secs(); + info!(stream, max_age_secs, "JetStream stream ready"); + Ok(()) +} + +pub fn router( + publisher: ClaimCheckPublisher, + config: &DatadogConfig, +) -> Router { + let state = AppState { + publisher, + webhook_token: config.webhook_token.clone(), + webhook_token_header: config.webhook_token_header.clone(), + subject_prefix: config.subject_prefix.clone(), + timestamp_tolerance: config.timestamp_tolerance, + nats_ack_timeout: config.nats_ack_timeout, + }; + + Router::new() + .route("/webhook", post(handle_webhook::)) + .layer(DefaultBodyLimit::max(HTTP_BODY_SIZE_MAX.as_usize())) + .with_state(state) +} + +#[instrument( + name = DATADOG_WEBHOOK, + skip_all, + fields( + event_type = tracing::field::Empty, + event_id = tracing::field::Empty, + subject = tracing::field::Empty, + ) +)] +async fn handle_webhook( + State(state): State>, + headers: HeaderMap, + body: Bytes, +) -> StatusCode { + let token = header_value(&headers, &state.webhook_token_header); + if let Err(error) = signature::verify(state.webhook_token.as_str(), token) { + warn!(reason = %error, "Datadog webhook token validation failed"); + return StatusCode::UNAUTHORIZED; + } + + let payload: serde_json::Value = match serde_json::from_slice(&body) { + Ok(value) => value, + Err(error) => { + warn!(error = %error, "Failed to parse Datadog webhook body as JSON"); + return publish_unroutable( + &state.publisher, + &state.subject_prefix, + None, + RejectReason::InvalidJson, + body, + state.nats_ack_timeout, + ) + .await; + } + }; + let webhook = DatadogWebhookWire::from_payload(&payload); + + if let Some(tolerance) = state.timestamp_tolerance { + let Some(timestamp_secs) = webhook.timestamp_secs else { + warn!("Missing or invalid timestamp in Datadog webhook payload while tolerance is enabled"); + return StatusCode::BAD_REQUEST; + }; + if !timestamp_is_fresh(timestamp_secs, tolerance) { + warn!(timestamp_secs, "Datadog webhook timestamp outside tolerance"); + return StatusCode::UNAUTHORIZED; + } + } + + let Some(raw_event_type) = webhook.event_type else { + warn!("Missing event_type in Datadog webhook payload"); + return publish_unroutable( + &state.publisher, + &state.subject_prefix, + webhook.event_id, + RejectReason::MissingEventType, + body, + state.nats_ack_timeout, + ) + .await; + }; + + let event_type = match DatadogEventType::new(raw_event_type) { + Ok(event_type) => event_type, + Err(error) => { + warn!(error = %error, "Invalid event_type in Datadog webhook payload"); + return publish_unroutable( + &state.publisher, + &state.subject_prefix, + webhook.event_id, + RejectReason::InvalidEventType, + body, + state.nats_ack_timeout, + ) + .await; + } + }; + + let subject = format!("{}.{}", state.subject_prefix, event_type); + let span = tracing::Span::current(); + span.record(trogon_semconv::attribute::EVENT_TYPE, event_type.as_str()); + if let Some(event_id) = webhook.event_id { + span.record(trogon_semconv::attribute::EVENT_ID, event_id); + } + span.record(trogon_semconv::attribute::SUBJECT, &subject); + + let mut nats_headers = async_nats::HeaderMap::new(); + nats_headers.insert(NATS_HEADER_EVENT_TYPE, event_type.as_str()); + if let Some(event_id) = webhook.event_id { + nats_headers.insert(async_nats::header::NATS_MESSAGE_ID, event_id); + nats_headers.insert(NATS_HEADER_EVENT_ID, event_id); + } + + let outcome = state + .publisher + .publish_event(subject, nats_headers, body, state.nats_ack_timeout.into()) + .await; + + outcome_to_status(outcome) +} + +#[cfg(test)] +mod tests; diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/server/tests.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/server/tests.rs new file mode 100644 index 000000000..1cc99f36e --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/server/tests.rs @@ -0,0 +1,336 @@ +use super::*; +use axum::body::Body; +use axum::http::Request; +use tower::ServiceExt; +use tracing_subscriber::util::SubscriberInitExt; +use trogon_nats::jetstream::StreamMaxAge; +use trogon_nats::jetstream::{ + ClaimCheckPublisher, MaxPayload, MockJetStreamContext, MockJetStreamPublisher, MockObjectStore, +}; + +const TEST_TOKEN: &str = "datadog-webhook-token"; +const DEFAULT_HEADER: &str = "x-datadog-webhook-token"; + +fn wrap_publisher(publisher: MockJetStreamPublisher) -> ClaimCheckPublisher { + ClaimCheckPublisher::new( + publisher, + MockObjectStore::new(), + "test-bucket".to_string(), + MaxPayload::from_server_limit(usize::MAX), + ) +} + +fn test_config() -> DatadogConfig { + config_with_tolerance(None) +} + +fn config_with_tolerance(timestamp_tolerance: Option) -> DatadogConfig { + DatadogConfig { + webhook_token: DatadogWebhookToken::new(TEST_TOKEN).unwrap(), + subject_prefix: NatsToken::new("datadog").unwrap(), + stream_name: NatsToken::new("DATADOG").unwrap(), + stream_max_age: StreamMaxAge::from_secs(3600).unwrap(), + nats_ack_timeout: NonZeroDuration::from_secs(10).unwrap(), + timestamp_tolerance, + webhook_token_header: HeaderName::from_static(DEFAULT_HEADER), + } +} + +fn tracing_guard() -> tracing::subscriber::DefaultGuard { + tracing_subscriber::fmt().with_test_writer().set_default() +} + +fn mock_app(publisher: MockJetStreamPublisher) -> Router { + router(wrap_publisher(publisher), &test_config()) +} + +fn mock_app_with_tolerance(publisher: MockJetStreamPublisher, tolerance_secs: u64) -> Router { + let config = config_with_tolerance(NonZeroDuration::from_secs(tolerance_secs).ok()); + router(wrap_publisher(publisher), &config) +} + +fn now_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() +} + +fn webhook_request(body: &[u8], token: Option<&str>) -> Request { + let mut builder = Request::builder().method("POST").uri("/webhook"); + if let Some(token) = token { + builder = builder.header(DEFAULT_HEADER, token); + } + builder.body(Body::from(body.to_vec())).unwrap() +} + +#[tokio::test] +async fn provision_creates_stream() { + let _guard = tracing_guard(); + let js = MockJetStreamContext::new(); + let config = test_config(); + + provision(&js, &config).await.unwrap(); + + let streams = js.created_streams(); + assert_eq!(streams.len(), 1); + assert_eq!(streams[0].name, "DATADOG"); + assert_eq!(streams[0].subjects, vec!["datadog.>"]); + assert_eq!(streams[0].max_age, Duration::from_secs(3600)); +} + +#[tokio::test] +async fn valid_event_publishes() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let app = mock_app(publisher.clone()); + let body = br#"{"event_type":"metric_alert_monitor","id":"7890","title":"CPU high"}"#; + + let resp = app.oneshot(webhook_request(body, Some(TEST_TOKEN))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let messages = publisher.published_messages(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].subject, "datadog.metric_alert_monitor"); + assert_eq!(messages[0].payload.as_ref(), body); + assert_eq!( + messages[0] + .headers + .get(async_nats::header::NATS_MESSAGE_ID) + .map(|v| v.as_str()), + Some("7890"), + ); + assert_eq!( + messages[0].headers.get(NATS_HEADER_EVENT_TYPE).map(|v| v.as_str()), + Some("metric_alert_monitor"), + ); + assert_eq!( + messages[0].headers.get(NATS_HEADER_EVENT_ID).map(|v| v.as_str()), + Some("7890"), + ); +} + +#[tokio::test] +async fn valid_event_without_id_publishes_without_message_id() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let app = mock_app(publisher.clone()); + let body = br#"{"event_type":"event_alert"}"#; + + let resp = app.oneshot(webhook_request(body, Some(TEST_TOKEN))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let messages = publisher.published_messages(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].subject, "datadog.event_alert"); + assert!(messages[0].headers.get(async_nats::header::NATS_MESSAGE_ID).is_none()); + assert!(messages[0].headers.get(NATS_HEADER_EVENT_ID).is_none()); +} + +#[tokio::test] +async fn invalid_json_publishes_unroutable_and_returns_ok() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let app = mock_app(publisher.clone()); + let body = b"{"; + + let resp = app.oneshot(webhook_request(body, Some(TEST_TOKEN))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let messages = publisher.published_messages(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].subject, "datadog.unroutable"); + assert_eq!( + messages[0].headers.get(NATS_HEADER_REJECT_REASON).map(|v| v.as_str()), + Some("invalid_json"), + ); + assert!(messages[0].headers.get(async_nats::header::NATS_MESSAGE_ID).is_none()); +} + +#[tokio::test] +async fn missing_event_type_publishes_unroutable_and_returns_ok() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let app = mock_app(publisher.clone()); + let body = br#"{"id":"42","title":"no type"}"#; + + let resp = app.oneshot(webhook_request(body, Some(TEST_TOKEN))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let messages = publisher.published_messages(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].subject, "datadog.unroutable"); + assert_eq!( + messages[0].headers.get(NATS_HEADER_REJECT_REASON).map(|v| v.as_str()), + Some("missing_event_type"), + ); + assert_eq!( + messages[0] + .headers + .get(async_nats::header::NATS_MESSAGE_ID) + .map(|v| v.as_str()), + Some("42"), + ); +} + +#[tokio::test] +async fn invalid_event_type_publishes_unroutable_and_returns_ok() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let app = mock_app(publisher.clone()); + let body = br#"{"event_type":"metric alert","id":"42"}"#; + + let resp = app.oneshot(webhook_request(body, Some(TEST_TOKEN))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let messages = publisher.published_messages(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].subject, "datadog.unroutable"); + assert_eq!( + messages[0].headers.get(NATS_HEADER_REJECT_REASON).map(|v| v.as_str()), + Some("invalid_event_type"), + ); +} + +#[tokio::test] +async fn missing_token_returns_401_and_publishes_nothing() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let app = mock_app(publisher.clone()); + let body = br#"{"event_type":"event_alert"}"#; + + let resp = app.oneshot(webhook_request(body, None)).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); + assert!(publisher.published_messages().is_empty()); +} + +#[tokio::test] +async fn wrong_token_returns_401_and_publishes_nothing() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let app = mock_app(publisher.clone()); + let body = br#"{"event_type":"event_alert"}"#; + + let resp = app.oneshot(webhook_request(body, Some("wrong-token"))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); + assert!(publisher.published_messages().is_empty()); +} + +#[tokio::test] +async fn publish_failure_returns_500() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + publisher.fail_next_js_publish(); + let app = mock_app(publisher); + let body = br#"{"event_type":"event_alert","id":"42"}"#; + + let resp = app.oneshot(webhook_request(body, Some(TEST_TOKEN))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR); +} + +#[tokio::test] +async fn unroutable_publish_failure_returns_500() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + publisher.fail_next_js_publish(); + let app = mock_app(publisher); + let body = br#"{"event_type":"metric alert"}"#; + + let resp = app.oneshot(webhook_request(body, Some(TEST_TOKEN))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR); +} + +#[tokio::test] +async fn fresh_timestamp_publishes_when_tolerance_enabled() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let app = mock_app_with_tolerance(publisher.clone(), 300); + let body = format!(r#"{{"event_type":"event_alert","timestamp":{}}}"#, now_secs()).into_bytes(); + + let resp = app.oneshot(webhook_request(&body, Some(TEST_TOKEN))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(publisher.published_subjects(), vec!["datadog.event_alert"]); +} + +#[tokio::test] +async fn fresh_timestamp_as_string_publishes() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let app = mock_app_with_tolerance(publisher.clone(), 300); + let body = format!(r#"{{"event_type":"event_alert","timestamp":"{}"}}"#, now_secs()).into_bytes(); + + let resp = app.oneshot(webhook_request(&body, Some(TEST_TOKEN))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(publisher.published_subjects(), vec!["datadog.event_alert"]); +} + +#[tokio::test] +async fn stale_timestamp_returns_401_and_publishes_nothing() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let app = mock_app_with_tolerance(publisher.clone(), 300); + let stale = now_secs() - 10_000; + let body = format!(r#"{{"event_type":"event_alert","timestamp":{stale}}}"#).into_bytes(); + + let resp = app.oneshot(webhook_request(&body, Some(TEST_TOKEN))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); + assert!(publisher.published_messages().is_empty()); +} + +#[tokio::test] +async fn missing_timestamp_returns_400_when_tolerance_enabled() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let app = mock_app_with_tolerance(publisher.clone(), 300); + let body = br#"{"event_type":"event_alert"}"#; + + let resp = app.oneshot(webhook_request(body, Some(TEST_TOKEN))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + assert!(publisher.published_messages().is_empty()); +} + +#[tokio::test] +async fn timestamp_ignored_when_tolerance_disabled() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let app = mock_app(publisher.clone()); + let body = br#"{"event_type":"event_alert","timestamp":1}"#; + + let resp = app.oneshot(webhook_request(body, Some(TEST_TOKEN))).await.unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(publisher.published_subjects(), vec!["datadog.event_alert"]); +} + +#[tokio::test] +async fn custom_webhook_token_header_is_honored() { + let _guard = tracing_guard(); + let publisher = MockJetStreamPublisher::new(); + let config = DatadogConfig { + webhook_token_header: HeaderName::from_static("x-acme-token"), + ..test_config() + }; + let app = router(wrap_publisher(publisher.clone()), &config); + let body = br#"{"event_type":"event_alert"}"#; + + let custom = Request::builder() + .method("POST") + .uri("/webhook") + .header("x-acme-token", TEST_TOKEN) + .body(Body::from(body.to_vec())) + .unwrap(); + let resp = app.clone().oneshot(custom).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(publisher.published_subjects(), vec!["datadog.event_alert"]); + + let default_header = app.oneshot(webhook_request(body, Some(TEST_TOKEN))).await.unwrap(); + assert_eq!(default_header.status(), StatusCode::UNAUTHORIZED); +} diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/signature.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/signature.rs new file mode 100644 index 000000000..dec7067d3 --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/signature.rs @@ -0,0 +1,30 @@ +use sha2::{Digest, Sha256}; + +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum SignatureError { + #[error("missing webhook token header")] + Missing, + #[error("webhook token mismatch")] + Mismatch, +} + +/// Verifies the Datadog webhook shared secret token using constant-time +/// comparison. +/// +/// Datadog does not sign webhook requests, so authentication relies on a shared +/// secret the operator delivers in a custom header. Both sides are hashed with +/// SHA-256 before comparing, ensuring equal-length slices regardless of input +/// length. This prevents leaking the secret's length via timing. +pub fn verify(secret: &str, token_header: Option<&str>) -> Result<(), SignatureError> { + let token = token_header.ok_or(SignatureError::Missing)?; + + let expected = Sha256::digest(secret.as_bytes()); + let provided = Sha256::digest(token.as_bytes()); + + let ok = subtle::ConstantTimeEq::ct_eq(expected.as_slice(), provided.as_slice()).unwrap_u8(); + if ok == 1 { Ok(()) } else { Err(SignatureError::Mismatch) } +} + +#[cfg(test)] +mod tests; diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/signature/tests.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/signature/tests.rs new file mode 100644 index 000000000..e016bc6e8 --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/signature/tests.rs @@ -0,0 +1,30 @@ +use super::*; + +#[test] +fn error_display_messages() { + assert_eq!(SignatureError::Missing.to_string(), "missing webhook token header"); + assert_eq!(SignatureError::Mismatch.to_string(), "webhook token mismatch"); +} + +#[test] +fn valid_token_passes() { + assert!(verify("my-secret", Some("my-secret")).is_ok()); +} + +#[test] +fn wrong_token_fails() { + assert!(matches!( + verify("correct-secret", Some("wrong-secret")), + Err(SignatureError::Mismatch) + )); +} + +#[test] +fn missing_token_fails() { + assert!(matches!(verify("secret", None), Err(SignatureError::Missing))); +} + +#[test] +fn empty_secret_does_not_match_nonempty_token() { + assert!(matches!(verify("", Some("something")), Err(SignatureError::Mismatch))); +} diff --git a/rsworkspace/crates/trogon-gateway/src/source/mod.rs b/rsworkspace/crates/trogon-gateway/src/source/mod.rs index aa41b11b6..dd85beab3 100644 --- a/rsworkspace/crates/trogon-gateway/src/source/mod.rs +++ b/rsworkspace/crates/trogon-gateway/src/source/mod.rs @@ -1,3 +1,4 @@ +pub mod datadog; pub mod discord; pub mod github; pub mod gitlab; diff --git a/rsworkspace/crates/trogon-gateway/src/source_plugin.rs b/rsworkspace/crates/trogon-gateway/src/source_plugin.rs index f88443ce7..e85bc2121 100644 --- a/rsworkspace/crates/trogon-gateway/src/source_plugin.rs +++ b/rsworkspace/crates/trogon-gateway/src/source_plugin.rs @@ -43,6 +43,7 @@ pub struct LinearPlugin; pub struct MicrosoftGraphPlugin; pub struct NotionPlugin; pub struct SentryPlugin; +pub struct DatadogPlugin; async fn provision_integrations( integrations: &[SourceIntegration], @@ -358,6 +359,31 @@ impl SourcePlugin for SentryPlugin { } } +impl SourcePlugin for DatadogPlugin { + fn id(&self) -> SourceId { + "datadog" + } + + async fn provision(&self, client: &C, config: &ResolvedConfig) -> Result<(), C::Error> { + provision_integrations(&config.datadog, self.id(), client, crate::source::datadog::provision).await + } + + fn mount(&self, app: Router, publisher: ClaimCheckPublisher, config: &ResolvedConfig) -> Router + where + P: JetStreamPublisher, + S: ObjectStorePut, + { + mount_integrations( + &config.datadog, + app, + publisher, + self.id(), + &self.path_prefix(), + |p, cfg| crate::source::datadog::router(p, cfg), + ) + } +} + /// Provision JetStream streams for every webhook source. Discord is provisioned separately by the caller. pub async fn provision_webhook_sources( client: &C, @@ -373,6 +399,7 @@ pub async fn provision_webhook_sources( MicrosoftGraphPlugin.provision(client, config).await?; NotionPlugin.provision(client, config).await?; SentryPlugin.provision(client, config).await?; + DatadogPlugin.provision(client, config).await?; Ok(()) } @@ -395,5 +422,6 @@ where app = LinearPlugin.mount(app, publisher.clone(), config); app = MicrosoftGraphPlugin.mount(app, publisher.clone(), config); app = NotionPlugin.mount(app, publisher.clone(), config); - SentryPlugin.mount(app, publisher, config) + app = SentryPlugin.mount(app, publisher.clone(), config); + DatadogPlugin.mount(app, publisher, config) } diff --git a/rsworkspace/crates/trogon-gateway/src/streams/tests.rs b/rsworkspace/crates/trogon-gateway/src/streams/tests.rs index 3554597b4..b439fc031 100644 --- a/rsworkspace/crates/trogon-gateway/src/streams/tests.rs +++ b/rsworkspace/crates/trogon-gateway/src/streams/tests.rs @@ -47,6 +47,9 @@ verification_token = "notion-verification-token-example" [sources.sentry.integrations.primary.webhook] client_secret = "sentry-client-secret" + +[sources.datadog.integrations.primary.webhook] +webhook_token = "datadog-webhook-token" "# .to_string() } @@ -69,7 +72,7 @@ async fn provision_all_sources_creates_all_streams() { provision(&js, &cfg).await.expect("provision should succeed"); - assert_eq!(js.created_streams().len(), 11); + assert_eq!(js.created_streams().len(), 12); } #[tokio::test] @@ -133,6 +136,9 @@ verification_token = "notion-verification-token-example" [sources.sentry.integrations.primary.webhook] client_secret = "sentry-client-secret" + +[sources.datadog.integrations.primary.webhook] +webhook_token = "datadog-webhook-token" "#; let f = write_toml(toml); let cfg = load(Some(f.path())).expect("load failed"); @@ -140,5 +146,5 @@ client_secret = "sentry-client-secret" provision(&js, &cfg).await.expect("provision should succeed"); - assert_eq!(js.created_streams().len(), 10); + assert_eq!(js.created_streams().len(), 11); } diff --git a/rsworkspace/crates/trogon-semconv/src/gen/attribute.rs b/rsworkspace/crates/trogon-semconv/src/gen/attribute.rs index e44f83c2a..7c61ddb9f 100644 --- a/rsworkspace/crates/trogon-semconv/src/gen/attribute.rs +++ b/rsworkspace/crates/trogon-semconv/src/gen/attribute.rs @@ -177,7 +177,7 @@ pub const DELIVERY: &str = "delivery"; /// Webhook event name (GitHub, GitLab) pub const EVENT: &str = "event"; -/// Webhook event identifier (Slack, Notion) +/// Webhook event identifier (Slack, Notion, Datadog) pub const EVENT_ID: &str = "event_id"; /// Event type discriminator. Reused by scheduler reconciliation and several gateway webhook spans with different value spaces diff --git a/rsworkspace/crates/trogon-semconv/src/gen/span.rs b/rsworkspace/crates/trogon-semconv/src/gen/span.rs index 2f2eb1226..48fd40284 100644 --- a/rsworkspace/crates/trogon-semconv/src/gen/span.rs +++ b/rsworkspace/crates/trogon-semconv/src/gen/span.rs @@ -120,6 +120,9 @@ pub const ACP_SESSION_SET_MODE: &str = "acp.session.set_mode"; /// ACP set_session_model handling pub const ACP_SESSION_SET_MODEL: &str = "acp.session.set_model"; +/// Datadog webhook ingestion +pub const DATADOG_WEBHOOK: &str = "datadog.webhook"; + /// ACP client method dispatch. Span name is `dispatch_client_method` (no namespace) today pub const DISPATCH_CLIENT_METHOD: &str = "dispatch_client_method";