diff --git a/src/agent/routine.rs b/src/agent/routine.rs index 296c1ff0..1b8ca96a 100644 --- a/src/agent/routine.rs +++ b/src/agent/routine.rs @@ -79,6 +79,13 @@ pub enum Trigger { #[serde(default)] filters: std::collections::HashMap, }, + /// Fire on incoming webhook POST to /api/webhooks/{path}. + Webhook { + /// Optional webhook path suffix (defaults to routine id). + path: Option, + /// Optional shared secret for HMAC validation. + secret: Option, + }, /// Only fires via tool call or CLI. Manual, } @@ -90,6 +97,7 @@ impl Trigger { Trigger::Cron { .. } => "cron", Trigger::Event { .. } => "event", Trigger::SystemEvent { .. } => "system_event", + Trigger::Webhook { .. } => "webhook", Trigger::Manual => "manual", } } @@ -171,6 +179,17 @@ impl Trigger { filters, }) } + "webhook" => { + let path = config + .get("path") + .and_then(|v| v.as_str()) + .map(String::from); + let secret = config + .get("secret") + .and_then(|v| v.as_str()) + .map(String::from); + Ok(Trigger::Webhook { path, secret }) + } "manual" => Ok(Trigger::Manual), other => Err(RoutineError::UnknownTriggerType { trigger_type: other.to_string(), @@ -198,6 +217,10 @@ impl Trigger { "event_type": event_type, "filters": filters, }), + Trigger::Webhook { path, secret } => serde_json::json!({ + "path": path, + "secret": secret, + }), Trigger::Manual => serde_json::json!({}), } } @@ -962,6 +985,14 @@ mod tests { .type_tag(), "system_event" ); + assert_eq!( + Trigger::Webhook { + path: None, + secret: None, + } + .type_tag(), + "webhook" + ); assert_eq!(Trigger::Manual.type_tag(), "manual"); } diff --git a/src/agent/routine_engine.rs b/src/agent/routine_engine.rs index 7cfdba20..16671239 100644 --- a/src/agent/routine_engine.rs +++ b/src/agent/routine_engine.rs @@ -720,6 +720,90 @@ impl RoutineEngine { Ok(run_id) } + /// Fire a routine from a webhook trigger. + /// + /// Similar to `fire_manual` but records the trigger as `"webhook"` with the + /// webhook path as detail. Skips ownership check (auth is via webhook secret). + /// Enforces enabled check, cooldown, and concurrent run limit. + pub async fn fire_webhook( + &self, + routine_id: Uuid, + webhook_path: &str, + ) -> Result { + let routine = self + .store + .get_routine(routine_id) + .await + .map_err(|e| RoutineError::Database { + reason: e.to_string(), + })? + .ok_or(RoutineError::NotFound { id: routine_id })?; + + if !routine.enabled { + return Err(RoutineError::Disabled { + name: routine.name.clone(), + }); + } + + if !self.check_cooldown(&routine) { + return Err(RoutineError::Cooldown { + name: routine.name.clone(), + }); + } + + if !self.check_concurrent(&routine).await { + return Err(RoutineError::MaxConcurrent { + name: routine.name.clone(), + }); + } + + if self.running_count.load(Ordering::Relaxed) >= self.config.max_concurrent_routines { + return Err(RoutineError::MaxConcurrent { + name: routine.name.clone(), + }); + } + + let run_id = Uuid::new_v4(); + let run = RoutineRun { + id: run_id, + routine_id: routine.id, + trigger_type: "webhook".to_string(), + trigger_detail: Some(webhook_path.to_string()), + started_at: Utc::now(), + completed_at: None, + status: RunStatus::Running, + result_summary: None, + tokens_used: None, + job_id: None, + created_at: Utc::now(), + }; + + if let Err(e) = self.store.create_routine_run(&run).await { + return Err(RoutineError::Database { + reason: format!("failed to create run record: {e}"), + }); + } + + let engine = EngineContext { + config: self.config.clone(), + store: self.store.clone(), + llm: self.llm.clone(), + workspace: self.workspace.clone(), + notify_tx: self.notify_tx.clone(), + running_count: self.running_count.clone(), + scheduler: self.scheduler.clone(), + tools: self.tools.clone(), + safety: self.safety.clone(), + sandbox_readiness: self.sandbox_readiness, + }; + + tokio::spawn(async move { + execute_routine(engine, routine, run).await; + }); + + Ok(run_id) + } + /// Spawn a fire in a background task. fn spawn_fire(&self, routine: Routine, trigger_type: &str, trigger_detail: Option) { let run = RoutineRun { diff --git a/src/channels/web/handlers/mod.rs b/src/channels/web/handlers/mod.rs index 0573a067..2f942058 100644 --- a/src/channels/web/handlers/mod.rs +++ b/src/channels/web/handlers/mod.rs @@ -26,3 +26,4 @@ pub mod routines; pub mod settings; #[allow(dead_code)] pub mod static_files; +pub mod webhooks; diff --git a/src/channels/web/handlers/routines.rs b/src/channels/web/handlers/routines.rs index 41bfee5a..368a28ae 100644 --- a/src/channels/web/handlers/routines.rs +++ b/src/channels/web/handlers/routines.rs @@ -303,7 +303,9 @@ fn routine_error_status(err: &RoutineError) -> StatusCode { match err { RoutineError::NotFound { .. } => StatusCode::NOT_FOUND, RoutineError::NotAuthorized { .. } => StatusCode::FORBIDDEN, - RoutineError::Disabled { .. } | RoutineError::MaxConcurrent { .. } => StatusCode::CONFLICT, + RoutineError::Disabled { .. } + | RoutineError::Cooldown { .. } + | RoutineError::MaxConcurrent { .. } => StatusCode::CONFLICT, _ => StatusCode::INTERNAL_SERVER_ERROR, } } diff --git a/src/channels/web/handlers/webhooks.rs b/src/channels/web/handlers/webhooks.rs new file mode 100644 index 00000000..7b041a06 --- /dev/null +++ b/src/channels/web/handlers/webhooks.rs @@ -0,0 +1,197 @@ +//! Public webhook trigger endpoint for routine webhook triggers. +//! +//! `POST /api/webhooks/{path}` — matches the path against routines with +//! `Trigger::Webhook { path, secret }`, validates the secret via constant-time +//! comparison, and fires the matching routine through the `RoutineEngine`. + +use std::sync::Arc; + +use axum::{ + Json, + extract::{Path, State}, + http::{HeaderMap, StatusCode}, +}; +use subtle::ConstantTimeEq; + +use crate::agent::routine::Trigger; +use crate::channels::web::server::GatewayState; + +/// Validate the webhook secret for a routine. +/// +/// Returns `Ok(())` if the routine has a configured secret and the provided +/// secret matches via constant-time comparison. Returns an appropriate HTTP +/// error if the secret is missing (403) or invalid (401). +fn validate_webhook_secret( + trigger: &Trigger, + provided_secret: &str, +) -> Result<(), (StatusCode, String)> { + // Require webhook secret — routines without a secret cannot be triggered via webhook + let expected_secret = match trigger { + Trigger::Webhook { + secret: Some(s), .. + } => s, + _ => { + return Err(( + StatusCode::FORBIDDEN, + "Webhook secret not configured for this routine. \ + Set a secret with: ironclaw routine update --webhook-secret " + .to_string(), + )); + } + }; + + if !bool::from(provided_secret.as_bytes().ct_eq(expected_secret.as_bytes())) { + return Err(( + StatusCode::UNAUTHORIZED, + "Invalid webhook secret".to_string(), + )); + } + + Ok(()) +} + +/// Handle incoming webhook POST to `/api/webhooks/{path}`. +/// +/// This endpoint is **public** (no gateway auth token required) but protected +/// by the per-routine webhook secret sent via the `X-Webhook-Secret` header. +pub async fn webhook_trigger_handler( + State(state): State>, + Path(path): Path, + headers: HeaderMap, +) -> Result, (StatusCode, String)> { + // Rate limit check + if !state.webhook_rate_limiter.check() { + return Err(( + StatusCode::TOO_MANY_REQUESTS, + "Rate limit exceeded. Try again shortly.".to_string(), + )); + } + + let store = state.store.as_ref().ok_or(( + StatusCode::SERVICE_UNAVAILABLE, + "Database not available".to_string(), + ))?; + + // Targeted query instead of loading all routines + let routine = store + .get_webhook_routine_by_path(&path) + .await + .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))? + .ok_or(( + StatusCode::NOT_FOUND, + "No routine matches this webhook path".to_string(), + ))?; + + let provided_secret = headers + .get("x-webhook-secret") + .and_then(|v| v.to_str().ok()) + .unwrap_or(""); + + validate_webhook_secret(&routine.trigger, provided_secret)?; + + // Fire through the RoutineEngine so guardrails, run tracking, + // notifications, and FullJob dispatch all work correctly. + let engine = { + let guard = state.routine_engine.read().await; + guard.as_ref().cloned().ok_or(( + StatusCode::SERVICE_UNAVAILABLE, + "Routine engine not available".to_string(), + ))? + }; + + let run_id = engine.fire_webhook(routine.id, &path).await.map_err(|e| { + let status = match &e { + crate::error::RoutineError::NotFound { .. } => StatusCode::NOT_FOUND, + crate::error::RoutineError::Disabled { .. } + | crate::error::RoutineError::Cooldown { .. } + | crate::error::RoutineError::MaxConcurrent { .. } => StatusCode::CONFLICT, + _ => StatusCode::INTERNAL_SERVER_ERROR, + }; + (status, e.to_string()) + })?; + + Ok(Json(serde_json::json!({ + "status": "triggered", + "routine_id": routine.id, + "routine_name": routine.name, + "run_id": run_id, + }))) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Routines with `secret: None` must be rejected with 403. + #[test] + fn test_validate_rejects_missing_secret() { + let trigger = Trigger::Webhook { + path: Some("my-hook".to_string()), + secret: None, + }; + let result = validate_webhook_secret(&trigger, "any-secret"); + let (status, msg) = result.unwrap_err(); + assert_eq!(status, StatusCode::FORBIDDEN); + assert!( + msg.contains("not configured"), + "Error should tell user to configure a secret, got: {msg}" + ); + } + + /// Non-webhook triggers must be rejected with 403. + #[test] + fn test_validate_rejects_non_webhook_trigger() { + let trigger = Trigger::Manual; + let result = validate_webhook_secret(&trigger, "any-secret"); + let (status, _) = result.unwrap_err(); + assert_eq!(status, StatusCode::FORBIDDEN); + } + + /// Correct secret passes validation. + #[test] + fn test_validate_accepts_correct_secret() { + let trigger = Trigger::Webhook { + path: Some("my-hook".to_string()), + secret: Some("s3cret-token".to_string()), + }; + assert!(validate_webhook_secret(&trigger, "s3cret-token").is_ok()); + } + + /// Wrong secret returns 401. + #[test] + fn test_validate_rejects_wrong_secret() { + let trigger = Trigger::Webhook { + path: Some("my-hook".to_string()), + secret: Some("correct-secret".to_string()), + }; + let result = validate_webhook_secret(&trigger, "wrong-secret"); + let (status, msg) = result.unwrap_err(); + assert_eq!(status, StatusCode::UNAUTHORIZED); + assert!(msg.contains("Invalid"), "Expected 'Invalid' in: {msg}"); + } + + /// Empty provided secret returns 401 (not a false positive). + #[test] + fn test_validate_rejects_empty_provided_secret() { + let trigger = Trigger::Webhook { + path: Some("my-hook".to_string()), + secret: Some("real-secret".to_string()), + }; + let result = validate_webhook_secret(&trigger, ""); + let (status, _) = result.unwrap_err(); + assert_eq!(status, StatusCode::UNAUTHORIZED); + } + + /// Constant-time comparison: secrets of different lengths are still rejected + /// (not short-circuited in a way that leaks length info). + #[test] + fn test_validate_rejects_different_length_secret() { + let trigger = Trigger::Webhook { + path: None, + secret: Some("short".to_string()), + }; + let result = validate_webhook_secret(&trigger, "a-much-longer-secret-value"); + let (status, _) = result.unwrap_err(); + assert_eq!(status, StatusCode::UNAUTHORIZED); + } +} diff --git a/src/channels/web/mod.rs b/src/channels/web/mod.rs index bfefc5c4..1fdb4455 100644 --- a/src/channels/web/mod.rs +++ b/src/channels/web/mod.rs @@ -98,6 +98,7 @@ impl GatewayChannel { skill_catalog: None, chat_rate_limiter: server::RateLimiter::new(30, 60), oauth_rate_limiter: server::RateLimiter::new(10, 60), + webhook_rate_limiter: server::RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)), @@ -136,6 +137,7 @@ impl GatewayChannel { skill_catalog: self.state.skill_catalog.clone(), chat_rate_limiter: server::RateLimiter::new(30, 60), oauth_rate_limiter: server::RateLimiter::new(10, 60), + webhook_rate_limiter: server::RateLimiter::new(10, 60), registry_entries: self.state.registry_entries.clone(), cost_guard: self.state.cost_guard.clone(), routine_engine: Arc::clone(&self.state.routine_engine), diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index 169bb0bf..63eafeab 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -190,6 +190,8 @@ pub struct GatewayState { pub chat_rate_limiter: RateLimiter, /// Rate limiter for OAuth callback endpoints (10 requests per 60 seconds). pub oauth_rate_limiter: RateLimiter, + /// Rate limiter for webhook trigger endpoints (10 requests per 60 seconds). + pub webhook_rate_limiter: RateLimiter, /// Registry catalog entries for the available extensions API. /// Populated at startup from `registry/` manifests, independent of extension manager. pub registry_entries: Vec, @@ -233,7 +235,11 @@ pub async fn start_server( "/oauth/slack/callback", get(slack_relay_oauth_callback_handler), ) - .route("/relay/events", post(relay_events_handler)); + .route("/relay/events", post(relay_events_handler)) + .route( + "/api/webhooks/{path}", + post(crate::channels::web::handlers::webhooks::webhook_trigger_handler), + ); // Protected routes (require auth) let auth_state = AuthState { token: auth_token }; @@ -2834,6 +2840,7 @@ mod tests { scheduler: None, chat_rate_limiter: RateLimiter::new(30, 60), oauth_rate_limiter: RateLimiter::new(10, 60), + webhook_rate_limiter: RateLimiter::new(10, 60), registry_entries: vec![], cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)), diff --git a/src/channels/web/test_helpers.rs b/src/channels/web/test_helpers.rs index 76b2a760..8751be6a 100644 --- a/src/channels/web/test_helpers.rs +++ b/src/channels/web/test_helpers.rs @@ -83,6 +83,7 @@ impl TestGatewayBuilder { scheduler: None, chat_rate_limiter: RateLimiter::new(30, 60), oauth_rate_limiter: RateLimiter::new(10, 60), + webhook_rate_limiter: RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)), diff --git a/src/channels/web/types.rs b/src/channels/web/types.rs index 861b5bd2..107ee05d 100644 --- a/src/channels/web/types.rs +++ b/src/channels/web/types.rs @@ -814,6 +814,14 @@ impl RoutineInfo { String::new(), format!("event: {}.{}", source, event_type), ), + crate::agent::routine::Trigger::Webhook { path, .. } => { + let p = path.as_deref().unwrap_or("default"); + ( + "webhook".to_string(), + String::new(), + format!("webhook: /api/webhooks/{}", p), + ) + } crate::agent::routine::Trigger::Manual => ( "manual".to_string(), String::new(), diff --git a/src/channels/web/ws.rs b/src/channels/web/ws.rs index 8efc69f6..470c3422 100644 --- a/src/channels/web/ws.rs +++ b/src/channels/web/ws.rs @@ -517,6 +517,7 @@ mod tests { skill_catalog: None, chat_rate_limiter: crate::channels::web::server::RateLimiter::new(30, 60), oauth_rate_limiter: crate::channels::web::server::RateLimiter::new(10, 60), + webhook_rate_limiter: crate::channels::web::server::RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)), diff --git a/src/db/libsql/routines.rs b/src/db/libsql/routines.rs index 3151e75b..6702cc1b 100644 --- a/src/db/libsql/routines.rs +++ b/src/db/libsql/routines.rs @@ -477,6 +477,34 @@ impl RoutineStore for LibSqlBackend { Ok(()) } + async fn get_webhook_routine_by_path( + &self, + path: &str, + ) -> Result, DatabaseError> { + let conn = self.connect().await?; + let mut rows = conn + .query( + &format!( + "SELECT {} FROM routines WHERE enabled = 1 AND trigger_type = 'webhook' \ + AND (json_extract(trigger_config, '$.path') = ?1 \ + OR (json_extract(trigger_config, '$.path') IS NULL AND CAST(id AS TEXT) = ?1))", + ROUTINE_COLUMNS + ), + params![path], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + match rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + Some(row) => Ok(Some(row_to_routine_libsql(&row)?)), + None => Ok(None), + } + } + async fn list_dispatched_routine_runs(&self) -> Result, DatabaseError> { let conn = self.connect().await?; let mut rows = conn diff --git a/src/db/mod.rs b/src/db/mod.rs index f1e8c276..d960ebaf 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -525,6 +525,10 @@ pub trait RoutineStore: Send + Sync { run_id: Uuid, job_id: Uuid, ) -> Result<(), DatabaseError>; + async fn get_webhook_routine_by_path( + &self, + path: &str, + ) -> Result, DatabaseError>; /// List routine runs that were dispatched as full_job but have not yet /// been finalized (status='running' with a linked job_id). diff --git a/src/db/postgres.rs b/src/db/postgres.rs index eaa6e049..e77452db 100644 --- a/src/db/postgres.rs +++ b/src/db/postgres.rs @@ -504,6 +504,13 @@ impl RoutineStore for PgBackend { self.store.link_routine_run_to_job(run_id, job_id).await } + async fn get_webhook_routine_by_path( + &self, + path: &str, + ) -> Result, DatabaseError> { + self.store.get_webhook_routine_by_path(path).await + } + async fn list_dispatched_routine_runs(&self) -> Result, DatabaseError> { self.store.list_dispatched_routine_runs().await } diff --git a/src/error.rs b/src/error.rs index 413bc8fd..ec378a80 100644 --- a/src/error.rs +++ b/src/error.rs @@ -376,6 +376,9 @@ pub enum RoutineError { #[error("Not authorized to trigger routine {id}")] NotAuthorized { id: Uuid }, + #[error("Routine {name} is in cooldown period")] + Cooldown { name: String }, + #[error("Routine {name} at max concurrent runs")] MaxConcurrent { name: String }, diff --git a/src/history/store.rs b/src/history/store.rs index 2deffab5..f0b593c2 100644 --- a/src/history/store.rs +++ b/src/history/store.rs @@ -1105,6 +1105,22 @@ impl Store { rows.iter().map(row_to_routine).collect() } + /// Find an enabled webhook routine by its configured path (or fallback to ID). + pub async fn get_webhook_routine_by_path( + &self, + path: &str, + ) -> Result, DatabaseError> { + let conn = self.conn().await?; + let row = conn + .query_opt( + "SELECT * FROM routines WHERE enabled AND trigger_type = 'webhook' \ + AND (trigger_config->>'path' = $1 OR (trigger_config->>'path' IS NULL AND id::text = $1))", + &[&path], + ) + .await?; + row.as_ref().map(row_to_routine).transpose() + } + /// List all enabled cron routines whose next_fire_at <= now. pub async fn list_due_cron_routines(&self) -> Result, DatabaseError> { let conn = self.conn().await?; diff --git a/src/tools/builtin/routine.rs b/src/tools/builtin/routine.rs index b37932ff..c197fe25 100644 --- a/src/tools/builtin/routine.rs +++ b/src/tools/builtin/routine.rs @@ -47,6 +47,10 @@ enum NormalizedTriggerRequest { event_type: String, filters: HashMap, }, + Webhook { + path: Option, + secret: Option, + }, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -827,6 +831,11 @@ fn parse_routine_trigger(params: &Value) -> Result { + let path = string_field(params, "request", "path", &["webhook_path"]); + let secret = string_field(params, "request", "secret", &["webhook_secret"]); + Ok(NormalizedTriggerRequest::Webhook { path, secret }) + } other => Err(ToolError::InvalidParameters(format!( "unknown request.kind: {other}" ))), @@ -915,6 +924,10 @@ fn build_routine_trigger(trigger: &NormalizedTriggerRequest) -> Trigger { event_type: event_type.clone(), filters: filters.clone(), }, + NormalizedTriggerRequest::Webhook { path, secret } => Trigger::Webhook { + path: path.clone(), + secret: secret.clone(), + }, } } diff --git a/tests/openai_compat_integration.rs b/tests/openai_compat_integration.rs index a1bc6a64..2a472d00 100644 --- a/tests/openai_compat_integration.rs +++ b/tests/openai_compat_integration.rs @@ -210,6 +210,7 @@ async fn start_test_server_with_provider( skill_catalog: None, chat_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(30, 60), oauth_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60), + webhook_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)), @@ -702,6 +703,7 @@ async fn test_no_llm_provider_returns_503() { skill_catalog: None, chat_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(30, 60), oauth_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60), + webhook_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)), diff --git a/tests/support/gateway_workflow_harness.rs b/tests/support/gateway_workflow_harness.rs index f5f01266..d33c6fe0 100644 --- a/tests/support/gateway_workflow_harness.rs +++ b/tests/support/gateway_workflow_harness.rs @@ -230,6 +230,7 @@ impl GatewayWorkflowHarness { skill_catalog: components.skill_catalog.clone(), chat_rate_limiter: RateLimiter::new(120, 60), oauth_rate_limiter: RateLimiter::new(10, 60), + webhook_rate_limiter: RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: Some(Arc::clone(&components.cost_guard)), routine_engine: Arc::clone(&routine_slot), diff --git a/tests/ws_gateway_integration.rs b/tests/ws_gateway_integration.rs index 6702d4ff..556c5dcc 100644 --- a/tests/ws_gateway_integration.rs +++ b/tests/ws_gateway_integration.rs @@ -58,6 +58,7 @@ async fn start_test_server() -> ( skill_catalog: None, chat_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(30, 60), oauth_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60), + webhook_rate_limiter: ironclaw::channels::web::server::RateLimiter::new(10, 60), registry_entries: Vec::new(), cost_guard: None, routine_engine: Arc::new(tokio::sync::RwLock::new(None)),