//! Webhooks (hub spec §12.2): queueing an event for every hook that wants //! it, signing and delivering with retries, and the payload envelope. use crate::db::{self, hooks as hdb}; use crate::Shared; use hub_api::WebhookEnvelope; /// Queue `event` for the hooks of a repository (and its organisation). pub async fn emit(app: &Shared, repo: &db::repos::RepoRow, event: &str, actor: &str, data: serde_json::Value) { // Live pages hear of it whether or not any hook does (oxygenhub#203). app.events.live.publish_hook(repo.id, event, &data); let hooks = match hdb::hooks_hearing(&app.db, repo.id, &repo.owner).await { Ok(h) => h, Err(e) => { tracing::warn!("hooks: {e}"); return; } }; if hooks.is_empty() { return; } let repository = hub_api::Repository { owner: repo.owner.clone(), name: repo.name.clone(), description: repo.description.clone(), default_bookmark: repo.default_bookmark.clone(), visibility: repo.visibility(), created_ms: repo.created_ms, updated_ms: repo.updated_ms, clone_url: format!("{}/{}", app.config.base_url(), repo.slug()), }; for h in hooks.into_iter().filter(|h| h.wants(event)) { let delivery = crate::auth::random_hex(16); let envelope = WebhookEnvelope { event: event.into(), delivery: delivery.clone(), repository: Some(repository.clone()), actor: actor.into(), at_ms: db::now_ms(), data: data.clone(), }; let payload = serde_json::to_string(&envelope).unwrap_or_default(); if let Err(e) = hdb::queue_delivery(&app.db, h.id, &delivery, event, &payload).await { tracing::warn!("queueing delivery: {e}"); } } app.events.notify_deliveries(); } /// What a chat message says about an event: a title, the page it links /// to (absolute) and a short detail. pub fn summary(env: &WebhookEnvelope, base: &str) -> (String, String, String) { let repo = env.repository.as_ref().map(|r| format!("{}/{}", r.owner, r.name)).unwrap_or_default(); let page = |path: &str| format!("{}/{repo}{path}", base.trim_end_matches('/')); let d = &env.data; let text = |v: &serde_json::Value| v.as_str().unwrap_or("").to_string(); let num = |v: &serde_json::Value| v.as_i64().unwrap_or(0); let actor = if env.actor.is_empty() { "Someone".to_string() } else { env.actor.clone() }; let issue_n = if d["issue"]["number"].is_i64() { num(&d["issue"]["number"]) } else { num(&d["number"]) }; let issue_title = text(&d["issue"]["title"]); match env.event.as_str() { "ping" => ( "OxygenHub is connected".into(), if repo.is_empty() { base.to_string() } else { page("") }, "Events from OxygenHub will arrive here.".into(), ), "push" => { let lines: Vec = d["updates"] .as_array() .map(|u| { u.iter() .map(|x| { let n = num(&x["changes"]); format!("{}: {n} change{}", text(&x["reference"]), if n == 1 { "" } else { "s" }) }) .collect() }) .unwrap_or_default(); let first = d["updates"][0]["reference"].as_str().unwrap_or("").to_string(); (format!("{actor} pushed to {repo}"), page(&format!("/changes/{first}")), lines.join("\n")) } "issue.opened" => { (format!("{actor} opened issue #{issue_n}: {issue_title}"), page(&format!("/issue/{issue_n}")), text(&d["issue"]["body"])) } "issue.closed" | "issue.reopened" | "issue.edited" => { let verb = env.event.trim_start_matches("issue."); (format!("{actor} {verb} issue #{issue_n}: {issue_title}"), page(&format!("/issue/{issue_n}")), String::new()) } "comment.issue" => { let body = if d["comment"]["body"].is_string() { text(&d["comment"]["body"]) } else { String::new() }; ( format!( "{actor} commented on issue #{issue_n}{}", if issue_title.is_empty() { String::new() } else { format!(": {issue_title}") } ), page(&format!("/issue/{issue_n}")), body, ) } "review" => ( format!("{actor} reviewed proposal #{}: {}", num(&d["number"]), text(&d["verdict"])), page(&format!("/proposal/{}", num(&d["number"]))), String::new(), ), "release.published" => ( format!("{actor} released {} ({})", text(&d["title"]), text(&d["tag"])), page(&format!("/releases/tag/{}", text(&d["tag"]))), String::new(), ), "check" => ( format!("Check {} {} on {repo}", text(&d["check"]["name"]), text(&d["check"]["status"])), text(&d["check"]["url"]).chars().take(500).collect::().replace(' ', "%20"), text(&d["check"]["summary"]), ), e if e.starts_with("proposal.") => { let n = num(&d["number"]); let verb = e.trim_start_matches("proposal."); let title = text(&d["title"]); ( format!("{actor} {verb} proposal #{n}{}", if title.is_empty() { String::new() } else { format!(": {title}") }), page(&format!("/proposal/{n}")), String::new(), ) } e => (format!("{e} on {repo}"), page(""), String::new()), } } fn cut(s: &str, max: usize) -> String { if s.chars().count() <= max { s.to_string() } else { format!("{}…", s.chars().take(max.saturating_sub(1)).collect::()) } } /// The body a hook receives in its format; the envelope itself for `oxygenhub`. pub fn body_for(format: &str, payload: &str, base: &str) -> String { let Ok(env) = serde_json::from_str::(payload) else { return payload.to_string() }; let (title, url, detail) = summary(&env, base); let repo = env.repository.as_ref().map(|r| format!("{}/{}", r.owner, r.name)).unwrap_or_else(|| "OxygenHub".into()); match format { "discord" => { let mut embed = serde_json::json!({ "title": cut(&title, 256), "description": cut(&detail, 2000), "color": 0x0064e0, "footer": { "text": cut(&format!("{repo} · {}", env.event), 200) }, "timestamp": crate::render::time::rfc3339(env.at_ms as u64), }); if url.starts_with("http") { embed["url"] = serde_json::Value::String(url); } // No @everyone, @here or role pings from anything a user typed. serde_json::json!({ "username": "OxygenHub", "allowed_mentions": { "parse": [] }, "embeds": [embed] }).to_string() } "slack" => { let esc = |s: &str| s.replace('&', "&").replace('<', "<").replace('>', ">"); let head = if url.starts_with("http") { format!("<{}|{}>", url, esc(&title)) } else { esc(&title) }; let text = if detail.is_empty() { head } else { format!("{head}\n>{}", esc(&cut(&detail, 1500)).replace('\n', "\n>")) }; serde_json::json!({ "text": text }).to_string() } _ => payload.to_string(), } } /// `X-Oxyhub-Signature` for a body: keyed BLAKE3 with the hook's secret. pub fn signature(secret: &str, body: &[u8]) -> String { let key = blake3::hash(secret.as_bytes()); format!("blake3-keyed={}", hex::encode(blake3::keyed_hash(key.as_bytes(), body).as_bytes())) } /// Deliver everything that is due; returns how many were attempted. pub async fn deliver_due(app: &Shared) -> usize { let due = match hdb::due_deliveries(&app.db, 50).await { Ok(d) => d, Err(_) => return 0, }; let mut n = 0; for d in due { let Ok(Some(hook)) = hdb::hook(&app.db, d.hook_id).await else { continue }; n += 1; // Only public addresses, pinned, no redirects (#219). let client = match crate::net::client(&hook.url, &app.config.outbound.allow, std::time::Duration::from_secs(20)).await { Ok(c) => c, Err(why) => { hdb::record_attempt(&app.db, d.id, d.attempts, None, &format!("not sent: {why}")).await.ok(); continue; } }; let body = body_for(&hook.format, &d.payload, app.config.base_url()); let (status, response) = match client .post(&hook.url) .header("content-type", "application/json") .header("x-oxyhub-event", &d.event) .header("x-oxyhub-delivery", &d.delivery) .header("x-oxyhub-signature", signature(&hook.secret, body.as_bytes())) .body(body) .send() .await { Ok(r) => { let status = r.status().as_u16() as i64; let text = r.text().await.unwrap_or_default(); (Some(status), text) } Err(e) => (None, e.to_string()), }; hdb::record_attempt(&app.db, d.id, d.attempts, status, &response).await.ok(); } n } /// The background worker: delivers when nudged and every minute anyway. pub async fn worker(app: Shared) { let mut rx = app.events.deliveries.subscribe(); loop { deliver_due(&app).await; let cutoff = db::now_ms() - app.config.retention.webhook_log_days as i64 * 86_400_000; hdb::prune_deliveries(&app.db, cutoff).await.ok(); tokio::select! { _ = rx.recv() => {} _ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {} } } } #[cfg(test)] mod tests { use super::*; fn envelope(event: &str, data: serde_json::Value) -> String { serde_json::to_string(&WebhookEnvelope { event: event.into(), delivery: "d".into(), repository: Some(hub_api::Repository { owner: "andymccall".into(), name: "oxygen".into(), description: String::new(), default_bookmark: "main".into(), visibility: hub_api::Visibility::Public, created_ms: 0, updated_ms: 0, clone_url: String::new(), }), actor: "andy".into(), at_ms: 1_790_000_000_000, data, }) .unwrap() } #[test] fn discord_gets_an_embed_without_mentions() { let p = envelope( "issue.opened", serde_json::json!({ "issue": { "number": 12, "title": "@everyone crash", "body": "x".repeat(3000) } }), ); let v: serde_json::Value = serde_json::from_str(&body_for("discord", &p, "https://hub.example:7343/")).unwrap(); let e = &v["embeds"][0]; assert_eq!(e["title"], "andy opened issue #12: @everyone crash"); assert_eq!(e["url"], "https://hub.example:7343/andymccall/oxygen/issue/12"); assert_eq!(e["description"].as_str().unwrap().chars().count(), 2000, "cut to Discord's limits"); assert_eq!(v["allowed_mentions"]["parse"], serde_json::json!([]), "no pings from user text"); assert_eq!(v["username"], "OxygenHub"); let push = envelope("push", serde_json::json!({ "updates": [{ "reference": "main", "changes": 3 }] })); let v: serde_json::Value = serde_json::from_str(&body_for("discord", &push, "https://h")).unwrap(); assert_eq!(v["embeds"][0]["title"], "andy pushed to andymccall/oxygen"); assert_eq!(v["embeds"][0]["description"], "main: 3 changes"); assert_eq!(v["embeds"][0]["url"], "https://h/andymccall/oxygen/changes/main"); let ping = envelope("ping", serde_json::json!({})); assert!(body_for("discord", &ping, "https://h").contains("OxygenHub is connected")); } #[test] fn slack_text_is_escaped_and_the_envelope_is_unchanged() { let p = envelope( "comment.issue", serde_json::json!({ "number": 3, "issue": { "number": 3, "title": "" }, "comment": { "body": "a & b" } }), ); let v: serde_json::Value = serde_json::from_str(&body_for("slack", &p, "https://h")).unwrap(); assert_eq!(v["text"], "\n>a & b"); assert_eq!(body_for("oxygenhub", &p, "https://h"), p); } #[test] fn formats_follow_the_url_and_urls_are_masked() { use crate::db::hooks::{format_for_url, masked_url}; assert_eq!(format_for_url("https://discord.com/api/webhooks/155/abc"), "discord"); assert_eq!(format_for_url("https://hooks.slack.com/services/T/B/x"), "slack"); assert_eq!(format_for_url("https://discord.com.evil.example/api/webhooks/1"), "oxygenhub"); assert_eq!(format_for_url("https://ci.example/hook"), "oxygenhub"); let m = masked_url("https://discord.com/api/webhooks/1554146511347126332/6g_WnXRdNOiVoo8sGXXW_OuIUceluD8lIUckxb9KHPwZ895XE"); assert_eq!(m, "https://discord.com/api/webhooks/1554…/6g_W…"); assert!(!m.contains("6g_WnX"), "the token is not shown"); assert_eq!(masked_url("https://ci.example/hook?token=secret"), "https://ci.example/hook?…"); } #[test] fn signatures_are_keyed() { let a = signature("s1", b"body"); assert!(a.starts_with("blake3-keyed=")); assert_ne!(a, signature("s2", b"body")); assert_ne!(a, signature("s1", b"other")); } }