Menu
Public repository / 3367b5f4b77e4085d5cc513b20e67403d2207543
AkurAI-Build / src/server/webhooks_api.rs
Latest change 3367b5f4b77e4085d5cc513b20e67403d2207543 - OLI-272: outbound webhooks with HMAC signatures, delivery log and redelivery; git Basic challenge by Óli
//! Outbound webhook routes (OLI-272) and the push event the post-receive hook posts.
use axum::{
Json,
extract::{Path, Query, State},
http::{HeaderMap, StatusCode},
};
use serde::Deserialize;
use super::access_api::{Identity, identify, role_on};
use super::{ApiError, AppState, Envelope, authorize, ok, unix_time};
use crate::access::Role;
use crate::webhooks::{self, Delivery, NewWebhook, Webhook};
fn require_admin(headers: &HeaderMap, state: &AppState, repository: &str) -> Result<(), ApiError> {
let identity = identify(headers, state)?;
match role_on(&identity, state, repository)? {
Some(role) if role.allows(Role::Admin) => Ok(()),
_ => Err(ApiError::forbidden(format!(
"admin access to {repository} is required"
))),
}
}
#[derive(Deserialize)]
pub(super) struct WebhookRequest {
url: String,
secret: String,
events: Vec<String>,
}
#[derive(Deserialize)]
pub(super) struct ActiveRequest {
active: bool,
}
#[derive(Deserialize)]
pub(super) struct DeliveriesQuery {
#[serde(default)]
limit: Option<usize>,
}
#[derive(Deserialize)]
pub(super) struct PushEvent {
reference: String,
before: String,
after: String,
#[serde(default)]
pusher: String,
}
pub(super) async fn list_webhooks(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
) -> Result<Json<Envelope<Vec<Webhook>>>, ApiError> {
require_admin(&headers, &state, &repository)?;
let database = state.database.clone();
let list = tokio::task::spawn_blocking(move || webhooks::list(&database, &repository))
.await
.map_err(ApiError::internal)?
.map_err(ApiError::bad)?;
Ok(ok(list))
}
pub(super) async fn create_webhook(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
Json(request): Json<WebhookRequest>,
) -> Result<(StatusCode, Json<Envelope<Webhook>>), ApiError> {
require_admin(&headers, &state, &repository)?;
let database = state.database.clone();
let created = tokio::task::spawn_blocking(move || {
webhooks::create(
&database,
&repository,
&NewWebhook {
url: request.url.trim(),
secret: &request.secret,
events: &request.events,
},
)
})
.await
.map_err(ApiError::internal)?
.map_err(ApiError::bad)?;
Ok((StatusCode::CREATED, ok(created)))
}
pub(super) async fn set_webhook_active(
State(state): State<AppState>,
headers: HeaderMap,
Path((repository, id)): Path<(String, i64)>,
Json(request): Json<ActiveRequest>,
) -> Result<Json<Envelope<bool>>, ApiError> {
require_admin(&headers, &state, &repository)?;
let database = state.database.clone();
let changed = tokio::task::spawn_blocking(move || {
webhooks::set_active(&database, &repository, id, request.active)
})
.await
.map_err(ApiError::internal)?
.map_err(ApiError::bad)?;
Ok(ok(changed))
}
pub(super) async fn delete_webhook(
State(state): State<AppState>,
headers: HeaderMap,
Path((repository, id)): Path<(String, i64)>,
) -> Result<Json<Envelope<bool>>, ApiError> {
require_admin(&headers, &state, &repository)?;
let database = state.database.clone();
let removed = tokio::task::spawn_blocking(move || webhooks::remove(&database, &repository, id))
.await
.map_err(ApiError::internal)?
.map_err(ApiError::bad)?;
Ok(ok(removed))
}
pub(super) async fn list_deliveries(
State(state): State<AppState>,
headers: HeaderMap,
Path((repository, id)): Path<(String, i64)>,
Query(query): Query<DeliveriesQuery>,
) -> Result<Json<Envelope<Vec<Delivery>>>, ApiError> {
require_admin(&headers, &state, &repository)?;
let database = state.database.clone();
let limit = query.limit.unwrap_or(50).clamp(1, 500);
let list = tokio::task::spawn_blocking(move || {
webhooks::deliveries(&database, &repository, id, limit)
})
.await
.map_err(ApiError::internal)?
.map_err(ApiError::bad)?;
Ok(ok(list))
}
pub(super) async fn redeliver(
State(state): State<AppState>,
headers: HeaderMap,
Path((repository, id, delivery)): Path<(String, i64, i64)>,
) -> Result<Json<Envelope<bool>>, ApiError> {
require_admin(&headers, &state, &repository)?;
let database = state.database.clone();
let now = unix_time()? as i64;
let queued = tokio::task::spawn_blocking(move || {
webhooks::redeliver(&database, &repository, id, delivery, now)
})
.await
.map_err(ApiError::internal)?
.map_err(ApiError::bad)?;
if !queued {
return Err(ApiError::not_found("delivery not found"));
}
Ok(ok(true))
}
/// `POST /api/repos/{repository}/events/push`: the post-receive hook reporting a
/// ref update (controller token only; the hook has it from the forge).
pub(super) async fn push_event(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
Json(event): Json<PushEvent>,
) -> Result<Json<Envelope<usize>>, ApiError> {
authorize(&headers, &state)?;
if !matches!(identify(&headers, &state)?, Identity::Master) {
return Err(ApiError::forbidden("only the forge reports pushes"));
}
let database = state.database.clone();
let now = unix_time()? as i64;
let payload = serde_json::json!({
"ref": event.reference,
"before": event.before,
"after": event.after,
"pusher": event.pusher,
"deleted": event.after.chars().all(|c| c == '0'),
"created": event.before.chars().all(|c| c == '0'),
});
let queued = tokio::task::spawn_blocking(move || {
webhooks::emit(&database, &repository, "push", &payload, now)
})
.await
.map_err(ApiError::internal)?
.map_err(ApiError::bad)?;
Ok(ok(queued.len()))
}
/// Fire-and-forget emit for handlers that must not fail on webhook trouble.
pub(crate) fn emit_quietly(
database: &crate::db::Database,
repository: &str,
event: &str,
payload: serde_json::Value,
) {
let now = unix_time().map(|t| t as i64).unwrap_or(0);
if let Err(error) = webhooks::emit(database, repository, event, &payload, now) {
tracing::warn!(%repository, event, %error, "webhook emit failed");
}
}
#[cfg(test)]
mod tests {
use super::*;
use axum::{
Router,
body::{Body, to_bytes},
http::{Request, header},
routing::{delete, get, post},
};
use tower::ServiceExt;
fn app() -> (Router, AppState) {
let state = crate::server::tests::test_state("api-token-123", "webhook-secret");
state
.database
.add_repository("demo", "https://example.com/demo.git", "main")
.expect("repo");
let router = Router::new()
.route(
"/api/repos/{repository}/webhooks",
get(list_webhooks).post(create_webhook),
)
.route(
"/api/repos/{repository}/webhooks/{id}",
delete(delete_webhook).patch(set_webhook_active),
)
.route(
"/api/repos/{repository}/webhooks/{id}/deliveries",
get(list_deliveries),
)
.route(
"/api/repos/{repository}/webhooks/{id}/deliveries/{delivery}/redeliver",
post(redeliver),
)
.route("/api/repos/{repository}/events/push", post(push_event))
.with_state(state.clone());
(router, state)
}
async fn send(
app: &Router,
method: &str,
uri: &str,
token: Option<&str>,
body: serde_json::Value,
) -> (StatusCode, serde_json::Value) {
let mut request = Request::builder()
.method(method)
.uri(uri)
.header(header::CONTENT_TYPE, "application/json");
if let Some(token) = token {
request = request.header(header::AUTHORIZATION, format!("Bearer {token}"));
}
let response = app
.clone()
.oneshot(request.body(Body::from(body.to_string())).expect("request"))
.await
.expect("response");
let status = response.status();
let bytes = to_bytes(response.into_body(), usize::MAX)
.await
.expect("body");
(
status,
serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null),
)
}
#[tokio::test]
async fn webhooks_are_managed_over_http_and_a_push_event_queues_a_signed_delivery() {
let (app, state) = app();
let master = Some("api-token-123");
let (status, _) = send(
&app,
"GET",
"/api/repos/demo/webhooks",
None,
serde_json::json!(null),
)
.await;
assert_eq!(status, StatusCode::UNAUTHORIZED);
let (status, created) = send(
&app,
"POST",
"/api/repos/demo/webhooks",
master,
serde_json::json!({"url": "https://hooks.example.com/in", "secret": "s3cret-s3cret", "events": ["push", "run"]}),
)
.await;
assert_eq!(status, StatusCode::CREATED, "{created}");
let id = created["data"]["id"].as_i64().expect("id");
assert!(
created["data"].get("secret").is_none(),
"secrets never leave the forge"
);
let (status, _) = send(&app, "POST", "/api/repos/demo/webhooks", master, serde_json::json!({"url": "http://evil/x", "secret": "s3cret-s3cret", "events": ["push"]})).await;
assert_eq!(status, StatusCode::BAD_REQUEST);
let (status, queued) = send(
&app,
"POST",
"/api/repos/demo/events/push",
master,
serde_json::json!({"reference": "refs/heads/main", "before": "a".repeat(40), "after": "b".repeat(40), "pusher": "ada@example.is"}),
)
.await;
assert_eq!(status, StatusCode::OK, "{queued}");
assert_eq!(queued["data"], 1);
let (_, deliveries) = send(
&app,
"GET",
&format!("/api/repos/demo/webhooks/{id}/deliveries"),
master,
serde_json::json!(null),
)
.await;
let delivery = &deliveries["data"][0];
assert_eq!(delivery["status"], "pending");
let payload: serde_json::Value =
serde_json::from_str(delivery["payload"].as_str().expect("payload")).expect("json");
assert_eq!(payload["data"]["pusher"], "ada@example.is");
assert_eq!(payload["data"]["created"], false);
let delivery_id = delivery["id"].as_i64().expect("id");
let (status, _) = send(
&app,
"POST",
&format!("/api/repos/demo/webhooks/{id}/deliveries/{delivery_id}/redeliver"),
master,
serde_json::json!(null),
)
.await;
assert_eq!(status, StatusCode::OK);
let (status, _) = send(
&app,
"POST",
&format!("/api/repos/demo/webhooks/{id}/deliveries/999/redeliver"),
master,
serde_json::json!(null),
)
.await;
assert_eq!(status, StatusCode::NOT_FOUND);
let (_, off) = send(
&app,
"PATCH",
&format!("/api/repos/demo/webhooks/{id}"),
master,
serde_json::json!({"active": false}),
)
.await;
assert_eq!(off["data"], true);
let (_, queued) = send(&app, "POST", "/api/repos/demo/events/push", master, serde_json::json!({"reference": "refs/heads/main", "before": "a".repeat(40), "after": "c".repeat(40)})).await;
assert_eq!(queued["data"], 0, "inactive hooks get nothing");
let (_, removed) = send(
&app,
"DELETE",
&format!("/api/repos/demo/webhooks/{id}"),
master,
serde_json::json!(null),
)
.await;
assert_eq!(removed["data"], true);
let _ = state;
}
}