From 45036ae623faf0f706f3d4f6a0b1004d8acc087a Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Wed, 1 Jul 2026 17:32:43 -0400 Subject: [PATCH 1/5] feat(gateway): add Datadog webhook source Datadog monitor and event alerts need to land on the NATS event bus alongside the other observability sources so downstream consumers can react to them. Signed-off-by: Yordis Prieto --- devops/docker/compose/.env.example | 3 + .../compose/services/trogon-gateway/README.md | 1 + .../services/trogon-gateway/gateway.toml | 10 + otel/semconv/registry/gateway.yaml | 11 +- rsworkspace/crates/trogon-gateway/README.md | 29 ++ .../crates/trogon-gateway/src/config.rs | 96 ++++++ .../crates/trogon-gateway/src/config/tests.rs | 90 +++++ .../crates/trogon-gateway/src/http/tests.rs | 3 + .../src/source/datadog/config.rs | 16 + .../src/source/datadog/constants.rs | 12 + .../src/source/datadog/datadog_event_type.rs | 56 ++++ .../datadog/datadog_event_type/tests.rs | 68 ++++ .../source/datadog/datadog_webhook_token.rs | 25 ++ .../datadog/datadog_webhook_token/tests.rs | 18 + .../trogon-gateway/src/source/datadog/mod.rs | 19 ++ .../src/source/datadog/server.rs | 260 +++++++++++++++ .../src/source/datadog/server/tests.rs | 309 ++++++++++++++++++ .../src/source/datadog/signature.rs | 30 ++ .../src/source/datadog/signature/tests.rs | 30 ++ .../crates/trogon-gateway/src/source/mod.rs | 1 + .../trogon-gateway/src/source_plugin.rs | 30 +- .../trogon-gateway/src/streams/tests.rs | 10 +- .../trogon-semconv/src/gen/attribute.rs | 2 +- .../crates/trogon-semconv/src/gen/span.rs | 3 + 24 files changed, 1127 insertions(+), 5 deletions(-) create mode 100644 rsworkspace/crates/trogon-gateway/src/source/datadog/config.rs create mode 100644 rsworkspace/crates/trogon-gateway/src/source/datadog/constants.rs create mode 100644 rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_event_type.rs create mode 100644 rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_event_type/tests.rs create mode 100644 rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_webhook_token.rs create mode 100644 rsworkspace/crates/trogon-gateway/src/source/datadog/datadog_webhook_token/tests.rs create mode 100644 rsworkspace/crates/trogon-gateway/src/source/datadog/mod.rs create mode 100644 rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs create mode 100644 rsworkspace/crates/trogon-gateway/src/source/datadog/server/tests.rs create mode 100644 rsworkspace/crates/trogon-gateway/src/source/datadog/signature.rs create mode 100644 rsworkspace/crates/trogon-gateway/src/source/datadog/signature/tests.rs 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..70fde3478 100644 --- a/devops/docker/compose/services/trogon-gateway/gateway.toml +++ b/devops/docker/compose/services/trogon-gateway/gateway.toml @@ -81,6 +81,16 @@ 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" } +# timestamp_tolerance_secs = 300 # optional anti-replay window; requires "timestamp": "$DATE_EPOCH" 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..8dbcfc2e6 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 the `X-Datadog-Webhook-Token` custom header (configured in the Datadog webhook's *Custom Headers*), verified in constant time. 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_EPOCH` (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,33 @@ 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" } +# timestamp_tolerance_secs = 300 # optional anti-replay window; requires "timestamp": "$DATE_EPOCH" 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_EPOCH", + "title": "$EVENT_TITLE", + "body": "$EVENT_MSG", + "alert_transition": "$ALERT_TRANSITION" +} +``` + +`timestamp` is only required when `timestamp_tolerance_secs` is set; `$DATE_EPOCH` 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..092ee40e9 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; @@ -350,6 +351,8 @@ struct SourcesConfig { notion: NotionConfig, #[config(nested)] sentry: SentryConfig, + #[config(nested)] + datadog: DatadogConfig, } #[derive(Config)] @@ -448,6 +451,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 +555,13 @@ struct SentryWebhookConfig { client_secret: Option, } +#[derive(serde::Deserialize)] +#[serde(deny_unknown_fields)] +struct DatadogWebhookConfig { + webhook_token: Option, + timestamp_tolerance_secs: Option, +} + pub struct ResolvedHttpServerConfig { pub port: u16, } @@ -573,6 +591,7 @@ pub struct ResolvedConfig { pub microsoft_graph: Vec>, pub notion: Vec>, pub sentry: Vec>, + pub datadog: Vec>, } impl ResolvedConfig { @@ -588,6 +607,7 @@ impl ResolvedConfig { || !self.microsoft_graph.is_empty() || !self.notion.is_empty() || !self.sentry.is_empty() + || !self.datadog.is_empty() } } @@ -620,6 +640,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 { + 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 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, + 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..e6ee68ba4 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,87 @@ 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_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 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..0f706f45a --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/config.rs @@ -0,0 +1,16 @@ +use super::datadog_webhook_token::DatadogWebhookToken; +use trogon_nats::NatsToken; +use trogon_nats::jetstream::StreamMaxAge; +use trogon_std::NonZeroDuration; + +pub struct DatadogConfig { + pub webhook_token: DatadogWebhookToken, + 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_EPOCH` (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..c7ae02e6d --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/constants.rs @@ -0,0 +1,12 @@ +use trogon_std::{ByteSize, HttpBodySizeMax}; + +pub const HTTP_BODY_SIZE_MAX: HttpBodySizeMax = HttpBodySizeMax::new(ByteSize::mib(2)).unwrap(); + +/// 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. +pub const 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..4eeb5cd29 --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs @@ -0,0 +1,260 @@ +use std::fmt; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use axum::{ + Router, body::Bytes, extract::DefaultBodyLimit, extract::State, http::HeaderMap, http::StatusCode, 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::{ + HEADER_WEBHOOK_TOKEN, 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<'a>(headers: &'a HeaderMap, name: &'static str) -> Option<&'a 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_EPOCH`) from the payload. Datadog does not send a timestamp header, so +/// the operator templates it into the body. Accepts a JSON number or a numeric +/// string. +fn body_timestamp_secs(payload: &serde_json::Value) -> Option { + let value = payload.get("timestamp")?; + if let Some(secs) = value.as_u64() { + return Some(secs); + } + if let Some(secs) = value.as_f64().filter(|secs| *secs >= 0.0) { + return Some(secs as u64); + } + 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() +} + +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, + 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(), + 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, HEADER_WEBHOOK_TOKEN); + 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; + } + }; + + if let Some(tolerance) = state.timestamp_tolerance { + let Some(timestamp_secs) = body_timestamp_secs(&payload) 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 event_id = body_event_id(&payload).map(str::to_owned); + + let Some(raw_event_type) = payload.get("event_type").and_then(serde_json::Value::as_str) else { + warn!("Missing event_type in Datadog webhook payload"); + return publish_unroutable( + &state.publisher, + &state.subject_prefix, + event_id.as_deref(), + 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, + event_id.as_deref(), + 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) = event_id.as_deref() { + 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) = event_id.as_deref() { + 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..1a860e7a1 --- /dev/null +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/server/tests.rs @@ -0,0 +1,309 @@ +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"; + +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, + } +} + +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(HEADER_WEBHOOK_TOKEN, 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"]); +} 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"; From 11ec28fb60417b72cb8a23d8f25dab951f02fee4 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Wed, 1 Jul 2026 17:43:19 -0400 Subject: [PATCH 2/5] feat(gateway): make Datadog webhook token header configurable Operators front Datadog behind proxies and conventions that dictate the custom header name, so the shared-secret header must be overridable while keeping a sensible default. Signed-off-by: Yordis Prieto --- .../services/trogon-gateway/gateway.toml | 1 + rsworkspace/crates/trogon-gateway/README.md | 3 +- .../crates/trogon-gateway/src/config.rs | 18 ++++++++++ .../crates/trogon-gateway/src/config/tests.rs | 36 +++++++++++++++++++ .../src/source/datadog/config.rs | 4 +++ .../src/source/datadog/constants.rs | 9 ++--- .../src/source/datadog/server.rs | 13 +++---- .../src/source/datadog/server/tests.rs | 29 ++++++++++++++- 8 files changed, 101 insertions(+), 12 deletions(-) diff --git a/devops/docker/compose/services/trogon-gateway/gateway.toml b/devops/docker/compose/services/trogon-gateway/gateway.toml index 70fde3478..ed0f4a986 100644 --- a/devops/docker/compose/services/trogon-gateway/gateway.toml +++ b/devops/docker/compose/services/trogon-gateway/gateway.toml @@ -89,6 +89,7 @@ webhook_secret = { env = "TROGON_GATEWAY_LOCAL_GITHUB_WEBHOOK_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_EPOCH" in the payload # # [sources.discord] diff --git a/rsworkspace/crates/trogon-gateway/README.md b/rsworkspace/crates/trogon-gateway/README.md index 8dbcfc2e6..6581237d9 100644 --- a/rsworkspace/crates/trogon-gateway/README.md +++ b/rsworkspace/crates/trogon-gateway/README.md @@ -105,7 +105,7 @@ 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 the `X-Datadog-Webhook-Token` custom header (configured in the Datadog webhook's *Custom Headers*), verified in constant time. 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 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_EPOCH` (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 @@ -170,6 +170,7 @@ 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_EPOCH" in the payload ``` diff --git a/rsworkspace/crates/trogon-gateway/src/config.rs b/rsworkspace/crates/trogon-gateway/src/config.rs index 092ee40e9..e1992b7d5 100644 --- a/rsworkspace/crates/trogon-gateway/src/config.rs +++ b/rsworkspace/crates/trogon-gateway/src/config.rs @@ -20,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; @@ -559,6 +560,7 @@ struct SentryWebhookConfig { #[serde(deny_unknown_fields)] struct DatadogWebhookConfig { webhook_token: Option, + webhook_token_header: Option, timestamp_tolerance_secs: Option, } @@ -1781,6 +1783,21 @@ fn resolve_datadog_integrations( ) 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()); @@ -1788,6 +1805,7 @@ fn resolve_datadog_integrations( id, crate::source::datadog::DatadogConfig { webhook_token, + webhook_token_header, subject_prefix, stream_name, stream_max_age, diff --git a/rsworkspace/crates/trogon-gateway/src/config/tests.rs b/rsworkspace/crates/trogon-gateway/src/config/tests.rs index e6ee68ba4..371ee5478 100644 --- a/rsworkspace/crates/trogon-gateway/src/config/tests.rs +++ b/rsworkspace/crates/trogon-gateway/src/config/tests.rs @@ -969,6 +969,42 @@ timestamp_tolerance_secs = 0 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/source/datadog/config.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/config.rs index 0f706f45a..921a35dc4 100644 --- a/rsworkspace/crates/trogon-gateway/src/source/datadog/config.rs +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/config.rs @@ -1,10 +1,14 @@ 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, diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/constants.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/constants.rs index c7ae02e6d..7798f4674 100644 --- a/rsworkspace/crates/trogon-gateway/src/source/datadog/constants.rs +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/constants.rs @@ -2,10 +2,11 @@ use trogon_std::{ByteSize, HttpBodySizeMax}; pub const HTTP_BODY_SIZE_MAX: HttpBodySizeMax = HttpBodySizeMax::new(ByteSize::mib(2)).unwrap(); -/// 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. -pub const HEADER_WEBHOOK_TOKEN: &str = "x-datadog-webhook-token"; +/// 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"; diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs index 4eeb5cd29..0b95aa095 100644 --- a/rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs @@ -2,7 +2,8 @@ use std::fmt; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use axum::{ - Router, body::Bytes, extract::DefaultBodyLimit, extract::State, http::HeaderMap, http::StatusCode, routing::post, + 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; @@ -15,9 +16,7 @@ use trogon_std::NonZeroDuration; use super::DatadogConfig; use super::DatadogEventType; use super::DatadogWebhookToken; -use super::constants::{ - HEADER_WEBHOOK_TOKEN, HTTP_BODY_SIZE_MAX, NATS_HEADER_EVENT_ID, NATS_HEADER_EVENT_TYPE, NATS_HEADER_REJECT_REASON, -}; +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)] @@ -37,7 +36,7 @@ impl RejectReason { } } -fn header_value<'a>(headers: &'a HeaderMap, name: &'static str) -> Option<&'a str> { +fn header_value(headers: &HeaderMap, name: impl AsHeaderName) -> Option<&str> { headers .get(name) .and_then(|value| value.to_str().ok()) @@ -117,6 +116,7 @@ async fn publish_unroutable( struct AppState { publisher: ClaimCheckPublisher, webhook_token: DatadogWebhookToken, + webhook_token_header: HeaderName, subject_prefix: NatsToken, timestamp_tolerance: Option, nats_ack_timeout: NonZeroDuration, @@ -144,6 +144,7 @@ pub fn 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, @@ -169,7 +170,7 @@ async fn handle_webhook( headers: HeaderMap, body: Bytes, ) -> StatusCode { - let token = header_value(&headers, HEADER_WEBHOOK_TOKEN); + 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; diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/server/tests.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/server/tests.rs index 1a860e7a1..1cc99f36e 100644 --- a/rsworkspace/crates/trogon-gateway/src/source/datadog/server/tests.rs +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/server/tests.rs @@ -9,6 +9,7 @@ use trogon_nats::jetstream::{ }; const TEST_TOKEN: &str = "datadog-webhook-token"; +const DEFAULT_HEADER: &str = "x-datadog-webhook-token"; fn wrap_publisher(publisher: MockJetStreamPublisher) -> ClaimCheckPublisher { ClaimCheckPublisher::new( @@ -31,6 +32,7 @@ fn config_with_tolerance(timestamp_tolerance: Option) -> Datado 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), } } @@ -57,7 +59,7 @@ fn now_secs() -> u64 { 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(HEADER_WEBHOOK_TOKEN, token); + builder = builder.header(DEFAULT_HEADER, token); } builder.body(Body::from(body.to_vec())).unwrap() } @@ -307,3 +309,28 @@ async fn timestamp_ignored_when_tolerance_disabled() { 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); +} From 32258d980337978de3934a742bc90569ef317513 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Wed, 1 Jul 2026 17:48:38 -0400 Subject: [PATCH 3/5] fix(gateway): harden Datadog webhook payload parsing Centralize untrusted-payload extraction into a single boundary type, reject fractional timestamps that could shift the replay-window decision, and correct the documented Datadog time variable to $DATE_POSIX so operators configure a timestamp the source actually accepts. Signed-off-by: Yordis Prieto --- .../services/trogon-gateway/gateway.toml | 2 +- rsworkspace/crates/trogon-gateway/README.md | 8 ++-- .../src/source/datadog/config.rs | 2 +- .../src/source/datadog/server.rs | 46 +++++++++++++------ 4 files changed, 38 insertions(+), 20 deletions(-) diff --git a/devops/docker/compose/services/trogon-gateway/gateway.toml b/devops/docker/compose/services/trogon-gateway/gateway.toml index ed0f4a986..3956d80a5 100644 --- a/devops/docker/compose/services/trogon-gateway/gateway.toml +++ b/devops/docker/compose/services/trogon-gateway/gateway.toml @@ -90,7 +90,7 @@ webhook_secret = { env = "TROGON_GATEWAY_LOCAL_GITHUB_WEBHOOK_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_EPOCH" in the payload +# timestamp_tolerance_secs = 300 # optional anti-replay window; requires "timestamp": "$DATE_POSIX" in the payload # # [sources.discord] # bot_token = { env = "DISCORD_BOT_TOKEN" } diff --git a/rsworkspace/crates/trogon-gateway/README.md b/rsworkspace/crates/trogon-gateway/README.md index 6581237d9..508db1ade 100644 --- a/rsworkspace/crates/trogon-gateway/README.md +++ b/rsworkspace/crates/trogon-gateway/README.md @@ -106,7 +106,7 @@ Source-specific extras: - 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_EPOCH` (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 +- 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 @@ -171,7 +171,7 @@ 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_EPOCH" in the payload +# 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: @@ -186,14 +186,14 @@ Configure the matching Datadog webhook (Integrations > Webhooks) with a custom h { "id": "$ID", "event_type": "$EVENT_TYPE", - "timestamp": "$DATE_EPOCH", + "timestamp": "$DATE_POSIX", "title": "$EVENT_TITLE", "body": "$EVENT_MSG", "alert_transition": "$ALERT_TRANSITION" } ``` -`timestamp` is only required when `timestamp_tolerance_secs` is set; `$DATE_EPOCH` is POSIX epoch seconds (`$DATE` is epoch milliseconds, which this source does not accept). +`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 diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/config.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/config.rs index 921a35dc4..358bcdce5 100644 --- a/rsworkspace/crates/trogon-gateway/src/source/datadog/config.rs +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/config.rs @@ -15,6 +15,6 @@ pub struct DatadogConfig { 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_EPOCH` (POSIX seconds) into the payload. + /// operator must template `$DATE_POSIX` (POSIX seconds) into the payload. pub timestamp_tolerance: Option, } diff --git a/rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs b/rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs index 0b95aa095..e65f61f76 100644 --- a/rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs +++ b/rsworkspace/crates/trogon-gateway/src/source/datadog/server.rs @@ -53,17 +53,15 @@ fn body_event_id(payload: &serde_json::Value) -> Option<&str> { } /// Reads the `timestamp` field (POSIX epoch seconds, e.g. Datadog's -/// `$DATE_EPOCH`) from the payload. Datadog does not send a timestamp header, so -/// the operator templates it into the body. Accepts a JSON number or a numeric -/// string. +/// `$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); } - if let Some(secs) = value.as_f64().filter(|secs| *secs >= 0.0) { - return Some(secs as u64); - } value.as_str().and_then(|secs| secs.trim().parse::().ok()) } @@ -75,6 +73,27 @@ fn timestamp_is_fresh(timestamp_secs: u64, tolerance: NonZeroDuration) -> bool { 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"); @@ -191,9 +210,10 @@ async fn handle_webhook( .await; } }; + let webhook = DatadogWebhookWire::from_payload(&payload); if let Some(tolerance) = state.timestamp_tolerance { - let Some(timestamp_secs) = body_timestamp_secs(&payload) else { + 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; }; @@ -203,14 +223,12 @@ async fn handle_webhook( } } - let event_id = body_event_id(&payload).map(str::to_owned); - - let Some(raw_event_type) = payload.get("event_type").and_then(serde_json::Value::as_str) else { + 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, - event_id.as_deref(), + webhook.event_id, RejectReason::MissingEventType, body, state.nats_ack_timeout, @@ -225,7 +243,7 @@ async fn handle_webhook( return publish_unroutable( &state.publisher, &state.subject_prefix, - event_id.as_deref(), + webhook.event_id, RejectReason::InvalidEventType, body, state.nats_ack_timeout, @@ -237,14 +255,14 @@ async fn handle_webhook( 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) = event_id.as_deref() { + 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) = event_id.as_deref() { + 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); } From 25c5dbf2f7cf9236af052cfc5dc6d6c93c211d85 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Wed, 1 Jul 2026 17:54:54 -0400 Subject: [PATCH 4/5] fix(gateway): reject enabled Datadog integration without a webhook block Silently dropping a configured-but-webhook-less integration lets the gateway start with an operator-declared integration missing, so it must fail config resolution loudly instead. Signed-off-by: Yordis Prieto --- rsworkspace/crates/trogon-gateway/src/config.rs | 1 + .../crates/trogon-gateway/src/config/tests.rs | 13 +++++++++++++ 2 files changed, 14 insertions(+) diff --git a/rsworkspace/crates/trogon-gateway/src/config.rs b/rsworkspace/crates/trogon-gateway/src/config.rs index e1992b7d5..0db0e3c42 100644 --- a/rsworkspace/crates/trogon-gateway/src/config.rs +++ b/rsworkspace/crates/trogon-gateway/src/config.rs @@ -1748,6 +1748,7 @@ fn resolve_datadog_integrations( continue; } let Some(webhook) = integration.webhook else { + errors.push(ConfigValidationError::missing_integration("datadog", &id, "webhook")); continue; }; let Some(token) = diff --git a/rsworkspace/crates/trogon-gateway/src/config/tests.rs b/rsworkspace/crates/trogon-gateway/src/config/tests.rs index 371ee5478..06111e3c3 100644 --- a/rsworkspace/crates/trogon-gateway/src/config/tests.rs +++ b/rsworkspace/crates/trogon-gateway/src/config/tests.rs @@ -935,6 +935,19 @@ fn datadog_empty_webhook_token_is_invalid() { ); } +#[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_timestamp_tolerance_defaults_to_disabled() { let f = write_toml(&datadog_toml("datadog-webhook-token")); From edcb685d11fa1afb3be6c09206c91d2b31db7e97 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Wed, 1 Jul 2026 20:58:01 -0400 Subject: [PATCH 5/5] chore(gateway): cover Datadog config resolver skip branches The coverage gate fails when a diff adds uncovered lines; the Datadog resolver's invalid-id, disabled-integration, and invalid-common-field branches had no tests. Signed-off-by: Yordis Prieto --- .../crates/trogon-gateway/src/config/tests.rs | 43 +++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/rsworkspace/crates/trogon-gateway/src/config/tests.rs b/rsworkspace/crates/trogon-gateway/src/config/tests.rs index 06111e3c3..b7c6cd72a 100644 --- a/rsworkspace/crates/trogon-gateway/src/config/tests.rs +++ b/rsworkspace/crates/trogon-gateway/src/config/tests.rs @@ -948,6 +948,49 @@ subject_prefix = "datadog-primary" ); } +#[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"));