diff --git a/cruciverb-server/src/activitypub/deliver.rs b/cruciverb-server/src/activitypub/deliver.rs new file mode 100644 index 0000000..2e97377 --- /dev/null +++ b/cruciverb-server/src/activitypub/deliver.rs @@ -0,0 +1,125 @@ +use std::sync::Arc; +use std::time::Duration; + +use crate::state::AppState; +use super::signatures; + +pub fn start_delivery_worker(state: Arc) { + tokio::spawn(async move { + loop { + if state.ap.is_some() { + // #16: skip processing when federation is frozen + if !super::is_frozen() { + process_queue(&state).await; + } + } + tokio::time::sleep(Duration::from_secs(30)).await; + } + }); +} + +async fn process_queue(state: &Arc) { + let ap = match state.ap.as_ref() { + Some(ap) => ap, + None => return, + }; + + let now = chrono::Utc::now().to_rfc3339(); + + let deliveries: Vec<(i64, String, String, i64)> = { + let dbi = state.db.lock().await; + let mut stmt = dbi.db.prepare( + "SELECT id, inbox_uri, payload, attempts FROM ap_delivery_queue + WHERE next_retry <= ?1 ORDER BY next_retry ASC LIMIT 20" + ).unwrap(); + stmt.query_map([&now], |row| { + Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)) + }).unwrap().filter_map(|r| r.ok()).collect() + }; + + if deliveries.is_empty() { return; } + + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build() + .unwrap(); + + for (id, inbox_uri, payload, attempts) in deliveries { + // #8: DNS rebinding protection on delivery targets + if crate::security::is_internal_url(&inbox_uri) { + tracing::warn!("refusing delivery to internal URL: {}", inbox_uri); + let dbi = state.db.lock().await; + dbi.db.execute("DELETE FROM ap_delivery_queue WHERE id = ?1", [id]).ok(); + continue; + } + + let host = inbox_uri.split("//").nth(1) + .and_then(|s| s.split('/').next()) + .unwrap_or(""); + let path = inbox_uri.split("//").nth(1) + .and_then(|s| s.find('/').map(|i| &s[i..])) + .unwrap_or("/inbox"); + + let key_id = format!("{}#main-key", ap.actor_uri); + let sig_headers = signatures::sign_request( + &ap.keys.private_key, + &key_id, + "POST", + path, + host, + Some(payload.as_bytes()), + ); + + let mut req = client.post(&inbox_uri) + .header("Content-Type", "application/activity+json"); + + for (name, value) in &sig_headers { + req = req.header(name.as_str(), value.as_str()); + } + + match req.body(payload.clone()).send().await { + Ok(resp) if resp.status().is_success() || resp.status().as_u16() == 202 => { + let dbi = state.db.lock().await; + dbi.db.execute("DELETE FROM ap_delivery_queue WHERE id = ?1", [id]).ok(); + crate::db::track_federation_activity(&dbi.db, host, "outbound"); + tracing::info!("delivered to {}", inbox_uri); + } + Ok(resp) => { + tracing::warn!("delivery to {} failed: {}", inbox_uri, resp.status()); + handle_failure(state, id, attempts).await; + } + Err(e) => { + tracing::warn!("delivery to {} error: {}", inbox_uri, e); + handle_failure(state, id, attempts).await; + } + } + } +} + +async fn handle_failure(state: &Arc, id: i64, attempts: i64) { + let dbi = state.db.lock().await; + let new_attempts = attempts + 1; + + if new_attempts >= 10 { + // give up after 10 attempts + dbi.db.execute("DELETE FROM ap_delivery_queue WHERE id = ?1", [id]).ok(); + tracing::info!("dropping delivery {} after {} attempts", id, new_attempts); + return; + } + + // exponential backoff: 1m, 5m, 30m, 2h, 12h, 24h cap + let delay_secs = match new_attempts { + 1 => 60, + 2 => 300, + 3 => 1800, + 4 => 7200, + 5 => 43200, + _ => 86400, + }; + + let next_retry = (chrono::Utc::now() + chrono::Duration::seconds(delay_secs)).to_rfc3339(); + dbi.db.execute( + "UPDATE ap_delivery_queue SET attempts = ?1, next_retry = ?2 WHERE id = ?3", + rusqlite::params![new_attempts, next_retry, id], + ).ok(); +}