ironflow_engine/notify/
webhook.rs

1//! [`WebhookSubscriber`] -- POSTs events as JSON to a URL.
2//!
3//! When a signing secret is configured, each outbound request includes an
4//! `X-Signature-256` header containing the HMAC-SHA256 digest of the JSON
5//! body, prefixed with `sha256=` (the same convention used by GitHub and
6//! GitLab). Receivers can verify the signature to ensure the payload was
7//! not tampered with in transit.
8
9use hmac::{Hmac, Mac};
10use reqwest::Client;
11use sha2::Sha256;
12
13use super::retry::{RetryConfig, deliver_with_retry, is_success_2xx};
14use super::{Event, EventSubscriber, SubscriberFuture};
15
16type HmacSha256 = Hmac<Sha256>;
17
18/// Header name used for outbound HMAC-SHA256 signatures.
19const SIGNATURE_HEADER: &str = "X-Signature-256";
20
21/// Subscriber that POSTs the event as JSON to a webhook URL.
22///
23/// Retries failed deliveries with exponential backoff (up to 3 attempts,
24/// 5 s timeout per attempt). The HTTP client is created once and reused.
25///
26/// When a signing secret is provided via [`WebhookSubscriber::with_signing_secret`],
27/// every outbound request includes an `X-Signature-256` header whose value
28/// is `sha256=<hex-encoded HMAC-SHA256>`, computed over the raw JSON body.
29/// This lets the receiver verify payload authenticity.
30///
31/// Event type filtering is handled by the
32/// [`EventPublisher`](super::EventPublisher) at subscription time -- this
33/// subscriber receives only events that already passed the filter.
34///
35/// # Examples
36///
37/// ```no_run
38/// use ironflow_engine::notify::{Event, EventPublisher, WebhookSubscriber};
39///
40/// let mut publisher = EventPublisher::new();
41///
42/// // Without signature
43/// publisher.subscribe(
44///     WebhookSubscriber::new("https://hooks.example.com/events"),
45///     &[Event::RUN_STATUS_CHANGED, Event::STEP_FAILED],
46/// );
47///
48/// // With HMAC-SHA256 signature
49/// publisher.subscribe(
50///     WebhookSubscriber::with_signing_secret(
51///         "https://hooks.example.com/signed",
52///         "my-webhook-secret",
53///     ),
54///     Event::ALL,
55/// );
56/// ```
57pub struct WebhookSubscriber {
58    url: String,
59    signing_secret: Option<String>,
60    client: Client,
61    retry_config: RetryConfig,
62}
63
64impl WebhookSubscriber {
65    /// Create a new webhook subscriber targeting the given URL.
66    ///
67    /// Uses the default [`RetryConfig`] (3 retries, 5 s timeout, 500 ms
68    /// base backoff). No outbound signature is added.
69    ///
70    /// # Panics
71    ///
72    /// Panics if the HTTP client cannot be built (TLS backend unavailable).
73    ///
74    /// # Examples
75    ///
76    /// ```
77    /// use ironflow_engine::notify::WebhookSubscriber;
78    ///
79    /// let subscriber = WebhookSubscriber::new("https://example.com/hook");
80    /// assert_eq!(subscriber.url(), "https://example.com/hook");
81    /// assert!(subscriber.signing_secret().is_none());
82    /// ```
83    pub fn new(url: &str) -> Self {
84        Self::with_retry_config(url, RetryConfig::default())
85    }
86
87    /// Create a webhook subscriber with a custom retry configuration.
88    ///
89    /// No outbound signature is added.
90    ///
91    /// # Panics
92    ///
93    /// Panics if the HTTP client cannot be built (TLS backend unavailable).
94    ///
95    /// # Examples
96    ///
97    /// ```
98    /// use ironflow_engine::notify::{RetryConfig, WebhookSubscriber};
99    ///
100    /// let config = RetryConfig::new(
101    ///     5,
102    ///     std::time::Duration::from_secs(10),
103    ///     std::time::Duration::from_secs(1),
104    /// );
105    /// let subscriber = WebhookSubscriber::with_retry_config("https://example.com/hook", config);
106    /// ```
107    pub fn with_retry_config(url: &str, retry_config: RetryConfig) -> Self {
108        Self::build(url, None, retry_config)
109    }
110
111    /// Create a webhook subscriber that signs outbound payloads with HMAC-SHA256.
112    ///
113    /// Each request will include an `X-Signature-256` header containing
114    /// `sha256=<hex-encoded digest>`, computed over the raw JSON body.
115    ///
116    /// Uses the default [`RetryConfig`].
117    ///
118    /// # Panics
119    ///
120    /// Panics if the HTTP client cannot be built (TLS backend unavailable).
121    ///
122    /// # Examples
123    ///
124    /// ```
125    /// use ironflow_engine::notify::WebhookSubscriber;
126    ///
127    /// let subscriber = WebhookSubscriber::with_signing_secret(
128    ///     "https://example.com/hook",
129    ///     "my-secret",
130    /// );
131    /// assert!(subscriber.signing_secret().is_some());
132    /// ```
133    pub fn with_signing_secret(url: &str, secret: &str) -> Self {
134        Self::with_signing_secret_and_retry(url, secret, RetryConfig::default())
135    }
136
137    /// Create a webhook subscriber with HMAC-SHA256 signing and custom retry config.
138    ///
139    /// # Panics
140    ///
141    /// Panics if the HTTP client cannot be built (TLS backend unavailable).
142    ///
143    /// # Examples
144    ///
145    /// ```
146    /// use ironflow_engine::notify::{RetryConfig, WebhookSubscriber};
147    ///
148    /// let config = RetryConfig::new(
149    ///     5,
150    ///     std::time::Duration::from_secs(10),
151    ///     std::time::Duration::from_secs(1),
152    /// );
153    /// let subscriber = WebhookSubscriber::with_signing_secret_and_retry(
154    ///     "https://example.com/hook",
155    ///     "my-secret",
156    ///     config,
157    /// );
158    /// ```
159    pub fn with_signing_secret_and_retry(
160        url: &str,
161        secret: &str,
162        retry_config: RetryConfig,
163    ) -> Self {
164        Self::build(url, Some(secret), retry_config)
165    }
166
167    fn build(url: &str, signing_secret: Option<&str>, retry_config: RetryConfig) -> Self {
168        let client = retry_config.build_client();
169        Self {
170            url: url.to_string(),
171            signing_secret: signing_secret.map(|s| s.to_string()),
172            client,
173            retry_config,
174        }
175    }
176
177    /// Returns the target URL.
178    pub fn url(&self) -> &str {
179        &self.url
180    }
181
182    /// Returns the signing secret, if configured.
183    pub fn signing_secret(&self) -> Option<&str> {
184        self.signing_secret.as_deref()
185    }
186
187    /// Compute the `sha256=<hex>` signature string for a given body.
188    fn compute_signature(secret: &str, body: &[u8]) -> String {
189        let mut mac =
190            HmacSha256::new_from_slice(secret.as_bytes()).expect("HMAC accepts any key size");
191        mac.update(body);
192        format!("sha256={}", hex::encode(mac.finalize().into_bytes()))
193    }
194}
195
196impl EventSubscriber for WebhookSubscriber {
197    fn name(&self) -> &str {
198        "webhook"
199    }
200
201    fn handle<'a>(&'a self, event: &'a Event) -> SubscriberFuture<'a> {
202        Box::pin(async move {
203            let body = serde_json::to_vec(event).expect("Event is always serializable");
204            let signature = self
205                .signing_secret
206                .as_deref()
207                .map(|secret| Self::compute_signature(secret, &body));
208
209            deliver_with_retry(
210                &self.retry_config,
211                || {
212                    let mut req = self
213                        .client
214                        .post(&self.url)
215                        .header("Content-Type", "application/json")
216                        .body(body.clone());
217                    if let Some(sig) = &signature {
218                        req = req.header(SIGNATURE_HEADER, sig.as_str());
219                    }
220                    req
221                },
222                is_success_2xx,
223                "webhook",
224                &self.url,
225            )
226            .await;
227        })
228    }
229}
230
231#[cfg(test)]
232mod tests {
233    use std::collections::HashMap;
234    use std::sync::Arc;
235    use std::time::Duration;
236
237    use axum::Router;
238    use axum::body::Bytes;
239    use axum::http::{HeaderMap, StatusCode};
240    use axum::routing::post;
241    use chrono::Utc;
242    use hmac::{Hmac, Mac};
243    use ironflow_store::models::RunStatus;
244    use rust_decimal::Decimal;
245    use sha2::Sha256;
246    use tokio::net::TcpListener;
247    use tokio::sync::Mutex;
248    use uuid::Uuid;
249
250    use super::*;
251    use crate::notify::{RunCreatedEvent, RunStatusChangedEvent};
252
253    type HmacSha256 = Hmac<Sha256>;
254    type CapturedRequest = Arc<Mutex<Option<(HeaderMap, Vec<u8>)>>>;
255
256    fn compute_expected_hmac(secret: &[u8], body: &[u8]) -> String {
257        let mut mac = HmacSha256::new_from_slice(secret).expect("HMAC key rejected");
258        mac.update(body);
259        format!("sha256={}", hex::encode(mac.finalize().into_bytes()))
260    }
261
262    #[test]
263    fn url_accessor() {
264        let sub = WebhookSubscriber::new("https://example.com/hook");
265        assert_eq!(sub.url(), "https://example.com/hook");
266    }
267
268    #[test]
269    fn name_is_webhook() {
270        let sub = WebhookSubscriber::new("https://example.com");
271        assert_eq!(sub.name(), "webhook");
272    }
273
274    #[test]
275    fn no_signing_secret_by_default() {
276        let sub = WebhookSubscriber::new("https://example.com/hook");
277        assert!(sub.signing_secret().is_none());
278    }
279
280    #[test]
281    fn with_signing_secret_stores_secret() {
282        let sub = WebhookSubscriber::with_signing_secret("https://example.com/hook", "my-secret");
283        assert_eq!(sub.signing_secret(), Some("my-secret"));
284    }
285
286    #[test]
287    fn with_signing_secret_and_retry_stores_secret() {
288        let config = RetryConfig::new(5, Duration::from_secs(10), Duration::from_secs(1));
289        let sub = WebhookSubscriber::with_signing_secret_and_retry(
290            "https://example.com/hook",
291            "my-secret",
292            config,
293        );
294        assert_eq!(sub.signing_secret(), Some("my-secret"));
295        assert_eq!(sub.url(), "https://example.com/hook");
296    }
297
298    #[test]
299    fn compute_signature_matches_hmac_sha256() {
300        let secret = "test-secret";
301        let body = b"{\"type\":\"run_created\"}";
302        let sig = WebhookSubscriber::compute_signature(secret, body);
303        let expected = compute_expected_hmac(secret.as_bytes(), body);
304        assert_eq!(sig, expected);
305    }
306
307    #[test]
308    fn compute_signature_empty_body() {
309        let secret = "test-secret";
310        let body = b"";
311        let sig = WebhookSubscriber::compute_signature(secret, body);
312        let expected = compute_expected_hmac(secret.as_bytes(), body);
313        assert_eq!(sig, expected);
314    }
315
316    #[test]
317    fn compute_signature_has_sha256_prefix() {
318        let sig = WebhookSubscriber::compute_signature("secret", b"body");
319        assert!(sig.starts_with("sha256="));
320        assert_eq!(sig.len(), 7 + 64); // "sha256=" + 64 hex chars
321    }
322
323    #[test]
324    fn compute_signature_rfc4231_test_vector() {
325        // RFC 4231 Test Case 2: Key = "Jefe", Data = "what do ya want for nothing?"
326        let key = "Jefe";
327        let data = b"what do ya want for nothing?";
328        let expected = "5bdcc146bf60754e6a042426089575c75a003f089d2739839dec58b964ec3843";
329
330        let sig = WebhookSubscriber::compute_signature(key, data);
331        assert_eq!(sig, format!("sha256={}", expected));
332    }
333
334    #[tokio::test]
335    async fn unsigned_webhook_does_not_send_signature_header() {
336        let received_headers: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
337        let captured = received_headers.clone();
338
339        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
340        let addr = listener.local_addr().unwrap();
341
342        let app = Router::new().route(
343            "/",
344            post(move |headers: HeaderMap, _body: Bytes| {
345                let captured = captured.clone();
346                async move {
347                    *captured.lock().await = Some(headers);
348                    StatusCode::OK
349                }
350            }),
351        );
352        tokio::spawn(async move {
353            axum::serve(listener, app).await.unwrap();
354        });
355
356        let sub = WebhookSubscriber::new(&format!("http://{}", addr));
357        let event = Event::RunCreated(RunCreatedEvent {
358            run_id: Uuid::now_v7(),
359            workflow_name: "deploy".to_string(),
360            at: Utc::now(),
361        });
362
363        sub.handle(&event).await;
364
365        let headers = received_headers.lock().await;
366        let headers = headers.as_ref().expect("request was received");
367        assert!(headers.get("X-Signature-256").is_none());
368    }
369
370    #[tokio::test]
371    async fn signed_webhook_sends_valid_signature_header() {
372        let secret = "webhook-secret-42";
373
374        let received: CapturedRequest = Arc::new(Mutex::new(None));
375        let captured = received.clone();
376
377        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
378        let addr = listener.local_addr().unwrap();
379
380        let app = Router::new().route(
381            "/",
382            post(move |headers: HeaderMap, body: Bytes| {
383                let captured = captured.clone();
384                async move {
385                    *captured.lock().await = Some((headers, body.to_vec()));
386                    StatusCode::OK
387                }
388            }),
389        );
390        tokio::spawn(async move {
391            axum::serve(listener, app).await.unwrap();
392        });
393
394        let sub = WebhookSubscriber::with_signing_secret(&format!("http://{}", addr), secret);
395        let event = Event::RunStatusChanged(RunStatusChangedEvent {
396            run_id: Uuid::now_v7(),
397            workflow_name: "deploy".to_string(),
398            from: RunStatus::Pending,
399            to: RunStatus::Running,
400            error: None,
401            cost_usd: Decimal::ZERO,
402            duration_ms: 0,
403            labels: HashMap::new(),
404            at: Utc::now(),
405        });
406
407        sub.handle(&event).await;
408
409        let guard = received.lock().await;
410        let (headers, body) = guard.as_ref().expect("request was received");
411
412        let sig_header = headers
413            .get("X-Signature-256")
414            .expect("X-Signature-256 header must be present")
415            .to_str()
416            .unwrap();
417
418        assert!(sig_header.starts_with("sha256="));
419
420        // Verify the signature is correct
421        let expected = compute_expected_hmac(secret.as_bytes(), body);
422        assert_eq!(sig_header, expected);
423    }
424
425    #[test]
426    fn different_secrets_produce_different_signatures() {
427        let body = b"{\"type\":\"run_created\"}";
428        let sig_a = WebhookSubscriber::compute_signature("secret-A", body);
429        let sig_b = WebhookSubscriber::compute_signature("secret-B", body);
430        assert_ne!(sig_a, sig_b);
431    }
432
433    #[test]
434    fn wrong_secret_does_not_match() {
435        let body = b"{\"type\":\"run_created\"}";
436        let sig = WebhookSubscriber::compute_signature("correct-secret", body);
437        let wrong = compute_expected_hmac(b"wrong-secret", body);
438        assert_ne!(sig, wrong);
439    }
440}
ironflow_engine/notify/webhook.rs

ironflow_engine/notify/
webhook.rs
