Menu
Public repository / main
AkurAI-Build / src/webhooks.rs
Latest change 3367b5f4b77e4085d5cc513b20e67403d2207543 - OLI-272: outbound webhooks with HMAC signatures, delivery log and redelivery; git Basic challenge by Óli
//! Outbound webhooks (OLI-272): per-repository subscriptions to `push`,
//! `pull_request` and `run` events, delivered as JSON POSTs signed with HMAC-SHA256
//! (`X-AkurAI-Signature-256: sha256=<hex>`), with a delivery log, retries with
//! backoff and redelivery on request.
use anyhow::{Context, Result, bail, ensure};
use hmac::{Hmac, Mac};
use rusqlite::{OptionalExtension, params};
use serde::Serialize;
use sha2::Sha256;
use std::process::{Command, Stdio};
use std::time::Duration;
use crate::db::Database;
pub const EVENTS: [&str; 3] = ["push", "pull_request", "run"];
/// Backoff after each failed attempt, in seconds; after the last one the delivery fails.
pub const BACKOFF_SECS: [i64; 5] = [0, 30, 120, 600, 3600];
const MAX_ATTEMPTS: i64 = BACKOFF_SECS.len() as i64;
const MAX_EXCERPT: usize = 240;
const TIMEOUT: Duration = Duration::from_secs(15);
#[derive(Clone, Debug, Serialize, PartialEq, Eq)]
pub struct Webhook {
pub id: i64,
pub repository: String,
pub url: String,
pub events: Vec<String>,
pub active: bool,
pub created_at: i64,
}
pub struct NewWebhook<'a> {
pub url: &'a str,
pub secret: &'a str,
pub events: &'a [String],
}
#[derive(Clone, Debug, Serialize, PartialEq, Eq)]
pub struct Delivery {
pub id: i64,
pub webhook_id: i64,
pub event: String,
pub payload: String,
pub status: String,
pub attempts: i64,
pub next_attempt_at: i64,
pub response_code: Option<i64>,
pub response_excerpt: Option<String>,
pub created_at: i64,
pub delivered_at: Option<i64>,
}
/// What the receiver answered (or why the request did not reach it).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Reply {
pub status: Option<u16>,
pub excerpt: String,
}
/// How a delivery is sent: the real one runs `curl`; tests inject their own.
pub trait Sender: Send + Sync {
fn send(&self, url: &str, headers: &[(String, String)], body: &[u8]) -> Reply;
}
fn repository_id(connection: &rusqlite::Connection, name: &str) -> Result<i64> {
connection
.query_row(
"SELECT id FROM repositories WHERE name = ?1",
[name],
|row| row.get(0),
)
.optional()?
.with_context(|| format!("repository not found: {name}"))
}
fn validate_url(url: &str) -> Result<()> {
ensure!(
url.starts_with("https://")
|| url.starts_with("http://127.0.0.1")
|| url.starts_with("http://localhost"),
"webhook url must be https (or loopback http)"
);
ensure!(
url.len() <= 2048 && !url.chars().any(char::is_whitespace),
"invalid webhook url"
);
Ok(())
}
fn normalise_events(events: &[String]) -> Result<Vec<String>> {
let mut out: Vec<String> = Vec::new();
for event in events {
let event = event.trim().to_ascii_lowercase();
ensure!(
EVENTS.contains(&event.as_str()),
"unknown event {event:?}; use push, pull_request or run"
);
if !out.contains(&event) {
out.push(event);
}
}
ensure!(!out.is_empty(), "at least one event is required");
Ok(out)
}
fn map_webhook(row: &rusqlite::Row<'_>) -> rusqlite::Result<Webhook> {
let events: String = row.get(3)?;
Ok(Webhook {
id: row.get(0)?,
repository: row.get(1)?,
url: row.get(2)?,
events: events.split(',').map(String::from).collect(),
active: row.get::<_, i64>(4)? != 0,
created_at: row.get(5)?,
})
}
const WEBHOOK_SELECT: &str = "SELECT w.id, r.name, w.url, w.events, w.active, w.created_at
FROM webhooks w JOIN repositories r ON r.id = w.repository_id";
pub fn create(database: &Database, repository: &str, new: &NewWebhook<'_>) -> Result<Webhook> {
validate_url(new.url)?;
ensure!(
new.secret.len() >= 8,
"secret must be at least 8 characters"
);
ensure!(new.secret.len() <= 256, "secret is too long");
let events = normalise_events(new.events)?;
let connection = database.connection()?;
let id = repository_id(&connection, repository)?;
connection.execute(
"INSERT INTO webhooks (repository_id, url, secret, events) VALUES (?1, ?2, ?3, ?4)",
params![id, new.url, new.secret, events.join(",")],
)?;
let webhook_id = connection.last_insert_rowid();
connection
.query_row(
&format!("{WEBHOOK_SELECT} WHERE w.id = ?1"),
[webhook_id],
map_webhook,
)
.map_err(Into::into)
}
pub fn list(database: &Database, repository: &str) -> Result<Vec<Webhook>> {
let connection = database.connection()?;
let id = repository_id(&connection, repository)?;
let mut statement = connection.prepare(&format!(
"{WEBHOOK_SELECT} WHERE w.repository_id = ?1 ORDER BY w.id"
))?;
let rows = statement.query_map([id], map_webhook)?;
Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
}
pub fn get(database: &Database, repository: &str, webhook_id: i64) -> Result<Option<Webhook>> {
let connection = database.connection()?;
let id = repository_id(&connection, repository)?;
connection
.query_row(
&format!("{WEBHOOK_SELECT} WHERE w.repository_id = ?1 AND w.id = ?2"),
params![id, webhook_id],
map_webhook,
)
.optional()
.map_err(Into::into)
}
pub fn set_active(
database: &Database,
repository: &str,
webhook_id: i64,
active: bool,
) -> Result<bool> {
let connection = database.connection()?;
let id = repository_id(&connection, repository)?;
Ok(connection.execute(
"UPDATE webhooks SET active = ?3 WHERE repository_id = ?1 AND id = ?2",
params![id, webhook_id, active as i64],
)? > 0)
}
pub fn remove(database: &Database, repository: &str, webhook_id: i64) -> Result<bool> {
let connection = database.connection()?;
let id = repository_id(&connection, repository)?;
Ok(connection.execute(
"DELETE FROM webhooks WHERE repository_id = ?1 AND id = ?2",
params![id, webhook_id],
)? > 0)
}
/// Queue the event for every active webhook of the repository subscribed to
/// it. Returns the delivery ids. Never fails the caller's operation: errors are
/// returned for logging only.
pub fn emit(
database: &Database,
repository: &str,
event: &str,
payload: &serde_json::Value,
now: i64,
) -> Result<Vec<i64>> {
ensure!(EVENTS.contains(&event), "unknown event {event:?}");
let body = serde_json::json!({
"event": event,
"repository": repository,
"at": now,
"data": payload,
})
.to_string();
let hooks: Vec<Webhook> = list(database, repository)?
.into_iter()
.filter(|h| h.active && h.events.iter().any(|e| e == event))
.collect();
let connection = database.connection()?;
let mut ids = Vec::new();
for hook in hooks {
connection.execute(
"INSERT INTO webhook_deliveries (webhook_id, event, payload, status, attempts, next_attempt_at)
VALUES (?1, ?2, ?3, 'pending', 0, ?4)",
params![hook.id, event, body, now],
)?;
ids.push(connection.last_insert_rowid());
}
Ok(ids)
}
fn map_delivery(row: &rusqlite::Row<'_>) -> rusqlite::Result<Delivery> {
Ok(Delivery {
id: row.get(0)?,
webhook_id: row.get(1)?,
event: row.get(2)?,
payload: row.get(3)?,
status: row.get(4)?,
attempts: row.get(5)?,
next_attempt_at: row.get(6)?,
response_code: row.get(7)?,
response_excerpt: row.get(8)?,
created_at: row.get(9)?,
delivered_at: row.get(10)?,
})
}
const DELIVERY_SELECT: &str =
"SELECT id, webhook_id, event, payload, status, attempts, next_attempt_at,
response_code, response_excerpt, created_at, delivered_at FROM webhook_deliveries";
/// The newest deliveries of a webhook (most recent first), at most `limit`.
pub fn deliveries(
database: &Database,
repository: &str,
webhook_id: i64,
limit: usize,
) -> Result<Vec<Delivery>> {
if get(database, repository, webhook_id)?.is_none() {
bail!("webhook not found");
}
let connection = database.connection()?;
let mut statement = connection.prepare(&format!(
"{DELIVERY_SELECT} WHERE webhook_id = ?1 ORDER BY id DESC LIMIT ?2"
))?;
let rows = statement.query_map(params![webhook_id, limit as i64], map_delivery)?;
Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
}
/// Put a delivery back in the queue for an immediate new attempt.
pub fn redeliver(
database: &Database,
repository: &str,
webhook_id: i64,
delivery_id: i64,
now: i64,
) -> Result<bool> {
if get(database, repository, webhook_id)?.is_none() {
bail!("webhook not found");
}
let connection = database.connection()?;
Ok(connection.execute(
"UPDATE webhook_deliveries SET status = 'pending', attempts = 0, next_attempt_at = ?3
WHERE id = ?1 AND webhook_id = ?2",
params![delivery_id, webhook_id, now],
)? > 0)
}
/// `sha256=<hex HMAC-SHA256 of the body>` for `X-AkurAI-Signature-256`.
pub fn signature(secret: &str, body: &[u8]) -> String {
let mut mac =
Hmac::<Sha256>::new_from_slice(secret.as_bytes()).expect("hmac accepts any key length");
mac.update(body);
let digest = mac.finalize().into_bytes();
let hex: String = digest.iter().map(|b| format!("{b:02x}")).collect();
format!("sha256={hex}")
}
fn due(database: &Database, now: i64, limit: usize) -> Result<Vec<(Delivery, String, String)>> {
let connection = database.connection()?;
let mut statement = connection.prepare(
"SELECT d.id, d.webhook_id, d.event, d.payload, d.status, d.attempts, d.next_attempt_at,
d.response_code, d.response_excerpt, d.created_at, d.delivered_at, w.url, w.secret
FROM webhook_deliveries d JOIN webhooks w ON w.id = d.webhook_id
WHERE d.status = 'pending' AND d.next_attempt_at <= ?1 AND w.active = 1
ORDER BY d.next_attempt_at, d.id LIMIT ?2",
)?;
let rows = statement.query_map(params![now, limit as i64], |row| {
Ok((
map_delivery(row)?,
row.get::<_, String>(11)?,
row.get::<_, String>(12)?,
))
})?;
Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
}
/// Send every due delivery once; failures are rescheduled with backoff until the
/// attempts run out. Returns (delivered, failed-this-round).
pub fn deliver_due(database: &Database, sender: &dyn Sender, now: i64) -> Result<(usize, usize)> {
let mut delivered = 0;
let mut failed = 0;
for (delivery, url, secret) in due(database, now, 50)? {
let body = delivery.payload.as_bytes();
let headers = vec![
("Content-Type".to_string(), "application/json".to_string()),
(
"User-Agent".to_string(),
"AkurAI-Build-Hookshot".to_string(),
),
("X-AkurAI-Event".to_string(), delivery.event.clone()),
("X-AkurAI-Delivery".to_string(), delivery.id.to_string()),
(
"X-AkurAI-Signature-256".to_string(),
signature(&secret, body),
),
];
let reply = sender.send(&url, &headers, body);
let attempts = delivery.attempts + 1;
let ok = reply.status.is_some_and(|s| (200..300).contains(&s));
let excerpt: String = reply.excerpt.chars().take(MAX_EXCERPT).collect();
let connection = database.connection()?;
if ok {
delivered += 1;
connection.execute(
"UPDATE webhook_deliveries SET status = 'delivered', attempts = ?2, response_code = ?3,
response_excerpt = ?4, delivered_at = ?5 WHERE id = ?1",
params![delivery.id, attempts, reply.status.map(i64::from), excerpt, now],
)?;
} else if attempts >= MAX_ATTEMPTS {
failed += 1;
connection.execute(
"UPDATE webhook_deliveries SET status = 'failed', attempts = ?2, response_code = ?3,
response_excerpt = ?4 WHERE id = ?1",
params![delivery.id, attempts, reply.status.map(i64::from), excerpt],
)?;
} else {
failed += 1;
let wait = BACKOFF_SECS[attempts as usize];
connection.execute(
"UPDATE webhook_deliveries SET attempts = ?2, response_code = ?3, response_excerpt = ?4,
next_attempt_at = ?5 WHERE id = ?1",
params![delivery.id, attempts, reply.status.map(i64::from), excerpt, now + wait],
)?;
}
}
Ok((delivered, failed))
}
/// Sends with `curl` (already the forge's outbound HTTP tool): no redirects,
/// bounded time, the status code read from the response.
pub struct CurlSender;
impl Sender for CurlSender {
fn send(&self, url: &str, headers: &[(String, String)], body: &[u8]) -> Reply {
let mut command = Command::new("curl");
command
.args([
"--silent",
"--show-error",
"--max-time",
&TIMEOUT.as_secs().to_string(),
])
.args(["--output", "/dev/null", "--write-out", "%{http_code}"])
.args(["-X", "POST", "--data-binary", "@-"]);
for (name, value) in headers {
command.arg("-H").arg(format!("{name}: {value}"));
}
command.arg(url);
let child = command
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn();
let mut child = match child {
Ok(child) => child,
Err(error) => {
return Reply {
status: None,
excerpt: format!("could not start curl: {error}"),
};
}
};
if let Some(mut stdin) = child.stdin.take() {
use std::io::Write;
let _ = stdin.write_all(body);
}
match child.wait_with_output() {
Ok(output) => {
let code = String::from_utf8_lossy(&output.stdout)
.trim()
.parse::<u16>()
.ok();
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
Reply {
status: code.filter(|c| *c != 0),
excerpt: if stderr.is_empty() {
code.map(|c| format!("HTTP {c}")).unwrap_or_default()
} else {
stderr
},
}
}
Err(error) => Reply {
status: None,
excerpt: error.to_string(),
},
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
const KEY: &str = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
fn database() -> Database {
let db = Database::memory(KEY).expect("db");
db.add_repository("demo", "https://example.com/demo.git", "main")
.expect("repo");
db
}
type Call = (String, Vec<(String, String)>, String);
struct Fake {
calls: Mutex<Vec<Call>>,
status: Mutex<Option<u16>>,
}
impl Fake {
fn answering(status: Option<u16>) -> Fake {
Fake {
calls: Mutex::new(Vec::new()),
status: Mutex::new(status),
}
}
}
impl Sender for Fake {
fn send(&self, url: &str, headers: &[(String, String)], body: &[u8]) -> Reply {
self.calls.lock().expect("lock").push((
url.to_string(),
headers.to_vec(),
String::from_utf8_lossy(body).to_string(),
));
let status = *self.status.lock().expect("lock");
Reply {
status,
excerpt: status.map_or("connection refused".into(), |s| format!("HTTP {s}")),
}
}
}
fn hook(db: &Database, events: &[&str]) -> Webhook {
let events: Vec<String> = events.iter().map(|e| e.to_string()).collect();
create(
db,
"demo",
&NewWebhook {
url: "https://hooks.example.com/in",
secret: "s3cret-s3cret",
events: &events,
},
)
.expect("hook")
}
#[test]
fn hooks_are_created_listed_validated_and_removed() {
let db = database();
let h = hook(&db, &["push", "run"]);
assert_eq!(h.events, vec!["push", "run"]);
assert!(h.active);
assert_eq!(list(&db, "demo").expect("list").len(), 1);
let bad = create(
&db,
"demo",
&NewWebhook {
url: "http://evil.example/x",
secret: "s3cret-s3cret",
events: &["push".into()],
},
);
assert!(bad.is_err(), "plain http to the internet is refused");
assert!(
create(
&db,
"demo",
&NewWebhook {
url: "https://x.example/",
secret: "short",
events: &["push".into()]
}
)
.is_err()
);
assert!(
create(
&db,
"demo",
&NewWebhook {
url: "https://x.example/",
secret: "s3cret-s3cret",
events: &["nope".into()]
}
)
.is_err()
);
assert!(set_active(&db, "demo", h.id, false).expect("ok"));
assert!(!get(&db, "demo", h.id).expect("ok").expect("hook").active);
assert!(remove(&db, "demo", h.id).expect("ok"));
assert!(get(&db, "demo", h.id).expect("ok").is_none());
}
#[test]
fn emit_queues_only_subscribed_active_hooks() {
let db = database();
let push_only = hook(&db, &["push"]);
let all = hook(&db, &["push", "pull_request", "run"]);
let off = hook(&db, &["run"]);
set_active(&db, "demo", off.id, false).expect("ok");
let ids = emit(&db, "demo", "run", &serde_json::json!({"id": 7}), 100).expect("emit");
assert_eq!(ids.len(), 1);
assert_eq!(deliveries(&db, "demo", all.id, 10).expect("ok").len(), 1);
assert!(
deliveries(&db, "demo", push_only.id, 10)
.expect("ok")
.is_empty()
);
assert!(deliveries(&db, "demo", off.id, 10).expect("ok").is_empty());
assert!(emit(&db, "demo", "bogus", &serde_json::json!({}), 100).is_err());
let d = &deliveries(&db, "demo", all.id, 10).expect("ok")[0];
let payload: serde_json::Value = serde_json::from_str(&d.payload).expect("json");
assert_eq!(payload["event"], "run");
assert_eq!(payload["repository"], "demo");
assert_eq!(payload["data"]["id"], 7);
}
#[test]
fn deliveries_are_signed_retried_with_backoff_and_redeliverable() {
let db = database();
let h = hook(&db, &["push"]);
emit(
&db,
"demo",
"push",
&serde_json::json!({"ref": "refs/heads/main"}),
1000,
)
.expect("emit");
let down = Fake::answering(None);
assert_eq!(deliver_due(&db, &down, 1000).expect("ok"), (0, 1));
let d = &deliveries(&db, "demo", h.id, 10).expect("ok")[0];
assert_eq!(
(d.status.as_str(), d.attempts, d.next_attempt_at),
("pending", 1, 1000 + BACKOFF_SECS[1])
);
assert_eq!(
deliver_due(&db, &down, 1001).expect("ok"),
(0, 0),
"not due yet"
);
let mut now = 1000;
for attempt in 2..=MAX_ATTEMPTS {
now += BACKOFF_SECS[(attempt - 1) as usize];
deliver_due(&db, &down, now).expect("ok");
}
let d = &deliveries(&db, "demo", h.id, 10).expect("ok")[0];
assert_eq!((d.status.as_str(), d.attempts), ("failed", MAX_ATTEMPTS));
assert!(redeliver(&db, "demo", h.id, d.id, now).expect("ok"));
let up = Fake::answering(Some(204));
assert_eq!(deliver_due(&db, &up, now).expect("ok"), (1, 0));
let d = &deliveries(&db, "demo", h.id, 10).expect("ok")[0];
assert_eq!(
(d.status.as_str(), d.response_code),
("delivered", Some(204))
);
let calls = up.calls.lock().expect("lock");
let (url, headers, body) = &calls[0];
assert_eq!(url, "https://hooks.example.com/in");
let header = |n: &str| {
headers
.iter()
.find(|(k, _)| k == n)
.map(|(_, v)| v.clone())
.expect(n)
};
assert_eq!(header("X-AkurAI-Event"), "push");
assert_eq!(header("X-AkurAI-Delivery"), d.id.to_string());
assert_eq!(
header("X-AkurAI-Signature-256"),
signature("s3cret-s3cret", body.as_bytes())
);
assert!(body.contains("refs/heads/main"));
}
#[test]
fn signature_is_github_style_hmac_sha256() {
assert_eq!(
signature("It's a Secret to Everybody", b"Hello, World!"),
"sha256=757107ea0eb2509fc211221cce984b8a37570b6d7586c22c46f4379c8b043e17"
);
}
}