| 1 |
|
| 2 |
|
| 3 |
|
| 4 |
use crate::db::{self, hooks as hdb}; |
| 5 |
use crate::Shared; |
| 6 |
use hub_api::WebhookEnvelope; |
| 7 |
|
| 8 |
|
| 9 |
pub async fn emit(app: &Shared, repo: &db::repos::RepoRow, event: &str, actor: &str, data: serde_json::Value) { |
| 10 |
|
| 11 |
app.events.live.publish_hook(repo.id, event, &data); |
| 12 |
let hooks = match hdb::hooks_hearing(&app.db, repo.id, &repo.owner).await { |
| 13 |
Ok(h) => h, |
| 14 |
Err(e) => { |
| 15 |
tracing::warn!("hooks: {e}"); |
| 16 |
return; |
| 17 |
} |
| 18 |
}; |
| 19 |
if hooks.is_empty() { |
| 20 |
return; |
| 21 |
} |
| 22 |
let repository = hub_api::Repository { |
| 23 |
owner: repo.owner.clone(), |
| 24 |
name: repo.name.clone(), |
| 25 |
description: repo.description.clone(), |
| 26 |
default_bookmark: repo.default_bookmark.clone(), |
| 27 |
visibility: repo.visibility(), |
| 28 |
created_ms: repo.created_ms, |
| 29 |
updated_ms: repo.updated_ms, |
| 30 |
clone_url: format!("{}/{}", app.config.base_url(), repo.slug()), |
| 31 |
}; |
| 32 |
for h in hooks.into_iter().filter(|h| h.wants(event)) { |
| 33 |
let delivery = crate::auth::random_hex(16); |
| 34 |
let envelope = WebhookEnvelope { |
| 35 |
event: event.into(), |
| 36 |
delivery: delivery.clone(), |
| 37 |
repository: Some(repository.clone()), |
| 38 |
actor: actor.into(), |
| 39 |
at_ms: db::now_ms(), |
| 40 |
data: data.clone(), |
| 41 |
}; |
| 42 |
let payload = serde_json::to_string(&envelope).unwrap_or_default(); |
| 43 |
if let Err(e) = hdb::queue_delivery(&app.db, h.id, &delivery, event, &payload).await { |
| 44 |
tracing::warn!("queueing delivery: {e}"); |
| 45 |
} |
| 46 |
} |
| 47 |
app.events.notify_deliveries(); |
| 48 |
} |
| 49 |
|
| 50 |
|
| 51 |
|
| 52 |
pub fn summary(env: &WebhookEnvelope, base: &str) -> (String, String, String) { |
| 53 |
let repo = env.repository.as_ref().map(|r| format!("{}/{}", r.owner, r.name)).unwrap_or_default(); |
| 54 |
let page = |path: &str| format!("{}/{repo}{path}", base.trim_end_matches('/')); |
| 55 |
let d = &env.data; |
| 56 |
let text = |v: &serde_json::Value| v.as_str().unwrap_or("").to_string(); |
| 57 |
let num = |v: &serde_json::Value| v.as_i64().unwrap_or(0); |
| 58 |
let actor = if env.actor.is_empty() { "Someone".to_string() } else { env.actor.clone() }; |
| 59 |
let issue_n = if d["issue"]["number"].is_i64() { num(&d["issue"]["number"]) } else { num(&d["number"]) }; |
| 60 |
let issue_title = text(&d["issue"]["title"]); |
| 61 |
match env.event.as_str() { |
| 62 |
"ping" => ( |
| 63 |
"OxygenHub is connected".into(), |
| 64 |
if repo.is_empty() { base.to_string() } else { page("") }, |
| 65 |
"Events from OxygenHub will arrive here.".into(), |
| 66 |
), |
| 67 |
"push" => { |
| 68 |
let lines: Vec<String> = d["updates"] |
| 69 |
.as_array() |
| 70 |
.map(|u| { |
| 71 |
u.iter() |
| 72 |
.map(|x| { |
| 73 |
let n = num(&x["changes"]); |
| 74 |
format!("{}: {n} change{}", text(&x["reference"]), if n == 1 { "" } else { "s" }) |
| 75 |
}) |
| 76 |
.collect() |
| 77 |
}) |
| 78 |
.unwrap_or_default(); |
| 79 |
let first = d["updates"][0]["reference"].as_str().unwrap_or("").to_string(); |
| 80 |
(format!("{actor} pushed to {repo}"), page(&format!("/changes/{first}")), lines.join("\n")) |
| 81 |
} |
| 82 |
"issue.opened" => { |
| 83 |
(format!("{actor} opened issue #{issue_n}: {issue_title}"), page(&format!("/issue/{issue_n}")), text(&d["issue"]["body"])) |
| 84 |
} |
| 85 |
"issue.closed" | "issue.reopened" | "issue.edited" => { |
| 86 |
let verb = env.event.trim_start_matches("issue."); |
| 87 |
(format!("{actor} {verb} issue #{issue_n}: {issue_title}"), page(&format!("/issue/{issue_n}")), String::new()) |
| 88 |
} |
| 89 |
"comment.issue" => { |
| 90 |
let body = if d["comment"]["body"].is_string() { text(&d["comment"]["body"]) } else { String::new() }; |
| 91 |
( |
| 92 |
format!( |
| 93 |
"{actor} commented on issue #{issue_n}{}", |
| 94 |
if issue_title.is_empty() { String::new() } else { format!(": {issue_title}") } |
| 95 |
), |
| 96 |
page(&format!("/issue/{issue_n}")), |
| 97 |
body, |
| 98 |
) |
| 99 |
} |
| 100 |
"review" => ( |
| 101 |
format!("{actor} reviewed proposal #{}: {}", num(&d["number"]), text(&d["verdict"])), |
| 102 |
page(&format!("/proposal/{}", num(&d["number"]))), |
| 103 |
String::new(), |
| 104 |
), |
| 105 |
"release.published" => ( |
| 106 |
format!("{actor} released {} ({})", text(&d["title"]), text(&d["tag"])), |
| 107 |
page(&format!("/releases/tag/{}", text(&d["tag"]))), |
| 108 |
String::new(), |
| 109 |
), |
| 110 |
"check" => ( |
| 111 |
format!("Check {} {} on {repo}", text(&d["check"]["name"]), text(&d["check"]["status"])), |
| 112 |
text(&d["check"]["url"]).chars().take(500).collect::<String>().replace(' ', "%20"), |
| 113 |
text(&d["check"]["summary"]), |
| 114 |
), |
| 115 |
e if e.starts_with("proposal.") => { |
| 116 |
let n = num(&d["number"]); |
| 117 |
let verb = e.trim_start_matches("proposal."); |
| 118 |
let title = text(&d["title"]); |
| 119 |
( |
| 120 |
format!("{actor} {verb} proposal #{n}{}", if title.is_empty() { String::new() } else { format!(": {title}") }), |
| 121 |
page(&format!("/proposal/{n}")), |
| 122 |
String::new(), |
| 123 |
) |
| 124 |
} |
| 125 |
e => (format!("{e} on {repo}"), page(""), String::new()), |
| 126 |
} |
| 127 |
} |
| 128 |
|
| 129 |
fn cut(s: &str, max: usize) -> String { |
| 130 |
if s.chars().count() <= max { |
| 131 |
s.to_string() |
| 132 |
} else { |
| 133 |
format!("{}…", s.chars().take(max.saturating_sub(1)).collect::<String>()) |
| 134 |
} |
| 135 |
} |
| 136 |
|
| 137 |
|
| 138 |
pub fn body_for(format: &str, payload: &str, base: &str) -> String { |
| 139 |
let Ok(env) = serde_json::from_str::<WebhookEnvelope>(payload) else { return payload.to_string() }; |
| 140 |
let (title, url, detail) = summary(&env, base); |
| 141 |
let repo = env.repository.as_ref().map(|r| format!("{}/{}", r.owner, r.name)).unwrap_or_else(|| "OxygenHub".into()); |
| 142 |
match format { |
| 143 |
"discord" => { |
| 144 |
let mut embed = serde_json::json!({ |
| 145 |
"title": cut(&title, 256), |
| 146 |
"description": cut(&detail, 2000), |
| 147 |
"color": 0x0064e0, |
| 148 |
"footer": { "text": cut(&format!("{repo} · {}", env.event), 200) }, |
| 149 |
"timestamp": crate::render::time::rfc3339(env.at_ms as u64), |
| 150 |
}); |
| 151 |
if url.starts_with("http") { |
| 152 |
embed["url"] = serde_json::Value::String(url); |
| 153 |
} |
| 154 |
|
| 155 |
serde_json::json!({ "username": "OxygenHub", "allowed_mentions": { "parse": [] }, "embeds": [embed] }).to_string() |
| 156 |
} |
| 157 |
"slack" => { |
| 158 |
let esc = |s: &str| s.replace('&', "&").replace('<', "<").replace('>', ">"); |
| 159 |
let head = if url.starts_with("http") { format!("<{}|{}>", url, esc(&title)) } else { esc(&title) }; |
| 160 |
let text = if detail.is_empty() { head } else { format!("{head}\n>{}", esc(&cut(&detail, 1500)).replace('\n', "\n>")) }; |
| 161 |
serde_json::json!({ "text": text }).to_string() |
| 162 |
} |
| 163 |
_ => payload.to_string(), |
| 164 |
} |
| 165 |
} |
| 166 |
|
| 167 |
|
| 168 |
pub fn signature(secret: &str, body: &[u8]) -> String { |
| 169 |
let key = blake3::hash(secret.as_bytes()); |
| 170 |
format!("blake3-keyed={}", hex::encode(blake3::keyed_hash(key.as_bytes(), body).as_bytes())) |
| 171 |
} |
| 172 |
|
| 173 |
|
| 174 |
pub async fn deliver_due(app: &Shared) -> usize { |
| 175 |
let due = match hdb::due_deliveries(&app.db, 50).await { |
| 176 |
Ok(d) => d, |
| 177 |
Err(_) => return 0, |
| 178 |
}; |
| 179 |
let mut n = 0; |
| 180 |
for d in due { |
| 181 |
let Ok(Some(hook)) = hdb::hook(&app.db, d.hook_id).await else { continue }; |
| 182 |
n += 1; |
| 183 |
|
| 184 |
let client = match crate::net::client(&hook.url, &app.config.outbound.allow, std::time::Duration::from_secs(20)).await { |
| 185 |
Ok(c) => c, |
| 186 |
Err(why) => { |
| 187 |
hdb::record_attempt(&app.db, d.id, d.attempts, None, &format!("not sent: {why}")).await.ok(); |
| 188 |
continue; |
| 189 |
} |
| 190 |
}; |
| 191 |
let body = body_for(&hook.format, &d.payload, app.config.base_url()); |
| 192 |
let (status, response) = match client |
| 193 |
.post(&hook.url) |
| 194 |
.header("content-type", "application/json") |
| 195 |
.header("x-oxyhub-event", &d.event) |
| 196 |
.header("x-oxyhub-delivery", &d.delivery) |
| 197 |
.header("x-oxyhub-signature", signature(&hook.secret, body.as_bytes())) |
| 198 |
.body(body) |
| 199 |
.send() |
| 200 |
.await |
| 201 |
{ |
| 202 |
Ok(r) => { |
| 203 |
let status = r.status().as_u16() as i64; |
| 204 |
let text = r.text().await.unwrap_or_default(); |
| 205 |
(Some(status), text) |
| 206 |
} |
| 207 |
Err(e) => (None, e.to_string()), |
| 208 |
}; |
| 209 |
hdb::record_attempt(&app.db, d.id, d.attempts, status, &response).await.ok(); |
| 210 |
} |
| 211 |
n |
| 212 |
} |
| 213 |
|
| 214 |
|
| 215 |
pub async fn worker(app: Shared) { |
| 216 |
let mut rx = app.events.deliveries.subscribe(); |
| 217 |
loop { |
| 218 |
deliver_due(&app).await; |
| 219 |
let cutoff = db::now_ms() - app.config.retention.webhook_log_days as i64 * 86_400_000; |
| 220 |
hdb::prune_deliveries(&app.db, cutoff).await.ok(); |
| 221 |
tokio::select! { |
| 222 |
_ = rx.recv() => {} |
| 223 |
_ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {} |
| 224 |
} |
| 225 |
} |
| 226 |
} |
| 227 |
|
| 228 |
#[cfg(test)] |
| 229 |
mod tests { |
| 230 |
use super::*; |
| 231 |
|
| 232 |
fn envelope(event: &str, data: serde_json::Value) -> String { |
| 233 |
serde_json::to_string(&WebhookEnvelope { |
| 234 |
event: event.into(), |
| 235 |
delivery: "d".into(), |
| 236 |
repository: Some(hub_api::Repository { |
| 237 |
owner: "andymccall".into(), |
| 238 |
name: "oxygen".into(), |
| 239 |
description: String::new(), |
| 240 |
default_bookmark: "main".into(), |
| 241 |
visibility: hub_api::Visibility::Public, |
| 242 |
created_ms: 0, |
| 243 |
updated_ms: 0, |
| 244 |
clone_url: String::new(), |
| 245 |
}), |
| 246 |
actor: "andy".into(), |
| 247 |
at_ms: 1_790_000_000_000, |
| 248 |
data, |
| 249 |
}) |
| 250 |
.unwrap() |
| 251 |
} |
| 252 |
|
| 253 |
#[test] |
| 254 |
fn discord_gets_an_embed_without_mentions() { |
| 255 |
let p = envelope( |
| 256 |
"issue.opened", |
| 257 |
serde_json::json!({ "issue": { "number": 12, "title": "@everyone crash", "body": "x".repeat(3000) } }), |
| 258 |
); |
| 259 |
let v: serde_json::Value = serde_json::from_str(&body_for("discord", &p, "https://hub.example:7343/")).unwrap(); |
| 260 |
let e = &v["embeds"][0]; |
| 261 |
assert_eq!(e["title"], "andy opened issue #12: @everyone crash"); |
| 262 |
assert_eq!(e["url"], "https://hub.example:7343/andymccall/oxygen/issue/12"); |
| 263 |
assert_eq!(e["description"].as_str().unwrap().chars().count(), 2000, "cut to Discord's limits"); |
| 264 |
assert_eq!(v["allowed_mentions"]["parse"], serde_json::json!([]), "no pings from user text"); |
| 265 |
assert_eq!(v["username"], "OxygenHub"); |
| 266 |
let push = envelope("push", serde_json::json!({ "updates": [{ "reference": "main", "changes": 3 }] })); |
| 267 |
let v: serde_json::Value = serde_json::from_str(&body_for("discord", &push, "https://h")).unwrap(); |
| 268 |
assert_eq!(v["embeds"][0]["title"], "andy pushed to andymccall/oxygen"); |
| 269 |
assert_eq!(v["embeds"][0]["description"], "main: 3 changes"); |
| 270 |
assert_eq!(v["embeds"][0]["url"], "https://h/andymccall/oxygen/changes/main"); |
| 271 |
let ping = envelope("ping", serde_json::json!({})); |
| 272 |
assert!(body_for("discord", &ping, "https://h").contains("OxygenHub is connected")); |
| 273 |
} |
| 274 |
|
| 275 |
#[test] |
| 276 |
fn slack_text_is_escaped_and_the_envelope_is_unchanged() { |
| 277 |
let p = envelope( |
| 278 |
"comment.issue", |
| 279 |
serde_json::json!({ "number": 3, "issue": { "number": 3, "title": "<b>" }, "comment": { "body": "a & b" } }), |
| 280 |
); |
| 281 |
let v: serde_json::Value = serde_json::from_str(&body_for("slack", &p, "https://h")).unwrap(); |
| 282 |
assert_eq!(v["text"], "<https://h/andymccall/oxygen/issue/3|andy commented on issue #3: <b>>\n>a & b"); |
| 283 |
assert_eq!(body_for("oxygenhub", &p, "https://h"), p); |
| 284 |
} |
| 285 |
|
| 286 |
#[test] |
| 287 |
fn formats_follow_the_url_and_urls_are_masked() { |
| 288 |
use crate::db::hooks::{format_for_url, masked_url}; |
| 289 |
assert_eq!(format_for_url("https://discord.com/api/webhooks/155/abc"), "discord"); |
| 290 |
assert_eq!(format_for_url("https://hooks.slack.com/services/T/B/x"), "slack"); |
| 291 |
assert_eq!(format_for_url("https://discord.com.evil.example/api/webhooks/1"), "oxygenhub"); |
| 292 |
assert_eq!(format_for_url("https://ci.example/hook"), "oxygenhub"); |
| 293 |
let m = masked_url("https://discord.com/api/webhooks/1554146511347126332/6g_WnXRdNOiVoo8sGXXW_OuIUceluD8lIUckxb9KHPwZ895XE"); |
| 294 |
assert_eq!(m, "https://discord.com/api/webhooks/1554…/6g_W…"); |
| 295 |
assert!(!m.contains("6g_WnX"), "the token is not shown"); |
| 296 |
assert_eq!(masked_url("https://ci.example/hook?token=secret"), "https://ci.example/hook?…"); |
| 297 |
} |
| 298 |
|
| 299 |
#[test] |
| 300 |
fn signatures_are_keyed() { |
| 301 |
let a = signature("s1", b"body"); |
| 302 |
assert!(a.starts_with("blake3-keyed=")); |
| 303 |
assert_ne!(a, signature("s2", b"body")); |
| 304 |
assert_ne!(a, signature("s1", b"other")); |
| 305 |
} |
| 306 |
} |