fix: clone reqwest client per task for concurrent webhook sends

Cargo.lock

···
  "sha2",
  "thiserror",
  "tokio",
+ "url",
 ]
 
 [[package]]

Cargo.toml

···
 hmac = "0.12"
 sha2 = "0.10"
 hex = "0.4"
+url = "2.5"
 
 [build-dependencies]
 askama = "0.13"

src/api/status.rs

···
                                             let did_for_event = did_string.clone();
                                             let uri = current_status.uri.clone();
                                             tokio::spawn(async move {
-                                                let event = StatusEvent {
-                                                    event: "status.deleted",
-                                                    did: &did_for_event,
-                                                    handle: None,
-                                                    status: None,
-                                                    text: None,
-                                                    uri: Some(&uri),
-                                                    since: None,
-                                                    expires: None,
-                                                };
-                                                send_status_event(pool, &did_for_event, event)
-                                                    .await;
+                                                crate::webhooks::emit_deleted(
+                                                    pool,
+                                                    &did_for_event,
+                                                    &uri,
+                                                )
+                                                .await;
                                             });
                                         }
 
···
 
                             let _ = status.save(db_pool.clone()).await;
 
-                            // Fire webhooks asynchronously
+                            // Fire webhooks asynchronously (helper keeps this file lean)
                             {
                                 let pool = db_pool.get_ref().clone();
-                                let did_for_event = status.author_did.clone();
-                                let emoji = status.status.clone();
-                                let text = status.text.clone();
-                                let uri = status.uri.clone();
-                                let since = status.started_at.to_rfc3339();
-                                let expires = status.expires_at.map(|e| e.to_rfc3339());
+                                let s = status.clone();
                                 tokio::spawn(async move {
-                                    let event = StatusEvent {
-                                        event: "status.created",
-                                        did: &did_for_event,
-                                        handle: None,
-                                        status: Some(&emoji),
-                                        text: text.as_deref(),
-                                        uri: Some(&uri),
-                                        since: Some(&since),
-                                        expires: expires.as_deref(),
-                                    };
-                                    send_status_event(pool, &did_for_event, event).await;
+                                    crate::webhooks::emit_created(pool, &s).await;
                                 });
                             }
                             Ok(Redirect::to("/")

src/api/webhooks.rs

···
-use crate::{db, error_handler::AppError};
+use crate::{config::Config, db, error_handler::AppError};
 use actix_session::Session;
 use actix_web::{HttpResponse, Responder, Result, delete, get, post, put, web};
 use async_sqlite::Pool;
 use atrium_api::types::string::Did;
 use serde::Deserialize;
 use std::sync::Arc;
+use url::Url;
 
 #[derive(Deserialize)]
 pub struct CreateWebhookRequest {
···
 pub async fn create_webhook(
     session: Session,
     db_pool: web::Data<Arc<Pool>>,
+    app_config: web::Data<Config>,
     payload: web::Json<CreateWebhookRequest>,
 ) -> Result<impl Responder> {
     let did = session.get::<Did>("did")?;
     if let Some(did) = did {
-        // Basic URL validation
-        if !(payload.url.starts_with("https://") || payload.url.starts_with("http://")) {
-            return Ok(web::Json(serde_json::json!({
-                "error": "URL must start with http:// or https://"
-            })));
+        // Robust URL + SSRF validation
+        if let Err(msg) = validate_url(&payload.url, &app_config) {
+            return Ok(web::Json(serde_json::json!({ "error": msg })));
+        }
+        // Events validation
+        if let Some(events_str) = &payload.events {
+            if let Err(msg) = validate_events(events_str) {
+                return Ok(web::Json(serde_json::json!({ "error": msg })));
+            }
         }
         let (id, secret) = db::create_webhook(
             &db_pool,
···
     match session.get::<Did>("did").unwrap_or(None) {
         Some(did) => {
             let id = path.into_inner();
+            if let Some(url) = &payload.url {
+                if Url::parse(url).is_err() {
+                    return HttpResponse::BadRequest()
+                        .json(serde_json::json!({ "error": "Invalid URL" }));
+                }
+            }
+            if let Some(events_str) = &payload.events {
+                if let Err(msg) = validate_events(events_str) {
+                    return HttpResponse::BadRequest().json(serde_json::json!({ "error": msg }));
+                }
+            }
             let res = db::update_webhook(
                 &db_pool,
                 did.as_str(),
···
             HttpResponse::Unauthorized().json(serde_json::json!({ "error": "Not authenticated" }))
         }
     }
+}
+
+fn validate_events(s: &str) -> Result<(), &'static str> {
+    if s.trim().is_empty() {
+        return Ok(());
+    }
+    const ALLOWED: &[&str] = &["status.created", "status.deleted"];
+    for ev in s.split(',').map(|e| e.trim()) {
+        if !ALLOWED.contains(&ev) {
+            return Err("Unsupported event type");
+        }
+    }
+    Ok(())
+}
+
+fn validate_url(raw: &str, cfg: &Config) -> Result<(), &'static str> {
+    let url = Url::parse(raw).map_err(|_| "Invalid URL")?;
+    let scheme = url.scheme();
+    let host = url.host_str().ok_or("Missing host")?.to_ascii_lowercase();
+
+    // Treat localhost explicitly
+    let host_is_localname = host == "localhost";
+
+    // If host is an IP literal, apply standard library checks
+    let ip_check_blocks = if let Ok(ip) = host.parse::<std::net::IpAddr>() {
+        match ip {
+            std::net::IpAddr::V4(v4) => {
+                v4.is_private()
+                    || v4.is_loopback()
+                    || v4.is_link_local()
+                    || v4.is_multicast()
+                    || v4.is_unspecified()
+            }
+            std::net::IpAddr::V6(v6) => {
+                v6.is_unique_local() || v6.is_loopback() || v6.is_multicast() || v6.is_unspecified()
+            }
+        }
+    } else {
+        false
+    };
+
+    // Enforce HTTPS in production
+    let is_production = !cfg.oauth_redirect_base.starts_with("http://localhost")
+        && !cfg.oauth_redirect_base.starts_with("http://127.0.0.1");
+    if is_production && scheme != "https" {
+        return Err("HTTPS required in production");
+    }
+
+    // Basic SSRF protection in production
+    if (host_is_localname || ip_check_blocks) && is_production {
+        return Err("Private/local hosts not allowed");
+    }
+
+    Ok(())
 }
 
 #[post("/api/webhooks/{id}/rotate")]

src/webhooks.rs

···
 use serde::Serialize;
 use sha2::Sha256;
 
-use crate::db::{Webhook, get_user_webhooks};
+use crate::db::{StatusFromDb, Webhook, get_user_webhooks};
+use futures_util::future;
 
 #[derive(Serialize)]
 pub struct StatusEvent<'a> {
···
         .any(|e| e.eq_ignore_ascii_case(event))
 }
 
+fn hmac_sig_hex(secret: &str, ts: &str, payload: &[u8]) -> String {
+    let mut mac =
+        Hmac::<Sha256>::new_from_slice(secret.as_bytes()).expect("HMAC can take key of any size");
+    mac.update(ts.as_bytes());
+    mac.update(b".");
+    mac.update(payload);
+    hex::encode(mac.finalize().into_bytes())
+}
+
 pub async fn send_status_event(pool: std::sync::Arc<Pool>, did: &str, event: StatusEvent<'_>) {
     let client = Client::new();
     let hooks = match get_user_webhooks(&pool, did).await {
···
     };
     let ts = chrono::Utc::now().timestamp().to_string();
 
-    for h in hooks.into_iter().filter(|h| should_send(h, event.event)) {
-        let mut mac = Hmac::<Sha256>::new_from_slice(h.secret.as_bytes())
-            .expect("HMAC can take key of any size");
-        mac.update(ts.as_bytes());
-        mac.update(b".");
-        mac.update(&payload);
-        let sig = hex::encode(mac.finalize().into_bytes());
+    let futures = hooks
+        .into_iter()
+        .filter(|h| should_send(h, event.event))
+        .map(|h| {
+            let payload = payload.clone();
+            let ts = ts.clone();
+            let client = client.clone();
+            async move {
+                let sig = hmac_sig_hex(&h.secret, &ts, &payload);
+                let res = client
+                    .post(&h.url)
+                    .header("User-Agent", "status-webhooks/1.0")
+                    .header("Content-Type", "application/json")
+                    .header("X-Status-Webhook-Timestamp", &ts)
+                    .header("X-Status-Webhook-Signature", format!("sha256={}", sig))
+                    .timeout(std::time::Duration::from_secs(5))
+                    .body(payload)
+                    .send()
+                    .await;
 
-        let res = client
-            .post(&h.url)
-            .header("User-Agent", "status-webhooks/1.0")
-            .header("Content-Type", "application/json")
-            .header("X-Status-Webhook-Timestamp", &ts)
-            .header("X-Status-Webhook-Signature", format!("sha256={}", sig))
-            .body(payload.clone())
-            .send()
-            .await;
-
-        match res {
-            Ok(resp) => {
-                if !resp.status().is_success() {
-                    log::warn!(
-                        "webhook delivery failed: {} -> status {}",
-                        &h.url,
-                        resp.status()
-                    );
+                match res {
+                    Ok(resp) => {
+                        if !resp.status().is_success() {
+                            log::warn!(
+                                "webhook delivery failed: {} -> status {}",
+                                &h.url,
+                                resp.status()
+                            );
+                        }
+                    }
+                    Err(e) => log::warn!("webhook delivery error to {}: {}", &h.url, e),
                 }
             }
-            Err(e) => log::warn!("webhook delivery error to {}: {}", &h.url, e),
-        }
+        });
+
+    future::join_all(futures).await;
+}
+
+pub async fn emit_created(pool: std::sync::Arc<Pool>, s: &StatusFromDb) {
+    let did = s.author_did.clone();
+    let emoji = s.status.clone();
+    let text = s.text.clone();
+    let uri = s.uri.clone();
+    let since = s.started_at.to_rfc3339();
+    let expires = s.expires_at.map(|e| e.to_rfc3339());
+    let event = StatusEvent {
+        event: "status.created",
+        did: &did,
+        handle: None,
+        status: Some(&emoji),
+        text: text.as_deref(),
+        uri: Some(&uri),
+        since: Some(&since),
+        expires: expires.as_deref(),
+    };
+    send_status_event(pool, &did, event).await;
+}
+
+pub async fn emit_deleted(pool: std::sync::Arc<Pool>, did: &str, uri: &str) {
+    let did_owned = did.to_string();
+    let uri_owned = uri.to_string();
+    let event = StatusEvent {
+        event: "status.deleted",
+        did: &did_owned,
+        handle: None,
+        status: None,
+        text: None,
+        uri: Some(&uri_owned),
+        since: None,
+        expires: None,
+    };
+    send_status_event(pool, &did_owned, event).await;
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_should_send_wildcard() {
+        let h = Webhook {
+            id: 1,
+            did: "d".into(),
+            url: "u".into(),
+            secret: "s".into(),
+            events: "*".into(),
+            active: true,
+            created_at: 0,
+            updated_at: 0,
+        };
+        assert!(should_send(&h, "status.created"));
+    }
+
+    #[test]
+    fn test_should_send_specific() {
+        let h = Webhook {
+            id: 1,
+            did: "d".into(),
+            url: "u".into(),
+            secret: "s".into(),
+            events: "status.deleted".into(),
+            active: true,
+            created_at: 0,
+            updated_at: 0,
+        };
+        assert!(should_send(&h, "status.deleted"));
+        assert!(!should_send(&h, "status.created"));
+    }
+
+    #[test]
+    fn test_hmac_sig_hex() {
+        let sig = hmac_sig_hex("secret", "1234567890", b"{\"a\":1}");
+        // Deterministic expected if inputs fixed
+        assert_eq!(sig.len(), 64);
     }
 }