1#![allow(deprecated)] use async_trait::async_trait;
64use chrono::{DateTime, Utc};
65use ipnet::IpNet;
66use rand::Rng;
67use serde::{Deserialize, Serialize};
68use std::collections::HashMap;
69use std::net::IpAddr;
70use std::time::Duration;
71use thiserror::Error;
72use url::Url;
73
74use crate::TaskId;
75
76#[derive(Debug, Error)]
87pub enum WebhookError {
88 #[error("Webhook request failed: {0}")]
98 RequestFailed(String),
99
100 #[error("Max retries exceeded for webhook")]
111 MaxRetriesExceeded,
112
113 #[error("Webhook serialization error: {0}")]
123 SerializationError(String),
124
125 #[error("Invalid webhook URL: {0}")]
135 InvalidUrl(String),
136
137 #[error("URL scheme not allowed: {0}. Only HTTPS is permitted for webhooks")]
147 SchemeNotAllowed(String),
148
149 #[error("Webhook URL resolves to blocked IP address: {0}")]
159 BlockedIpAddress(String),
160
161 #[error("DNS resolution failed for webhook URL host: {0}")]
171 DnsResolutionFailed(String),
172}
173
174const BLOCKED_IP_RANGES: &[&str] = &[
182 "127.0.0.0/8",
184 "10.0.0.0/8",
186 "172.16.0.0/12",
187 "192.168.0.0/16",
188 "169.254.0.0/16",
190 "::1/128",
192 "fe80::/10",
194 "fc00::/7",
196];
197
198pub fn is_blocked_ip(ip: &IpAddr) -> bool {
221 BLOCKED_IP_RANGES.iter().any(|range| {
222 range
223 .parse::<IpNet>()
224 .map(|net| net.contains(ip))
225 .unwrap_or(false)
226 })
227}
228
229pub fn validate_webhook_url(url_str: &str) -> Result<Url, WebhookError> {
262 let parsed_url =
264 Url::parse(url_str).map_err(|e| WebhookError::InvalidUrl(format!("{}: {}", url_str, e)))?;
265
266 if parsed_url.scheme() != "https" {
268 return Err(WebhookError::SchemeNotAllowed(
269 parsed_url.scheme().to_string(),
270 ));
271 }
272
273 let host = parsed_url
275 .host_str()
276 .ok_or_else(|| WebhookError::InvalidUrl("URL has no host".to_string()))?;
277
278 let host_for_parse = host
282 .strip_prefix('[')
283 .and_then(|s| s.strip_suffix(']'))
284 .unwrap_or(host);
285
286 if let Ok(ip) = host_for_parse.parse::<IpAddr>() {
287 if is_blocked_ip(&ip) {
288 return Err(WebhookError::BlockedIpAddress(ip.to_string()));
289 }
290 return Ok(parsed_url);
291 }
292
293 let host_lower = host.to_lowercase();
296 if host_lower == "localhost" || host_lower.ends_with(".localhost") {
297 return Err(WebhookError::BlockedIpAddress("localhost".to_string()));
298 }
299
300 if host_lower.ends_with(".internal") || host_lower.ends_with(".local") {
302 return Err(WebhookError::BlockedIpAddress(format!(
303 "internal hostname: {}",
304 host
305 )));
306 }
307
308 Ok(parsed_url)
309}
310
311pub async fn validate_resolved_ips(url: &Url) -> Result<(), WebhookError> {
324 let host = url
325 .host_str()
326 .ok_or_else(|| WebhookError::InvalidUrl("URL has no host".to_string()))?;
327
328 let host_for_parse = host
331 .strip_prefix('[')
332 .and_then(|s| s.strip_suffix(']'))
333 .unwrap_or(host);
334
335 if host_for_parse.parse::<IpAddr>().is_ok() {
336 return Ok(());
337 }
338
339 let port = url.port().unwrap_or(443);
340
341 let addrs = tokio::net::lookup_host(format!("{}:{}", host, port))
343 .await
344 .map_err(|e| WebhookError::DnsResolutionFailed(format!("{}: {}", host, e)))?;
345
346 for addr in addrs {
348 if is_blocked_ip(&addr.ip()) {
349 return Err(WebhookError::BlockedIpAddress(format!(
350 "{} resolves to {}",
351 host,
352 addr.ip()
353 )));
354 }
355 }
356
357 Ok(())
358}
359
360#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
371#[serde(rename_all = "lowercase")]
372pub enum TaskStatus {
373 Success,
375 Failed,
377 Cancelled,
379}
380
381#[derive(Debug, Clone, Serialize, Deserialize)]
408pub struct WebhookEvent {
409 pub task_id: TaskId,
411 pub task_name: String,
413 pub status: TaskStatus,
415 pub result: Option<String>,
417 pub error: Option<String>,
419 pub started_at: DateTime<Utc>,
421 pub completed_at: DateTime<Utc>,
423 pub duration_ms: u64,
425}
426
427#[deprecated(
446 since = "0.2.0",
447 note = "Use `WebhookRetrySettings` with the `#[settings]` macro instead."
448)]
449#[derive(Debug, Clone)]
450pub struct RetryConfig {
451 pub max_retries: u32,
453 pub initial_backoff: Duration,
455 pub max_backoff: Duration,
457 pub backoff_multiplier: f64,
459}
460
461impl Default for RetryConfig {
462 fn default() -> Self {
463 Self {
464 max_retries: 3,
465 initial_backoff: Duration::from_millis(100),
466 max_backoff: Duration::from_secs(30),
467 backoff_multiplier: 2.0,
468 }
469 }
470}
471
472#[deprecated(
496 since = "0.2.0",
497 note = "Use `WebhookSettings` with the `#[settings]` macro instead."
498)]
499#[derive(Debug, Clone)]
500pub struct WebhookConfig {
501 pub url: String,
503 pub method: String,
505 pub headers: HashMap<String, String>,
507 pub timeout: Duration,
509 pub retry_config: RetryConfig,
511}
512
513impl Default for WebhookConfig {
514 fn default() -> Self {
515 Self {
516 url: String::new(),
517 method: "POST".to_string(),
518 headers: HashMap::new(),
519 timeout: Duration::from_secs(5),
520 retry_config: RetryConfig::default(),
521 }
522 }
523}
524
525#[async_trait]
554pub trait WebhookSender: Send + Sync {
555 async fn send(&self, event: &WebhookEvent) -> Result<(), WebhookError>;
557}
558
559pub struct HttpWebhookSender {
578 client: reqwest::Client,
579 config: WebhookConfig,
580}
581
582impl HttpWebhookSender {
583 pub fn new(config: WebhookConfig) -> Self {
593 let client = reqwest::Client::builder()
594 .timeout(config.timeout)
595 .build()
596 .unwrap_or_else(|_| reqwest::Client::new());
597
598 Self { client, config }
599 }
600
601 pub fn calculate_backoff(&self, retry_count: u32) -> Duration {
622 let retry_config = &self.config.retry_config;
623
624 let backoff_ms = retry_config.initial_backoff.as_millis() as f64
626 * retry_config.backoff_multiplier.powi(retry_count as i32);
627
628 let mut rng = rand::rng();
630 let jitter = rng.random_range(-0.25..=0.25);
631 let backoff_with_jitter = backoff_ms * (1.0 + jitter);
632
633 let capped_backoff = backoff_with_jitter.min(retry_config.max_backoff.as_millis() as f64);
635
636 Duration::from_millis(capped_backoff.max(0.0) as u64)
637 }
638
639 async fn send_with_retry(&self, event: &WebhookEvent) -> Result<(), WebhookError> {
641 let mut retry_count = 0;
642 let max_retries = self.config.retry_config.max_retries;
643
644 loop {
645 match self.send_request(event).await {
646 Ok(_) => return Ok(()),
647 Err(e) => {
648 if retry_count >= max_retries {
649 return Err(WebhookError::MaxRetriesExceeded);
650 }
651
652 let backoff = self.calculate_backoff(retry_count);
653 tracing::warn!(
654 attempt = retry_count + 1,
655 max_attempts = max_retries + 1,
656 error = %e,
657 backoff = ?backoff,
658 "Webhook request failed, retrying"
659 );
660
661 tokio::time::sleep(backoff).await;
663 retry_count += 1;
664 }
665 }
666 }
667 }
668
669 async fn send_request(&self, event: &WebhookEvent) -> Result<(), WebhookError> {
671 let json_body = serde_json::to_string(event)
672 .map_err(|e| WebhookError::SerializationError(e.to_string()))?;
673
674 let mut request = match self.config.method.to_uppercase().as_str() {
675 "POST" => self.client.post(&self.config.url),
676 "PUT" => self.client.put(&self.config.url),
677 "PATCH" => self.client.patch(&self.config.url),
678 _ => self.client.post(&self.config.url),
679 };
680
681 for (key, value) in &self.config.headers {
683 request = request.header(key, value);
684 }
685
686 let response = request
688 .header("Content-Type", "application/json")
689 .body(json_body)
690 .send()
691 .await
692 .map_err(|e| WebhookError::RequestFailed(e.to_string()))?;
693
694 if !response.status().is_success() {
696 return Err(WebhookError::RequestFailed(format!(
697 "HTTP {}: {}",
698 response.status(),
699 response
700 .text()
701 .await
702 .unwrap_or_else(|_| "No response body".to_string())
703 )));
704 }
705
706 Ok(())
707 }
708}
709
710#[async_trait]
711impl WebhookSender for HttpWebhookSender {
712 async fn send(&self, event: &WebhookEvent) -> Result<(), WebhookError> {
713 let validated_url = validate_webhook_url(&self.config.url)?;
715 validate_resolved_ips(&validated_url).await?;
716
717 self.send_with_retry(event).await
718 }
719}
720
721#[cfg(test)]
722mod tests {
723 use super::*;
724 use rstest::rstest;
725 use std::time::Duration;
726
727 #[rstest]
728 fn test_task_status_serialization() {
729 let status = TaskStatus::Success;
731
732 let json = serde_json::to_string(&status).unwrap();
734
735 assert_eq!(json, r#""success""#);
737
738 let status: TaskStatus = serde_json::from_str(r#""failed""#).unwrap();
740
741 assert_eq!(status, TaskStatus::Failed);
743 }
744
745 #[rstest]
746 fn test_webhook_event_serialization() {
747 let now = Utc::now();
749 let event = WebhookEvent {
750 task_id: TaskId::new(),
751 task_name: "test_task".to_string(),
752 status: TaskStatus::Success,
753 result: Some("OK".to_string()),
754 error: None,
755 started_at: now,
756 completed_at: now,
757 duration_ms: 1000,
758 };
759
760 let json = serde_json::to_string(&event).unwrap();
762
763 assert!(json.contains("test_task"));
765 assert!(json.contains(r#""status":"success""#));
766
767 let deserialized: WebhookEvent = serde_json::from_str(&json).unwrap();
769
770 assert_eq!(deserialized.task_name, "test_task");
772 assert_eq!(deserialized.status, TaskStatus::Success);
773 }
774
775 #[rstest]
776 fn test_retry_config_default() {
777 let config = RetryConfig::default();
779
780 assert_eq!(config.max_retries, 3);
782 assert_eq!(config.initial_backoff, Duration::from_millis(100));
783 assert_eq!(config.max_backoff, Duration::from_secs(30));
784 assert_eq!(config.backoff_multiplier, 2.0);
785 }
786
787 #[rstest]
788 fn test_webhook_config_default() {
789 let config = WebhookConfig::default();
791
792 assert_eq!(config.url, "");
794 assert_eq!(config.method, "POST");
795 assert_eq!(config.timeout, Duration::from_secs(5));
796 assert!(config.headers.is_empty());
797 }
798
799 #[rstest]
800 fn test_calculate_backoff() {
801 let config = WebhookConfig {
803 url: "https://example.com".to_string(),
804 method: "POST".to_string(),
805 headers: HashMap::new(),
806 timeout: Duration::from_secs(5),
807 retry_config: RetryConfig {
808 max_retries: 3,
809 initial_backoff: Duration::from_millis(100),
810 max_backoff: Duration::from_secs(10),
811 backoff_multiplier: 2.0,
812 },
813 };
814 let sender = HttpWebhookSender::new(config);
815
816 let backoff0 = sender.calculate_backoff(0);
818 let backoff1 = sender.calculate_backoff(1);
819 let backoff2 = sender.calculate_backoff(2);
820
821 assert!(backoff0.as_millis() >= 75 && backoff0.as_millis() <= 125); assert!(backoff1.as_millis() >= 150 && backoff1.as_millis() <= 250); assert!(backoff2.as_millis() >= 300 && backoff2.as_millis() <= 500); let backoff_large = sender.calculate_backoff(100);
828 assert!(backoff_large <= Duration::from_secs(10));
829 }
830
831 #[rstest]
832 fn test_webhook_error_display() {
833 let error = WebhookError::RequestFailed("Connection timeout".to_string());
835 assert_eq!(
836 error.to_string(),
837 "Webhook request failed: Connection timeout"
838 );
839
840 let error = WebhookError::MaxRetriesExceeded;
841 assert_eq!(error.to_string(), "Max retries exceeded for webhook");
842
843 let error = WebhookError::SerializationError("Invalid JSON".to_string());
844 assert_eq!(
845 error.to_string(),
846 "Webhook serialization error: Invalid JSON"
847 );
848 }
849
850 #[rstest]
851 #[tokio::test]
852 async fn test_http_webhook_sender_creation() {
853 let config = WebhookConfig::default();
855 let sender = HttpWebhookSender::new(config);
856
857 assert_eq!(sender.config.method, "POST");
859 }
860
861 #[rstest]
862 #[tokio::test]
863 async fn test_webhook_event_creation() {
864 let now = Utc::now();
866 let started = now - chrono::Duration::seconds(5);
867
868 let event = WebhookEvent {
870 task_id: TaskId::new(),
871 task_name: "test_task".to_string(),
872 status: TaskStatus::Success,
873 result: Some("Task completed successfully".to_string()),
874 error: None,
875 started_at: started,
876 completed_at: now,
877 duration_ms: 5000,
878 };
879
880 assert_eq!(event.task_name, "test_task");
882 assert_eq!(event.status, TaskStatus::Success);
883 assert!(event.result.is_some());
884 assert!(event.error.is_none());
885 assert_eq!(event.duration_ms, 5000);
886 }
887
888 #[rstest]
889 #[tokio::test]
890 async fn test_webhook_failed_event() {
891 let now = Utc::now();
893
894 let event = WebhookEvent {
896 task_id: TaskId::new(),
897 task_name: "failed_task".to_string(),
898 status: TaskStatus::Failed,
899 result: None,
900 error: Some("Database connection failed".to_string()),
901 started_at: now,
902 completed_at: now,
903 duration_ms: 100,
904 };
905
906 assert_eq!(event.status, TaskStatus::Failed);
908 assert!(event.result.is_none());
909 assert!(event.error.is_some());
910 assert_eq!(
911 event.error.unwrap(),
912 "Database connection failed".to_string()
913 );
914 }
915
916 #[rstest]
921 #[tokio::test]
922 async fn test_webhook_send_success() {
923 let mut server = mockito::Server::new_async().await;
925 let mock = server
926 .mock("POST", "/webhook")
927 .with_status(200)
928 .with_header("content-type", "application/json")
929 .with_body(r#"{"status":"ok"}"#)
930 .create_async()
931 .await;
932
933 let config = WebhookConfig {
934 url: format!("{}/webhook", server.url()),
935 method: "POST".to_string(),
936 headers: HashMap::new(),
937 timeout: Duration::from_secs(5),
938 retry_config: RetryConfig {
939 max_retries: 0,
940 initial_backoff: Duration::from_millis(10),
941 max_backoff: Duration::from_secs(1),
942 backoff_multiplier: 2.0,
943 },
944 };
945
946 let sender = HttpWebhookSender::new(config);
947
948 let now = Utc::now();
949 let event = WebhookEvent {
950 task_id: TaskId::new(),
951 task_name: "test_task".to_string(),
952 status: TaskStatus::Success,
953 result: Some("OK".to_string()),
954 error: None,
955 started_at: now,
956 completed_at: now,
957 duration_ms: 100,
958 };
959
960 let result = sender.send_with_retry(&event).await;
962
963 assert!(result.is_ok());
965 mock.assert_async().await;
966 }
967
968 #[rstest]
969 #[tokio::test]
970 async fn test_webhook_send_retry_then_success() {
971 let mut server = mockito::Server::new_async().await;
973
974 let mock1 = server
976 .mock("POST", "/webhook")
977 .with_status(500)
978 .expect(1)
979 .create_async()
980 .await;
981
982 let mock2 = server
983 .mock("POST", "/webhook")
984 .with_status(503)
985 .expect(1)
986 .create_async()
987 .await;
988
989 let mock3 = server
990 .mock("POST", "/webhook")
991 .with_status(200)
992 .expect(1)
993 .create_async()
994 .await;
995
996 let config = WebhookConfig {
997 url: format!("{}/webhook", server.url()),
998 method: "POST".to_string(),
999 headers: HashMap::new(),
1000 timeout: Duration::from_secs(5),
1001 retry_config: RetryConfig {
1002 max_retries: 3,
1003 initial_backoff: Duration::from_millis(10),
1004 max_backoff: Duration::from_secs(1),
1005 backoff_multiplier: 2.0,
1006 },
1007 };
1008
1009 let sender = HttpWebhookSender::new(config);
1010
1011 let now = Utc::now();
1012 let event = WebhookEvent {
1013 task_id: TaskId::new(),
1014 task_name: "test_task".to_string(),
1015 status: TaskStatus::Success,
1016 result: Some("OK".to_string()),
1017 error: None,
1018 started_at: now,
1019 completed_at: now,
1020 duration_ms: 100,
1021 };
1022
1023 let result = sender.send_with_retry(&event).await;
1025
1026 assert!(result.is_ok());
1028 mock1.assert_async().await;
1029 mock2.assert_async().await;
1030 mock3.assert_async().await;
1031 }
1032
1033 #[rstest]
1034 #[tokio::test]
1035 async fn test_webhook_send_max_retries_exceeded() {
1036 let mut server = mockito::Server::new_async().await;
1038
1039 let mock = server
1041 .mock("POST", "/webhook")
1042 .with_status(500)
1043 .expect(4) .create_async()
1045 .await;
1046
1047 let config = WebhookConfig {
1048 url: format!("{}/webhook", server.url()),
1049 method: "POST".to_string(),
1050 headers: HashMap::new(),
1051 timeout: Duration::from_secs(5),
1052 retry_config: RetryConfig {
1053 max_retries: 3,
1054 initial_backoff: Duration::from_millis(10),
1055 max_backoff: Duration::from_secs(1),
1056 backoff_multiplier: 2.0,
1057 },
1058 };
1059
1060 let sender = HttpWebhookSender::new(config);
1061
1062 let now = Utc::now();
1063 let event = WebhookEvent {
1064 task_id: TaskId::new(),
1065 task_name: "test_task".to_string(),
1066 status: TaskStatus::Success,
1067 result: Some("OK".to_string()),
1068 error: None,
1069 started_at: now,
1070 completed_at: now,
1071 duration_ms: 100,
1072 };
1073
1074 let result = sender.send_with_retry(&event).await;
1076
1077 assert!(result.is_err());
1079 assert!(matches!(
1080 result.unwrap_err(),
1081 WebhookError::MaxRetriesExceeded
1082 ));
1083 mock.assert_async().await;
1084 }
1085
1086 #[rstest]
1087 #[tokio::test]
1088 async fn test_webhook_custom_headers() {
1089 let mut server = mockito::Server::new_async().await;
1091
1092 let mock = server
1093 .mock("POST", "/webhook")
1094 .match_header("Authorization", "Bearer test-token")
1095 .match_header("X-Custom-Header", "custom-value")
1096 .with_status(200)
1097 .create_async()
1098 .await;
1099
1100 let mut headers = HashMap::new();
1101 headers.insert("Authorization".to_string(), "Bearer test-token".to_string());
1102 headers.insert("X-Custom-Header".to_string(), "custom-value".to_string());
1103
1104 let config = WebhookConfig {
1105 url: format!("{}/webhook", server.url()),
1106 method: "POST".to_string(),
1107 headers,
1108 timeout: Duration::from_secs(5),
1109 retry_config: RetryConfig {
1110 max_retries: 0,
1111 initial_backoff: Duration::from_millis(10),
1112 max_backoff: Duration::from_secs(1),
1113 backoff_multiplier: 2.0,
1114 },
1115 };
1116
1117 let sender = HttpWebhookSender::new(config);
1118
1119 let now = Utc::now();
1120 let event = WebhookEvent {
1121 task_id: TaskId::new(),
1122 task_name: "test_task".to_string(),
1123 status: TaskStatus::Success,
1124 result: Some("OK".to_string()),
1125 error: None,
1126 started_at: now,
1127 completed_at: now,
1128 duration_ms: 100,
1129 };
1130
1131 let result = sender.send_with_retry(&event).await;
1133
1134 assert!(result.is_ok());
1136 mock.assert_async().await;
1137 }
1138
1139 #[rstest]
1140 #[tokio::test]
1141 async fn test_webhook_retry_loop_sleeps_between_retries() {
1142 let mut server = mockito::Server::new_async().await;
1145
1146 let _mock = server
1148 .mock("POST", "/webhook")
1149 .with_status(500)
1150 .expect(3) .create_async()
1152 .await;
1153
1154 let config = WebhookConfig {
1155 url: format!("{}/webhook", server.url()),
1156 method: "POST".to_string(),
1157 headers: HashMap::new(),
1158 timeout: Duration::from_secs(5),
1159 retry_config: RetryConfig {
1160 max_retries: 2,
1161 initial_backoff: Duration::from_millis(50),
1162 max_backoff: Duration::from_secs(1),
1163 backoff_multiplier: 2.0,
1164 },
1165 };
1166
1167 let sender = HttpWebhookSender::new(config);
1168
1169 let now = Utc::now();
1170 let event = WebhookEvent {
1171 task_id: TaskId::new(),
1172 task_name: "test_task".to_string(),
1173 status: TaskStatus::Success,
1174 result: None,
1175 error: None,
1176 started_at: now,
1177 completed_at: now,
1178 duration_ms: 0,
1179 };
1180
1181 let start = std::time::Instant::now();
1183 let result = sender.send_with_retry(&event).await;
1184 let elapsed = start.elapsed();
1185
1186 assert!(result.is_err());
1190 assert!(
1191 elapsed >= Duration::from_millis(80),
1192 "Expected at least 80ms delay from retry backoff sleep, got {:?}",
1193 elapsed
1194 );
1195 }
1196
1197 #[rstest]
1200 #[case("127.0.0.1", true)]
1201 #[case("127.0.0.2", true)]
1202 #[case("127.255.255.255", true)]
1203 #[case("10.0.0.1", true)]
1204 #[case("10.255.255.255", true)]
1205 #[case("172.16.0.1", true)]
1206 #[case("172.31.255.255", true)]
1207 #[case("192.168.0.1", true)]
1208 #[case("192.168.255.255", true)]
1209 #[case("169.254.169.254", true)]
1210 #[case("169.254.170.2", true)]
1211 #[case("::1", true)]
1212 #[case("fe80::1", true)]
1213 #[case("fc00::1", true)]
1214 #[case("8.8.8.8", false)]
1215 #[case("1.1.1.1", false)]
1216 #[case("203.0.113.1", false)]
1217 #[case("2001:db8::1", false)]
1218 fn test_is_blocked_ip(#[case] ip_str: &str, #[case] expected: bool) {
1219 let ip: IpAddr = ip_str.parse().unwrap();
1221
1222 let result = is_blocked_ip(&ip);
1224
1225 assert_eq!(
1227 result, expected,
1228 "IP {} should be blocked={}",
1229 ip_str, expected
1230 );
1231 }
1232
1233 #[rstest]
1234 #[case("https://example.com/webhook", true)]
1235 #[case("https://api.example.com/hooks/123", true)]
1236 #[case("https://hooks.slack.com/services/T00/B00/xxx", true)]
1237 fn test_validate_webhook_url_accepts_valid_urls(#[case] url: &str, #[case] _valid: bool) {
1238 let result = validate_webhook_url(url);
1240
1241 assert!(
1243 result.is_ok(),
1244 "URL {} should be valid: {:?}",
1245 url,
1246 result.err()
1247 );
1248 }
1249
1250 #[rstest]
1251 #[case("http://example.com/webhook", "SchemeNotAllowed")]
1252 #[case("ftp://example.com/file", "SchemeNotAllowed")]
1253 #[case("not-a-url", "InvalidUrl")]
1254 #[case("https://127.0.0.1/webhook", "BlockedIpAddress")]
1255 #[case("https://10.0.0.1/webhook", "BlockedIpAddress")]
1256 #[case("https://172.16.0.1/webhook", "BlockedIpAddress")]
1257 #[case("https://192.168.1.1/webhook", "BlockedIpAddress")]
1258 #[case("https://169.254.169.254/latest/meta-data/", "BlockedIpAddress")]
1259 #[case("https://[::1]/webhook", "BlockedIpAddress")]
1260 #[case("https://[fe80::1]/webhook", "BlockedIpAddress")]
1261 #[case("https://[fc00::1]/webhook", "BlockedIpAddress")]
1262 #[case("https://localhost/webhook", "BlockedIpAddress")]
1263 #[case("https://sub.localhost/webhook", "BlockedIpAddress")]
1264 #[case("https://service.internal/webhook", "BlockedIpAddress")]
1265 #[case("https://printer.local/webhook", "BlockedIpAddress")]
1266 fn test_validate_webhook_url_rejects_unsafe_urls(
1267 #[case] url: &str,
1268 #[case] expected_error: &str,
1269 ) {
1270 let result = validate_webhook_url(url);
1272
1273 assert!(result.is_err(), "URL {} should be rejected", url);
1275 let err = result.unwrap_err();
1276 let err_name = match &err {
1277 WebhookError::InvalidUrl(_) => "InvalidUrl",
1278 WebhookError::SchemeNotAllowed(_) => "SchemeNotAllowed",
1279 WebhookError::BlockedIpAddress(_) => "BlockedIpAddress",
1280 WebhookError::DnsResolutionFailed(_) => "DnsResolutionFailed",
1281 _ => "Other",
1282 };
1283 assert_eq!(
1284 err_name, expected_error,
1285 "URL {} should produce {} error, got: {}",
1286 url, expected_error, err
1287 );
1288 }
1289
1290 #[rstest]
1291 fn test_validate_webhook_url_blocks_cloud_metadata_endpoint() {
1292 let metadata_urls = [
1294 "https://169.254.169.254/latest/meta-data/",
1295 "https://169.254.169.254/computeMetadata/v1/",
1296 "https://169.254.170.2/v2/credentials",
1297 ];
1298
1299 for url in &metadata_urls {
1300 let result = validate_webhook_url(url);
1302
1303 assert!(
1305 result.is_err(),
1306 "Cloud metadata URL {} should be blocked",
1307 url
1308 );
1309 assert!(
1310 matches!(result.unwrap_err(), WebhookError::BlockedIpAddress(_)),
1311 "Cloud metadata URL {} should produce BlockedIpAddress error",
1312 url
1313 );
1314 }
1315 }
1316
1317 #[rstest]
1318 fn test_webhook_error_display_ssrf_variants() {
1319 let error = WebhookError::InvalidUrl("bad-url".to_string());
1321 assert_eq!(error.to_string(), "Invalid webhook URL: bad-url");
1322
1323 let error = WebhookError::SchemeNotAllowed("http".to_string());
1324 assert_eq!(
1325 error.to_string(),
1326 "URL scheme not allowed: http. Only HTTPS is permitted for webhooks"
1327 );
1328
1329 let error = WebhookError::BlockedIpAddress("127.0.0.1".to_string());
1330 assert_eq!(
1331 error.to_string(),
1332 "Webhook URL resolves to blocked IP address: 127.0.0.1"
1333 );
1334
1335 let error = WebhookError::DnsResolutionFailed("bad.host".to_string());
1336 assert_eq!(
1337 error.to_string(),
1338 "DNS resolution failed for webhook URL host: bad.host"
1339 );
1340 }
1341
1342 #[rstest]
1343 #[tokio::test]
1344 async fn test_send_rejects_http_url_via_ssrf_validation() {
1345 let config = WebhookConfig {
1347 url: "http://example.com/webhook".to_string(),
1348 method: "POST".to_string(),
1349 headers: HashMap::new(),
1350 timeout: Duration::from_secs(5),
1351 retry_config: RetryConfig::default(),
1352 };
1353 let sender = HttpWebhookSender::new(config);
1354 let now = Utc::now();
1355 let event = WebhookEvent {
1356 task_id: TaskId::new(),
1357 task_name: "test_task".to_string(),
1358 status: TaskStatus::Success,
1359 result: None,
1360 error: None,
1361 started_at: now,
1362 completed_at: now,
1363 duration_ms: 0,
1364 };
1365
1366 let result = sender.send(&event).await;
1368
1369 assert!(result.is_err());
1371 assert!(matches!(
1372 result.unwrap_err(),
1373 WebhookError::SchemeNotAllowed(_)
1374 ));
1375 }
1376
1377 #[rstest]
1378 #[tokio::test]
1379 async fn test_send_rejects_private_ip_via_ssrf_validation() {
1380 let config = WebhookConfig {
1382 url: "https://192.168.1.1/webhook".to_string(),
1383 method: "POST".to_string(),
1384 headers: HashMap::new(),
1385 timeout: Duration::from_secs(5),
1386 retry_config: RetryConfig::default(),
1387 };
1388 let sender = HttpWebhookSender::new(config);
1389 let now = Utc::now();
1390 let event = WebhookEvent {
1391 task_id: TaskId::new(),
1392 task_name: "test_task".to_string(),
1393 status: TaskStatus::Success,
1394 result: None,
1395 error: None,
1396 started_at: now,
1397 completed_at: now,
1398 duration_ms: 0,
1399 };
1400
1401 let result = sender.send(&event).await;
1403
1404 assert!(result.is_err());
1406 assert!(matches!(
1407 result.unwrap_err(),
1408 WebhookError::BlockedIpAddress(_)
1409 ));
1410 }
1411
1412 #[rstest]
1422 #[case(1, Duration::from_millis(30), Duration::from_millis(50))]
1423 #[case(2, Duration::from_millis(80), Duration::from_millis(50))]
1424 #[case(3, Duration::from_millis(200), Duration::from_millis(50))]
1425 #[tokio::test]
1426 async fn test_webhook_retry_sleep_is_called_between_attempts(
1427 #[case] max_retries: u32,
1428 #[case] min_elapsed: Duration,
1429 #[case] initial_backoff: Duration,
1430 ) {
1431 let mut server = mockito::Server::new_async().await;
1433
1434 let _mock = server
1436 .mock("POST", "/webhook")
1437 .with_status(500)
1438 .expect((max_retries + 1) as usize)
1439 .create_async()
1440 .await;
1441
1442 let config = WebhookConfig {
1443 url: format!("{}/webhook", server.url()),
1444 method: "POST".to_string(),
1445 headers: HashMap::new(),
1446 timeout: Duration::from_secs(5),
1447 retry_config: RetryConfig {
1448 max_retries,
1449 initial_backoff,
1450 max_backoff: Duration::from_secs(1),
1451 backoff_multiplier: 2.0,
1452 },
1453 };
1454
1455 let sender = HttpWebhookSender::new(config);
1456 let now = Utc::now();
1457 let event = WebhookEvent {
1458 task_id: TaskId::new(),
1459 task_name: "regression_742".to_string(),
1460 status: TaskStatus::Success,
1461 result: None,
1462 error: None,
1463 started_at: now,
1464 completed_at: now,
1465 duration_ms: 0,
1466 };
1467
1468 let start = std::time::Instant::now();
1470 let result = sender.send_with_retry(&event).await;
1471 let elapsed = start.elapsed();
1472
1473 assert!(
1477 result.is_err(),
1478 "expected MaxRetriesExceeded after all retries"
1479 );
1480 assert!(
1481 elapsed >= min_elapsed,
1482 "Regression #742: expected sleep between retries (>={:?}), got {:?}",
1483 min_elapsed,
1484 elapsed
1485 );
1486 }
1487}