crates/gunnar-server/src/webhook.rs
//! The event stream's OUTBOUND leg: deliver each event to an external URL (webhooks).
//!
//! ```text
//! git push ──▶ gunnar ──▶ event bus ──▶ this subscriber ──▶ per-endpoint queue ──▶ POST ──▶ your CI
//! │ │
//! └── the client is already done here ──────────────────────────────┘
//! ```
//!
//! # It is a subscriber to the SAME bus, and that is the whole architecture
//!
//! A webhook is a consumer of the event broadcast bus, exactly as the
//! replicator ([`crate::mirror`]) and the console's SSE fan-out are (one implementation). It
//! is NOT a second event source: nothing in the push path calls into this
//! module, `receive-pack` publishes to the bus as it always did, and
//! [`WebhookDelivery::spawn`] subscribes. So the push path cannot be made
//! slower, or made to fail, by a webhook — which is the same hard requirement
//! replication has, for the same reason: a push that succeeded locally has
//! succeeded, whatever a subscriber is doing with the news.
//!
//! Three things follow from being a subscriber:
//!
//! * `EventBus::publish` is a `broadcast::send`: it never blocks and never
//! fails, with or without a webhook attached.
//! * a delivery that cannot keep up **lags the bus**, and the bus answers a
//! lagging subscriber with an `EventGap` rather than with memory. A gap is
//! surfaced per endpoint as a `dropped` count — the receiver missed events
//! and should reconcile over the API, the same drop-with-a-gap contract every
//! event-stream subscriber gets.
//! * the POST runs on a **background worker**, off the broadcast receiver. The
//! subscribe task only ever ENQUEUES, into a bounded queue, and returns. A
//! dead endpoint therefore cannot wedge the bus: the worker blocks on the
//! socket, the subscribe task does not, and past the queue bound events are
//! dropped-with-a-gap rather than buffered without limit. A webhook that
//! wedges the server is worse than no webhook.
//!
//! # The queue bound
//!
//! Each endpoint has its own bounded queue ([`DEFAULT_QUEUE_CAPACITY`]). Past
//! the bound the NEWEST event is refused and the endpoint's `dropped` counter
//! climbs: a receiver that has fallen a thousand events behind is not a
//! queueing problem, it is a receiver that is down, and the answer to a down
//! receiver is a bounded gap it can reconcile from, not an unbounded queue that
//! turns one slow endpoint into an OOM. That is the shape that OOM'd a machine
//! in this tree on 2026-08-03, and it is designed out here by construction.
//!
//! # Signed, and why
//!
//! Each delivery is signed with an `sshsig` under the `gunnar-webhook`
//! namespace ([`gunnar_core::webhook`]), over the exact request body. The
//! receiver verifies with the server's published public key and no shared
//! secret — see the module docs there. An optional HMAC header is set when an
//! endpoint configures a secret, for GitHub-webhook-compatible receivers.
use std::collections::VecDeque;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use anyhow::{Context, Result};
use gunnar_core::ssh_key;
use gunnar_grpc::{EventBus, EventFilter};
use gunnar_proto::{event, Event, EventKind, RepoPath};
use serde::{Deserialize, Serialize};
use crate::stores::ServerStores;
/// The on-disk configuration file, in the data root beside `replication.json`.
pub const WEBHOOK_CONFIG: &str = "webhooks.json";
/// Per-endpoint queue bound. See the module header.
pub const DEFAULT_QUEUE_CAPACITY: usize = 1024;
/// First retry delay and the ceiling it doubles up to. The backoff ladder
/// itself is [`crate::durable::backoff`], shared with the replicator (one implementation).
const RETRY_BASE: Duration = Duration::from_secs(2);
const RETRY_MAX: Duration = Duration::from_secs(300);
/// **A delivery attempt is bounded.** Unlike a replication job — where a dead
/// standby retries forever because the second copy still has to be made — a
/// missed webhook is a missed notification, and retrying a dead endpoint
/// forever with a growing queue is the wedge this module exists to avoid. After
/// this many failures the event is dropped-with-a-gap (the `dropped` counter),
/// the receiver reconciles over the API, and the queue keeps draining.
const MAX_ATTEMPTS: u32 = 5;
/// Per-attempt ceiling on the whole POST, so a black-holing endpoint costs one
/// timeout rather than a stuck worker.
const DELIVER_TIMEOUT: Duration = Duration::from_secs(15);
const TARGET_WEBHOOK: &str = "gunnar::webhook";
// ───────────────────────────────────────────────────────────────────────────
// Configuration — durable, same shape and hazards as replication.json
// ───────────────────────────────────────────────────────────────────────────
/// The set of configured endpoints. One JSON document in the data root.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WebhookConfig {
#[serde(default)]
pub endpoints: Vec<Endpoint>,
}
/// One outbound endpoint, as an operator declares it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Endpoint {
/// The operator's name for it — appears in every log line and status row.
pub name: String,
/// Where deliveries are POSTed. `http://` or `https://`.
pub url: String,
/// Event-kind names to deliver (e.g. `ref.updated`). Empty = every kind.
#[serde(default)]
pub kinds: Vec<String>,
/// Deliver only events about this namespace. Empty = every namespace.
#[serde(default)]
pub namespace: String,
/// Deliver only events about this repository. `None` = every repository.
#[serde(default)]
pub repo: Option<RepoPath>,
/// Optional HMAC shared secret for `X-Hub-Signature-256`. Stored, never
/// returned over the control plane.
#[serde(default)]
pub secret: Option<String>,
#[serde(default)]
pub added_at_unix_ms: i64,
}
impl WebhookConfig {
/// Read `<root>/webhooks.json`, or the empty configuration when it is
/// absent. A missing file is not an error — most servers have no webhooks —
/// but a **malformed** one is, so a typo cannot silently deliver nowhere.
/// An empty or whitespace-only file is the power-cut torn-write artifact and is
/// treated as "no webhooks configured", exactly as `replication.json` is.
pub fn load(root: impl AsRef<Path>) -> Result<Self> {
let path = root.as_ref().join(WEBHOOK_CONFIG);
let text = match std::fs::read_to_string(&path) {
Ok(text) => text,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
return Ok(WebhookConfig::default())
}
Err(err) => return Err(err).with_context(|| format!("reading {}", path.display())),
};
if text.trim().is_empty() {
tracing::warn!(
target: TARGET_WEBHOOK,
event = "webhook.config_empty",
path = %path.display(),
"webhooks.json is empty — the signature of a power cut during a save. Treating \
it as no webhooks configured and coming up serving; it is rewritten on the next \
apply."
);
return Ok(WebhookConfig::default());
}
serde_json::from_str(&text).with_context(|| {
format!(
"parsing {}; gunnar will not deliver to a misconfigured endpoint because of a \
syntax error",
path.display()
)
})
}
/// Write the configuration back, atomically and durably, through
/// the shared writer `replication.json` uses (one implementation).
pub fn save(&self, root: impl AsRef<Path>) -> Result<()> {
let text = serde_json::to_string_pretty(self)?;
crate::durable::write_atomic(root.as_ref(), WEBHOOK_CONFIG, text.as_bytes())
}
/// [`WebhookConfig::save`], **and a copy in the metadata mirror on the slow
/// disk** — the same treatment `replication.json` gets, for the same
/// reason: the endpoints an operator configured are a handful of bytes, and
/// losing them with the stripe means nobody downstream is told anything
/// until somebody remembers to reconfigure it by hand.
pub fn save_mirrored(
&self,
root: impl AsRef<Path>,
slow: &crate::slow::SlowRoot,
) -> Result<()> {
let text = serde_json::to_string_pretty(self)?;
crate::durable::write_atomic_mirrored(root.as_ref(), slow, WEBHOOK_CONFIG, text.as_bytes())
}
/// Add or replace an endpoint by name. Replace, so a retried `add` on a
/// flaky link does not leave two.
pub fn set_endpoint(&mut self, endpoint: Endpoint) {
match self.endpoints.iter_mut().find(|e| e.name == endpoint.name) {
Some(existing) => *existing = endpoint,
None => self.endpoints.push(endpoint),
}
}
/// Remove an endpoint by name. `true` when one was actually removed.
pub fn remove_endpoint(&mut self, name: &str) -> bool {
let before = self.endpoints.len();
self.endpoints.retain(|e| e.name != name);
self.endpoints.len() != before
}
}
// ───────────────────────────────────────────────────────────────────────────
// Event-kind names — the stable strings a `--kinds` filter and the payload use
// ───────────────────────────────────────────────────────────────────────────
/// The stable wire name of an event kind, e.g. `ref.updated`. This is what a
/// `--kinds` filter names and what the delivered body's `kind` field carries.
pub fn kind_name(kind: EventKind) -> &'static str {
match kind {
EventKind::Unspecified => "unspecified",
EventKind::RefUpdated => "ref.updated",
EventKind::RefDeleted => "ref.deleted",
EventKind::PushRejected => "push.rejected",
EventKind::RepoCreated => "repo.created",
EventKind::RepoDeleted => "repo.deleted",
EventKind::MirrorSynced => "mirror.synced",
EventKind::KeyUsed => "key.used",
EventKind::PushAttestation => "push.attestation",
EventKind::Gap => "gap",
EventKind::ReplicationFailed => "replication.failed",
EventKind::PullRequestOpened => "pull_request.opened",
EventKind::PullRequestMerged => "pull_request.merged",
EventKind::PullRequestClosed => "pull_request.closed",
EventKind::StreamOpened => "stream.opened",
EventKind::VisibilityChanged => "repo.visibility_changed",
EventKind::EncryptedChanged => "repo.encrypted_changed",
}
}
/// The kind a `--kinds` name selects, or `None` for a name that matches no
/// kind — which the control plane rejects at add time, because a typo here is a
/// webhook that silently delivers nothing.
pub fn kind_from_name(name: &str) -> Option<EventKind> {
let name = name.trim();
[
EventKind::RefUpdated,
EventKind::RefDeleted,
EventKind::PushRejected,
EventKind::RepoCreated,
EventKind::RepoDeleted,
EventKind::MirrorSynced,
EventKind::KeyUsed,
EventKind::PushAttestation,
EventKind::ReplicationFailed,
EventKind::PullRequestOpened,
EventKind::PullRequestMerged,
EventKind::PullRequestClosed,
EventKind::VisibilityChanged,
EventKind::EncryptedChanged,
]
.into_iter()
.find(|k| kind_name(*k) == name)
}
/// The delivered request body: a stable JSON object over the event.
///
/// Built by hand rather than from a prost serialisation, for the same reason
/// provenance signs canonical bytes: the receiver reads THESE bytes and the
/// signature is over them, so the shape has to be one gunnar controls and not
/// whatever a serialiser happens to emit. Every delivery carries `kind`, `seq`
/// and `at_unix_ms`; repo-scoped kinds carry `namespace`/`repo`; and the common
/// kinds carry the fields a CI trigger actually reads.
pub fn event_body(event: &Event) -> Vec<u8> {
let kind = EventKind::try_from(event.kind).unwrap_or(EventKind::Unspecified);
let mut obj = serde_json::Map::new();
obj.insert("kind".into(), kind_name(kind).into());
obj.insert("seq".into(), event.seq.into());
obj.insert("at_unix_ms".into(), event.at_unix_ms.into());
if let Some(path) = event.repo_path() {
obj.insert("namespace".into(), path.namespace.clone().into());
obj.insert("repo".into(), path.repo.clone().into());
}
let mut payload = serde_json::Map::new();
match event.payload.as_ref() {
Some(event::Payload::RefUpdated(e)) => {
payload.insert("ref".into(), e.ref_name.clone().into());
payload.insert("old_oid".into(), e.old_oid.clone().into());
payload.insert("new_oid".into(), e.new_oid.clone().into());
payload.insert("principal".into(), e.principal_id.clone().into());
payload.insert("forced".into(), e.forced.into());
}
Some(event::Payload::RefDeleted(e)) => {
payload.insert("ref".into(), e.ref_name.clone().into());
payload.insert("old_oid".into(), e.old_oid.clone().into());
payload.insert("principal".into(), e.principal_id.clone().into());
}
Some(event::Payload::RepoDeleted(e)) => {
payload.insert("principal".into(), e.principal_id.clone().into());
}
Some(event::Payload::RepoCreated(e)) => {
payload.insert("principal".into(), e.principal_id.clone().into());
}
Some(event::Payload::MirrorSynced(e)) => {
payload.insert("remote_url".into(), e.remote_url.clone().into());
payload.insert("refs_changed".into(), e.refs_changed.into());
}
Some(event::Payload::PushAttestation(e)) => {
payload.insert("attestation_id".into(), e.attestation_id.clone().into());
payload.insert("new_object_count".into(), e.new_object_count.into());
payload.insert("new_object_bytes".into(), e.new_object_bytes.into());
payload.insert("pack_hash".into(), e.pack_hash.clone().into());
}
Some(event::Payload::VisibilityChanged(e)) => {
let name = |v: i32| match gunnar_proto::Visibility::try_from(v) {
Ok(gunnar_proto::Visibility::Public) => "public",
Ok(gunnar_proto::Visibility::Private) => "private",
_ => "unspecified",
};
payload.insert("from".into(), name(e.from).into());
payload.insert("to".into(), name(e.to).into());
payload.insert("principal".into(), e.principal_id.clone().into());
}
Some(event::Payload::EncryptedChanged(e)) => {
payload.insert("from".into(), e.from.into());
payload.insert("to".into(), e.to.into());
payload.insert("principal".into(), e.principal_id.clone().into());
}
Some(event::Payload::PullRequestOpened(_))
| Some(event::Payload::PullRequestMerged(_))
| Some(event::Payload::PullRequestClosed(_)) => {}
_ => {}
}
obj.insert("payload".into(), serde_json::Value::Object(payload));
// to_vec, not to_vec_pretty: the bytes are what the signature covers, and a
// compact stable rendering is what the receiver hashes.
serde_json::to_vec(&serde_json::Value::Object(obj)).unwrap_or_default()
}
/// The `EventFilter` that decides which events reach one endpoint.
fn filter_for(endpoint: &Endpoint) -> EventFilter {
EventFilter {
namespaces: if endpoint.namespace.is_empty() {
Vec::new()
} else {
vec![endpoint.namespace.clone()]
},
repos: endpoint.repo.clone().into_iter().collect(),
kinds: endpoint
.kinds
.iter()
.filter_map(|k| kind_from_name(k))
.collect(),
include_attestation_object_ids: false,
}
}
// ───────────────────────────────────────────────────────────────────────────
// The running delivery: one subscriber, one bounded queue + worker per endpoint
// ───────────────────────────────────────────────────────────────────────────
/// Per-endpoint counters, surfaced in `webhook list`.
#[derive(Debug, Default)]
struct Counters {
delivered: AtomicU64,
failed: AtomicU64,
dropped: AtomicU64,
last_error: Mutex<String>,
}
/// One endpoint's queue, worker wake-up and counters.
struct Runner {
endpoint: Endpoint,
filter: EventFilter,
queue: Mutex<VecDeque<Arc<Event>>>,
notify: tokio::sync::Notify,
counters: Counters,
}
impl Runner {
fn new(endpoint: Endpoint) -> Arc<Self> {
Arc::new(Runner {
filter: filter_for(&endpoint),
endpoint,
queue: Mutex::new(VecDeque::new()),
notify: tokio::sync::Notify::new(),
counters: Counters::default(),
})
}
/// Enqueue an event for delivery. **Never blocks** — that is what keeps the
/// bus off the socket. Past the bound the newest is dropped-with-a-gap.
fn enqueue(&self, event: Arc<Event>) {
let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
if queue.len() >= DEFAULT_QUEUE_CAPACITY {
drop(queue);
self.counters.dropped.fetch_add(1, Ordering::Relaxed);
tracing::warn!(
target: TARGET_WEBHOOK,
event = "webhook.queue_full",
endpoint = %self.endpoint.name,
"the per-endpoint queue is full; dropping the newest event. The receiver is \
behind and should reconcile over the API."
);
return;
}
queue.push_back(event);
drop(queue);
self.notify.notify_one();
}
fn pop(&self) -> Option<Arc<Event>> {
self.queue
.lock()
.unwrap_or_else(|e| e.into_inner())
.pop_front()
}
#[cfg_attr(not(test), allow(dead_code))]
fn queue_len(&self) -> usize {
self.queue.lock().unwrap_or_else(|e| e.into_inner()).len()
}
}
/// Handles for the spawned tasks; aborting them stops delivery.
pub struct WebhookTasks {
tasks: Vec<tokio::task::JoinHandle<()>>,
}
impl WebhookTasks {
pub fn abort(&self) {
for task in &self.tasks {
task.abort();
}
}
pub fn is_empty(&self) -> bool {
self.tasks.is_empty()
}
}
impl Drop for WebhookTasks {
fn drop(&mut self) {
self.abort();
}
}
/// The running webhook delivery: the signing key, the bus, and one runner per
/// endpoint.
pub struct WebhookDelivery {
bus: Arc<EventBus>,
signing_key: Arc<ssh_key::PrivateKey>,
config: WebhookConfig,
runners: Vec<Arc<Runner>>,
}
impl WebhookDelivery {
pub fn new(
bus: Arc<EventBus>,
signing_key: Arc<ssh_key::PrivateKey>,
config: WebhookConfig,
) -> Arc<Self> {
let runners = config.endpoints.iter().cloned().map(Runner::new).collect();
Arc::new(WebhookDelivery {
bus,
signing_key,
config,
runners,
})
}
pub fn config(&self) -> &WebhookConfig {
&self.config
}
/// Start the event-stream subscriber and one worker per endpoint.
pub fn spawn(self: &Arc<Self>) -> WebhookTasks {
let mut tasks = Vec::new();
for runner in &self.runners {
let me = Arc::clone(self);
let runner = Arc::clone(runner);
// A32-14, stays a task: the webhook workers; their handles are kept and aborted at the dispatcher's stop.
tasks.push(tokio::spawn(async move { worker(me, runner).await }));
}
let me = Arc::clone(self);
tasks.push(tokio::spawn(async move { subscribe(me).await }));
WebhookTasks { tasks }
}
/// The endpoints, with live counters, as the control plane reports them.
fn endpoint_rows(&self) -> Vec<gunnar_proto::WebhookEndpoint> {
self.runners
.iter()
.map(|r| gunnar_proto::WebhookEndpoint {
name: r.endpoint.name.clone(),
url: r.endpoint.url.clone(),
kinds: r.endpoint.kinds.clone(),
namespace: r.endpoint.namespace.clone(),
repo: r.endpoint.repo.clone(),
has_secret: r.endpoint.secret.is_some(),
added_at_unix_ms: r.endpoint.added_at_unix_ms,
delivered: r.counters.delivered.load(Ordering::Relaxed),
failed: r.counters.failed.load(Ordering::Relaxed),
dropped: r.counters.dropped.load(Ordering::Relaxed),
last_error: r
.counters
.last_error
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone(),
})
.collect()
}
}
/// The event-stream subscriber. **The only coupling to the push path**, and it only
/// ever enqueues — never delivers — so the bus is never behind a socket.
async fn subscribe(delivery: Arc<WebhookDelivery>) {
use tokio_stream::StreamExt;
// Watch everything and fan out per endpoint's filter. One subscription for
// all endpoints keeps the bus's subscriber count — and its lag accounting —
// bounded by the server, not by how many webhooks an operator configured.
let mut stream = Box::pin(delivery.bus.subscribe(EventFilter::default()));
while let Some(next) = stream.next().await {
let event = match next {
Ok(event) => event,
Err(status) => {
tracing::warn!(target: TARGET_WEBHOOK, %status, "webhook subscription error; continuing");
continue;
}
};
if let Some(event::Payload::Gap(gap)) = event.payload.as_ref() {
// A lagged subscriber lost events. Every endpoint missed them; count
// it so `webhook list` shows a non-zero `dropped` and the receiver
// knows to reconcile. This is the bus's drop-with-a-gap contract,
// surfaced per endpoint.
for runner in &delivery.runners {
runner
.counters
.dropped
.fetch_add(gap.dropped, Ordering::Relaxed);
}
tracing::error!(
target: TARGET_WEBHOOK,
event = "webhook.bus_gap",
dropped = gap.dropped,
"the webhook subscriber fell behind the event bus and lost events; every endpoint \
is credited a gap"
);
continue;
}
// One `Arc` shared across every matching endpoint's queue, so a push
// watched by ten webhooks is one allocation and not ten.
let event = Arc::new(event);
for runner in &delivery.runners {
if runner.filter.matches(&event) {
runner.enqueue(Arc::clone(&event));
}
}
}
tracing::warn!(target: TARGET_WEBHOOK, "the event stream ended; no more deliveries");
}
/// One endpoint's delivery loop.
async fn worker(delivery: Arc<WebhookDelivery>, runner: Arc<Runner>) {
loop {
let Some(event) = runner.pop() else {
runner.notify.notified().await;
continue;
};
deliver_with_retry(&delivery, &runner, &event).await;
}
}
/// Deliver one event, retrying with bounded backoff. A dead endpoint costs at
/// most [`MAX_ATTEMPTS`] tries and then drops-with-a-gap — it never blocks the
/// worker forever, and it never grows the queue (the queue keeps draining while
/// this awaits its own sleeps).
async fn deliver_with_retry(delivery: &WebhookDelivery, runner: &Runner, event: &Event) {
let body = event_body(event);
let kind = EventKind::try_from(event.kind).unwrap_or(EventKind::Unspecified);
let signed = match gunnar_core::webhook::sign_delivery(&delivery.signing_key, &body) {
Ok(signed) => signed,
Err(err) => {
// Signing is the point of a gunnar webhook; an UNSIGNED delivery is
// not a fallback we ship. Count it failed and drop it.
runner.counters.failed.fetch_add(1, Ordering::Relaxed);
*runner
.counters
.last_error
.lock()
.unwrap_or_else(|e| e.into_inner()) = format!("signing the delivery failed: {err}");
tracing::error!(target: TARGET_WEBHOOK, endpoint = %runner.endpoint.name, %err, "could not sign a delivery; dropping it");
return;
}
};
let delivery_id = format!("{}:{}", delivery.bus.epoch(), event.seq);
for attempt in 1..=MAX_ATTEMPTS {
let result = tokio::time::timeout(
DELIVER_TIMEOUT,
post(&runner.endpoint, &body, &signed, kind, &delivery_id),
)
.await;
match result {
Ok(Ok(status)) if (200..300).contains(&status) => {
runner.counters.delivered.fetch_add(1, Ordering::Relaxed);
runner
.counters
.last_error
.lock()
.unwrap_or_else(|e| e.into_inner())
.clear();
return;
}
Ok(Ok(status)) => {
set_error(runner, format!("endpoint answered HTTP {status}"));
}
Ok(Err(err)) => set_error(runner, format!("{err:#}")),
Err(_) => set_error(runner, format!("timed out after {DELIVER_TIMEOUT:?}")),
}
runner.counters.failed.fetch_add(1, Ordering::Relaxed);
if attempt < MAX_ATTEMPTS {
let wait = crate::durable::backoff(attempt, RETRY_BASE, RETRY_MAX);
tokio::time::sleep(wait).await;
}
}
// Exhausted: drop-with-a-gap so the receiver reconciles rather than the
// worker retrying a dead endpoint forever behind a growing queue.
runner.counters.dropped.fetch_add(1, Ordering::Relaxed);
tracing::warn!(
target: TARGET_WEBHOOK,
event = "webhook.delivery_dropped",
endpoint = %runner.endpoint.name,
seq = event.seq,
"a delivery failed {MAX_ATTEMPTS} times and is dropped; the receiver should reconcile"
);
}
fn set_error(runner: &Runner, msg: String) {
*runner
.counters
.last_error
.lock()
.unwrap_or_else(|e| e.into_inner()) = msg;
}
// ───────────────────────────────────────────────────────────────────────────
// The HTTP POST — hyper over a raw connection, http:// and https://
// ───────────────────────────────────────────────────────────────────────────
/// POST `body` to the endpoint with the signature and event headers, returning
/// the HTTP status. One connection per delivery: webhooks are low-rate, and a
/// fresh connection is one fewer piece of state to keep bounded.
async fn post(
endpoint: &Endpoint,
body: &[u8],
signed: &gunnar_core::webhook::SignedDelivery,
kind: EventKind,
delivery_id: &str,
) -> Result<u16> {
use http_body_util::Full;
use hyper::body::Bytes;
let target = Url::parse(&endpoint.url)?;
let mut builder = hyper::Request::builder()
.method("POST")
.uri(&target.path_and_query)
.header("host", target.host_header())
.header("content-type", "application/json")
.header(
"user-agent",
format!("gunnar/{}", env!("CARGO_PKG_VERSION")),
)
.header(gunnar_core::webhook::EVENT_HEADER, kind_name(kind))
.header(gunnar_core::webhook::DELIVERY_HEADER, delivery_id)
.header(
gunnar_core::webhook::NAMESPACE_HEADER,
gunnar_core::webhook::WEBHOOK_NAMESPACE,
)
.header(
gunnar_core::webhook::KEY_FINGERPRINT_HEADER,
&signed.key_fingerprint,
)
.header(
gunnar_core::webhook::SIGNATURE_HEADER,
signed.signature_header(),
);
if let Some(secret) = endpoint.secret.as_deref() {
builder = builder.header(
gunnar_core::webhook::HMAC_HEADER,
gunnar_core::webhook::hmac_sha256_header(secret.as_bytes(), body),
);
}
let request = builder
.body(Full::new(Bytes::copy_from_slice(body)))
.context("building the delivery request")?;
let stream = target.connect().await?;
let status = send_over(stream, request).await?;
Ok(status)
}
/// Drive one HTTP/1.1 request/response over an established byte stream.
async fn send_over<S>(
stream: S,
request: hyper::Request<http_body_util::Full<hyper::body::Bytes>>,
) -> Result<u16>
where
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
{
let io = hyper_util::rt::TokioIo::new(stream);
let (mut sender, conn) = hyper::client::conn::http1::handshake(io)
.await
.context("HTTP handshake with the endpoint")?;
// The connection future must be driven; drop it when the response is in.
// A32-14, stays a task: drives one HTTP connection; aborted when the response is in.
let conn = tokio::spawn(async move {
let _ = conn.await;
});
let response = sender
.send_request(request)
.await
.context("sending the delivery")?;
let status = response.status().as_u16();
conn.abort();
Ok(status)
}
/// A parsed endpoint URL: enough of one to open a socket and write a request.
struct Url {
https: bool,
host: String,
port: u16,
path_and_query: String,
}
impl Url {
fn parse(raw: &str) -> Result<Self> {
let (scheme, rest) = raw
.split_once("://")
.ok_or_else(|| anyhow::anyhow!("webhook url {raw:?} has no scheme"))?;
let https = match scheme {
"http" => false,
"https" => true,
other => anyhow::bail!("webhook url scheme {other:?} is not http or https"),
};
let (authority, path) = match rest.find('/') {
Some(i) => (&rest[..i], &rest[i..]),
None => (rest, "/"),
};
let (host, port) = match authority.rsplit_once(':') {
Some((h, p)) => (
h.to_owned(),
p.parse::<u16>()
.with_context(|| format!("webhook url port {p:?}"))?,
),
None => (authority.to_owned(), if https { 443 } else { 80 }),
};
anyhow::ensure!(!host.is_empty(), "webhook url {raw:?} has no host");
Ok(Url {
https,
host,
port,
path_and_query: if path.is_empty() {
"/".to_owned()
} else {
path.to_owned()
},
})
}
fn host_header(&self) -> String {
let default = if self.https { 443 } else { 80 };
if self.port == default {
self.host.clone()
} else {
format!("{}:{}", self.host, self.port)
}
}
/// Open the transport: a plain TCP stream, or a TLS one for https.
async fn connect(&self) -> Result<Box<dyn Transport>> {
let tcp = match nordisk_pairing::egress::proxy() {
// **A box behind its front** (the lean estate, A19 row 14): the endpoint is
// reached through the front's egress door, end to end, the front naming the host.
Some(door) => {
let (door, host, port) = (door.to_string(), self.host.clone(), self.port);
let tunnel = tokio::task::spawn_blocking(move || {
nordisk_pairing::egress::dial(&door, &host, port)
})
.await
.context("the egress door's dial")?
.with_context(|| {
format!(
"connecting to {}:{} through the front's egress door",
self.host, self.port
)
})?;
tunnel.set_nonblocking(true).context("the egress tunnel")?;
tokio::net::TcpStream::from_std(tunnel).context("the egress tunnel")?
}
None => tokio::net::TcpStream::connect((self.host.as_str(), self.port))
.await
.with_context(|| format!("connecting to {}:{}", self.host, self.port))?,
};
if !self.https {
return Ok(Box::new(tcp));
}
let connector = tls_connector()?;
let server_name = tokio_rustls::rustls::pki_types::ServerName::try_from(self.host.clone())
.with_context(|| format!("{:?} is not a valid TLS server name", self.host))?;
let tls = connector
.connect(server_name, tcp)
.await
.with_context(|| format!("TLS handshake with {}", self.host))?;
Ok(Box::new(tls))
}
}
/// The byte-stream shape `send_over` needs, so http and https share one path.
trait Transport: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static {}
impl<T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static> Transport for T {}
/// A rustls client using the OS trust store, built once.
fn tls_connector() -> Result<tokio_rustls::TlsConnector> {
use std::sync::OnceLock;
static CONNECTOR: OnceLock<tokio_rustls::TlsConnector> = OnceLock::new();
if let Some(c) = CONNECTOR.get() {
return Ok(c.clone());
}
gunnar_grpc::ensure_crypto_provider();
let mut roots = tokio_rustls::rustls::RootCertStore::empty();
let native = rustls_native_certs::load_native_certs();
for cert in native.certs {
let _ = roots.add(cert);
}
anyhow::ensure!(
!roots.is_empty(),
"no system TLS roots were loadable, so an https webhook cannot be verified; use an \
http:// endpoint behind a TLS-terminating proxy, or install a CA bundle"
);
let config = tokio_rustls::rustls::ClientConfig::builder()
.with_root_certificates(roots)
.with_no_client_auth();
let connector = tokio_rustls::TlsConnector::from(Arc::new(config));
Ok(CONNECTOR.get_or_init(|| connector).clone())
}
// ───────────────────────────────────────────────────────────────────────────
// The live slot — set over the control plane, takes effect without a reboot
// ───────────────────────────────────────────────────────────────────────────
/// Owns the delivery's lifecycle, the way [`crate::mirror::ReplicationSlot`]
/// owns the replicator's: the appliance has no shell, so an endpoint has to be
/// settable over the control plane and take effect without a reboot. `apply`
/// writes the file FIRST, then swaps the running delivery, so a box that
/// acknowledged the change and lost power comes back with it.
pub struct WebhookSlot {
stores: Arc<ServerStores>,
bus: Arc<EventBus>,
runtime: tokio::runtime::Handle,
running: Mutex<Option<(Arc<WebhookDelivery>, WebhookTasks)>>,
}
impl WebhookSlot {
pub fn new(
stores: Arc<ServerStores>,
bus: Arc<EventBus>,
runtime: tokio::runtime::Handle,
) -> Arc<Self> {
Arc::new(WebhookSlot {
stores,
bus,
runtime,
running: Mutex::new(None),
})
}
fn delivery(&self) -> Option<Arc<WebhookDelivery>> {
self.running
.lock()
.unwrap_or_else(|e| e.into_inner())
.as_ref()
.map(|(d, _)| Arc::clone(d))
}
/// The configuration an operator is asking about: the running one, else the
/// file (so a `list` before anything is configured reads the file rather
/// than inventing an answer).
pub fn config(&self) -> Result<WebhookConfig> {
if let Some(delivery) = self.delivery() {
return Ok(delivery.config().clone());
}
WebhookConfig::load(self.stores.root()).context("reading the webhook configuration")
}
/// The server's webhook signing public key, one OpenSSH line — what a
/// receiver pins. Available whether or not any endpoint is configured.
pub fn signing_public_key(&self) -> Result<String> {
let key = self.stores.webhook_key()?;
Ok(key
.public_key()
.to_openssh()
.context("rendering the webhook public key")?
.trim()
.to_owned())
}
/// Persist `config` and make it the running one. Disk first.
pub fn apply(&self, config: WebhookConfig) -> Result<()> {
// A twin keeps no per-box mirror (C7, `crate::twin::forget_mirrors`): the file alone.
let saved = if self.stores.is_twin() {
config.save(self.stores.root())
} else {
config.save_mirrored(self.stores.root(), self.stores.slow())
};
saved.context("writing the webhook configuration")?;
let key = Arc::new(self.stores.webhook_key()?);
let mut running = self.running.lock().unwrap_or_else(|e| e.into_inner());
// Take (and thereby abort) the old tasks before subscribing the new
// delivery, so two subscribers are never on the bus at once.
if let Some((_, tasks)) = running.take() {
tasks.abort();
}
if config.endpoints.is_empty() {
tracing::info!(target: TARGET_WEBHOOK, event = "webhook.stopped", "the last webhook endpoint was removed; delivery is stopped");
return Ok(());
}
let delivery = WebhookDelivery::new(Arc::clone(&self.bus), key, config);
let tasks = {
let _guard = self.runtime.enter();
// A32-14, stays a task: the delivery's workers on the git runtime; their handles are kept (`running`) and stopped with it.
delivery.spawn()
};
*running = Some((delivery, tasks));
Ok(())
}
/// Stop delivering without touching the file (shutdown / drop). The config
/// stays on disk so a restart comes back delivering to the same endpoints.
pub fn stop(&self) {
if let Some((_, tasks)) = self
.running
.lock()
.unwrap_or_else(|e| e.into_inner())
.take()
{
tasks.abort();
}
}
/// Build the `WebhookListResponse` the control plane returns.
pub fn list_response(&self) -> Result<gunnar_proto::WebhookListResponse> {
let endpoints = if let Some(delivery) = self.delivery() {
delivery.endpoint_rows()
} else {
// Nothing running: render the on-disk endpoints with zero counters.
self.config()?
.endpoints
.into_iter()
.map(|e| gunnar_proto::WebhookEndpoint {
name: e.name,
url: e.url,
kinds: e.kinds,
namespace: e.namespace,
repo: e.repo,
has_secret: e.secret.is_some(),
added_at_unix_ms: e.added_at_unix_ms,
delivered: 0,
failed: 0,
dropped: 0,
last_error: String::new(),
})
.collect()
};
Ok(gunnar_proto::WebhookListResponse {
endpoints,
signing_public_key: self.signing_public_key().unwrap_or_default(),
signature_namespace: gunnar_core::webhook::WEBHOOK_NAMESPACE.to_owned(),
})
}
/// Add or replace an endpoint by name, validate it, persist, and start.
pub fn add(&self, endpoint: Endpoint) -> Result<gunnar_proto::WebhookListResponse> {
// Validate the URL up front so a bad one is refused at add time rather
// than failing silently on every delivery.
Url::parse(&endpoint.url)?;
// A --kinds name that matches no kind is a typo that would deliver
// nothing; refuse it by name.
for name in &endpoint.kinds {
anyhow::ensure!(
kind_from_name(name).is_some(),
"unknown event kind {name:?}; a --kinds filter that names nothing delivers nothing"
);
}
let mut config = self.config()?;
config.set_endpoint(endpoint);
self.apply(config)?;
self.list_response()
}
/// Remove an endpoint by name.
pub fn remove(&self, name: &str) -> Result<Option<gunnar_proto::WebhookListResponse>> {
let mut config = self.config()?;
if !config.remove_endpoint(name) {
return Ok(None);
}
self.apply(config)?;
Ok(Some(self.list_response()?))
}
}
#[cfg(test)]
mod tests;