OXYGENHUB
# 7dbb41b09a99 oxygenhub / crates / hub / src / hooks.rs
14.7 KiB · Rust · last changed Live pages L1: an event bus of topics, versions and a replay buffer 20 days ago source blame history raw
1
//! Webhooks (hub spec §12.2): queueing an event for every hook that wants
2
//! it, signing and delivering with retries, and the payload envelope.
3
4
use crate::db::{self, hooks as hdb};
5
use crate::Shared;
6
use hub_api::WebhookEnvelope;
7
8
/// Queue `event` for the hooks of a repository (and its organisation).
9
pub async fn emit(app: &Shared, repo: &db::repos::RepoRow, event: &str, actor: &str, data: serde_json::Value) {
10
    // Live pages hear of it whether or not any hook does (oxygenhub#203).
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
        match hdb::queue_delivery(&app.db, h.id, &delivery, event, &payload).await {
44
            Ok(Some(_)) => {}
45
            Ok(None) => tracing::warn!("webhook {} has {} deliveries waiting: {event} not queued", h.id, hdb::MAX_WAITING),
46
            Err(e) => tracing::warn!("queueing delivery: {e}"),
47
        }
48
    }
49
    app.events.notify_deliveries();
50
}
51
52
/// What a chat message says about an event: a title, the page it links
53
/// to (absolute) and a short detail.
54
pub fn summary(env: &WebhookEnvelope, base: &str) -> (String, String, String) {
55
    let repo = env.repository.as_ref().map(|r| format!("{}/{}", r.owner, r.name)).unwrap_or_default();
56
    let page = |path: &str| format!("{}/{repo}{path}", base.trim_end_matches('/'));
57
    let d = &env.data;
58
    let text = |v: &serde_json::Value| v.as_str().unwrap_or("").to_string();
59
    let num = |v: &serde_json::Value| v.as_i64().unwrap_or(0);
60
    let actor = if env.actor.is_empty() { "Someone".to_string() } else { env.actor.clone() };
61
    let issue_n = if d["issue"]["number"].is_i64() { num(&d["issue"]["number"]) } else { num(&d["number"]) };
62
    let issue_title = text(&d["issue"]["title"]);
63
    match env.event.as_str() {
64
        "ping" => (
65
            "OxygenHub is connected".into(),
66
            if repo.is_empty() { base.to_string() } else { page("") },
67
            "Events from OxygenHub will arrive here.".into(),
68
        ),
69
        "push" => {
70
            let lines: Vec<String> = d["updates"]
71
                .as_array()
72
                .map(|u| {
73
                    u.iter()
74
                        .map(|x| {
75
                            let n = num(&x["changes"]);
76
                            format!("{}: {n} change{}", text(&x["reference"]), if n == 1 { "" } else { "s" })
77
                        })
78
                        .collect()
79
                })
80
                .unwrap_or_default();
81
            let first = d["updates"][0]["reference"].as_str().unwrap_or("").to_string();
82
            (format!("{actor} pushed to {repo}"), page(&format!("/changes/{first}")), lines.join("\n"))
83
        }
84
        "issue.opened" => {
85
            (format!("{actor} opened issue #{issue_n}: {issue_title}"), page(&format!("/issue/{issue_n}")), text(&d["issue"]["body"]))
86
        }
87
        "issue.closed" | "issue.reopened" | "issue.edited" => {
88
            let verb = env.event.trim_start_matches("issue.");
89
            (format!("{actor} {verb} issue #{issue_n}: {issue_title}"), page(&format!("/issue/{issue_n}")), String::new())
90
        }
91
        "comment.issue" => {
92
            let body = if d["comment"]["body"].is_string() { text(&d["comment"]["body"]) } else { String::new() };
93
            (
94
                format!(
95
                    "{actor} commented on issue #{issue_n}{}",
96
                    if issue_title.is_empty() { String::new() } else { format!(": {issue_title}") }
97
                ),
98
                page(&format!("/issue/{issue_n}")),
99
                body,
100
            )
101
        }
102
        "review" => (
103
            format!("{actor} reviewed proposal #{}: {}", num(&d["number"]), text(&d["verdict"])),
104
            page(&format!("/proposal/{}", num(&d["number"]))),
105
            String::new(),
106
        ),
107
        "release.published" => (
108
            format!("{actor} released {} ({})", text(&d["title"]), text(&d["tag"])),
109
            page(&format!("/releases/tag/{}", text(&d["tag"]))),
110
            String::new(),
111
        ),
112
        "check" => (
113
            format!("Check {} {} on {repo}", text(&d["check"]["name"]), text(&d["check"]["status"])),
114
            text(&d["check"]["url"]).chars().take(500).collect::<String>().replace(' ', "%20"),
115
            text(&d["check"]["summary"]),
116
        ),
117
        e if e.starts_with("proposal.") => {
118
            let n = num(&d["number"]);
119
            let verb = e.trim_start_matches("proposal.");
120
            let title = text(&d["title"]);
121
            (
122
                format!("{actor} {verb} proposal #{n}{}", if title.is_empty() { String::new() } else { format!(": {title}") }),
123
                page(&format!("/proposal/{n}")),
124
                String::new(),
125
            )
126
        }
127
        e => (format!("{e} on {repo}"), page(""), String::new()),
128
    }
129
}
130
131
fn cut(s: &str, max: usize) -> String {
132
    if s.chars().count() <= max {
133
        s.to_string()
134
    } else {
135
        format!("{}…", s.chars().take(max.saturating_sub(1)).collect::<String>())
136
    }
137
}
138
139
/// The body a hook receives in its format; the envelope itself for `oxygenhub`.
140
pub fn body_for(format: &str, payload: &str, base: &str) -> String {
141
    let Ok(env) = serde_json::from_str::<WebhookEnvelope>(payload) else { return payload.to_string() };
142
    let (title, url, detail) = summary(&env, base);
143
    let repo = env.repository.as_ref().map(|r| format!("{}/{}", r.owner, r.name)).unwrap_or_else(|| "OxygenHub".into());
144
    match format {
145
        "discord" => {
146
            let mut embed = serde_json::json!({
147
                "title": cut(&title, 256),
148
                "description": cut(&detail, 2000),
149
                "color": 0x0064e0,
150
                "footer": { "text": cut(&format!("{repo} · {}", env.event), 200) },
151
                "timestamp": crate::render::time::rfc3339(env.at_ms as u64),
152
            });
153
            if url.starts_with("http") {
154
                embed["url"] = serde_json::Value::String(url);
155
            }
156
            // No @everyone, @here or role pings from anything a user typed.
157
            serde_json::json!({ "username": "OxygenHub", "allowed_mentions": { "parse": [] }, "embeds": [embed] }).to_string()
158
        }
159
        "slack" => {
160
            let esc = |s: &str| s.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;");
161
            let head = if url.starts_with("http") { format!("<{}|{}>", url, esc(&title)) } else { esc(&title) };
162
            let text = if detail.is_empty() { head } else { format!("{head}\n>{}", esc(&cut(&detail, 1500)).replace('\n', "\n>")) };
163
            serde_json::json!({ "text": text }).to_string()
164
        }
165
        _ => payload.to_string(),
166
    }
167
}
168
169
/// `X-Oxyhub-Signature` for a body: keyed BLAKE3 with the hook's secret.
170
pub fn signature(secret: &str, body: &[u8]) -> String {
171
    let key = blake3::hash(secret.as_bytes());
172
    format!("blake3-keyed={}", hex::encode(blake3::keyed_hash(key.as_bytes(), body).as_bytes()))
173
}
174
175
/// Deliver everything that is due; returns how many were attempted.
176
pub async fn deliver_due(app: &Shared) -> usize {
177
    let due = match hdb::due_deliveries(&app.db, 50).await {
178
        Ok(d) => d,
179
        Err(_) => return 0,
180
    };
181
    let mut n = 0;
182
    for d in due {
183
        let Ok(Some(hook)) = hdb::hook(&app.db, d.hook_id).await else { continue };
184
        n += 1;
185
        // Only public addresses, pinned, no redirects (#219).
186
        let client = match crate::net::client(&hook.url, &app.config.outbound.allow, std::time::Duration::from_secs(20)).await {
187
            Ok(c) => c,
188
            Err(why) => {
189
                hdb::record_attempt(&app.db, d.id, d.attempts, None, &format!("not sent: {why}")).await.ok();
190
                continue;
191
            }
192
        };
193
        let body = body_for(&hook.format, &d.payload, app.config.base_url());
194
        let (status, response) = match client
195
            .post(&hook.url)
196
            .header("content-type", "application/json")
197
            .header("x-oxyhub-event", &d.event)
198
            .header("x-oxyhub-delivery", &d.delivery)
199
            .header("x-oxyhub-signature", signature(&hook.secret, body.as_bytes()))
200
            .body(body)
201
            .send()
202
            .await
203
        {
204
            Ok(r) => {
205
                let status = r.status().as_u16() as i64;
206
                (Some(status), capped_text(r, RESPONSE_LIMIT).await)
207
            }
208
            Err(e) => (None, e.to_string()),
209
        };
210
        hdb::record_attempt(&app.db, d.id, d.attempts, status, &response).await.ok();
211
    }
212
    n
213
}
214
215
/// How much of a webhook's response is read (#220): enough to show why a
216
/// delivery failed, never so much that an endpoint can fill memory.
217
const RESPONSE_LIMIT: usize = 64 * 1024;
218
219
/// A response's body as text, read in chunks and stopped at `limit` bytes.
220
async fn capped_text(mut r: reqwest::Response, limit: usize) -> String {
221
    let mut body = Vec::new();
222
    while let Ok(Some(chunk)) = r.chunk().await {
223
        let room = limit - body.len();
224
        body.extend_from_slice(&chunk[..chunk.len().min(room)]);
225
        if body.len() >= limit {
226
            break;
227
        }
228
    }
229
    String::from_utf8_lossy(&body).into_owned()
230
}
231
232
/// The background worker: delivers when nudged and every minute anyway.
233
pub async fn worker(app: Shared) {
234
    let mut rx = app.events.deliveries.subscribe();
235
    loop {
236
        deliver_due(&app).await;
237
        let cutoff = db::now_ms() - app.config.retention.webhook_log_days as i64 * 86_400_000;
238
        hdb::prune_deliveries(&app.db, cutoff).await.ok();
239
        tokio::select! {
240
            _ = rx.recv() => {}
241
            _ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {}
242
        }
243
    }
244
}
245
246
#[cfg(test)]
247
mod tests {
248
    use super::*;
249
250
    #[tokio::test]
251
    async fn a_response_is_read_only_up_to_the_limit() {
252
        // #220: an endpoint answering 100 KiB is read to 64 KiB, no further.
253
        let big = reqwest::Response::from(axum::http::Response::new("x".repeat(100 * 1024)));
254
        assert_eq!(capped_text(big, RESPONSE_LIMIT).await.len(), RESPONSE_LIMIT);
255
        let small = reqwest::Response::from(axum::http::Response::new("ok".to_string()));
256
        assert_eq!(capped_text(small, RESPONSE_LIMIT).await, "ok");
257
    }
258
259
    fn envelope(event: &str, data: serde_json::Value) -> String {
260
        serde_json::to_string(&WebhookEnvelope {
261
            event: event.into(),
262
            delivery: "d".into(),
263
            repository: Some(hub_api::Repository {
264
                owner: "andymccall".into(),
265
                name: "oxygen".into(),
266
                description: String::new(),
267
                default_bookmark: "main".into(),
268
                visibility: hub_api::Visibility::Public,
269
                created_ms: 0,
270
                updated_ms: 0,
271
                clone_url: String::new(),
272
            }),
273
            actor: "andy".into(),
274
            at_ms: 1_790_000_000_000,
275
            data,
276
        })
277
        .unwrap()
278
    }
279
280
    #[test]
281
    fn discord_gets_an_embed_without_mentions() {
282
        let p = envelope(
283
            "issue.opened",
284
            serde_json::json!({ "issue": { "number": 12, "title": "@everyone crash", "body": "x".repeat(3000) } }),
285
        );
286
        let v: serde_json::Value = serde_json::from_str(&body_for("discord", &p, "https://hub.example:7343/")).unwrap();
287
        let e = &v["embeds"][0];
288
        assert_eq!(e["title"], "andy opened issue #12: @everyone crash");
289
        assert_eq!(e["url"], "https://hub.example:7343/andymccall/oxygen/issue/12");
290
        assert_eq!(e["description"].as_str().unwrap().chars().count(), 2000, "cut to Discord's limits");
291
        assert_eq!(v["allowed_mentions"]["parse"], serde_json::json!([]), "no pings from user text");
292
        assert_eq!(v["username"], "OxygenHub");
293
        let push = envelope("push", serde_json::json!({ "updates": [{ "reference": "main", "changes": 3 }] }));
294
        let v: serde_json::Value = serde_json::from_str(&body_for("discord", &push, "https://h")).unwrap();
295
        assert_eq!(v["embeds"][0]["title"], "andy pushed to andymccall/oxygen");
296
        assert_eq!(v["embeds"][0]["description"], "main: 3 changes");
297
        assert_eq!(v["embeds"][0]["url"], "https://h/andymccall/oxygen/changes/main");
298
        let ping = envelope("ping", serde_json::json!({}));
299
        assert!(body_for("discord", &ping, "https://h").contains("OxygenHub is connected"));
300
    }
301
302
    #[test]
303
    fn slack_text_is_escaped_and_the_envelope_is_unchanged() {
304
        let p = envelope(
305
            "comment.issue",
306
            serde_json::json!({ "number": 3, "issue": { "number": 3, "title": "<b>" }, "comment": { "body": "a & b" } }),
307
        );
308
        let v: serde_json::Value = serde_json::from_str(&body_for("slack", &p, "https://h")).unwrap();
309
        assert_eq!(v["text"], "<https://h/andymccall/oxygen/issue/3|andy commented on issue #3: &lt;b&gt;>\n>a &amp; b");
310
        assert_eq!(body_for("oxygenhub", &p, "https://h"), p);
311
    }
312
313
    #[test]
314
    fn formats_follow_the_url_and_urls_are_masked() {
315
        use crate::db::hooks::{format_for_url, masked_url};
316
        assert_eq!(format_for_url("https://discord.com/api/webhooks/155/abc"), "discord");
317
        assert_eq!(format_for_url("https://hooks.slack.com/services/T/B/x"), "slack");
318
        assert_eq!(format_for_url("https://discord.com.evil.example/api/webhooks/1"), "oxygenhub");
319
        assert_eq!(format_for_url("https://ci.example/hook"), "oxygenhub");
320
        let m = masked_url("https://discord.com/api/webhooks/1554146511347126332/6g_WnXRdNOiVoo8sGXXW_OuIUceluD8lIUckxb9KHPwZ895XE");
321
        assert_eq!(m, "https://discord.com/api/webhooks/1554…/6g_W…");
322
        assert!(!m.contains("6g_WnX"), "the token is not shown");
323
        assert_eq!(masked_url("https://ci.example/hook?token=secret"), "https://ci.example/hook?…");
324
    }
325
326
    #[test]
327
    fn signatures_are_keyed() {
328
        let a = signature("s1", b"body");
329
        assert!(a.starts_with("blake3-keyed="));
330
        assert_ne!(a, signature("s2", b"body"));
331
        assert_ne!(a, signature("s1", b"other"));
332
    }
333
}