From e424144b68dc9e3f3e64e7b0dbc0727f387addd6 Mon Sep 17 00:00:00 2001 From: Shiv Rossi Date: Sun, 6 Sep 2026 21:47:08 -0600 Subject: [PATCH] feat: add Uptime Kuma event source --- Cargo.lock | 2 + Cargo.toml | 1 + README.md | 13 +- crates/rite-server/Cargo.toml | 1 + crates/rite-server/src/dispatch.rs | 22 ++- crates/rite-server/src/lib.rs | 121 +++++++++++++-- crates/rite-sources/Cargo.toml | 1 + crates/rite-sources/src/lib.rs | 1 + crates/rite-sources/src/uptime_kuma.rs | 196 +++++++++++++++++++++++++ rite.example.toml | 5 + 10 files changed, 338 insertions(+), 25 deletions(-) create mode 100644 crates/rite-sources/src/uptime_kuma.rs diff --git a/Cargo.lock b/Cargo.lock index d3816d2..69b848e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1111,6 +1111,7 @@ version = "0.1.0" dependencies = [ "async-trait", "axum", + "base64", "chrono", "hex", "hmac", @@ -1133,6 +1134,7 @@ name = "rite-sources" version = "0.1.0" dependencies = [ "async-trait", + "base64", "chrono", "futures-util", "hex", diff --git a/Cargo.toml b/Cargo.toml index 642853b..612ec24 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,7 @@ authors = ["TechGodHQ"] [workspace.dependencies] anyhow = "1" async-trait = "0.1" +base64 = "0.22" axum = "0.8" chrono = { version = "0.4", features = ["serde"] } clap = { version = "4", features = ["derive", "env"] } diff --git a/README.md b/README.md index a3761ce..347b4f9 100644 --- a/README.md +++ b/README.md @@ -5,8 +5,9 @@ Rite is a self-hostable, minimal event-to-action runtime. It receives authenticated events, normalizes them, matches TOML-configured handlers, and forwards matching events to actions. ```text -GitHub webhook ──> Rite ──> configured HTTP action -Iris event ──> Rite ──> configured HTTP action +GitHub webhook ──> Rite ──> configured HTTP action +Uptime Kuma webhook ──> Rite ──> configured HTTP action +Iris event ──> Rite ──> configured HTTP action ``` ## Quick start @@ -23,6 +24,11 @@ action = { type = "http_post", url = "https://example.test/hooks/pr" } [sources.iris] enabled = true base_url = "http://127.0.0.1:3000" + +# Required when this ingress source is enabled. +[sources.uptime_kuma] +enabled = true +secret = "change-me" ``` Run it: @@ -36,6 +42,7 @@ Endpoints: - `GET /health` — returns `ok` - `GET /sources` — configured source adapters - `POST /event/github` — authenticated GitHub webhook ingress using `X-Hub-Signature-256` +- `POST /event/uptime_kuma` — authenticated Uptime Kuma webhook ingress using base64 `Signature` HMAC-SHA256 ## CLI and MCP @@ -109,4 +116,4 @@ cargo clippy --all-targets -- -D warnings cargo fmt --all -- --check ``` -Rite currently normalizes GitHub `push` and `pull_request` payloads, matches configured handlers, and executes `http_post` actions with the normalized event as JSON. +Rite normalizes GitHub `push` and `pull_request` payloads plus Uptime Kuma heartbeats, matches configured handlers, and executes `http_post` actions with the normalized event as JSON. Uptime Kuma heartbeats require a configured secret; `status` maps deterministically to `up` (info), `down` (critical), `pending` (warning), or `maintenance` (info). Matching-safe metadata includes `monitor_name`, `monitor_id`, `status`, plus present URL/timing fields; Kuma's `msg` becomes the optional event body. diff --git a/crates/rite-server/Cargo.toml b/crates/rite-server/Cargo.toml index 1b2ab8d..db2a9b6 100644 --- a/crates/rite-server/Cargo.toml +++ b/crates/rite-server/Cargo.toml @@ -20,6 +20,7 @@ tracing.workspace = true url.workspace = true [dev-dependencies] +base64.workspace = true hmac.workspace = true hex.workspace = true sha2.workspace = true diff --git a/crates/rite-server/src/dispatch.rs b/crates/rite-server/src/dispatch.rs index 8fd6c99..56bcf7e 100644 --- a/crates/rite-server/src/dispatch.rs +++ b/crates/rite-server/src/dispatch.rs @@ -8,7 +8,7 @@ //! the same status/message pair. use axum::{Json, http::StatusCode, response::IntoResponse}; -use rite_core::{EventSource, RiteAction}; +use rite_core::RiteAction; use serde_json::{Value, json}; use std::collections::BTreeMap; @@ -154,13 +154,16 @@ pub async fn execute_raw_operation_http( /// 404 unknown source, 401 failed verification, 400 unparseable payload, /// 202 with a JSON ack once handlers have run. async fn receive_event(state: &AppState, input: RawOperationInput) -> axum::response::Response { - if input.path.get("source").map(String::as_str) != Some("github") { + let Some(source_id) = input.path.get("source") else { return (StatusCode::NOT_FOUND, "unknown event source").into_response(); - } - if let Err(error) = state.github.verify(&input.headers, &input.raw_body).await { + }; + let Some(source) = state.sources.get(source_id) else { + return (StatusCode::NOT_FOUND, "unknown event source").into_response(); + }; + if let Err(error) = source.verify(&input.headers, &input.raw_body).await { return (StatusCode::UNAUTHORIZED, error.to_string()).into_response(); } - let event = match state.github.parse(&input.headers, &input.raw_body).await { + let event = match source.parse(&input.headers, &input.raw_body).await { Ok(event) => event, Err(error) => return (StatusCode::BAD_REQUEST, error.to_string()).into_response(), }; @@ -244,7 +247,14 @@ async fn receive_event(state: &AppState, input: RawOperationInput) -> axum::resp #[allow(clippy::unused_async)] async fn list_sources(state: &AppState) -> Result { - let mut sources = vec![json!({"id": "github", "name": "GitHub"})]; + let mut sources = state + .sources + .values() + .map(|source| { + let metadata = source.metadata(); + json!({"id": metadata.id, "name": metadata.name}) + }) + .collect::>(); if state.iris.is_some() { sources.push(json!({"id": "iris", "name": "Iris"})); } diff --git a/crates/rite-server/src/lib.rs b/crates/rite-server/src/lib.rs index 1f2587a..c283c7b 100644 --- a/crates/rite-server/src/lib.rs +++ b/crates/rite-server/src/lib.rs @@ -3,6 +3,7 @@ pub mod dispatch; use std::{ + collections::BTreeMap, sync::{ Arc, atomic::{AtomicU64, Ordering}, @@ -11,8 +12,8 @@ use std::{ }; use axum::{Json, Router, routing::get}; -use rite_core::{RiteAction, RiteHandler}; -use rite_sources::{github::GitHubSource, iris::IrisSource}; +use rite_core::{EventSource, RiteAction, RiteHandler}; +use rite_sources::{github::GitHubSource, iris::IrisSource, uptime_kuma::UptimeKumaSource}; use serde::Deserialize; /// Severity emitted while checking a loaded configuration before the server starts. @@ -47,6 +48,16 @@ pub struct RiteConfig { pub struct SourcesConfig { /// Optional Iris SSE subscription source. pub iris: Option, + /// Optional authenticated Uptime Kuma webhook source. + pub uptime_kuma: Option, +} + +/// Uptime Kuma webhook configuration. +#[derive(Debug, Clone, Deserialize)] +pub struct UptimeKumaConfig { + #[serde(default)] + pub enabled: bool, + pub secret: String, } /// Iris subscription configuration. @@ -61,7 +72,8 @@ pub struct IrisConfig { #[derive(Clone)] pub struct AppState { pub handlers: Arc>, - pub github: Arc, + /// Ingress sources keyed by their stable source ID. + pub sources: Arc>>, pub client: reqwest::Client, pub iris: Option, pub metrics: Arc, @@ -113,10 +125,14 @@ pub fn load_config(input: &str) -> Result { #[must_use] pub fn validate(config: &RiteConfig) -> Vec { let mut diagnostics = Vec::new(); - let configured_sources = [Some("github"), config.sources.iris.as_ref().map(|_| "iris")] - .into_iter() - .flatten() - .collect::>(); + let configured_sources = [ + Some("github"), + config.sources.iris.as_ref().map(|_| "iris"), + config.sources.uptime_kuma.as_ref().map(|_| "uptime_kuma"), + ] + .into_iter() + .flatten() + .collect::>(); if config.rites.is_empty() { diagnostics.push(Diagnostic { @@ -134,6 +150,16 @@ pub fn validate(config: &RiteConfig) -> Vec { }); } + if let Some(kuma) = &config.sources.uptime_kuma + && kuma.enabled + && kuma.secret.trim().is_empty() + { + diagnostics.push(Diagnostic { + level: DiagnosticLevel::Error, + message: "enabled source 'uptime_kuma' has an empty secret".into(), + }); + } + let mut names = std::collections::BTreeSet::new(); for handler in &config.rites { if !configured_sources.contains(handler.source.as_str()) { @@ -158,14 +184,23 @@ pub fn validate(config: &RiteConfig) -> Vec { /// A secret-free configuration inventory for startup logs. #[must_use] pub fn startup_summary(config: &RiteConfig) -> String { - let source_count = 1 + usize::from(config.sources.iris.is_some()); - let enabled_count = 1 + usize::from( - config - .sources - .iris - .as_ref() - .is_some_and(|source| source.enabled), - ); + let source_count = 1 + + usize::from(config.sources.iris.is_some()) + + usize::from(config.sources.uptime_kuma.is_some()); + let enabled_count = + 1 + usize::from( + config + .sources + .iris + .as_ref() + .is_some_and(|source| source.enabled), + ) + usize::from( + config + .sources + .uptime_kuma + .as_ref() + .is_some_and(|source| source.enabled), + ); format!( "rite: {source_count} sources ({enabled_count} enabled), {} handlers loaded", config.rites.len() @@ -197,9 +232,17 @@ pub fn configured_state(secret: &str, config: RiteConfig) -> rite_core::Result> = BTreeMap::new(); + sources.insert("github".into(), Arc::new(GitHubSource::new(secret)?)); + if let Some(kuma) = config.sources.uptime_kuma.filter(|config| config.enabled) { + sources.insert( + "uptime_kuma".into(), + Arc::new(UptimeKumaSource::new(kuma.secret)?), + ); + } Ok(AppState { handlers: Arc::new(config.rites), - github: Arc::new(GitHubSource::new(secret)?), + sources: Arc::new(sources), client: reqwest::Client::new(), iris, metrics: Arc::new(Metrics::default()), @@ -286,6 +329,14 @@ mod tests { format!("sha256={}", hex::encode(mac.finalize().into_bytes())) } + fn kuma_signature(secret: &str, body: &[u8]) -> String { + use base64::{Engine as _, engine::general_purpose::STANDARD}; + + let mut mac = Hmac::::new_from_slice(secret.as_bytes()).expect("valid key"); + mac.update(body); + STANDARD.encode(mac.finalize().into_bytes()) + } + #[test] fn validation_reports_invalid_configuration() { let config = load_config( @@ -384,6 +435,44 @@ mod tests { assert_eq!(status["actions_failed"], 0); } + #[tokio::test] + async fn uptime_kuma_ingress_requires_signature_and_acks_valid_heartbeats() { + let config = + load_config("[sources.uptime_kuma]\nenabled = true\nsecret = \"kuma-secret\"\n") + .expect("config parses"); + let app = app(configured_state("github-secret", config).expect("valid state")); + let body = br#"{"monitor":{"id":7,"name":"API"},"heartbeat":{"status":0,"msg":"connection refused"}}"#; + let response = app + .clone() + .oneshot( + Request::post("/event/uptime_kuma") + .header("Signature", kuma_signature("kuma-secret", body)) + .body(Body::from(body.as_slice())) + .expect("request"), + ) + .await + .expect("response"); + assert_eq!(response.status(), StatusCode::ACCEPTED); + let bytes = to_bytes(response.into_body(), usize::MAX) + .await + .expect("body"); + assert!( + std::str::from_utf8(&bytes) + .expect("utf8") + .contains("uptime_kuma") + ); + + let rejected = app + .oneshot( + Request::post("/event/uptime_kuma") + .body(Body::from(body.as_slice())) + .expect("request"), + ) + .await + .expect("response"); + assert_eq!(rejected.status(), StatusCode::UNAUTHORIZED); + } + #[tokio::test] async fn invalid_signature_is_unauthorized() { let state = configured_state("secret", RiteConfig::default()).expect("valid state"); diff --git a/crates/rite-sources/Cargo.toml b/crates/rite-sources/Cargo.toml index f7c4d28..a01f809 100644 --- a/crates/rite-sources/Cargo.toml +++ b/crates/rite-sources/Cargo.toml @@ -6,6 +6,7 @@ license.workspace = true [dependencies] async-trait.workspace = true +base64.workspace = true chrono.workspace = true futures-util.workspace = true hex.workspace = true diff --git a/crates/rite-sources/src/lib.rs b/crates/rite-sources/src/lib.rs index 18d8f70..8a00187 100644 --- a/crates/rite-sources/src/lib.rs +++ b/crates/rite-sources/src/lib.rs @@ -2,3 +2,4 @@ pub mod github; pub mod iris; +pub mod uptime_kuma; diff --git a/crates/rite-sources/src/uptime_kuma.rs b/crates/rite-sources/src/uptime_kuma.rs new file mode 100644 index 0000000..4d961be --- /dev/null +++ b/crates/rite-sources/src/uptime_kuma.rs @@ -0,0 +1,196 @@ +//! Uptime Kuma webhook verification and normalization. + +use std::collections::BTreeMap; + +use async_trait::async_trait; +use base64::{Engine as _, engine::general_purpose::STANDARD}; +use chrono::Utc; +use hmac::{Hmac, Mac}; +use rite_core::{EventSource, Result, RiteError, RiteEvent, Severity, SourceMetadata}; +use serde_json::{Value, json}; +use sha2::Sha256; + +const METADATA: SourceMetadata = SourceMetadata { + id: "uptime_kuma", + name: "Uptime Kuma", + capabilities: &["webhook", "hmac-sha256", "heartbeat"], +}; + +/// Uptime Kuma webhook source authenticated with its required signature secret. +#[derive(Debug, Clone)] +pub struct UptimeKumaSource { + secret: Vec, +} + +impl UptimeKumaSource { + /// Construct an Uptime Kuma source from its configured webhook secret. + pub fn new(secret: impl AsRef) -> Result { + let secret = secret.as_ref().trim(); + if secret.is_empty() { + return Err(RiteError::Config( + "uptime_kuma webhook secret is required".into(), + )); + } + Ok(Self { + secret: secret.as_bytes().to_vec(), + }) + } + + fn header<'a>(headers: &'a [(String, String)], name: &str) -> Option<&'a str> { + headers + .iter() + .find_map(|(key, value)| key.eq_ignore_ascii_case(name).then_some(value.as_str())) + } + + fn action_and_severity(status: i64) -> Result<(&'static str, Severity)> { + match status { + 0 => Ok(("down", Severity::Critical)), + 1 => Ok(("up", Severity::Info)), + 2 => Ok(("pending", Severity::Warning)), + 3 => Ok(("maintenance", Severity::Info)), + _ => Err(RiteError::Parse(format!( + "Uptime Kuma heartbeat has unknown status: {status}" + ))), + } + } +} + +#[async_trait] +impl EventSource for UptimeKumaSource { + fn metadata(&self) -> &SourceMetadata { + &METADATA + } + + async fn verify(&self, headers: &[(String, String)], body: &[u8]) -> Result<()> { + let signature = Self::header(headers, "signature") + .ok_or_else(|| RiteError::Authentication("missing Signature".into()))?; + let provided = STANDARD + .decode(signature) + .map_err(|_| RiteError::Authentication("Signature is not base64".into()))?; + let mut mac = Hmac::::new_from_slice(&self.secret) + .map_err(|_| RiteError::Authentication("invalid webhook secret".into()))?; + mac.update(body); + mac.verify_slice(&provided) + .map_err(|_| RiteError::Authentication("Uptime Kuma signature mismatch".into())) + } + + async fn parse(&self, _headers: &[(String, String)], body: &[u8]) -> Result { + let payload: Value = + serde_json::from_slice(body).map_err(|error| RiteError::Parse(error.to_string()))?; + let monitor = payload + .get("monitor") + .and_then(Value::as_object) + .ok_or_else(|| RiteError::Parse("Uptime Kuma payload missing monitor".into()))?; + let heartbeat = payload + .get("heartbeat") + .and_then(Value::as_object) + .ok_or_else(|| RiteError::Parse("Uptime Kuma payload missing heartbeat".into()))?; + let monitor_name = monitor + .get("name") + .and_then(Value::as_str) + .filter(|name| !name.trim().is_empty()) + .ok_or_else(|| RiteError::Parse("Uptime Kuma payload missing monitor.name".into()))?; + let status = heartbeat + .get("status") + .and_then(Value::as_i64) + .ok_or_else(|| { + RiteError::Parse("Uptime Kuma payload missing heartbeat.status".into()) + })?; + let (action, severity) = Self::action_and_severity(status)?; + let mut metadata = BTreeMap::from([ + ("monitor_name".into(), json!(monitor_name)), + ("status".into(), json!(action)), + ]); + for (payload, metadata_key) in [ + (monitor.get("id"), "monitor_id"), + (monitor.get("url"), "monitor_url"), + (heartbeat.get("ping"), "ping_ms"), + (heartbeat.get("time"), "heartbeat_time"), + ] { + if let Some(value) = payload.filter(|value| !value.is_null()) { + metadata.insert(metadata_key.into(), value.clone()); + } + } + if let Some(duration_seconds) = heartbeat.get("duration").and_then(Value::as_f64) { + metadata.insert("duration_ms".into(), json!(duration_seconds * 1_000.0)); + } + Ok(RiteEvent { + source: "uptime_kuma".into(), + event_type: "heartbeat".into(), + action: Some(action.into()), + timestamp: Utc::now(), + severity, + title: monitor_name.into(), + body: heartbeat + .get("msg") + .and_then(Value::as_str) + .filter(|message| !message.is_empty()) + .map(ToOwned::to_owned), + metadata, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn headers(secret: &str, body: &[u8]) -> Vec<(String, String)> { + let mut mac = Hmac::::new_from_slice(secret.as_bytes()).expect("valid key"); + mac.update(body); + vec![( + "Signature".into(), + STANDARD.encode(mac.finalize().into_bytes()), + )] + } + + #[tokio::test] + async fn verifies_and_normalizes_heartbeat() { + let source = UptimeKumaSource::new("secret").expect("source"); + let body = br#"{"monitor":{"id":4,"name":"API","url":"https://api.test"},"heartbeat":{"status":0,"msg":"connection refused","ping":12,"duration":5,"time":"2026-09-06 12:00:00"}}"#; + let headers = headers("secret", body); + source + .verify(&headers, body) + .await + .expect("signature accepted"); + let event = source.parse(&headers, body).await.expect("payload parses"); + assert_eq!(event.action.as_deref(), Some("down")); + assert_eq!(event.severity, Severity::Critical); + assert_eq!(event.metadata["monitor_name"], "API"); + assert_eq!(event.metadata["status"], "down"); + assert_eq!(event.metadata["ping_ms"], 12); + assert_eq!(event.metadata["duration_ms"], 5_000.0); + } + + #[tokio::test] + async fn rejects_bad_or_missing_signature() { + let source = UptimeKumaSource::new("secret").expect("source"); + assert!( + source + .verify(&headers("wrong", b"{}"), b"{}") + .await + .is_err() + ); + assert!(source.verify(&[], b"{}").await.is_err()); + } + + #[tokio::test] + async fn rejects_malformed_or_unknown_heartbeat() { + let source = UptimeKumaSource::new("secret").expect("source"); + assert!(source.parse(&[], b"not json").await.is_err()); + assert!( + source + .parse( + &[], + br#"{"monitor":{"name":"API"},"heartbeat":{"status":99}}"# + ) + .await + .is_err() + ); + } + + #[test] + fn requires_a_secret() { + assert!(UptimeKumaSource::new(" ").is_err()); + } +} diff --git a/rite.example.toml b/rite.example.toml index da9ee57..34e3007 100644 --- a/rite.example.toml +++ b/rite.example.toml @@ -8,6 +8,11 @@ action = { type = "http_post", url = "https://example.test/hooks/pr" } enabled = true base_url = "http://127.0.0.1:3000" +# Uptime Kuma sends base64 HMAC-SHA256 in its Signature header. A secret is required. +[sources.uptime_kuma] +enabled = true +secret = "change-me" + [[rites]] name = "discord-urgent-alert" source = "iris"