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    /// Send webhook with retry logic
101    async fn send_with_retry(&self, url: &Url, json: &str, auth: Option<&str>) -> Result<(), WebhookError> {
102        let mut last_error = None;
103
104        for attempt in 0..=self.max_retries {
105            if attempt > 0 {
106                let delay = self.retry_delay_ms * 2u64.pow(attempt - 1); // Exponential backoff
107                debug!("Retrying webhook delivery after {}ms (attempt {})", delay, attempt);
108                tokio::time::sleep(Duration::from_millis(delay)).await;
109            }
110
111            match self.send_request(url, json, auth).await {
112                Ok(()) => {
113                    debug!("Webhook delivered successfully");
114                    return Ok(());
115                }
116                Err(e) => {
117                    warn!("Webhook delivery attempt {} failed: {}", attempt + 1, e);
118                    last_error = Some(e);
119                }
120            }
121        }
122
123        Err(last_error.unwrap_or(WebhookError::Unknown))
124    }
125
126    /// Send a single HTTP request
127    async fn send_request(&self, url: &Url, json: &str, auth: Option<&str>) -> Result<(), WebhookError> {
128        let Some(client) = self.client.as_ref() else {
129            return Err(WebhookError::ClientUnavailable);
130        };
131
132        let mut request = client
133            .post(url.clone())
134            .header("Content-Type", "application/json")
135            .header("User-Agent", "VT Code-A2A/1.0");
136
137        if let Some(auth_header) = auth {
138            request = request.header("Authorization", auth_header);
139        }
140
141        let response = request
142            .body(json.to_string())
143            .send()
144            .await
145            .map_err(|e| WebhookError::Network(e.to_string()))?;
146
147        if response.status().is_success() {
148            Ok(())
149        } else {
150            Err(WebhookError::HttpError(response.status().as_u16()))
151        }
152    }
153}
154
155/// Webhook delivery errors
156#[derive(Debug, Clone, thiserror::Error)]
157pub enum WebhookError {
158    /// Network error
159    #[error("Network error: {0}")]
160    Network(String),
161    /// HTTP error status code
162    #[error("HTTP error: {0}")]
163    HttpError(u16),
164    /// JSON serialization error
165    #[error("Serialization error: {0}")]
166    Serialization(String),
167    /// Webhook URL failed validation
168    #[error("Invalid webhook URL: {0}")]
169    InvalidUrl(String),
170    /// HTTP client could not be configured safely
171    #[error("Webhook HTTP client is unavailable")]
172    ClientUnavailable,
173    /// Unknown error
174    #[error("Unknown error")]
175    Unknown,
176}
177
178#[cfg(test)]
179mod tests {
180    use super::*;
181    use crate::types::{TaskState, TaskStatus};
182
183    #[test]
184    fn test_webhook_notifier_creation() {
185        let notifier = WebhookNotifier::new();
186        assert_eq!(notifier.max_retries, 3);
187        assert_eq!(notifier.retry_delay_ms, 1000);
188    }
189
190    #[test]
191    fn test_webhook_notifier_with_settings() {
192        let notifier = WebhookNotifier::with_settings(5, 2000);
193        assert_eq!(notifier.max_retries, 5);
194        assert_eq!(notifier.retry_delay_ms, 2000);
195    }
196
197    #[test]
198    fn test_parse_webhook_url_requires_safe_scheme_and_host() {
199        for url in [
200            "https://example.com/webhook",
201            "http://localhost/webhook",
202            "http://127.0.0.1:8080/webhook",
203            "http://[::1]:8080/webhook",
204        ] {
205            assert!(parse_webhook_url(url).is_ok(), "URL should be accepted: {url}");
206        }
207
208        for url in [
209            "http://localhost.evil.example/webhook",
210            "http://localhost@evil.example/webhook",
211            "http://example.com/webhook",
212            "ftp://example.com/webhook",
213            "https://user:password@example.com/webhook",
214            "https://example.com/webhook#fragment",
215        ] {
216            assert!(parse_webhook_url(url).is_err(), "URL should be rejected: {url}");
217        }
218    }
219
220    #[tokio::test]
221    async fn test_webhook_error_display() {
222        let err = WebhookError::Network("Connection refused".to_string());
223        assert!(err.to_string().contains("Network error"));
224
225        let err = WebhookError::HttpError(404);
226        assert!(err.to_string().contains("404"));
227    }
228
229    #[tokio::test]
230    async fn test_send_event_serialization() {
231        let notifier = WebhookNotifier::new();
232        let config = TaskPushNotificationConfig {
233            task_id: "task-1".to_string(),
234            url: "https://example.com/webhook".to_string(),
235            authentication: None,
236        };
237
238        let event = StreamingEvent::TaskStatus {
239            task_id: "task-1".to_string(),
240            context_id: None,
241            status: TaskStatus::new(TaskState::Completed),
242            kind: "status-update".to_string(),
243            r#final: true,
244        };
245
246        // This will fail with network error since the URL doesn't exist,
247        // but we're testing that serialization works
248        let result = notifier.send_event(&config, event).await;
249        assert!(result.is_err());
250
251        if let Err(WebhookError::Serialization(_)) = result {
252            panic!("Unexpected serialization error");
253        }
254    }
255
256    #[tokio::test]
257    async fn test_send_event_rejects_invalid_url_before_network_access() {
258        let notifier = WebhookNotifier::new();
259        let config = TaskPushNotificationConfig {
260            task_id: "task-1".to_string(),
261            url: "http://localhost.evil.example/webhook".to_string(),
262            authentication: None,
263        };
264        let event = StreamingEvent::Unknown;
265
266        let result = notifier.send_event(&config, event).await;
267        assert!(matches!(result, Err(WebhookError::InvalidUrl(_))));
268    }
269
270    #[tokio::test]
271    async fn test_webhook_client_does_not_follow_redirects() {
272        use tokio::io::{AsyncReadExt, AsyncWriteExt};
273        use tokio::net::TcpListener;
274
275        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind test listener");
276        let address = listener.local_addr().expect("read test listener address");
277        let server = tokio::spawn(async move {
278            let (mut socket, _) = listener.accept().await.expect("accept initial webhook");
279            let mut request = [0; 1024];
280            let _ignored = socket.read(&mut request).await.expect("read initial webhook");
281            let redirect = format!(
282                "HTTP/1.1 302 Found\r\nLocation: http://{address}/redirect-target\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
283            );
284            socket.write_all(redirect.as_bytes()).await.expect("send redirect");
285
286            match tokio::time::timeout(Duration::from_millis(250), listener.accept()).await {
287                Ok(Ok((mut redirected_socket, _))) => {
288                    let mut request = [0; 1024];
289                    let _ignored = redirected_socket.read(&mut request).await.expect("read redirected webhook");
290                    let response = b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
291                    redirected_socket.write_all(response).await.expect("send redirected response");
292                    true
293                }
294                Ok(Err(_)) | Err(_) => false,
295            }
296        });
297
298        let notifier = WebhookNotifier::with_settings(0, 0);
299        let config = TaskPushNotificationConfig {
300            task_id: "task-1".to_string(),
301            url: format!("http://{address}/hook"),
302            authentication: None,
303        };
304        let result = notifier.send_event(&config, StreamingEvent::Unknown).await;
305        assert!(matches!(result, Err(WebhookError::HttpError(302))));
306        assert!(!server.await.expect("join test server"));
307    }
308}
vtcode_a2a/webhook.rs

vtcode_a2a/
webhook.rs
