From 5051ab4d2cf2766dd33e02a38a6ae91e2375e352 Mon Sep 17 00:00:00 2001 From: devfive Date: Mon, 31 Aug 2026 07:36:03 +0900 Subject: [PATCH 1/2] Add signed webhook delivery Reuse the durable event outbox so integrations receive exact event envelopes with bounded retries, dead-letter recovery, per-delivery timeouts, exact filters, and SSRF-safe endpoint resolution. --- Cargo.lock | 25 + Cargo.toml | 1 + apps/example-app/Cargo.toml | 4 + apps/example-app/src/main.rs | 2 +- apps/example-app/tests/openapi.rs | 4 + apps/example-app/tests/system.rs | 381 +++++- bun.lock | 15 + crates/core/src/events.rs | 65 +- plugins/webhooks/Cargo.toml | 25 + plugins/webhooks/app/(webhooks)/page.tsx | 1092 +++++++++++++++++ .../webhooks/app/(webhooks)/route.meta.json | 6 + ...k_endpoints_and_deliveries.vespertide.json | 163 +++ plugins/webhooks/models/deliveries.json | 78 ++ plugins/webhooks/models/endpoints.json | 73 ++ plugins/webhooks/package.json | 19 + plugins/webhooks/src/delivery.rs | 561 +++++++++ plugins/webhooks/src/lib.rs | 18 + plugins/webhooks/src/models/deliveries.rs | 41 + plugins/webhooks/src/models/endpoints.rs | 32 + plugins/webhooks/src/models/mod.rs | 2 + plugins/webhooks/src/routes/mod.rs | 538 ++++++++ plugins/webhooks/tsconfig.json | 11 + plugins/webhooks/vespertide.json | 16 + 23 files changed, 3160 insertions(+), 12 deletions(-) create mode 100644 plugins/webhooks/Cargo.toml create mode 100644 plugins/webhooks/app/(webhooks)/page.tsx create mode 100644 plugins/webhooks/app/(webhooks)/route.meta.json create mode 100644 plugins/webhooks/migrations/0001_create_webhook_endpoints_and_deliveries.vespertide.json create mode 100644 plugins/webhooks/models/deliveries.json create mode 100644 plugins/webhooks/models/endpoints.json create mode 100644 plugins/webhooks/package.json create mode 100644 plugins/webhooks/src/delivery.rs create mode 100644 plugins/webhooks/src/lib.rs create mode 100644 plugins/webhooks/src/models/deliveries.rs create mode 100644 plugins/webhooks/src/models/endpoints.rs create mode 100644 plugins/webhooks/src/models/mod.rs create mode 100644 plugins/webhooks/src/routes/mod.rs create mode 100644 plugins/webhooks/tsconfig.json create mode 100644 plugins/webhooks/vespertide.json diff --git a/Cargo.lock b/Cargo.lock index 9f84b5a..0b92a08 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1345,19 +1345,23 @@ dependencies = [ "anyhow", "audit-log", "auth", + "axum", "content", "example-memo-plugin", "example-plugin", + "hmac 0.13.0", "media", "reqwest", "sea-orm", "serde", "serde_json", + "sha2 0.11.0", "tempfile", "tokio", "tracing", "tracing-subscriber", "vespera", + "webhooks", "yeollin-app", ] @@ -4883,6 +4887,27 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "webhooks" +version = "0.1.0" +dependencies = [ + "anyhow", + "axum", + "chrono", + "hmac 0.13.0", + "rand 0.10.2", + "reqwest", + "sea-orm", + "serde", + "serde_json", + "sha2 0.11.0", + "tokio", + "tracing", + "vespera", + "vespertide", + "yeollin-plugin", +] + [[package]] name = "webpki-root-certs" version = "1.0.9" diff --git a/Cargo.toml b/Cargo.toml index 038a057..68910e2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -62,6 +62,7 @@ tokio-util = { version = "0.7", features = ["io"] } jsonwebtoken = { version = "11", features = ["rust_crypto"] } argon2 = "0.6" rand = "0.10" +hmac = "0.13" sha2 = "0.11" # Internal crates diff --git a/apps/example-app/Cargo.toml b/apps/example-app/Cargo.toml index 7e29f11..8e88406 100644 --- a/apps/example-app/Cargo.toml +++ b/apps/example-app/Cargo.toml @@ -26,7 +26,11 @@ sea-orm = { workspace = true } audit-log = { path = "../../plugins/audit-log" } media = { path = "../../plugins/media" } content = { path = "../../plugins/content" } +webhooks = { path = "../../plugins/webhooks" } [dev-dependencies] reqwest = { workspace = true, features = ["json", "multipart"] } tempfile = { workspace = true } +axum = { workspace = true } +hmac = { workspace = true } +sha2 = { workspace = true } diff --git a/apps/example-app/src/main.rs b/apps/example-app/src/main.rs index aa9cea7..8608de6 100644 --- a/apps/example-app/src/main.rs +++ b/apps/example-app/src/main.rs @@ -42,7 +42,7 @@ async fn main() -> anyhow::Result<()> { // Create app builder using yeollin_app! macro // This macro handles both register_plugin() and vespera merge in one call let app = yeollin::yeollin_app! { - plugins: [audit_log, auth, content, example_memo_plugin, example_plugin, media], + plugins: [audit_log, auth, content, example_memo_plugin, example_plugin, media, webhooks], openapi: "openapi.json", title: "Example CMS API", version: "1.0.0", diff --git a/apps/example-app/tests/openapi.rs b/apps/example-app/tests/openapi.rs index 509ac2a..d3b73aa 100644 --- a/apps/example-app/tests/openapi.rs +++ b/apps/example-app/tests/openapi.rs @@ -125,6 +125,10 @@ fn every_plugin_route_lives_under_its_declared_namespace() { "/api/media", "/api/media/file", "/api/media/{id}", + "/api/webhooks", + "/api/webhooks/deliveries", + "/api/webhooks/deliveries/{id}/retry", + "/api/webhooks/{id}", ] ); diff --git a/apps/example-app/tests/system.rs b/apps/example-app/tests/system.rs index fae501a..a7b0d74 100644 --- a/apps/example-app/tests/system.rs +++ b/apps/example-app/tests/system.rs @@ -4,14 +4,28 @@ //! plugin registration, migrations, the auth middleware, and the auth //! routes all have to agree for these to pass. +use std::fmt::Write; use std::net::TcpListener; use std::process::Stdio; use std::time::Duration; +use axum::{ + body::Bytes, + extract::State, + http::{HeaderMap, StatusCode}, + routing::post, + Router, +}; +use hmac::{Hmac, KeyInit, Mac}; use sea_orm::{ConnectionTrait, Database, DbBackend, Statement}; use serde_json::Value; +use sha2::Sha256; use tempfile::TempDir; -use tokio::process::{Child, Command}; +use tokio::{ + process::{Child, Command}, + sync::mpsc, + task::JoinHandle, +}; const ADMIN: &str = "admin"; const PASSWORD: &str = "system-test-password"; @@ -24,6 +38,77 @@ struct Server { _workdir: TempDir, } +#[derive(Debug)] +struct CapturedWebhook { + headers: HeaderMap, + body: Vec, +} + +#[derive(Clone)] +struct WebhookReceiverState { + captured: mpsc::UnboundedSender, +} + +struct WebhookReceiver { + url: String, + captured: mpsc::UnboundedReceiver, + task: JoinHandle<()>, +} + +impl Drop for WebhookReceiver { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn capture_webhook( + State(state): State, + headers: HeaderMap, + body: Bytes, +) -> StatusCode { + let _ = state.captured.send(CapturedWebhook { + headers, + body: body.to_vec(), + }); + StatusCode::NO_CONTENT +} + +async fn start_webhook_receiver() -> WebhookReceiver { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind webhook receiver"); + let address = listener.local_addr().expect("webhook receiver address"); + let (captured, receiver) = mpsc::unbounded_channel(); + let state = WebhookReceiverState { captured }; + let task = tokio::spawn(async move { + axum::serve( + listener, + Router::new() + .route("/hook", post(capture_webhook)) + .with_state(state), + ) + .await + .expect("serve webhook receiver"); + }); + WebhookReceiver { + url: format!("http://{address}/hook"), + captured: receiver, + task, + } +} + +fn webhook_signature(secret: &str, body: &[u8]) -> String { + let mut mac = Hmac::::new_from_slice(secret.as_bytes()).expect("HMAC key"); + mac.update(body); + mac.finalize() + .into_bytes() + .iter() + .fold(String::with_capacity(64), |mut output, byte| { + let _ = write!(output, "{byte:02x}"); + output + }) +} + impl Drop for Server { fn drop(&mut self) { let _ = self.child.start_kill(); @@ -85,6 +170,10 @@ impl Server { .replace('\\', "/"); format!("sqlite://{path}?mode=ro") } + + fn writable_database_url(&self) -> String { + self.database_url().replace("mode=ro", "mode=rw") + } } async fn login_as( @@ -142,7 +231,11 @@ async fn assembled_system_enforces_authentication() { .send() .await .unwrap(); - assert_ne!(dev_asset.status(), 200, "dev asset paths must not bypass auth"); + assert_ne!( + dev_asset.status(), + 200, + "dev asset paths must not bypass auth" + ); // Prefix widening: /health is public, /healthz is not. let widened = client.get(server.url("/healthz")).send().await.unwrap(); @@ -196,7 +289,11 @@ async fn assembled_system_rotates_and_revokes_refresh_tokens() { .send() .await .unwrap(); - assert_eq!(replayed.status(), 401, "a spent refresh token must not work"); + assert_eq!( + replayed.status(), + 401, + "a spent refresh token must not work" + ); let logout = client .post(server.url("/api/auth/logout")) @@ -220,10 +317,18 @@ async fn assembled_system_enforces_role_on_admin_routes() { let server = start().await; let client = reqwest::Client::new(); - let anonymous = client.get(server.url("/api/auth/users")).send().await.unwrap(); + let anonymous = client + .get(server.url("/api/auth/users")) + .send() + .await + .unwrap(); assert_eq!(anonymous.status(), 401, "the roster must not be public"); - let tokens: Value = login(&client, &server, PASSWORD).await.json().await.unwrap(); + let tokens: Value = login(&client, &server, PASSWORD) + .await + .json() + .await + .unwrap(); let access = tokens["access_token"].as_str().expect("access token"); let admin = client @@ -637,7 +742,11 @@ async fn assembled_system_manages_typed_content_publication() { .send() .await .unwrap(); - assert_eq!(protected.status(), 401, "collection management is protected"); + assert_eq!( + protected.status(), + 401, + "collection management is protected" + ); let absent_public = client .get(server.url("/api/content/pages/published?slug=first-page")) @@ -799,7 +908,10 @@ async fn assembled_system_manages_typed_content_publication() { assert_eq!(updated.status(), 200); let updated: Value = updated.json().await.unwrap(); assert_eq!(updated["status"], "published"); - assert_eq!(updated["fields"]["heroImage"], "media:0123456789abcdef0123456789abcdef"); + assert_eq!( + updated["fields"]["heroImage"], + "media:0123456789abcdef0123456789abcdef" + ); let unpublished = client .post(server.url(&format!("/api/content/pages/{first_id}/unpublish"))) @@ -858,7 +970,10 @@ async fn assembled_system_manages_typed_content_publication() { .await .unwrap(); assert_eq!(audit["total"], 2); - assert_eq!(audit["events"][0]["payload"]["content"]["collection"], "pages"); + assert_eq!( + audit["events"][0]["payload"]["content"]["collection"], + "pages" + ); let deleted = client .delete(server.url(&format!("/api/content/pages/{first_id}"))) @@ -876,6 +991,245 @@ async fn assembled_system_manages_typed_content_publication() { assert_eq!(gone.status(), 404); } +#[tokio::test] +async fn assembled_system_configures_signs_and_retries_webhooks() { + let server = start().await; + let client = reqwest::Client::new(); + let mut receiver = start_webhook_receiver().await; + + let anonymous = client + .get(server.url("/api/webhooks")) + .send() + .await + .unwrap(); + assert_eq!( + anonymous.status(), + 401, + "webhook configuration is protected" + ); + + let token = admin_token(&client, &server).await; + let account = client + .post(server.url("/api/auth/users")) + .bearer_auth(&token) + .json(&serde_json::json!({ + "username": "webhook-reader", + "password": "webhook-reader-password", + "role": "user", + })) + .send() + .await + .unwrap(); + assert_eq!(account.status(), 200); + let user_tokens: Value = login_as( + &client, + &server, + "webhook-reader", + "webhook-reader-password", + ) + .await + .json() + .await + .unwrap(); + let forbidden = client + .get(server.url("/api/webhooks")) + .bearer_auth(user_tokens["access_token"].as_str().unwrap()) + .send() + .await + .unwrap(); + assert_eq!( + forbidden.status(), + 403, + "webhook secrets require admin access" + ); + + let short_secret = client + .post(server.url("/api/webhooks")) + .bearer_auth(&token) + .json(&serde_json::json!({ + "name": "Invalid", + "url": receiver.url, + "secret": "too-short", + "allowPrivateNetworks": true, + })) + .send() + .await + .unwrap(); + assert_eq!(short_secret.status(), 400); + + let signing_secret = "0123456789abcdef0123456789abcdef"; + let created = client + .post(server.url("/api/webhooks")) + .bearer_auth(&token) + .json(&serde_json::json!({ + "name": " Local receiver ", + "url": receiver.url, + "secret": signing_secret, + "eventNames": ["memo.created"], + "allowPrivateNetworks": true, + "timeoutSeconds": 2, + "enabled": true, + })) + .send() + .await + .unwrap(); + assert_eq!(created.status(), 200); + let webhook: Value = created.json().await.unwrap(); + let webhook_id = webhook["id"].as_str().unwrap(); + assert_eq!(webhook["name"], "Local receiver"); + assert_eq!(webhook["hasSecret"], true); + assert!( + webhook.get("secret").is_none(), + "signing secrets are write-only" + ); + + let duplicate = client + .post(server.url("/api/webhooks")) + .bearer_auth(&token) + .json(&serde_json::json!({ + "name": "Local receiver", + "url": "https://example.com/hook", + "secret": signing_secret, + })) + .send() + .await + .unwrap(); + assert_eq!(duplicate.status(), 409, "endpoint names remain unique"); + + let memo: Value = client + .post(server.url("/api/example-memo-plugin")) + .bearer_auth(&token) + .json(&serde_json::json!({ + "title": "Webhook source", + "content": "Signed after commit", + })) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + let memo_id = memo["id"] + .as_i64() + .unwrap_or_else(|| panic!("memo creation failed: {memo}")); + let captured = tokio::time::timeout(Duration::from_secs(3), receiver.captured.recv()) + .await + .expect("webhook delivery timeout") + .expect("webhook receiver closed"); + assert_eq!(captured.headers["x-yeollin-event"], "memo.created"); + assert_eq!( + captured.headers["x-yeollin-signature"], + format!( + "sha256={}", + webhook_signature(signing_secret, &captured.body) + ) + ); + let envelope: Value = serde_json::from_slice(&captured.body).unwrap(); + assert_eq!(envelope["name"], "memo.created"); + assert_eq!(envelope["payload"]["memo"]["id"], memo["id"]); + + let updated = client + .patch(server.url(&format!("/api/example-memo-plugin/{memo_id}"))) + .bearer_auth(&token) + .json(&serde_json::json!({ + "title": "Webhook source updated", + "content": "The exact filter excludes this event", + })) + .send() + .await + .unwrap(); + assert_eq!(updated.status(), 200); + assert!( + tokio::time::timeout(Duration::from_millis(500), receiver.captured.recv()) + .await + .is_err(), + "memo.updated must not pass the memo.created filter" + ); + + let listed: Value = client + .get(server.url("/api/webhooks/deliveries")) + .bearer_auth(&token) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(listed["total"], 1); + assert_eq!(listed["deliveries"][0]["status"], "delivered"); + assert_eq!(listed["deliveries"][0]["attempts"], 1); + let delivery_id = listed["deliveries"][0]["id"].as_str().unwrap(); + + let db = Database::connect(server.writable_database_url()) + .await + .unwrap(); + db.execute_raw(Statement::from_sql_and_values( + DbBackend::Sqlite, + "UPDATE webhook_deliveries SET status = ?, attempts = ? WHERE id = ?", + [ + "dead_letter".into(), + 5_i32.into(), + delivery_id.to_string().into(), + ], + )) + .await + .unwrap(); + db.close().await.unwrap(); + + let retried = client + .post(server.url(&format!("/api/webhooks/deliveries/{delivery_id}/retry"))) + .bearer_auth(&token) + .send() + .await + .unwrap(); + assert_eq!(retried.status(), 200); + let retried: Value = retried.json().await.unwrap(); + assert_eq!(retried["status"], "pending"); + assert_eq!(retried["attempts"], 0); + tokio::time::timeout(Duration::from_secs(3), receiver.captured.recv()) + .await + .expect("manual retry delivery timeout") + .expect("webhook receiver closed"); + + let replaced = client + .put(server.url(&format!("/api/webhooks/{webhook_id}"))) + .bearer_auth(&token) + .json(&serde_json::json!({ + "name": "Local receiver", + "url": receiver.url, + "secret": null, + "eventNames": [], + "allowPrivateNetworks": true, + "timeoutSeconds": 3, + "enabled": false, + })) + .send() + .await + .unwrap(); + assert_eq!(replaced.status(), 200); + let replaced: Value = replaced.json().await.unwrap(); + assert_eq!(replaced["hasSecret"], true, "omission retains the secret"); + assert_eq!(replaced["enabled"], false); + + let deleted = client + .delete(server.url(&format!("/api/webhooks/{webhook_id}"))) + .bearer_auth(&token) + .send() + .await + .unwrap(); + assert_eq!(deleted.status(), 200); + let after_delete: Value = client + .get(server.url("/api/webhooks/deliveries")) + .bearer_auth(&token) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(after_delete["total"], 0, "delete removes delivery history"); +} + #[tokio::test] async fn assembled_system_manages_accounts() { let server = start().await; @@ -1027,7 +1381,10 @@ async fn assembled_system_refuses_to_lock_itself_out() { .json() .await .unwrap(); - assert_eq!(identity["role"], "admin", "the refusals must not half-apply"); + assert_eq!( + identity["role"], "admin", + "the refusals must not half-apply" + ); } #[tokio::test] @@ -1035,7 +1392,11 @@ async fn assembled_system_ends_sessions_when_a_password_changes() { let server = start().await; let client = reqwest::Client::new(); - let tokens: Value = login(&client, &server, PASSWORD).await.json().await.unwrap(); + let tokens: Value = login(&client, &server, PASSWORD) + .await + .json() + .await + .unwrap(); let access = tokens["access_token"].as_str().expect("access token"); let refresh = tokens["refresh_token"].as_str().expect("refresh token"); diff --git a/bun.lock b/bun.lock index a34dfe7..3f4bc82 100644 --- a/bun.lock +++ b/bun.lock @@ -119,6 +119,19 @@ "vinext": "^1.0.0-beta.8", }, }, + "plugins/webhooks": { + "name": "@yeollin-plugin/webhooks", + "version": "0.1.0", + "dependencies": { + "@devup-ui/react": "^1.0.41", + "react": "^19.2.8", + }, + "devDependencies": { + "@types/react": "^19", + "typescript": "^7.0", + "vinext": "^1.0.0-beta.8", + }, + }, }, "packages": { "@babel/code-frame": ["@babel/code-frame@7.29.7", "", { "dependencies": { "@babel/helper-validator-identifier": "^7.29.7", "js-tokens": "^4.0.0", "picocolors": "^1.1.1" } }, "sha512-Aup7aUOfpbAUg2ROOJN6Iw5f9DMBlzu0mIkm/malLQFN/YQgO48wCj0Kxa3sEHJvPVFg7siR+qRInwXd2qhQKw=="], @@ -527,6 +540,8 @@ "@yeollin-plugin/media": ["@yeollin-plugin/media@workspace:plugins/media"], + "@yeollin-plugin/webhooks": ["@yeollin-plugin/webhooks@workspace:plugins/webhooks"], + "@yeollin/app": ["@yeollin/app@workspace:packages/app"], "@yeollin/types": ["@yeollin/types@workspace:packages/types"], diff --git a/crates/core/src/events.rs b/crates/core/src/events.rs index 6baa7bf..75f9c89 100644 --- a/crates/core/src/events.rs +++ b/crates/core/src/events.rs @@ -17,6 +17,7 @@ tokio::task_local! { } const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(1); +const MAX_RETRY_DELAY: Duration = Duration::from_secs(5 * 60); const BATCH_SIZE: u64 = 100; const MAX_ERROR_LENGTH: usize = 1_024; @@ -339,8 +340,9 @@ impl EventBus { if let Some(error) = failure { active.last_error = Set(Some(truncate_error(error))); + let attempts = envelope_attempts(&active); active.available_at = Set((chrono::Utc::now() - + chrono::Duration::from_std(self.inner.poll_interval) + + chrono::Duration::from_std(retry_delay(self.inner.poll_interval, attempts)) .unwrap_or_else(|_| chrono::Duration::seconds(1))) .into()); } else { @@ -378,6 +380,27 @@ impl EventBus { Ok((events, total)) } + /// Make one persisted event immediately eligible for deferred delivery again. + /// + /// Administrative dead-letter tooling uses this after resetting its own + /// subscriber state. The event payload is never rewritten. + pub async fn requeue(&self, event_id: i64) -> Result { + let Some(stored) = events::Entity::find_by_id(event_id) + .one(&self.inner.db) + .await? + else { + return Ok(false); + }; + let mut active: events::ActiveModel = stored.into(); + active.processed_at = Set(None); + active.delivery_attempts = Set(0); + active.available_at = Set(chrono::Utc::now().into()); + active.last_error = Set(None); + active.update(&self.inner.db).await?; + self.wake(); + Ok(true) + } + /// Apply an audit retention cutoff without touching pending or non-audit rows. pub async fn purge_audited_before( &self, @@ -437,6 +460,13 @@ fn truncate_error(error: String) -> String { error.chars().take(MAX_ERROR_LENGTH).collect() } +fn retry_delay(base: Duration, attempts: i32) -> Duration { + let exponent = u32::try_from(attempts.saturating_sub(1)) + .unwrap_or_default() + .min(31); + base.saturating_mul(1_u32 << exponent).min(MAX_RETRY_DELAY) +} + /// An application transaction that records events and wakes delivery only after commit. pub struct EventTransaction { bus: EventBus, @@ -711,6 +741,30 @@ mod tests { assert!(stored.last_error.unwrap().contains("temporary failure")); } + #[tokio::test] + async fn an_administrator_can_requeue_a_persisted_event() { + let db = database().await; + let bus = EventBus::new(db.clone(), []).unwrap(); + let mut transaction = bus.begin().await.unwrap(); + let emitted = transaction + .emit(&TestEvent { value: "manual" }) + .await + .unwrap(); + transaction.commit().await.unwrap(); + assert_eq!(bus.drain_once().await.unwrap(), 1); + + assert!(bus.requeue(emitted.id).await.unwrap()); + let stored = events::Entity::find_by_id(emitted.id) + .one(&db) + .await + .unwrap() + .unwrap(); + assert!(stored.processed_at.is_none()); + assert_eq!(stored.delivery_attempts, 0); + assert!(stored.last_error.is_none()); + assert!(!bus.requeue(i64::MAX).await.unwrap()); + } + #[tokio::test] async fn inline_subscribers_cannot_start_a_nested_event_transaction() { let db = database().await; @@ -762,4 +816,13 @@ mod tests { Err(EventError::InvalidRegistration(_)) )); } + + #[test] + fn deferred_retries_back_off_exponentially_and_are_capped() { + let base = Duration::from_secs(1); + assert_eq!(retry_delay(base, 1), Duration::from_secs(1)); + assert_eq!(retry_delay(base, 2), Duration::from_secs(2)); + assert_eq!(retry_delay(base, 3), Duration::from_secs(4)); + assert_eq!(retry_delay(base, 20), MAX_RETRY_DELAY); + } } diff --git a/plugins/webhooks/Cargo.toml b/plugins/webhooks/Cargo.toml new file mode 100644 index 0000000..200f662 --- /dev/null +++ b/plugins/webhooks/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "webhooks" +version = "0.1.0" +edition = "2021" +description = "Signed event webhooks with SSRF-safe delivery" + +[lib] +path = "src/lib.rs" + +[dependencies] +yeollin-plugin = { workspace = true } +vespera = { workspace = true } +vespertide = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +axum = { workspace = true } +chrono = { workspace = true } +rand = { workspace = true } +sea-orm = { workspace = true } +tokio = { workspace = true } +anyhow = { workspace = true } +tracing = { workspace = true } +reqwest = { workspace = true } +hmac = { workspace = true } +sha2 = { workspace = true } diff --git a/plugins/webhooks/app/(webhooks)/page.tsx b/plugins/webhooks/app/(webhooks)/page.tsx new file mode 100644 index 0000000..79e3cf9 --- /dev/null +++ b/plugins/webhooks/app/(webhooks)/page.tsx @@ -0,0 +1,1092 @@ +'use client' + +import { Box, Flex, Grid, Text, VStack } from '@devup-ui/react' +import { useCallback, useEffect, useState } from 'react' + +type DeliveryStatus = 'pending' | 'delivered' | 'dead_letter' + +interface Webhook { + id: string + name: string + url: string + eventNames: string[] + allowPrivateNetworks: boolean + timeoutSeconds: number + enabled: boolean + hasSecret: boolean + createdAt: string + updatedAt: string +} + +interface Delivery { + id: string + webhookId: string + eventId: number + eventName: string + status: DeliveryStatus + attempts: number + maxAttempts: number + responseStatus: number | null + lastError: string | null + createdAt: string + updatedAt: string + deliveredAt: string | null +} + +interface DeliveryPage { + deliveries: Delivery[] + total: number + page: number + pageSize: number +} + +interface FormState { + name: string + url: string + secret: string + eventNames: string + timeoutSeconds: number + allowPrivateNetworks: boolean + enabled: boolean +} + +interface Feedback { + kind: 'success' | 'error' + message: string +} + +type ApiResult = + { ok: true; data: unknown } | { ok: false; status: number; message: string } + +const EMPTY_FORM: FormState = { + name: '', + url: '', + secret: '', + eventNames: '', + timeoutSeconds: 5, + allowPrivateNetworks: false, + enabled: true, +} + +const EMPTY_DELIVERY_PAGE: DeliveryPage = { + deliveries: [], + total: 0, + page: 1, + pageSize: 25, +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null +} + +function isDeliveryStatus(value: unknown): value is DeliveryStatus { + return value === 'pending' || value === 'delivered' || value === 'dead_letter' +} + +function parseWebhook(value: unknown): Webhook | null { + if (!isRecord(value)) return null + if ( + typeof value.id !== 'string' || + typeof value.name !== 'string' || + typeof value.url !== 'string' || + !Array.isArray(value.eventNames) || + !value.eventNames.every((name) => typeof name === 'string') || + typeof value.allowPrivateNetworks !== 'boolean' || + typeof value.timeoutSeconds !== 'number' || + typeof value.enabled !== 'boolean' || + typeof value.hasSecret !== 'boolean' || + typeof value.createdAt !== 'string' || + typeof value.updatedAt !== 'string' + ) { + return null + } + return { + id: value.id, + name: value.name, + url: value.url, + eventNames: value.eventNames as string[], + allowPrivateNetworks: value.allowPrivateNetworks, + timeoutSeconds: value.timeoutSeconds, + enabled: value.enabled, + hasSecret: value.hasSecret, + createdAt: value.createdAt, + updatedAt: value.updatedAt, + } +} + +function parseDelivery(value: unknown): Delivery | null { + if (!isRecord(value)) return null + if ( + typeof value.id !== 'string' || + typeof value.webhookId !== 'string' || + typeof value.eventId !== 'number' || + typeof value.eventName !== 'string' || + !isDeliveryStatus(value.status) || + typeof value.attempts !== 'number' || + typeof value.maxAttempts !== 'number' || + (value.responseStatus !== null && + typeof value.responseStatus !== 'number') || + (value.lastError !== null && typeof value.lastError !== 'string') || + typeof value.createdAt !== 'string' || + typeof value.updatedAt !== 'string' || + (value.deliveredAt !== null && typeof value.deliveredAt !== 'string') + ) { + return null + } + return { + id: value.id, + webhookId: value.webhookId, + eventId: value.eventId, + eventName: value.eventName, + status: value.status, + attempts: value.attempts, + maxAttempts: value.maxAttempts, + responseStatus: value.responseStatus, + lastError: value.lastError, + createdAt: value.createdAt, + updatedAt: value.updatedAt, + deliveredAt: value.deliveredAt, + } +} + +function parseWebhooks(value: unknown): Webhook[] { + if (!isRecord(value) || !Array.isArray(value.webhooks)) { + throw new Error('The server returned invalid webhook data.') + } + return value.webhooks + .map(parseWebhook) + .filter((webhook): webhook is Webhook => webhook !== null) +} + +function parseDeliveryPage(value: unknown): DeliveryPage { + if (!isRecord(value) || !Array.isArray(value.deliveries)) { + throw new Error('The server returned invalid delivery data.') + } + return { + deliveries: value.deliveries + .map(parseDelivery) + .filter((delivery): delivery is Delivery => delivery !== null), + total: typeof value.total === 'number' ? value.total : 0, + page: typeof value.page === 'number' ? value.page : 1, + pageSize: typeof value.pageSize === 'number' ? value.pageSize : 25, + } +} + +function errorMessage(value: unknown, fallback: string): string { + return isRecord(value) && typeof value.error === 'string' + ? value.error + : fallback +} + +async function request(path: string, init?: RequestInit): Promise { + try { + const response = await fetch(path, init) + const data = (await response.json().catch(() => null)) as unknown + if (!response.ok) { + return { + ok: false, + status: response.status, + message: errorMessage( + data, + `The server rejected the request (HTTP ${response.status}).`, + ), + } + } + return { ok: true, data } + } catch { + return { + ok: false, + status: 0, + message: 'Could not reach the server. Check your connection and retry.', + } + } +} + +function jsonRequest(method: string, value: unknown): RequestInit { + return { + method, + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(value), + } +} + +function eventNames(value: string): string[] { + return value + .split(/[\n,]/) + .map((name) => name.trim()) + .filter((name) => name !== '') +} + +function formatDate(value: string): string { + const date = new Date(value) + return Number.isNaN(date.getTime()) ? 'Unknown time' : date.toLocaleString() +} + +function statusLabel(status: DeliveryStatus): string { + if (status === 'dead_letter') return 'Dead letter' + return status === 'delivered' ? 'Delivered' : 'Pending' +} + +export default function WebhooksPage() { + const [webhooks, setWebhooks] = useState([]) + const [deliveryPage, setDeliveryPage] = + useState(EMPTY_DELIVERY_PAGE) + const [deliveryPageNumber, setDeliveryPageNumber] = useState(1) + const [statusFilter, setStatusFilter] = useState('') + const [loading, setLoading] = useState(true) + const [forbidden, setForbidden] = useState(false) + const [loadError, setLoadError] = useState('') + const [feedback, setFeedback] = useState(null) + const [form, setForm] = useState(EMPTY_FORM) + const [editingId, setEditingId] = useState('') + const [saving, setSaving] = useState(false) + const [busyId, setBusyId] = useState('') + + const loadData = useCallback(async () => { + setLoading(true) + const params = new URLSearchParams({ + page: String(deliveryPageNumber), + pageSize: '25', + }) + if (statusFilter !== '') params.set('status', statusFilter) + const [webhookResult, deliveryResult] = await Promise.all([ + request('/api/webhooks'), + request(`/api/webhooks/deliveries?${params}`), + ]) + if (!webhookResult.ok) { + setForbidden(webhookResult.status === 403) + setLoadError(webhookResult.message) + setLoading(false) + return + } + if (!deliveryResult.ok) { + setForbidden(deliveryResult.status === 403) + setLoadError(deliveryResult.message) + setLoading(false) + return + } + + try { + setWebhooks(parseWebhooks(webhookResult.data)) + setDeliveryPage(parseDeliveryPage(deliveryResult.data)) + setForbidden(false) + setLoadError('') + } catch (cause) { + setLoadError( + cause instanceof Error ? cause.message : 'Could not load webhooks.', + ) + } finally { + setLoading(false) + } + }, [deliveryPageNumber, statusFilter]) + + useEffect(() => { + const timer = window.setTimeout(() => void loadData(), 0) + return () => window.clearTimeout(timer) + }, [loadData]) + + function resetForm() { + setForm(EMPTY_FORM) + setEditingId('') + } + + function beginEdit(webhook: Webhook) { + setEditingId(webhook.id) + setForm({ + name: webhook.name, + url: webhook.url, + secret: '', + eventNames: webhook.eventNames.join('\n'), + timeoutSeconds: webhook.timeoutSeconds, + allowPrivateNetworks: webhook.allowPrivateNetworks, + enabled: webhook.enabled, + }) + setFeedback(null) + window.scrollTo({ top: 0, behavior: 'smooth' }) + } + + async function saveWebhook(event: React.FormEvent) { + event.preventDefault() + setSaving(true) + setFeedback(null) + const updating = editingId !== '' + const payload = { + name: form.name, + url: form.url, + secret: updating && form.secret === '' ? null : form.secret, + eventNames: eventNames(form.eventNames), + allowPrivateNetworks: form.allowPrivateNetworks, + timeoutSeconds: form.timeoutSeconds, + enabled: form.enabled, + } + const result = await request( + updating ? `/api/webhooks/${editingId}` : '/api/webhooks', + jsonRequest(updating ? 'PUT' : 'POST', payload), + ) + if (!result.ok) { + setFeedback({ kind: 'error', message: result.message }) + setSaving(false) + return + } + const name = form.name.trim() + resetForm() + setFeedback({ + kind: 'success', + message: updating + ? `Updated “${name}”.` + : `Created “${name}”. Its secret will not be shown again.`, + }) + await loadData() + setSaving(false) + } + + async function deleteWebhook(webhook: Webhook) { + if ( + !window.confirm( + `Delete “${webhook.name}” and all of its delivery history? This cannot be undone.`, + ) + ) { + return + } + setBusyId(webhook.id) + setFeedback(null) + const result = await request(`/api/webhooks/${webhook.id}`, { + method: 'DELETE', + }) + if (result.ok) { + if (editingId === webhook.id) resetForm() + setFeedback({ + kind: 'success', + message: `Deleted “${webhook.name}”.`, + }) + await loadData() + } else { + setFeedback({ kind: 'error', message: result.message }) + } + setBusyId('') + } + + async function retryDelivery(delivery: Delivery) { + setBusyId(delivery.id) + setFeedback(null) + const result = await request( + `/api/webhooks/deliveries/${delivery.id}/retry`, + { method: 'POST' }, + ) + setFeedback( + result.ok + ? { + kind: 'success', + message: `Requeued ${delivery.eventName}. Refresh to follow its result.`, + } + : { kind: 'error', message: result.message }, + ) + await loadData() + setBusyId('') + } + + const webhookNames = new Map( + webhooks.map((webhook) => [webhook.id, webhook.name]), + ) + const enabledCount = webhooks.filter((webhook) => webhook.enabled).length + const pageCount = Math.max( + 1, + Math.ceil(deliveryPage.total / deliveryPage.pageSize), + ) + + return ( + + + + + Webhooks + + Send committed CMS events to signed HTTP endpoints. + + + void loadData()} + px={4} + py={3} + type="button" + > + {loading ? 'Refreshing...' : 'Refresh'} + + + + {forbidden ? ( + + + + Administrator access required + + + Endpoint secrets and delivery payloads are restricted to + administrators. + + + + ) : ( + <> + + + + + + + + + + + + + {editingId === '' ? 'Add endpoint' : 'Edit endpoint'} + + + Secrets are write-only and must contain at least 32 + bytes. + + + {editingId !== '' ? ( + + Cancel edit + + ) : null} + + + + + + setForm((current) => ({ ...current, name: value })) + } + placeholder="Publishing pipeline" + required + type="text" + value={form.name} + /> + + + + setForm((current) => ({ ...current, url: value })) + } + placeholder="https://example.com/hooks/yeollin" + required + type="url" + value={form.url} + /> + + + + + + + setForm((current) => ({ ...current, secret: value })) + } + placeholder={ + editingId === '' + ? 'At least 32 bytes' + : 'Unchanged when blank' + } + required={editingId === ''} + type="password" + value={form.secret} + /> + + + , + ) => + setForm((current) => ({ + ...current, + eventNames: event.target.value, + })) + } + outline="none" + p={3} + placeholder={'content.published\nmedia.uploaded'} + resize="vertical" + value={form.eventNames} + /> + + + + + + + setForm((current) => ({ + ...current, + timeoutSeconds: Number(value), + })) + } + required + type="number" + value={String(form.timeoutSeconds)} + /> + + + + setForm((current) => ({ + ...current, + enabled: checked, + })) + } + /> + + + setForm((current) => ({ + ...current, + allowPrivateNetworks: checked, + })) + } + /> + + Opt out of SSRF protection only for a trusted internal + receiver. + + + + + + + + {saving + ? 'Saving...' + : editingId === '' + ? 'Add webhook' + : 'Save changes'} + + + + + + + {feedback !== null ? ( + + + {feedback.message} + + + ) : null} + + {loadError !== '' ? ( + + + {loadError} + + + ) : null} + + + Configured endpoints + {loading && webhooks.length === 0 ? ( + Loading endpoints... + ) : webhooks.length === 0 ? ( + + No endpoints yet. Add one above to begin delivering events. + + ) : ( + + {webhooks.map((webhook) => ( + + + + + + {webhook.name} + + + {webhook.enabled ? 'Enabled' : 'Disabled'} + + + + {webhook.timeoutSeconds}s timeout + + + + {webhook.url} + + + {webhook.eventNames.length === 0 + ? 'All event names' + : webhook.eventNames.join(', ')} + + {webhook.allowPrivateNetworks ? ( + + Private-network delivery allowed + + ) : null} + + beginEdit(webhook)}> + Edit + + void deleteWebhook(webhook)} + > + {busyId === webhook.id ? 'Deleting...' : 'Delete'} + + + + + ))} + + )} + + + + + + Delivery history + + Failed deliveries retry with exponential backoff and stop + after five attempts. + + + + ) => { + setDeliveryPageNumber(1) + setStatusFilter(event.target.value) + }} + p={3} + value={statusFilter} + > + + + + + + + + + {loading && deliveryPage.deliveries.length === 0 ? ( + Loading deliveries... + ) : deliveryPage.deliveries.length === 0 ? ( + No deliveries match this view. + ) : ( + + {deliveryPage.deliveries.map((delivery) => ( + + + + + + {delivery.eventName} + + + + {statusLabel(delivery.status)} + + + + + {webhookNames.get(delivery.webhookId) ?? + 'Deleted endpoint'}{' '} + · attempt {delivery.attempts}/{delivery.maxAttempts} + {delivery.responseStatus === null + ? '' + : ` · HTTP ${delivery.responseStatus}`} + + + {formatDate(delivery.updatedAt)} · event # + {delivery.eventId} + + {delivery.lastError !== null ? ( + + {delivery.lastError} + + ) : null} + + {delivery.status === 'dead_letter' ? ( + void retryDelivery(delivery)} + > + {busyId === delivery.id ? 'Requeuing...' : 'Retry'} + + ) : null} + + + ))} + + )} + + {deliveryPage.total > 0 ? ( + + setDeliveryPageNumber((page) => page - 1)} + > + Previous + + + {deliveryPage.total} deliveries · Page {deliveryPageNumber}{' '} + of {pageCount} + + = pageCount || loading} + onClick={() => setDeliveryPageNumber((page) => page + 1)} + > + Next + + + ) : null} + + + )} + + + ) +} + +function Summary({ label, value }: { label: string; value: number }) { + return ( + + + + {label} + + {value} + + + ) +} + +function Field({ + children, + hint, + htmlFor, + label, +}: { + children: React.ReactNode + hint?: string + htmlFor: string + label: string +}) { + return ( + + + {label} + + {children} + {hint === undefined ? null : ( + + {hint} + + )} + + ) +} + +function Input({ + onChange, + ...props +}: Omit, 'onChange'> & { + onChange: (value: string) => void +}) { + return ( + ) => + onChange(event.target.value) + } + outline="none" + p={3} + /> + ) +} + +function Checkbox({ + checked, + label, + onChange, +}: { + checked: boolean + label: string + onChange: (checked: boolean) => void +}) { + return ( + + ) => + onChange(event.target.checked) + } + type="checkbox" + /> + {label} + + ) +} + +function EmptyCard({ children }: { children: React.ReactNode }) { + return ( + + + {children} + + + ) +} + +function PrimaryButton({ + children, + disabled = false, + type = 'button', +}: { + children: React.ReactNode + disabled?: boolean + type?: 'button' | 'submit' +}) { + return ( + + {children} + + ) +} + +function SecondaryButton({ + children, + disabled = false, + onClick, +}: { + children: React.ReactNode + disabled?: boolean + onClick: () => void +}) { + return ( + + {children} + + ) +} + +function DangerButton({ + children, + disabled, + onClick, +}: { + children: React.ReactNode + disabled: boolean + onClick: () => void +}) { + return ( + + {children} + + ) +} diff --git a/plugins/webhooks/app/(webhooks)/route.meta.json b/plugins/webhooks/app/(webhooks)/route.meta.json new file mode 100644 index 0000000..d6a7538 --- /dev/null +++ b/plugins/webhooks/app/(webhooks)/route.meta.json @@ -0,0 +1,6 @@ +{ + "label": "Webhooks", + "order": 40, + "access": "authenticated", + "menu": true +} diff --git a/plugins/webhooks/migrations/0001_create_webhook_endpoints_and_deliveries.vespertide.json b/plugins/webhooks/migrations/0001_create_webhook_endpoints_and_deliveries.vespertide.json new file mode 100644 index 0000000..bfbf71b --- /dev/null +++ b/plugins/webhooks/migrations/0001_create_webhook_endpoints_and_deliveries.vespertide.json @@ -0,0 +1,163 @@ +{ + "$schema": "https://raw.githubusercontent.com/dev-five-git/vespertide/refs/heads/main/schemas/migration.schema.json", + "actions": [ + { + "columns": [ + { + "name": "id", + "nullable": false, + "primary_key": true, + "type": "text" + }, + { + "index": true, + "name": "webhook_id", + "nullable": false, + "type": "text" + }, + { + "index": true, + "name": "event_id", + "nullable": false, + "type": "big_int" + }, + { + "index": true, + "name": "event_name", + "nullable": false, + "type": "text" + }, + { + "index": true, + "name": "status", + "nullable": false, + "type": "text" + }, + { + "default": 0, + "name": "attempts", + "nullable": false, + "type": "integer" + }, + { + "name": "response_status", + "nullable": true, + "type": "integer" + }, + { + "name": "last_error", + "nullable": true, + "type": "text" + }, + { + "default": "NOW()", + "index": true, + "name": "created_at", + "nullable": false, + "type": "timestamptz" + }, + { + "default": "NOW()", + "name": "updated_at", + "nullable": false, + "type": "timestamptz" + }, + { + "name": "delivered_at", + "nullable": true, + "type": "timestamptz" + } + ], + "constraints": [ + { + "columns": [ + "webhook_id", + "event_id" + ], + "name": "uq_webhook_deliveries_endpoint_event", + "type": "unique" + } + ], + "table": "deliveries", + "type": "create_table" + }, + { + "columns": [ + { + "name": "id", + "nullable": false, + "primary_key": true, + "type": "text" + }, + { + "index": true, + "name": "name", + "nullable": false, + "type": "text" + }, + { + "name": "url", + "nullable": false, + "type": "text" + }, + { + "name": "secret", + "nullable": false, + "type": "text" + }, + { + "name": "event_names", + "nullable": false, + "type": "json" + }, + { + "default": false, + "name": "allow_private_networks", + "nullable": false, + "type": "boolean" + }, + { + "default": 5, + "name": "timeout_seconds", + "nullable": false, + "type": "integer" + }, + { + "default": true, + "index": true, + "name": "enabled", + "nullable": false, + "type": "boolean" + }, + { + "default": "NOW()", + "index": true, + "name": "created_at", + "nullable": false, + "type": "timestamptz" + }, + { + "default": "NOW()", + "name": "updated_at", + "nullable": false, + "type": "timestamptz" + } + ], + "constraints": [ + { + "columns": [ + "name" + ], + "name": "uq_webhook_endpoints_name", + "type": "unique" + } + ], + "table": "endpoints", + "type": "create_table" + } + ], + "comment": "create webhook endpoints and deliveries", + "created_at": "2026-08-30T21:51:35Z", + "id": "36dbdbd2-148f-458d-8129-cc0a47febc91", + "version": 1 +} \ No newline at end of file diff --git a/plugins/webhooks/models/deliveries.json b/plugins/webhooks/models/deliveries.json new file mode 100644 index 0000000..22f0636 --- /dev/null +++ b/plugins/webhooks/models/deliveries.json @@ -0,0 +1,78 @@ +{ + "$schema": "https://raw.githubusercontent.com/dev-five-git/vespertide/refs/heads/main/schemas/model.schema.json", + "name": "deliveries", + "description": "Per-endpoint webhook delivery and dead-letter state", + "columns": [ + { + "name": "id", + "type": "text", + "nullable": false, + "primary_key": true + }, + { + "name": "webhook_id", + "type": "text", + "nullable": false, + "index": true + }, + { + "name": "event_id", + "type": "big_int", + "nullable": false, + "index": true + }, + { + "name": "event_name", + "type": "text", + "nullable": false, + "index": true + }, + { + "name": "status", + "type": "text", + "nullable": false, + "index": true + }, + { + "name": "attempts", + "type": "integer", + "nullable": false, + "default": 0 + }, + { + "name": "response_status", + "type": "integer", + "nullable": true + }, + { + "name": "last_error", + "type": "text", + "nullable": true + }, + { + "name": "created_at", + "type": "timestamptz", + "nullable": false, + "default": "NOW()", + "index": true + }, + { + "name": "updated_at", + "type": "timestamptz", + "nullable": false, + "default": "NOW()" + }, + { + "name": "delivered_at", + "type": "timestamptz", + "nullable": true + } + ], + "constraints": [ + { + "type": "unique", + "name": "uq_webhook_deliveries_endpoint_event", + "columns": ["webhook_id", "event_id"] + } + ] +} diff --git a/plugins/webhooks/models/endpoints.json b/plugins/webhooks/models/endpoints.json new file mode 100644 index 0000000..d62bf75 --- /dev/null +++ b/plugins/webhooks/models/endpoints.json @@ -0,0 +1,73 @@ +{ + "$schema": "https://raw.githubusercontent.com/dev-five-git/vespertide/refs/heads/main/schemas/model.schema.json", + "name": "endpoints", + "description": "Administrator-configured signed webhook endpoints", + "columns": [ + { + "name": "id", + "type": "text", + "nullable": false, + "primary_key": true + }, + { + "name": "name", + "type": "text", + "nullable": false, + "index": true + }, + { + "name": "url", + "type": "text", + "nullable": false + }, + { + "name": "secret", + "type": "text", + "nullable": false + }, + { + "name": "event_names", + "type": "json", + "nullable": false + }, + { + "name": "allow_private_networks", + "type": "boolean", + "nullable": false, + "default": false + }, + { + "name": "timeout_seconds", + "type": "integer", + "nullable": false, + "default": 5 + }, + { + "name": "enabled", + "type": "boolean", + "nullable": false, + "default": true, + "index": true + }, + { + "name": "created_at", + "type": "timestamptz", + "nullable": false, + "default": "NOW()", + "index": true + }, + { + "name": "updated_at", + "type": "timestamptz", + "nullable": false, + "default": "NOW()" + } + ], + "constraints": [ + { + "type": "unique", + "name": "uq_webhook_endpoints_name", + "columns": ["name"] + } + ] +} diff --git a/plugins/webhooks/package.json b/plugins/webhooks/package.json new file mode 100644 index 0000000..482373e --- /dev/null +++ b/plugins/webhooks/package.json @@ -0,0 +1,19 @@ +{ + "name": "@yeollin-plugin/webhooks", + "version": "0.1.0", + "private": true, + "scripts": { + "dev": "cargo run -p yeollin-cli -- dev", + "build": "cargo run -p yeollin-cli -- build", + "typecheck": "tsc --noEmit" + }, + "dependencies": { + "@devup-ui/react": "^1.0.41", + "react": "^19.2.8" + }, + "devDependencies": { + "@types/react": "^19", + "typescript": "^7.0", + "vinext": "^1.0.0-beta.8" + } +} diff --git a/plugins/webhooks/src/delivery.rs b/plugins/webhooks/src/delivery.rs new file mode 100644 index 0000000..6b4ed16 --- /dev/null +++ b/plugins/webhooks/src/delivery.rs @@ -0,0 +1,561 @@ +//! Deferred, signed delivery for committed event envelopes. + +use std::{ + fmt::Write, + net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, + time::Duration, +}; + +use anyhow::Context; +use hmac::{Hmac, KeyInit, Mac}; +use reqwest::{redirect::Policy, Url}; +use sea_orm::{ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, Set}; +use sha2::Sha256; +use yeollin_plugin::EventEnvelope; + +use crate::models::{deliveries, endpoints}; + +pub(crate) const STATUS_PENDING: &str = "pending"; +pub(crate) const STATUS_DELIVERED: &str = "delivered"; +pub(crate) const STATUS_DEAD_LETTER: &str = "dead_letter"; +pub(crate) const MAX_ATTEMPTS: i32 = 5; +const ID_BYTES: usize = 16; +const MAX_ERROR_LENGTH: usize = 1_024; + +pub(crate) async fn deliver_event( + event: EventEnvelope, + db: DatabaseConnection, +) -> anyhow::Result<()> { + let configured = endpoints::Entity::find() + .filter(endpoints::Column::Enabled.eq(true)) + .all(&db) + .await?; + let mut retryable_failures = Vec::new(); + + for endpoint in configured { + if !matches_event(&endpoint.event_names, &event.name)? { + continue; + } + + let delivery = find_or_create_delivery(&db, &endpoint, &event).await?; + if matches!( + delivery.status.as_str(), + STATUS_DELIVERED | STATUS_DEAD_LETTER + ) { + continue; + } + + let attempt = delivery.attempts.saturating_add(1); + let now = chrono::Utc::now(); + let result = send(&endpoint, &delivery.id, &event).await; + let mut active: deliveries::ActiveModel = delivery.into(); + active.attempts = Set(attempt); + active.updated_at = Set(now.into()); + + match result { + Ok(status) => { + active.status = Set(STATUS_DELIVERED.to_string()); + active.response_status = Set(Some(i32::from(status.as_u16()))); + active.last_error = Set(None); + active.delivered_at = Set(Some(now.into())); + } + Err(failure) => { + let message = truncate_error(failure.message); + active.response_status = Set(failure.response_status); + active.last_error = Set(Some(message.clone())); + active.delivered_at = Set(None); + if attempt >= MAX_ATTEMPTS { + active.status = Set(STATUS_DEAD_LETTER.to_string()); + tracing::error!( + delivery_id = %active_id(&active), + webhook_id = %endpoint.id, + event_id = event.id, + attempts = attempt, + error = %message, + "Webhook delivery entered the dead-letter state" + ); + } else { + active.status = Set(STATUS_PENDING.to_string()); + retryable_failures.push(format!("{}: {message}", endpoint.name)); + } + } + } + active.update(&db).await?; + } + + if retryable_failures.is_empty() { + Ok(()) + } else { + anyhow::bail!("webhook delivery failed: {}", retryable_failures.join("; ")) + } +} + +async fn find_or_create_delivery( + db: &DatabaseConnection, + endpoint: &endpoints::Model, + event: &EventEnvelope, +) -> anyhow::Result { + if let Some(delivery) = deliveries::Entity::find() + .filter(deliveries::Column::WebhookId.eq(&endpoint.id)) + .filter(deliveries::Column::EventId.eq(event.id)) + .one(db) + .await? + { + return Ok(delivery); + } + + let now = chrono::Utc::now(); + Ok(deliveries::ActiveModel { + id: Set(random_id()), + webhook_id: Set(endpoint.id.clone()), + event_id: Set(event.id), + event_name: Set(event.name.clone()), + status: Set(STATUS_PENDING.to_string()), + attempts: Set(0), + response_status: Set(None), + last_error: Set(None), + created_at: Set(now.into()), + updated_at: Set(now.into()), + delivered_at: Set(None), + } + .insert(db) + .await?) +} + +async fn send( + endpoint: &endpoints::Model, + delivery_id: &str, + event: &EventEnvelope, +) -> Result { + let url = validate_url(&endpoint.url).map_err(DeliveryFailure::without_status)?; + let resolved = resolve_and_validate(&url, endpoint.allow_private_networks) + .await + .map_err(DeliveryFailure::without_status)?; + let host = url + .host_str() + .ok_or_else(|| DeliveryFailure::without_status("Webhook URL has no host"))?; + let mut client = + reqwest::Client::builder() + .redirect(Policy::none()) + .timeout(Duration::from_secs( + u64::try_from(endpoint.timeout_seconds).unwrap_or(1), + )); + if host.parse::().is_err() { + client = client.resolve_to_addrs(host, &resolved); + } + let client = client + .build() + .map_err(|error| DeliveryFailure::without_status(error.to_string()))?; + let body = serde_json::to_vec(event) + .map_err(|error| DeliveryFailure::without_status(error.to_string()))?; + let signature = signature(&endpoint.secret, &body) + .map_err(|error| DeliveryFailure::without_status(error.to_string()))?; + let response = client + .post(url) + .header("content-type", "application/json") + .header("user-agent", "Yeollin-CMS-Webhooks/1.0") + .header("x-yeollin-event", &event.name) + .header("x-yeollin-delivery", delivery_id) + .header("x-yeollin-signature", format!("sha256={signature}")) + .body(body) + .send() + .await + .map_err(|error| DeliveryFailure::without_status(error.to_string()))?; + let status = response.status(); + if status.is_success() { + Ok(status) + } else { + Err(DeliveryFailure { + response_status: Some(i32::from(status.as_u16())), + message: format!("Endpoint returned HTTP {}", status.as_u16()), + }) + } +} + +pub(crate) fn validate_url(value: &str) -> Result { + let url = Url::parse(value.trim()).map_err(|_| "URL must be absolute".to_string())?; + if !matches!(url.scheme(), "http" | "https") { + return Err("URL must use http or https".to_string()); + } + if url.host_str().is_none() { + return Err("URL must include a host".to_string()); + } + if !url.username().is_empty() || url.password().is_some() { + return Err("URL must not contain credentials".to_string()); + } + if url.fragment().is_some() { + return Err("URL must not contain a fragment".to_string()); + } + Ok(url) +} + +async fn resolve_and_validate(url: &Url, allow_private: bool) -> Result, String> { + let host = url + .host_str() + .ok_or_else(|| "URL has no host".to_string())?; + let port = url + .port_or_known_default() + .ok_or_else(|| "URL has no usable port".to_string())?; + let addresses = if let Ok(ip) = host.parse::() { + vec![SocketAddr::new(ip, port)] + } else { + tokio::net::lookup_host((host, port)) + .await + .with_context(|| format!("Could not resolve `{host}`")) + .map_err(|error| error.to_string())? + .collect() + }; + if addresses.is_empty() { + return Err("Webhook host resolved to no addresses".to_string()); + } + if !allow_private && addresses.iter().any(|address| blocked_ip(address.ip())) { + return Err( + "Webhook host resolves to a private, loopback, or link-local address".to_string(), + ); + } + Ok(addresses) +} + +fn blocked_ip(ip: IpAddr) -> bool { + match ip { + IpAddr::V4(ip) => blocked_ipv4(ip), + IpAddr::V6(ip) => blocked_ipv6(ip), + } +} + +fn blocked_ipv4(ip: Ipv4Addr) -> bool { + ip.is_private() + || ip.is_loopback() + || ip.is_link_local() + || ip.is_unspecified() + || ip.is_multicast() + || ip.is_broadcast() +} + +fn blocked_ipv6(ip: Ipv6Addr) -> bool { + let octets = ip.octets(); + ip.is_loopback() + || ip.is_unspecified() + || ip.is_multicast() + || ip.is_unicast_link_local() + || octets[0] & 0xfe == 0xfc + || ip.to_ipv4_mapped().is_some_and(blocked_ipv4) +} + +fn matches_event(value: &serde_json::Value, event_name: &str) -> anyhow::Result { + let names: Vec = + serde_json::from_value(value.clone()).context("stored webhook event filter is invalid")?; + Ok(names.is_empty() || names.iter().any(|name| name == event_name)) +} + +fn signature(secret: &str, body: &[u8]) -> anyhow::Result { + let mut mac = Hmac::::new_from_slice(secret.as_bytes()) + .context("webhook secret could not initialize HMAC")?; + mac.update(body); + Ok(mac + .finalize() + .into_bytes() + .iter() + .fold(String::with_capacity(64), |mut output, byte| { + let _ = write!(output, "{byte:02x}"); + output + })) +} + +fn random_id() -> String { + rand::random::<[u8; ID_BYTES]>().iter().fold( + String::with_capacity(ID_BYTES * 2), + |mut output, byte| { + let _ = write!(output, "{byte:02x}"); + output + }, + ) +} + +fn truncate_error(error: String) -> String { + error.chars().take(MAX_ERROR_LENGTH).collect() +} + +fn active_id(active: &deliveries::ActiveModel) -> &str { + match &active.id { + sea_orm::ActiveValue::Set(id) | sea_orm::ActiveValue::Unchanged(id) => id, + sea_orm::ActiveValue::NotSet => "unknown", + } +} + +struct DeliveryFailure { + response_status: Option, + message: String, +} + +impl DeliveryFailure { + fn without_status(message: impl Into) -> Self { + Self { + response_status: None, + message: message.into(), + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::{ + atomic::{AtomicU16, Ordering}, + Arc, + }; + + use axum::{ + body::Bytes, + extract::State, + http::{HeaderMap, StatusCode}, + routing::post, + Router, + }; + use sea_orm::PaginatorTrait; + use tokio::{sync::mpsc, task::JoinHandle}; + + use super::*; + + #[derive(Debug)] + struct CapturedRequest { + headers: HeaderMap, + body: Vec, + } + + #[derive(Clone)] + struct ReceiverState { + status: Arc, + delay: Duration, + captured: mpsc::UnboundedSender, + } + + async fn capture( + State(state): State, + headers: HeaderMap, + body: Bytes, + ) -> StatusCode { + if !state.delay.is_zero() { + tokio::time::sleep(state.delay).await; + } + let _ = state.captured.send(CapturedRequest { + headers, + body: body.to_vec(), + }); + StatusCode::from_u16(state.status.load(Ordering::SeqCst)).unwrap() + } + + async fn receiver( + status: StatusCode, + delay: Duration, + ) -> ( + String, + Arc, + mpsc::UnboundedReceiver, + JoinHandle<()>, + ) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let (captured, received) = mpsc::unbounded_channel(); + let status = Arc::new(AtomicU16::new(status.as_u16())); + let state = ReceiverState { + status: Arc::clone(&status), + delay, + captured, + }; + let server = tokio::spawn(async move { + axum::serve( + listener, + Router::new() + .route("/hook", post(capture)) + .with_state(state), + ) + .await + .unwrap(); + }); + (format!("http://{address}/hook"), status, received, server) + } + + async fn database() -> DatabaseConnection { + let db = sea_orm::Database::connect("sqlite::memory:").await.unwrap(); + let metadata = crate::metadata(); + (metadata.on_init.as_ref().unwrap())(db.clone()) + .await + .unwrap(); + db + } + + async fn endpoint( + db: &DatabaseConnection, + id: &str, + url: String, + event_names: &[&str], + allow_private: bool, + timeout_seconds: i32, + ) { + let now = chrono::Utc::now(); + endpoints::ActiveModel { + id: Set(id.to_string()), + name: Set(format!("Endpoint {id}")), + url: Set(url), + secret: Set("0123456789abcdef0123456789abcdef".to_string()), + event_names: Set(serde_json::json!(event_names)), + allow_private_networks: Set(allow_private), + timeout_seconds: Set(timeout_seconds), + enabled: Set(true), + created_at: Set(now.into()), + updated_at: Set(now.into()), + } + .insert(db) + .await + .unwrap(); + } + + fn event(id: i64, name: &str) -> EventEnvelope { + EventEnvelope { + id, + name: name.to_string(), + payload: serde_json::json!({ "value": "signed" }), + audit: true, + created_at: chrono::Utc::now().into(), + } + } + + #[test] + fn hmac_signature_is_stable() { + assert_eq!( + signature("secret", b"body").unwrap(), + "dc46983557fea127b43af721467eb9b3fde2338fe3e14f51952aa8478c13d355" + ); + } + + #[test] + fn blocks_non_public_address_classes() { + for address in [ + "127.0.0.1", + "10.0.0.1", + "172.16.0.1", + "192.168.1.1", + "169.254.1.1", + "0.0.0.0", + "::1", + "fe80::1", + "fd00::1", + "::ffff:127.0.0.1", + ] { + assert!( + blocked_ip(address.parse().unwrap()), + "did not block {address}" + ); + } + assert!(!blocked_ip("1.1.1.1".parse().unwrap())); + assert!(!blocked_ip("2606:4700:4700::1111".parse().unwrap())); + } + + #[test] + fn webhook_urls_exclude_ambiguous_authorities() { + assert!(validate_url("https://example.com/hooks?id=1").is_ok()); + for invalid in [ + "ftp://example.com/hook", + "https://user:pass@example.com/hook", + "https://example.com/hook#fragment", + "/relative", + ] { + assert!(validate_url(invalid).is_err(), "accepted {invalid}"); + } + } + + #[tokio::test] + async fn exact_filter_delivers_a_signed_envelope_once() { + let db = database().await; + let (url, _status, mut received, server) = + receiver(StatusCode::NO_CONTENT, Duration::ZERO).await; + endpoint(&db, "matching", url.clone(), &["memo.created"], true, 2).await; + endpoint(&db, "filtered", url, &["memo.updated"], true, 2).await; + let envelope = event(41, "memo.created"); + + deliver_event(envelope.clone(), db.clone()).await.unwrap(); + let captured = tokio::time::timeout(Duration::from_secs(1), received.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(captured.body, serde_json::to_vec(&envelope).unwrap()); + assert_eq!(captured.headers["x-yeollin-event"], "memo.created"); + assert!( + captured.headers["x-yeollin-delivery"] + .to_str() + .unwrap() + .len() + == ID_BYTES * 2 + ); + let expected = format!( + "sha256={}", + signature("0123456789abcdef0123456789abcdef", &captured.body).unwrap() + ); + assert_eq!(captured.headers["x-yeollin-signature"], expected); + assert!(received.try_recv().is_err()); + assert_eq!(deliveries::Entity::find().count(&db).await.unwrap(), 1); + + deliver_event(envelope, db).await.unwrap(); + assert!( + received.try_recv().is_err(), + "successful delivery was resent" + ); + server.abort(); + } + + #[tokio::test] + async fn failures_stop_retrying_in_the_dead_letter_state() { + let db = database().await; + let (url, _status, mut received, server) = + receiver(StatusCode::SERVICE_UNAVAILABLE, Duration::ZERO).await; + endpoint(&db, "failing", url, &[], true, 2).await; + let envelope = event(42, "memo.created"); + + for attempt in 1..=MAX_ATTEMPTS { + let result = deliver_event(envelope.clone(), db.clone()).await; + assert_eq!(result.is_err(), attempt < MAX_ATTEMPTS); + received.recv().await.unwrap(); + } + let delivery = deliveries::Entity::find().one(&db).await.unwrap().unwrap(); + assert_eq!(delivery.status, STATUS_DEAD_LETTER); + assert_eq!(delivery.attempts, MAX_ATTEMPTS); + assert_eq!(delivery.response_status, Some(503)); + + deliver_event(envelope, db).await.unwrap(); + assert!(received.try_recv().is_err(), "dead letter was sent again"); + server.abort(); + } + + #[tokio::test] + async fn local_addresses_require_the_explicit_opt_out() { + let db = database().await; + let (url, _status, mut received, server) = + receiver(StatusCode::NO_CONTENT, Duration::ZERO).await; + endpoint(&db, "blocked", url, &[], false, 2).await; + + let error = deliver_event(event(43, "memo.created"), db.clone()) + .await + .unwrap_err(); + assert!(error + .to_string() + .contains("private, loopback, or link-local")); + assert!(received.try_recv().is_err()); + let delivery = deliveries::Entity::find().one(&db).await.unwrap().unwrap(); + assert_eq!(delivery.status, STATUS_PENDING); + assert_eq!(delivery.attempts, 1); + server.abort(); + } + + #[tokio::test] + async fn per_delivery_timeout_is_enforced() { + let db = database().await; + let (url, _status, _received, server) = + receiver(StatusCode::NO_CONTENT, Duration::from_secs(3)).await; + endpoint(&db, "slow", url, &[], true, 1).await; + let started = tokio::time::Instant::now(); + + assert!(deliver_event(event(44, "memo.created"), db).await.is_err()); + assert!(started.elapsed() < Duration::from_secs(2)); + server.abort(); + } +} diff --git a/plugins/webhooks/src/lib.rs b/plugins/webhooks/src/lib.rs new file mode 100644 index 0000000..bf163b7 --- /dev/null +++ b/plugins/webhooks/src/lib.rs @@ -0,0 +1,18 @@ +//! Signed event webhooks with SSRF-safe delivery. + +mod delivery; +pub mod models; +mod routes; + +use yeollin_plugin::SubscriberRegistration; + +yeollin_plugin::yeollin_plugin! { + name: "webhooks", + author: "DevFive", + description: "Signed event webhooks with SSRF-safe delivery", + subscribers: [SubscriberRegistration::deferred( + "deliver", + [], + delivery::deliver_event, + )], +} diff --git a/plugins/webhooks/src/models/deliveries.rs b/plugins/webhooks/src/models/deliveries.rs new file mode 100644 index 0000000..70b3363 --- /dev/null +++ b/plugins/webhooks/src/models/deliveries.rs @@ -0,0 +1,41 @@ +use sea_orm::entity::prelude::*; + +/// Per-endpoint webhook delivery and dead-letter state +#[sea_orm::model] +#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)] +#[sea_orm(table_name = "webhook_deliveries")] +pub struct Model { + #[sea_orm(primary_key)] + pub id: String, + #[sea_orm(indexed)] + pub webhook_id: String, + #[sea_orm(indexed)] + pub event_id: i64, + #[sea_orm(indexed)] + pub event_name: String, + #[sea_orm(indexed)] + pub status: String, + #[sea_orm(default_value = 0)] + pub attempts: i32, + pub response_status: Option, + pub last_error: Option, + #[sea_orm(indexed, default_value = "NOW()")] + pub created_at: DateTimeWithTimeZone, + #[sea_orm(default_value = "NOW()")] + pub updated_at: DateTimeWithTimeZone, + pub delivered_at: Option, +} + +// Index definitions (SeaORM uses Statement builders externally) +// (unnamed) on [webhook_id] +// (unnamed) on [event_id] +// (unnamed) on [event_name] +// (unnamed) on [status] +// (unnamed) on [created_at] + +/// Composite unique constraints — declare in migrations or use Statement builder. +pub const COMPOSITE_UNIQUES: &[&[&str]] = &[ + &["webhook_id", "event_id"], // uq_webhook_deliveries_endpoint_event +]; +vespera::schema_type!(Schema from Model, name = "DeliveriesSchema"); +impl ActiveModelBehavior for ActiveModel {} diff --git a/plugins/webhooks/src/models/endpoints.rs b/plugins/webhooks/src/models/endpoints.rs new file mode 100644 index 0000000..0ae7364 --- /dev/null +++ b/plugins/webhooks/src/models/endpoints.rs @@ -0,0 +1,32 @@ +use sea_orm::entity::prelude::*; + +/// Administrator-configured signed webhook endpoints +#[sea_orm::model] +#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)] +#[sea_orm(table_name = "webhook_endpoints")] +pub struct Model { + #[sea_orm(primary_key)] + pub id: String, + #[sea_orm(unique)] + pub name: String, + pub url: String, + pub secret: String, + pub event_names: Json, + #[sea_orm(default_value = false)] + pub allow_private_networks: bool, + #[sea_orm(default_value = 5)] + pub timeout_seconds: i32, + #[sea_orm(indexed, default_value = true)] + pub enabled: bool, + #[sea_orm(indexed, default_value = "NOW()")] + pub created_at: DateTimeWithTimeZone, + #[sea_orm(default_value = "NOW()")] + pub updated_at: DateTimeWithTimeZone, +} + +// Index definitions (SeaORM uses Statement builders externally) +// (unnamed) on [name] +// (unnamed) on [enabled] +// (unnamed) on [created_at] +vespera::schema_type!(Schema from Model, name = "EndpointsSchema"); +impl ActiveModelBehavior for ActiveModel {} diff --git a/plugins/webhooks/src/models/mod.rs b/plugins/webhooks/src/models/mod.rs new file mode 100644 index 0000000..0d8515e --- /dev/null +++ b/plugins/webhooks/src/models/mod.rs @@ -0,0 +1,2 @@ +pub mod deliveries; +pub mod endpoints; diff --git a/plugins/webhooks/src/routes/mod.rs b/plugins/webhooks/src/routes/mod.rs new file mode 100644 index 0000000..e4c1d6f --- /dev/null +++ b/plugins/webhooks/src/routes/mod.rs @@ -0,0 +1,538 @@ +//! Administrator webhook configuration and delivery history APIs. + +use std::fmt::Write; + +use axum::{extract::Query, Extension, Json}; +use sea_orm::{ + ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, Order, PaginatorTrait, + QueryFilter, QueryOrder, Set, TransactionTrait, +}; +use serde::{Deserialize, Serialize}; +use vespera::Schema; +use yeollin_plugin::{Authorize, CurrentUser, EventBus, PluginError}; + +use crate::{ + delivery::{validate_url, MAX_ATTEMPTS, STATUS_DEAD_LETTER, STATUS_DELIVERED, STATUS_PENDING}, + models::{deliveries, endpoints}, +}; + +const DEFAULT_PAGE_SIZE: u64 = 25; +const MAX_PAGE_SIZE: u64 = 100; +const ID_BYTES: usize = 16; +const MIN_SECRET_BYTES: usize = 32; +const MAX_SECRET_BYTES: usize = 512; +const MAX_NAME_CHARS: usize = 100; +const MAX_EVENT_NAMES: usize = 100; +const MAX_EVENT_NAME_CHARS: usize = 128; +const MAX_URL_CHARS: usize = 2_048; +const MIN_TIMEOUT_SECONDS: i32 = 1; +const MAX_TIMEOUT_SECONDS: i32 = 30; + +#[derive(Clone, Debug, Serialize, Schema)] +#[serde(rename_all = "camelCase")] +pub struct WebhookResponse { + pub id: String, + pub name: String, + pub url: String, + pub event_names: Vec, + pub allow_private_networks: bool, + pub timeout_seconds: i32, + pub enabled: bool, + pub has_secret: bool, + pub created_at: String, + pub updated_at: String, +} + +impl TryFrom for WebhookResponse { + type Error = PluginError; + + fn try_from(model: endpoints::Model) -> Result { + let event_names = serde_json::from_value(model.event_names).map_err(|error| { + tracing::error!(%error, webhook_id = %model.id, "Stored webhook filter is invalid"); + PluginError::internal() + })?; + Ok(Self { + id: model.id, + name: model.name, + url: model.url, + event_names, + allow_private_networks: model.allow_private_networks, + timeout_seconds: model.timeout_seconds, + enabled: model.enabled, + has_secret: !model.secret.is_empty(), + created_at: model.created_at.to_rfc3339(), + updated_at: model.updated_at.to_rfc3339(), + }) + } +} + +#[derive(Debug, Serialize, Schema)] +#[serde(rename_all = "camelCase")] +pub struct ListWebhooksResponse { + pub webhooks: Vec, +} + +#[derive(Debug, Deserialize, Schema)] +#[serde(rename_all = "camelCase")] +pub struct CreateWebhookRequest { + pub name: String, + pub url: String, + pub secret: String, + #[serde(default)] + #[schema(default = "[]")] + pub event_names: Vec, + #[serde(default)] + pub allow_private_networks: bool, + #[serde(default = "default_timeout_seconds")] + pub timeout_seconds: i32, + #[serde(default = "default_enabled")] + pub enabled: bool, +} + +#[derive(Debug, Deserialize, Schema)] +#[serde(rename_all = "camelCase")] +pub struct UpdateWebhookRequest { + pub name: String, + pub url: String, + pub secret: Option, + #[serde(default)] + #[schema(default = "[]")] + pub event_names: Vec, + #[serde(default)] + pub allow_private_networks: bool, + pub timeout_seconds: i32, + pub enabled: bool, +} + +#[derive(Debug, Serialize, Schema)] +#[serde(rename_all = "camelCase")] +pub struct DeleteWebhookResponse { + pub success: bool, + pub deleted_id: String, +} + +#[derive(Clone, Debug, Serialize, Schema)] +#[serde(rename_all = "camelCase")] +pub struct DeliveryResponse { + pub id: String, + pub webhook_id: String, + pub event_id: i64, + pub event_name: String, + pub status: String, + pub attempts: i32, + pub max_attempts: i32, + pub response_status: Option, + pub last_error: Option, + pub created_at: String, + pub updated_at: String, + pub delivered_at: Option, +} + +impl From for DeliveryResponse { + fn from(model: deliveries::Model) -> Self { + Self { + id: model.id, + webhook_id: model.webhook_id, + event_id: model.event_id, + event_name: model.event_name, + status: model.status, + attempts: model.attempts, + max_attempts: MAX_ATTEMPTS, + response_status: model.response_status, + last_error: model.last_error, + created_at: model.created_at.to_rfc3339(), + updated_at: model.updated_at.to_rfc3339(), + delivered_at: model.delivered_at.map(|value| value.to_rfc3339()), + } + } +} + +#[derive(Debug, Default, Deserialize, Schema)] +#[serde(rename_all = "camelCase")] +pub struct ListDeliveriesQuery { + pub page: Option, + pub page_size: Option, + pub webhook_id: Option, + pub status: Option, +} + +#[derive(Debug, Serialize, Schema)] +#[serde(rename_all = "camelCase")] +pub struct ListDeliveriesResponse { + pub deliveries: Vec, + pub total: u64, + pub page: u64, + pub page_size: u64, +} + +/// List configured endpoints without returning signing secrets. +#[vespera::route(get, tags = ["webhooks"])] +pub async fn list_webhooks( + Extension(db): Extension, + Extension(current): Extension, +) -> Result, PluginError> { + current.require_role("admin")?; + let webhooks = endpoints::Entity::find() + .order_by(endpoints::Column::CreatedAt, Order::Desc) + .all(&db) + .await? + .into_iter() + .map(WebhookResponse::try_from) + .collect::>()?; + Ok(Json(ListWebhooksResponse { webhooks })) +} + +/// Create an endpoint with a write-only signing secret. +#[vespera::route(post, tags = ["webhooks"])] +pub async fn create_webhook( + Extension(db): Extension, + Extension(current): Extension, + Json(request): Json, +) -> Result, PluginError> { + current.require_role("admin")?; + let values = validate_values( + request.name, + request.url, + request.event_names, + request.timeout_seconds, + )?; + let secret = validate_secret(request.secret)?; + ensure_name_available(&db, &values.name, None).await?; + let now = chrono::Utc::now(); + let model = endpoints::ActiveModel { + id: Set(random_id()), + name: Set(values.name), + url: Set(values.url), + secret: Set(secret), + event_names: Set(serde_json::json!(values.event_names)), + allow_private_networks: Set(request.allow_private_networks), + timeout_seconds: Set(values.timeout_seconds), + enabled: Set(request.enabled), + created_at: Set(now.into()), + updated_at: Set(now.into()), + } + .insert(&db) + .await?; + Ok(Json(WebhookResponse::try_from(model)?)) +} + +/// Replace endpoint configuration; omit `secret` to retain the existing value. +#[vespera::route(put, path = "/{id}", tags = ["webhooks"])] +pub async fn update_webhook( + Extension(db): Extension, + Extension(current): Extension, + axum::extract::Path(id): axum::extract::Path, + Json(request): Json, +) -> Result, PluginError> { + current.require_role("admin")?; + let Some(existing) = endpoints::Entity::find_by_id(&id).one(&db).await? else { + return Err(PluginError::not_found("Webhook not found")); + }; + let values = validate_values( + request.name, + request.url, + request.event_names, + request.timeout_seconds, + )?; + ensure_name_available(&db, &values.name, Some(&id)).await?; + let secret = request.secret.map(validate_secret).transpose()?; + let mut active: endpoints::ActiveModel = existing.into(); + active.name = Set(values.name); + active.url = Set(values.url); + if let Some(secret) = secret { + active.secret = Set(secret); + } + active.event_names = Set(serde_json::json!(values.event_names)); + active.allow_private_networks = Set(request.allow_private_networks); + active.timeout_seconds = Set(values.timeout_seconds); + active.enabled = Set(request.enabled); + active.updated_at = Set(chrono::Utc::now().into()); + let model = active.update(&db).await?; + Ok(Json(WebhookResponse::try_from(model)?)) +} + +/// Delete an endpoint and its per-delivery history. +#[vespera::route(delete, path = "/{id}", tags = ["webhooks"])] +pub async fn delete_webhook( + Extension(db): Extension, + Extension(current): Extension, + axum::extract::Path(id): axum::extract::Path, +) -> Result, PluginError> { + current.require_role("admin")?; + if endpoints::Entity::find_by_id(&id).one(&db).await?.is_none() { + return Err(PluginError::not_found("Webhook not found")); + } + let transaction = db.begin().await?; + deliveries::Entity::delete_many() + .filter(deliveries::Column::WebhookId.eq(&id)) + .exec(&transaction) + .await?; + endpoints::Entity::delete_by_id(&id) + .exec(&transaction) + .await?; + transaction.commit().await?; + Ok(Json(DeleteWebhookResponse { + success: true, + deleted_id: id, + })) +} + +/// List recent endpoint deliveries and dead letters. +#[vespera::route(get, path = "/deliveries", tags = ["webhooks"])] +pub async fn list_deliveries( + Extension(db): Extension, + Extension(current): Extension, + Query(query): Query, +) -> Result, PluginError> { + current.require_role("admin")?; + let page = query.page.unwrap_or(1).max(1); + let page_size = query + .page_size + .unwrap_or(DEFAULT_PAGE_SIZE) + .clamp(1, MAX_PAGE_SIZE); + let mut find = deliveries::Entity::find(); + if let Some(webhook_id) = query + .webhook_id + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + { + find = find.filter(deliveries::Column::WebhookId.eq(webhook_id)); + } + if let Some(status) = query + .status + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + { + if !matches!( + status, + STATUS_PENDING | STATUS_DELIVERED | STATUS_DEAD_LETTER + ) { + return Err(PluginError::bad_request("Unknown delivery status")); + } + find = find.filter(deliveries::Column::Status.eq(status)); + } + let paginator = find + .order_by(deliveries::Column::CreatedAt, Order::Desc) + .paginate(&db, page_size); + let total = paginator.num_items().await?; + let deliveries = paginator + .fetch_page(page - 1) + .await? + .into_iter() + .map(DeliveryResponse::from) + .collect(); + Ok(Json(ListDeliveriesResponse { + deliveries, + total, + page, + page_size, + })) +} + +/// Reset a dead letter and immediately requeue its immutable source event. +#[vespera::route(post, path = "/deliveries/{id}/retry", tags = ["webhooks"])] +pub async fn retry_delivery( + Extension(db): Extension, + Extension(events): Extension, + Extension(current): Extension, + axum::extract::Path(id): axum::extract::Path, +) -> Result, PluginError> { + current.require_role("admin")?; + let Some(delivery) = deliveries::Entity::find_by_id(&id).one(&db).await? else { + return Err(PluginError::not_found("Webhook delivery not found")); + }; + if delivery.status != STATUS_DEAD_LETTER { + return Err(PluginError::conflict( + "Only dead-letter deliveries can be retried", + )); + } + let Some(endpoint) = endpoints::Entity::find_by_id(&delivery.webhook_id) + .one(&db) + .await? + else { + return Err(PluginError::bad_request("The webhook no longer exists")); + }; + if !endpoint.enabled { + return Err(PluginError::bad_request( + "Enable the webhook before retrying its delivery", + )); + } + + let event_id = delivery.event_id; + let previous = delivery.clone(); + let mut active: deliveries::ActiveModel = delivery.into(); + active.status = Set(STATUS_PENDING.to_string()); + active.attempts = Set(0); + active.response_status = Set(None); + active.last_error = Set(None); + active.updated_at = Set(chrono::Utc::now().into()); + active.delivered_at = Set(None); + let reset = active.update(&db).await?; + match events.requeue(event_id).await { + Ok(true) => {} + result => { + let mut restore: deliveries::ActiveModel = reset.clone().into(); + restore.status = Set(previous.status); + restore.attempts = Set(previous.attempts); + restore.response_status = Set(previous.response_status); + restore.last_error = Set(previous.last_error); + restore.updated_at = Set(chrono::Utc::now().into()); + restore.delivered_at = Set(previous.delivered_at); + restore.update(&db).await?; + if let Err(error) = result { + return Err(error.into()); + } + return Err(PluginError::bad_request( + "The source event is no longer available", + )); + } + } + Ok(Json(DeliveryResponse::from(reset))) +} + +struct ValidatedValues { + name: String, + url: String, + event_names: Vec, + timeout_seconds: i32, +} + +fn validate_values( + name: String, + url: String, + event_names: Vec, + timeout_seconds: i32, +) -> Result { + let name = name.trim().to_string(); + if name.is_empty() || name.chars().count() > MAX_NAME_CHARS { + return Err(PluginError::bad_request(format!( + "Name must contain 1 to {MAX_NAME_CHARS} characters" + ))); + } + if url.chars().count() > MAX_URL_CHARS { + return Err(PluginError::bad_request("Webhook URL is too long")); + } + let url = validate_url(&url) + .map_err(PluginError::bad_request)? + .to_string(); + let event_names = normalize_event_names(event_names)?; + if !(MIN_TIMEOUT_SECONDS..=MAX_TIMEOUT_SECONDS).contains(&timeout_seconds) { + return Err(PluginError::bad_request(format!( + "timeoutSeconds must be between {MIN_TIMEOUT_SECONDS} and {MAX_TIMEOUT_SECONDS}" + ))); + } + Ok(ValidatedValues { + name, + url, + event_names, + timeout_seconds, + }) +} + +fn normalize_event_names(names: Vec) -> Result, PluginError> { + if names.len() > MAX_EVENT_NAMES { + return Err(PluginError::bad_request(format!( + "At most {MAX_EVENT_NAMES} event names are allowed" + ))); + } + let mut normalized = Vec::with_capacity(names.len()); + for name in names { + let name = name.trim(); + if name.is_empty() { + continue; + } + if name.chars().count() > MAX_EVENT_NAME_CHARS + || !name.chars().all(|character| { + character.is_ascii_alphanumeric() || matches!(character, '.' | '-' | '_' | ':') + }) + { + return Err(PluginError::bad_request(format!( + "Invalid event name `{name}`" + ))); + } + normalized.push(name.to_string()); + } + normalized.sort(); + normalized.dedup(); + Ok(normalized) +} + +fn validate_secret(secret: String) -> Result { + let length = secret.len(); + if !(MIN_SECRET_BYTES..=MAX_SECRET_BYTES).contains(&length) { + return Err(PluginError::bad_request(format!( + "Secret must contain {MIN_SECRET_BYTES} to {MAX_SECRET_BYTES} bytes" + ))); + } + Ok(secret) +} + +async fn ensure_name_available( + db: &DatabaseConnection, + name: &str, + except_id: Option<&str>, +) -> Result<(), PluginError> { + let mut find = endpoints::Entity::find().filter(endpoints::Column::Name.eq(name)); + if let Some(id) = except_id { + find = find.filter(endpoints::Column::Id.ne(id)); + } + if find.one(db).await?.is_some() { + return Err(PluginError::conflict( + "A webhook with this name already exists", + )); + } + Ok(()) +} + +fn random_id() -> String { + rand::random::<[u8; ID_BYTES]>().iter().fold( + String::with_capacity(ID_BYTES * 2), + |mut output, byte| { + let _ = write!(output, "{byte:02x}"); + output + }, + ) +} + +const fn default_timeout_seconds() -> i32 { + 5 +} + +const fn default_enabled() -> bool { + true +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn filters_are_exact_normalized_and_deduplicated() { + assert_eq!( + normalize_event_names(vec![ + " memo.updated ".to_string(), + "memo.created".to_string(), + "memo.updated".to_string(), + String::new(), + ]) + .unwrap(), + ["memo.created", "memo.updated"] + ); + assert!(normalize_event_names(vec!["memo created".to_string()]).is_err()); + } + + #[test] + fn secrets_and_timeouts_are_bounded() { + assert!(validate_secret("short".to_string()).is_err()); + assert!(validate_secret("s".repeat(MIN_SECRET_BYTES)).is_ok()); + assert!(validate_values( + "Hook".to_string(), + "https://example.com/hook".to_string(), + Vec::new(), + MAX_TIMEOUT_SECONDS + 1, + ) + .is_err()); + } +} diff --git a/plugins/webhooks/tsconfig.json b/plugins/webhooks/tsconfig.json new file mode 100644 index 0000000..437c3d2 --- /dev/null +++ b/plugins/webhooks/tsconfig.json @@ -0,0 +1,11 @@ +{ + "extends": "../../packages/app/tsconfig.json", + "compilerOptions": { + "paths": { + "@/*": ["../../packages/app/src/*"] + }, + "noEmit": true + }, + "include": ["app/**/*.ts", "app/**/*.tsx"], + "exclude": ["node_modules"] +} diff --git a/plugins/webhooks/vespertide.json b/plugins/webhooks/vespertide.json new file mode 100644 index 0000000..3de0534 --- /dev/null +++ b/plugins/webhooks/vespertide.json @@ -0,0 +1,16 @@ +{ + "modelsDir": "models", + "migrationsDir": "migrations", + "tableNamingCase": "snake", + "columnNamingCase": "snake", + "modelFormat": "json", + "migrationFormat": "json", + "migrationFilenamePattern": "%04v_%m", + "modelExportDir": "src/models", + "seaorm": { + "extraEnumDerives": ["vespera::Schema"], + "extraModelDerives": [], + "enumNamingCase": "camel" + }, + "prefix": "webhook_" +} From 10a3e8604a2041933cb0b23e409f64fb63bde457 Mon Sep 17 00:00:00 2001 From: devfive Date: Mon, 31 Aug 2026 07:36:10 +0900 Subject: [PATCH 2/2] Document webhook delivery guarantees Give operators and plugin authors the signing contract, retry lifecycle, filtering semantics, and private-network safety model needed to consume webhooks correctly. --- CHANGELOG.md | 3 +++ README.md | 2 +- docs/architecture.md | 11 +++++++++ docs/plugin-authoring.md | 52 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 67 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index eac766e..591f649 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ The project is pre-1.0, so breaking changes can appear in any release. ### Security +- Webhook delivery signs the exact event envelope with HMAC-SHA256, disables redirects, enforces a per-request timeout, and blocks private, loopback, link-local, unspecified, multicast, and IPv6 unique-local destinations by default. Validated DNS answers are pinned to prevent rebinding; private-network delivery requires an explicit per-endpoint opt-out. - Authentication middleware no longer exempts any path ending in `.ico`. Only exact-match files such as `/favicon.ico` are exempt. - Vite dev-server asset paths (`/@`, `/__vite_hmr`, `/node_modules/`, `/src/`, `/df/`) now skip authentication only while the dev proxy is active. - Public and guest route matching is whole-path exact instead of prefix-based, and the server refuses to start unless `JWT_SECRET` is at least 32 bytes. @@ -21,6 +22,7 @@ The project is pre-1.0, so breaking changes can appear in any release. ### Added +- The `webhooks` plugin adds administrator endpoint CRUD, write-only signing secrets, exact event-name filters, per-endpoint delivery history, five-attempt dead letters, and explicit manual retries. Successful endpoints are not resent when another endpoint fails. - Compile-time typed content collections through `yeollin_content_collection!` and the `collections` plugin declaration. Collection field types drive concrete handlers, validation, exported schemas, and generated editor pages while the framework owns IDs, collection-scoped slugs, author, and timestamps. - A shared draft/published content repository with paginated administrator CRUD, transactional `content.created` / `updated` / `published` / `unpublished` / `deleted` audit events, and an exact public-by-slug endpoint that exposes only published entries. The reference `content` plugin ships a typed `pages` collection with media-reference validation. - The `media` plugin provides an administrator media library, typed multipart image uploads, paginated metadata, deletion, and a public serving route. JPEG, PNG, GIF, and WebP are verified from their signatures; upload size has a 10 MiB hard ceiling and a typed 1–10 MiB setting. @@ -41,6 +43,7 @@ The project is pre-1.0, so breaking changes can appear in any release. ### Changed +- Deferred outbox failures now retry with exponential backoff capped at five minutes instead of a fixed polling delay. - Passwords must be at least 12 **characters** — counted as characters, not bytes, so a short multi-byte password cannot pass. This applies to the bootstrap administrator too, so a deployment whose `YEOLLIN_ADMIN_PASSWORD` is shorter now fails at startup with the reason rather than seeding a weak account. - Changing or resetting a password ends every session for that account. A refresh token minted before the change stops working, which is what makes a password change useful for containing a compromise. diff --git a/README.md b/README.md index 346e00b..1de85cd 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,7 @@ Yeollin CMS is a Tauri-inspired, plugin-based CMS *framework* rather than a fini |------|----------| | `crates/` | The Rust workspace crates: `core` (shared types), `auth` (JWT, Argon2, middleware), `plugin` (`PluginMetadata`, `FrontendAssets`), `plugin-macros` (`yeollin_plugin!`, `yeollin_app!`), `app` (`YeollinAppBuilder` runtime), `cli` (`init`, `prebuild`, `dev`, `build`). | | `packages/` | The Node workspace. `packages/app` is the vinext frontend template that gets extracted into `.yeollin/app/` at prebuild time. It is a template, not the running app. | -| `plugins/` | Plugin crates. `auth` owns accounts and sessions; `audit-log` reads explicitly marked outbox events; `media` owns runtime image uploads; `content` demonstrates compile-time typed draft/publish collections; `example-plugin` is a minimal library plugin; `example-memo-plugin` demonstrates database CRUD, typed settings, and audited events. | +| `plugins/` | Plugin crates. `auth` owns accounts and sessions; `audit-log` reads explicitly marked outbox events; `media` owns runtime image uploads; `content` demonstrates compile-time typed draft/publish collections; `webhooks` delivers signed events with retry and dead-letter history; `example-plugin` is a minimal library plugin; `example-memo-plugin` demonstrates database CRUD, typed settings, and audited events. | | `apps/` | Standalone application crates. `apps/example-app` wires the example plugins together with `yeollin_app!` and is the entry point used for local development. | `.yeollin/` is generated during prebuild and is gitignored. Never edit it by hand. diff --git a/docs/architecture.md b/docs/architecture.md index 06c2492..a365521 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -136,6 +136,17 @@ in place. Its retention pass deletes only processed, marked rows so the outbox remains the single source of truth and pending delivery is never treated as a disposable log. +The `webhooks` plugin attaches one Deferred subscriber to that drainer. It +materializes a stable row per `(endpoint, event)`, which is the idempotency +boundary: endpoints already marked delivered are skipped while a failed peer +retries. The shared outbox schedules failures with capped exponential backoff; +the endpoint row records the response status, error, attempts, and terminal +dead-letter state. A manual retry resets that row and makes the immutable source +event immediately available again. HMAC-SHA256 covers the exact envelope bytes. +Network delivery disables redirects, applies a per-endpoint timeout, validates +every resolved address against the default private/loopback/link-local denylist, +and pins accepted DNS answers for the connection. + The same core migration owns `content_entries`. A plugin collection registers a concrete Rust field type, generated handlers, and its build-time schema. Runtime writes round-trip that concrete type through the shared JSON field while the diff --git a/docs/plugin-authoring.md b/docs/plugin-authoring.md index e6c3e81..3521e72 100644 --- a/docs/plugin-authoring.md +++ b/docs/plugin-authoring.md @@ -450,6 +450,58 @@ event transaction uncommittable and rolls it back. Deferred delivery is at-least-once, so Deferred handlers must be idempotent; a crash after the handler succeeds but before the outbox row is marked can deliver it again. +### Signed webhooks + +The `webhooks` plugin is the standard external Deferred subscriber. Its +administrator page at `/webhooks` configures endpoints and shows per-endpoint +delivery history. The API lives at `/api/webhooks`; signing secrets are accepted +on create or replacement but are never serialized back to a caller. + +Each matching endpoint receives the serialized `EventEnvelope` as the exact +request body with these headers: + +| Header | Value | +|--------|-------| +| `Content-Type` | `application/json` | +| `X-Yeollin-Event` | Exact event name, such as `content.published` | +| `X-Yeollin-Delivery` | Stable per-endpoint delivery ID | +| `X-Yeollin-Signature` | `sha256=` | + +Verify the signature over the raw request bytes before parsing JSON. The +conceptual check is `HMAC-SHA256(endpoint_secret, raw_body)`. Compare the +supplied and calculated digests in constant time. Do not calculate the digest +over re-serialized JSON, because whitespace and object-key ordering can change +the byte sequence. + +Endpoint event filters use whole-name exact matching. An empty filter receives +every event; `content` does not match `content.published`. A successful endpoint +is not sent again when another endpoint needs a retry. Failures retry through +the core outbox with delays of 1, 2, 4, 8 seconds and so on, capped at five +minutes. Each endpoint stops after five attempts and enters `dead_letter`. +Administrators can explicitly retry a dead letter while its immutable source +event still exists. + +Every delivery has its own timeout (1–30 seconds), accepts only HTTP or HTTPS, +and refuses redirects. Before connecting, the plugin resolves the hostname, +blocks private, loopback, and link-local addresses by default, and pins the +validated DNS answers into the HTTP client to prevent DNS rebinding between the +check and connection. Unspecified, multicast, and IPv6 unique-local addresses +are refused too. `allowPrivateNetworks` is an explicit per-endpoint opt-out for +a trusted internal receiver; enabling it removes the address-range guard, not +the signature, timeout, or redirect protections. + +The management routes are administrator-only: + +| Method | Path | Purpose | +|--------|------|---------| +| `GET`, `POST` | `/api/webhooks` | List endpoints or create one. | +| `PUT`, `DELETE` | `/api/webhooks/{id}` | Replace or delete endpoint configuration. | +| `GET` | `/api/webhooks/deliveries` | Paginated history with endpoint/status filters. | +| `POST` | `/api/webhooks/deliveries/{id}/retry` | Reset a dead letter and requeue its source event. | + +Treat event payloads as a public contract with the receiving system. Never put +passwords, tokens, signing secrets, or unnecessary personal data in an event. + ## API routes: how URLs are derived Every plugin API lives under `/api/`: