Coverage / crates/hub/src/hooks.rs 91.5%

pub async fn emit(app: &Shared, repo: &db::repos::RepoRow, event: &str, actor: &str, data: serde_json::Value) {

    app.bus.hook(repo.id, event, &data);

    if event == "check" {

        if let Some(state) = data["check"]["state"].as_str() {

            let (app, repo, state) = (app.clone(), repo.clone(), state.to_string());

            tokio::spawn(crate::autoland::on_check_task(app, repo, state));

        }

    }

    let hooks = match hdb::hooks_hearing(&app.db, repo.id, &repo.owner).await {

        Ok(h) => h,

    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();

        match hdb::queue_delivery(&app.db, h.id, &delivery, event, &payload).await {

            Ok(Some(_)) => {}

    app.bus.delivery_waiting();

}

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<String> = 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" => {

        "comment.issue" => {

            let body = if d["comment"]["body"].is_string() { text(&d["comment"]["body"]) } else { String::new() };

                format!(

                    if issue_title.is_empty() { String::new() } else { format!(": {issue_title}") }

                page(&format!("/issue/{issue_n}")),

                body,

        "review" => (

        "release.published" => (

        "check" => (

            format!("Check {} {} on {repo}", text(&d["check"]["name"]), text(&d["check"]["status"])),

            text(&d["check"]["url"]).chars().take(500).collect::<String>().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(),

}

fn cut(s: &str, max: usize) -> String {

    if s.chars().count() <= max {

        s.to_string()

        format!("{}…", s.chars().take(max.saturating_sub(1)).collect::<String>())

}

pub fn body_for(format: &str, payload: &str, base: &str) -> String {

    let Ok(env) = serde_json::from_str::<WebhookEnvelope>(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);

            }

            serde_json::json!({ "username": "OxygenHub", "allowed_mentions": { "parse": [] }, "embeds": [embed] }).to_string()

        "slack" => {

            let esc = |s: &str| s.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;");

            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(),

}

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()))

}

pub async fn deliver_due(app: &Shared) -> usize {

    let due = match hdb::due_deliveries(&app.db, 50).await {

        Ok(d) => d,

    let mut n = 0;

    for d in due {

        let Ok(Some(hook)) = hdb::hook(&app.db, d.hook_id).await else { continue };

        n += 1;

        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;

                (Some(status), capped_text(r, RESPONSE_LIMIT).await)

        hdb::record_attempt(&app.db, d.id, d.attempts, status, &response).await.ok();

    n

}

async fn capped_text(mut r: reqwest::Response, limit: usize) -> String {

    let mut body = Vec::new();

    while let Ok(Some(chunk)) = r.chunk().await {

        let room = limit - body.len();

        body.extend_from_slice(&chunk[..chunk.len().min(room)]);

        if body.len() >= limit {

            break;

        }

    String::from_utf8_lossy(&body).into_owned()

}

pub async fn worker(app: Shared) {

    let mut rx = app.bus.subscribe(crate::bus::Topic::Deliveries);

        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)) => {}

    async fn a_response_is_read_only_up_to_the_limit() {

        let big = reqwest::Response::from(axum::http::Response::new("x".repeat(100 * 1024)));

        assert_eq!(capped_text(big, RESPONSE_LIMIT).await.len(), RESPONSE_LIMIT);

        let small = reqwest::Response::from(axum::http::Response::new("ok".to_string()));

        assert_eq!(capped_text(small, RESPONSE_LIMIT).await, "ok");

    }

    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()

    }

    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"));

    }

    fn slack_text_is_escaped_and_the_envelope_is_unchanged() {

        let p = envelope(

            "comment.issue",

            serde_json::json!({ "number": 3, "issue": { "number": 3, "title": "<b>" }, "comment": { "body": "a & b" } }),

        let v: serde_json::Value = serde_json::from_str(&body_for("slack", &p, "https://h")).unwrap();

        assert_eq!(v["text"], "<https://h/andymccall/oxygen/issue/3|andy commented on issue #3: &lt;b&gt;>\n>a &amp; b");

        assert_eq!(body_for("oxygenhub", &p, "https://h"), p);

    }

    fn formats_follow_the_url_and_urls_are_masked() {

        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?…");

    }

    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"));

    }