vtcode_a2a/
webhook.rs

1//! Webhook delivery for A2A push notifications
2//!
3//! Handles HTTP POST delivery of streaming events to configured webhook URLs
4//! with retry logic, authentication, and error handling.
5
6use super::rpc::{SendStreamingMessageResponse, StreamingEvent, TaskPushNotificationConfig};
7use reqwest::Client;
8use std::time::Duration;
9use tracing::{debug, warn};
10use url::{Host, Url};
11
12/// Parse and validate a webhook destination before it is stored or requested.
13///
14/// HTTPS is required for remote destinations. Plain HTTP is limited to the
15/// exact localhost name or an IP loopback address, never a hostname suffix.
16pub(crate) fn parse_webhook_url(raw_url: &str) -> Result<Url, String> {
17    let url = Url::parse(raw_url).map_err(|error| format!("Webhook URL must be a valid absolute URL: {error}"))?;
18
19    if url.username() != "" || url.password().is_some() {
20        return Err("Webhook URL must not contain credentials".to_string());
21    }
22    if url.fragment().is_some() {
23        return Err("Webhook URL must not contain a fragment".to_string());
24    }
25
26    let is_loopback = matches!(
27        url.host(),
28        Some(Host::Domain(host)) if host.eq_ignore_ascii_case("localhost")
29    ) || matches!(url.host(), Some(Host::Ipv4(address)) if address.is_loopback())
30        || matches!(url.host(), Some(Host::Ipv6(address)) if address.is_loopback());
31
32    match url.scheme() {
33        "https" => Ok(url),
34        "http" if is_loopback => Ok(url),
35        "http" => Err("Webhook URL must use HTTPS unless it targets localhost".to_string()),
36        _ => Err("Webhook URL must use HTTPS or HTTP localhost".to_string()),
37    }
38}
39
40/// Webhook notifier for delivering A2A events
41#[derive(Debug, Clone)]
42pub struct WebhookNotifier {
43    client: Option<Client>,
44    max_retries: u32,
45    retry_delay_ms: u64,
46}
47
48impl Default for WebhookNotifier {
49    fn default() -> Self {
50        Self::new()
51    }
52}
53
54impl WebhookNotifier {
55    fn build_http_client() -> Option<Client> {
56        match Client::builder()
57            .timeout(Duration::from_secs(10))
58            .redirect(reqwest::redirect::Policy::none())
59            .build()
60        {
61            Ok(client) => Some(client),
62            Err(error) => {
63                warn!(error = %error, "Failed to configure webhook HTTP client; webhook delivery disabled");
64                None
65            }
66        }
67    }
68
69    /// Create a new webhook notifier with default settings
70    pub(crate) fn new() -> Self {
71        Self {
72            client: Self::build_http_client(),
73            max_retries: 3,
74            retry_delay_ms: 1000,
75        }
76    }
77
78    /// Create a webhook notifier with custom settings
79    fn with_settings(max_retries: u32, retry_delay_ms: u64) -> Self {
80        Self {
81            client: Self::build_http_client(),
82            max_retries,
83            retry_delay_ms,
84        }
85    }
86
87    /// Deliver a streaming event to a webhook URL
88    pub(crate) async fn send_event(
89        &self,
90        config: &TaskPushNotificationConfig,
91        event: StreamingEvent,
92    ) -> Result<(), WebhookError> {
93        let response = SendStreamingMessageResponse { event };
94        let json = serde_json::to_string(&response).map_err(|e| WebhookError::Serialization(e.to_string()))?;
95        let url = parse_webhook_url(&config.url).map_err(WebhookError::InvalidUrl)?;
96
97        self.send_with_retry(&url, &json, config.authentication.as_deref()).await
98    }
99
100    /// Retry delay before `attempt` (1-based) with unbounded exponential growth
101    /// guarded against overflow.
102    fn retry_delay(&self, attempt: u32) -> Duration {
103        let exponent = attempt.saturating_sub(1);
104        Duration::from_millis(self.retry_delay_ms.saturating_mul(2_u64.saturating_pow(exponent)))
105    }
106
107    /// Send webhook with retry logic
108    async fn send_with_retry(&self, url: &Url, json: &str, auth: Option<&str>) -> Result<(), WebhookError> {
109        let mut last_error = None;
110
111        for attempt in 0..=self.max_retries {
112            if attempt > 0 {
113                let delay = self.retry_delay(attempt); // Exponential backoff
114                debug!("Retrying webhook delivery after {:?} (attempt {})", delay, attempt);
115                tokio::time::sleep(delay).await;
116            }
117
118            match self.send_request(url, json, auth).await {
119                Ok(()) => {
120                    debug!("Webhook delivered successfully");
121                    return Ok(());
122                }
123                Err(e) => {
124                    warn!("Webhook delivery attempt {} failed: {}", attempt + 1, e);
125                    last_error = Some(e);
126                }
127            }
128        }
129
130        Err(last_error.unwrap_or(WebhookError::Unknown))
131    }
132
133    /// Send a single HTTP request
134    async fn send_request(&self, url: &Url, json: &str, auth: Option<&str>) -> Result<(), WebhookError> {
135        let Some(client) = self.client.as_ref() else {
136            return Err(WebhookError::ClientUnavailable);
137        };
138
139        let mut request = client
140            .post(url.clone())
141            .header("Content-Type", "application/json")
142            .header("User-Agent", "VT Code-A2A/1.0");
143
144        if let Some(auth_header) = auth {
145            request = request.header("Authorization", auth_header);
146        }
147
148        let response = request
149            .body(json.to_string())
150            .send()
151            .await
152            .map_err(|e| WebhookError::Network(e.to_string()))?;
153
154        if response.status().is_success() {
155            Ok(())
156        } else {
157            Err(WebhookError::HttpError(response.status().as_u16()))
158        }
159    }
160}
161
162/// Webhook delivery errors
163#[derive(Debug, Clone, thiserror::Error)]
164pub enum WebhookError {
165    /// Network error
166    #[error("Network error: {0}")]
167    Network(String),
168    /// HTTP error status code
169    #[error("HTTP error: {0}")]
170    HttpError(u16),
171    /// JSON serialization error
172    #[error("Serialization error: {0}")]
173    Serialization(String),
174    /// Webhook URL failed validation
175    #[error("Invalid webhook URL: {0}")]
176    InvalidUrl(String),
177    /// HTTP client could not be configured safely
178    #[error("Webhook HTTP client is unavailable")]
179    ClientUnavailable,
180    /// Unknown error
181    #[error("Unknown error")]
182    Unknown,
183}
184
185#[cfg(test)]
186mod tests {
187    use super::*;
188    use crate::types::{TaskState, TaskStatus};
189
190    #[test]
191    fn test_webhook_notifier_creation() {
192        let notifier = WebhookNotifier::new();
193        assert_eq!(notifier.max_retries, 3);
194        assert_eq!(notifier.retry_delay_ms, 1000);
195    }
196
197    #[test]
198    fn test_webhook_notifier_with_settings() {
199        let notifier = WebhookNotifier::with_settings(5, 2000);
200        assert_eq!(notifier.max_retries, 5);
201        assert_eq!(notifier.retry_delay_ms, 2000);
202    }
203
204    #[test]
205    fn test_webhook_retry_delay_doubles_and_saturates() {
206        let notifier = WebhookNotifier::with_settings(3, 1_000);
207        assert_eq!(notifier.retry_delay(1), Duration::from_millis(1_000));
208        assert_eq!(notifier.retry_delay(2), Duration::from_millis(2_000));
209        assert_eq!(notifier.retry_delay(3), Duration::from_millis(4_000));
210        assert_eq!(
211            notifier.retry_delay(u32::MAX),
212            Duration::from_millis(u64::MAX),
213            "an overflowing exponent must saturate instead of panicking"
214        );
215    }
216
217    #[test]
218    fn test_parse_webhook_url_requires_safe_scheme_and_host() {
219        for url in [
220            "https://example.com/webhook",
221            "http://localhost/webhook",
222            "http://127.0.0.1:8080/webhook",
223            "http://[::1]:8080/webhook",
224        ] {
225            assert!(parse_webhook_url(url).is_ok(), "URL should be accepted: {url}");
226        }
227
228        for url in [
229            "http://localhost.evil.example/webhook",
230            "http://localhost@evil.example/webhook",
231            "http://example.com/webhook",
232            "ftp://example.com/webhook",
233            "https://user:password@example.com/webhook",
234            "https://example.com/webhook#fragment",
235        ] {
236            assert!(parse_webhook_url(url).is_err(), "URL should be rejected: {url}");
237        }
238    }
239
240    #[tokio::test]
241    async fn test_webhook_error_display() {
242        let err = WebhookError::Network("Connection refused".to_string());
243        assert!(err.to_string().contains("Network error"));
244
245        let err = WebhookError::HttpError(404);
246        assert!(err.to_string().contains("404"));
247    }
248
249    #[tokio::test]
250    async fn test_send_event_serialization() {
251        let notifier = WebhookNotifier::new();
252        let config = TaskPushNotificationConfig {
253            task_id: "task-1".to_string(),
254            url: "https://example.com/webhook".to_string(),
255            authentication: None,
256        };
257
258        let event = StreamingEvent::TaskStatus {
259            task_id: "task-1".to_string(),
260            context_id: None,
261            status: TaskStatus::new(TaskState::Completed),
262            kind: "status-update".to_string(),
263            r#final: true,
264        };
265
266        // This will fail with network error since the URL doesn't exist,
267        // but we're testing that serialization works
268        let result = notifier.send_event(&config, event).await;
269        assert!(result.is_err());
270
271        if let Err(WebhookError::Serialization(_)) = result {
272            panic!("Unexpected serialization error");
273        }
274    }
275
276    #[tokio::test]
277    async fn test_send_event_rejects_invalid_url_before_network_access() {
278        let notifier = WebhookNotifier::new();
279        let config = TaskPushNotificationConfig {
280            task_id: "task-1".to_string(),
281            url: "http://localhost.evil.example/webhook".to_string(),
282            authentication: None,
283        };
284        let event = StreamingEvent::Unknown;
285
286        let result = notifier.send_event(&config, event).await;
287        assert!(matches!(result, Err(WebhookError::InvalidUrl(_))));
288    }
289
290    #[tokio::test]
291    async fn test_webhook_client_does_not_follow_redirects() {
292        use tokio::io::{AsyncReadExt, AsyncWriteExt};
293        use tokio::net::TcpListener;
294
295        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind test listener");
296        let address = listener.local_addr().expect("read test listener address");
297        let server = tokio::spawn(async move {
298            let (mut socket, _) = listener.accept().await.expect("accept initial webhook");
299            let mut request = [0; 1024];
300            let _ignored = socket.read(&mut request).await.expect("read initial webhook");
301            let redirect = format!(
302                "HTTP/1.1 302 Found\r\nLocation: http://{address}/redirect-target\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
303            );
304            socket.write_all(redirect.as_bytes()).await.expect("send redirect");
305
306            match tokio::time::timeout(Duration::from_millis(250), listener.accept()).await {
307                Ok(Ok((mut redirected_socket, _))) => {
308                    let mut request = [0; 1024];
309                    let _ignored = redirected_socket.read(&mut request).await.expect("read redirected webhook");
310                    let response = b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
311                    redirected_socket.write_all(response).await.expect("send redirected response");
312                    true
313                }
314                Ok(Err(_)) | Err(_) => false,
315            }
316        });
317
318        let notifier = WebhookNotifier::with_settings(0, 0);
319        let config = TaskPushNotificationConfig {
320            task_id: "task-1".to_string(),
321            url: format!("http://{address}/hook"),
322            authentication: None,
323        };
324        let result = notifier.send_event(&config, StreamingEvent::Unknown).await;
325        assert!(matches!(result, Err(WebhookError::HttpError(302))));
326        assert!(!server.await.expect("join test server"));
327    }
328}
vtcode_a2a/webhook.rs

vtcode_a2a/
webhook.rs
