From 86afea7cbb658232fa5997ceb0c4ff36e77b3f82 Mon Sep 17 00:00:00 2001 From: Bill Raj-Derouin Date: Wed, 14 Oct 2026 17:12:44 -0700 Subject: [PATCH 1/3] rust(feat): add list_webhooks and webhook_ids on create_rule and update_rule Agents could not see or attach webhooks, so a request to alert someone when a rule fires lost its notification step. - New list_webhooks tool over ListWebhooks. The backend returns credentials in plaintext, so the tool keeps only the target URL's scheme and the last two labels of its host, header names, and no body template. - create_rule and update_rule take webhook_ids. On update_rule it replaces the rule's set, like asset_ids; omitted, the current webhooks are kept, even through a conditions_json rewrite, and next_step names any webhook detached. - Both tools confirm each webhook exists and is not archived before writing, since Sift accepts an archived webhook on a rule and never calls it. Webhook actions written into the conditions are refused so none skips that check. - create_rule documents the annotation action shape, and both tools check for exactly one condition. Sift answers a rule with none with an internal error. --- rust/crates/sift_cli/CHANGELOG.md | 13 + .../sift_cli/assets/skills/sift/SKILL.md | 8 +- rust/crates/sift_mcp/src/server/mod.rs | 7 +- rust/crates/sift_mcp/src/service/mod.rs | 1 + rust/crates/sift_mcp/src/service/rules/mod.rs | 114 +++++- .../crates/sift_mcp/src/service/rules/test.rs | 285 ++++++++++++- .../sift_mcp/src/service/webhooks/mod.rs | 165 ++++++++ .../sift_mcp/src/service/webhooks/test.rs | 285 +++++++++++++ rust/crates/sift_mcp/src/tool/mod.rs | 1 + rust/crates/sift_mcp/src/tool/rules/mod.rs | 209 +++++++++- rust/crates/sift_mcp/src/tool/rules/test.rs | 382 +++++++++++++++++- rust/crates/sift_mcp/src/tool/webhooks/mod.rs | 83 ++++ .../crates/sift_mcp/src/tool/webhooks/test.rs | 97 +++++ rust/crates/sift_mcp/src/tool_events.json | 1 + rust/crates/sift_test_util/src/mock/mod.rs | 1 + .../sift_test_util/src/mock/webhooks/mod.rs | 1 + .../sift_test_util/src/mock/webhooks/v1.rs | 98 +++++ 17 files changed, 1707 insertions(+), 44 deletions(-) create mode 100644 rust/crates/sift_mcp/src/service/webhooks/mod.rs create mode 100644 rust/crates/sift_mcp/src/service/webhooks/test.rs create mode 100644 rust/crates/sift_mcp/src/tool/webhooks/mod.rs create mode 100644 rust/crates/sift_mcp/src/tool/webhooks/test.rs create mode 100644 rust/crates/sift_test_util/src/mock/webhooks/mod.rs create mode 100644 rust/crates/sift_test_util/src/mock/webhooks/v1.rs diff --git a/rust/crates/sift_cli/CHANGELOG.md b/rust/crates/sift_cli/CHANGELOG.md index 4f1ed7ac2c..fbe48ccc66 100644 --- a/rust/crates/sift_cli/CHANGELOG.md +++ b/rust/crates/sift_cli/CHANGELOG.md @@ -7,6 +7,19 @@ This project adheres to [Semantic Versioning](http://semver.org/). ### What's New +- New `list_webhooks` tool lists the webhooks a rule can call. It withholds + credentials: `target_url` keeps only its scheme and the last two labels of + its host, each header keeps only its name, and the request body template is + left out. +- `create_rule` and `update_rule` take `webhook_ids`, so an agent can make a + rule call a webhook when it fires. On `update_rule` it replaces the rule's + set, like `asset_ids`, and the result names any webhook it detached. Both + tools check that each webhook exists and is not archived before saving the + rule, and refuse webhook actions written into the rule's conditions. +- `create_rule` now documents the annotation action and says a rule takes + exactly one condition. Agents no longer need to copy an existing rule to get + the actions right. + ## [v0.7.4] - October 14, 2026 ### What's New diff --git a/rust/crates/sift_cli/assets/skills/sift/SKILL.md b/rust/crates/sift_cli/assets/skills/sift/SKILL.md index d409e2797d..6ed1c71f71 100644 --- a/rust/crates/sift_cli/assets/skills/sift/SKILL.md +++ b/rust/crates/sift_cli/assets/skills/sift/SKILL.md @@ -85,7 +85,7 @@ tool. connectivity check; when it fails, expect every other Sift tool to fail too. - **Discovery:** `list_assets`, `list_runs`, `list_channels`, `list_reports`, `list_report_templates`, `list_rules`, `list_rule_versions`, `list_annotations`, - `list_artifacts`, `list_artifact_versions`. + `list_artifacts`, `list_artifact_versions`, `list_webhooks`. - **Artifacts:** `download_artifact` (latest version, or pin `artifact_version_id`), `list_artifact_versions` for one artifact's history, newest first. The artifact tools, including `list_artifacts` and the artifact writes below, @@ -251,6 +251,12 @@ tool. cannot be attached to a template) — skip the template and pass `rule_ids`, `rule_client_keys`, or `rule_version_ids` directly to `create_report`. Track progress via `list_report_rule_summaries`. +- **Alert outside Sift when a rule fires.** Find the webhook with + `list_webhooks`, then pass its id in `webhook_ids` on `create_rule` or + `update_rule`. On `update_rule` it replaces the rule's set, like `asset_ids`, + so include the webhooks the rule already calls. Do not write webhook actions + into `conditions`. These tools cannot create a webhook; when none fits, tell + the user to create one in the Sift web app. ## Rules that always apply diff --git a/rust/crates/sift_mcp/src/server/mod.rs b/rust/crates/sift_mcp/src/server/mod.rs index 0ecf101b41..3299165532 100644 --- a/rust/crates/sift_mcp/src/server/mod.rs +++ b/rust/crates/sift_mcp/src/server/mod.rs @@ -53,6 +53,7 @@ use crate::service::{ rule_evaluation::RuleEvaluationService, rules::RuleService, runs::RunService, test_reports::TestReportService, url::UrlService, user_defined_functions::UserDefinedFunctionService, users::UserService, + webhooks::WebhookService, }; #[derive(Clone)] @@ -78,6 +79,7 @@ pub struct SiftMcpServer { pub docs_service: DocsService, pub user_defined_function_service: UserDefinedFunctionService, pub user_service: UserService, + pub webhook_service: WebhookService, pub allow_create: bool, pub allow_destructive: bool, @@ -234,6 +236,7 @@ impl SiftMcpServer { tool_router.merge(Self::docs_router()); tool_router.merge(Self::user_defined_functions_router()); tool_router.merge(Self::users_router()); + tool_router.merge(Self::webhooks_router()); tool_router.merge(Self::update_router()); let registered_tools: HashSet = tool_router .list_all() @@ -300,7 +303,8 @@ impl SiftMcpServer { let docs_service = DocsService::new(channel.clone(), retry_policy.clone()); let user_defined_function_service = UserDefinedFunctionService::new(channel.clone(), retry_policy.clone()); - let user_service = UserService::new(channel.clone(), retry_policy); + let user_service = UserService::new(channel.clone(), retry_policy.clone()); + let webhook_service = WebhookService::new(channel.clone(), retry_policy); Ok(Self { annotation_service, @@ -321,6 +325,7 @@ impl SiftMcpServer { docs_service, user_defined_function_service, user_service, + webhook_service, tool_router, prompt_router, allow_create, diff --git a/rust/crates/sift_mcp/src/service/mod.rs b/rust/crates/sift_mcp/src/service/mod.rs index 8ccaa277c6..1ef14d60a8 100644 --- a/rust/crates/sift_mcp/src/service/mod.rs +++ b/rust/crates/sift_mcp/src/service/mod.rs @@ -17,5 +17,6 @@ pub mod test_reports; pub mod url; pub mod user_defined_functions; pub mod users; +pub mod webhooks; pub(crate) mod common; diff --git a/rust/crates/sift_mcp/src/service/rules/mod.rs b/rust/crates/sift_mcp/src/service/rules/mod.rs index e16f963850..ddc16e5151 100644 --- a/rust/crates/sift_mcp/src/service/rules/mod.rs +++ b/rust/crates/sift_mcp/src/service/rules/mod.rs @@ -4,10 +4,11 @@ use anyhow::{Context, Result}; use sift_rs::{ SiftChannel, rules::v1::{ - ArchiveRuleRequest, CreateRuleRequest, GetRuleRequest, ListRuleVersionsRequest, + ActionKind, ArchiveRuleRequest, CreateRuleRequest, GetRuleRequest, ListRuleVersionsRequest, ListRuleVersionsResponse, ListRulesRequest, ListRulesResponse, Rule, RuleAction, - RuleCondition, RuleVersion, UnarchiveRuleRequest, UpdateActionRequest, - UpdateConditionRequest, UpdateRuleRequest, rule_service_client::RuleServiceClient, + RuleActionConfiguration, RuleCondition, RuleVersion, UnarchiveRuleRequest, + UpdateActionRequest, UpdateConditionRequest, UpdateRuleRequest, WebhookActionConfiguration, + rule_action_configuration::Configuration, rule_service_client::RuleServiceClient, }, }; @@ -24,6 +25,17 @@ pub struct RuleUpdate { pub asset_ids: Option>, pub tag_ids: Option>, pub conditions: Option>, + /// Replaces the set of webhooks the rule calls. `None` keeps the current + /// set, including when `conditions` replaces the actions they sat on. + pub webhook_ids: Option>, +} + +/// The outcome of `update_rule`. +#[derive(Debug)] +pub struct UpdatedRule { + pub rule_id: String, + /// Webhooks the rule called before the update and no longer calls. + pub detached_webhook_ids: Vec, } #[cfg(test)] @@ -110,8 +122,15 @@ impl RuleService { }) } - /// Creates a rule from a full rule definition. Returns the new `rule_id`. - pub async fn create_rule(&self, update: UpdateRuleRequest) -> Result { + /// Creates a rule from a full rule definition that calls each of + /// `webhook_ids` when it fires. Returns the new `rule_id`. + pub async fn create_rule( + &self, + mut update: UpdateRuleRequest, + webhook_ids: Vec, + ) -> Result { + set_webhooks(&mut update.conditions, &webhook_ids); + let channel = self.channel.clone(); let resp = with_retry(&self.policy, move || { @@ -136,14 +155,15 @@ impl RuleService { /// Safely updates an existing rule by `rule_id`. The underlying `UpdateRule` /// RPC is a full replace, so this fetches the current rule, overlays only the /// fields set in `changes`, and sends the merged definition back. Fields not - /// named in `changes` are preserved. Returns the `rule_id`. - pub async fn update_rule(&self, rule_id: String, changes: RuleUpdate) -> Result { + /// named in `changes` are preserved. + pub async fn update_rule(&self, rule_id: String, changes: RuleUpdate) -> Result { let current = self .get_rule(rule_id.clone()) .await? .ok_or_else(|| anyhow::anyhow!("rule '{rule_id}' not found"))?; let mut update = rule_to_update_request(current); + let current_webhook_ids = webhook_ids_in(&update.conditions); let RuleUpdate { name, @@ -153,6 +173,7 @@ impl RuleService { asset_ids, tag_ids, conditions, + webhook_ids, } = changes; if let Some(name) = name { @@ -180,6 +201,12 @@ impl RuleService { if let Some(conditions) = conditions { update.conditions = conditions; } + let webhook_ids = webhook_ids.unwrap_or_else(|| current_webhook_ids.clone()); + set_webhooks(&mut update.conditions, &webhook_ids); + let detached_webhook_ids = current_webhook_ids + .into_iter() + .filter(|id| !webhook_ids.contains(id)) + .collect(); let channel = self.channel.clone(); let resp = with_retry(&self.policy, move || { @@ -196,7 +223,10 @@ impl RuleService { .await .context("failed to update rule")?; - Ok(resp.rule_id) + Ok(UpdatedRule { + rule_id: resp.rule_id, + detached_webhook_ids, + }) } /// Retrieves the latest version of a rule by id, or `None` if it does not exist. @@ -384,3 +414,71 @@ fn action_to_update(action: RuleAction) -> UpdateActionRequest { configuration: action.configuration, } } + +/// Makes every condition call exactly `webhook_ids`. Sift fires a webhook from +/// the condition that holds it and allows one condition per rule, so every +/// condition is that one. An action for a kept webhook keeps its +/// `rule_action_id`, since Sift deletes any action an update leaves out, and a +/// repeated id is attached once because Sift rejects the rule otherwise. +/// Non-webhook actions are left alone. +fn set_webhooks(conditions: &mut [UpdateConditionRequest], webhook_ids: &[String]) { + for condition in conditions { + condition.actions.retain(|action| { + webhook_id(action).is_none_or(|id| webhook_ids.iter().any(|w| w == id)) + }); + + for id in webhook_ids { + let attached = condition + .actions + .iter() + .any(|action| webhook_id(action) == Some(id.as_str())); + if !attached { + condition.actions.push(webhook_action(id.clone())); + } + } + } +} + +/// Whether any condition carries a webhook action. The rule tools take +/// webhooks only through `webhook_ids`, so each one is checked before saving. +pub fn has_webhook_action(conditions: &[UpdateConditionRequest]) -> bool { + conditions + .iter() + .flat_map(|condition| &condition.actions) + .any(|action| webhook_id(action).is_some()) +} + +/// The webhooks the conditions call, in order, each once. +fn webhook_ids_in(conditions: &[UpdateConditionRequest]) -> Vec { + let mut ids: Vec = Vec::new(); + for id in conditions + .iter() + .flat_map(|condition| &condition.actions) + .filter_map(webhook_id) + { + if !ids.iter().any(|existing| existing == id) { + ids.push(id.to_string()); + } + } + ids +} + +/// The webhook an action calls, or `None` for any other kind of action. +fn webhook_id(action: &UpdateActionRequest) -> Option<&str> { + match action.configuration.as_ref()?.configuration.as_ref()? { + Configuration::Webhook(webhook) => Some(webhook.webhook_id.as_str()), + _ => None, + } +} + +fn webhook_action(webhook_id: String) -> UpdateActionRequest { + UpdateActionRequest { + rule_action_id: None, + action_type: ActionKind::Webhook as i32, + configuration: Some(RuleActionConfiguration { + configuration: Some(Configuration::Webhook(WebhookActionConfiguration { + webhook_id, + })), + }), + } +} diff --git a/rust/crates/sift_mcp/src/service/rules/test.rs b/rust/crates/sift_mcp/src/service/rules/test.rs index 275e405d23..a1db896f40 100644 --- a/rust/crates/sift_mcp/src/service/rules/test.rs +++ b/rust/crates/sift_mcp/src/service/rules/test.rs @@ -1,7 +1,10 @@ use sift_rs::rules::v1::{ - ArchiveRuleResponse, CreateRuleResponse, GetRuleResponse, ListRuleVersionsResponse, - ListRulesResponse, Rule, RuleAssetConfiguration, RuleVersion, UnarchiveRuleResponse, - UpdateRuleRequest, UpdateRuleResponse, rule_service_server::RuleServiceServer, + ActionKind, AnnotationActionConfiguration, ArchiveRuleResponse, CreateRuleResponse, + GetRuleResponse, ListRuleVersionsResponse, ListRulesResponse, Rule, RuleAction, + RuleActionConfiguration, RuleAssetConfiguration, RuleCondition, RuleVersion, + UnarchiveRuleResponse, UpdateActionRequest, UpdateConditionRequest, UpdateRuleRequest, + UpdateRuleResponse, WebhookActionConfiguration, rule_action_configuration::Configuration, + rule_service_server::RuleServiceServer, }; use super::RuleUpdate; @@ -266,11 +269,14 @@ async fn create_rule_forwards_definition_and_returns_id() { let (service, _h) = service_with_mock(mock).await; let rule_id = service - .create_rule(UpdateRuleRequest { - name: "overtemp".into(), - description: "engine over temperature".into(), - ..Default::default() - }) + .create_rule( + UpdateRuleRequest { + name: "overtemp".into(), + description: "engine over temperature".into(), + ..Default::default() + }, + Vec::new(), + ) .await .expect("create_rule failed"); @@ -320,7 +326,7 @@ async fn update_rule_merges_changes_over_current_definition() { let (service, _h) = service_with_mock(mock).await; - let rule_id = service + let updated = service .update_rule( "rule-1".to_string(), RuleUpdate { @@ -331,7 +337,8 @@ async fn update_rule_merges_changes_over_current_definition() { .await .expect("update_rule failed"); - assert_eq!(rule_id, "rule-1"); + assert_eq!(updated.rule_id, "rule-1"); + assert!(updated.detached_webhook_ids.is_empty()); } #[tokio::test] @@ -422,9 +429,265 @@ async fn create_rule_propagates_grpc_error() { let (service, _h) = service_with_mock(mock).await; let err = service - .create_rule(UpdateRuleRequest::default()) + .create_rule(UpdateRuleRequest::default(), Vec::new()) .await .expect_err("expected error"); assert!(err.to_string().contains("failed to create rule")); } + +fn annotation_config() -> RuleActionConfiguration { + RuleActionConfiguration { + configuration: Some(Configuration::Annotation( + AnnotationActionConfiguration::default(), + )), + } +} + +fn webhook_config(webhook_id: &str) -> RuleActionConfiguration { + RuleActionConfiguration { + configuration: Some(Configuration::Webhook(WebhookActionConfiguration { + webhook_id: webhook_id.into(), + })), + } +} + +/// The webhook ids an action list calls, in order. +fn webhook_ids(actions: &[UpdateActionRequest]) -> Vec<&str> { + actions + .iter() + .filter_map( + |a| match a.configuration.as_ref()?.configuration.as_ref()? { + Configuration::Webhook(w) => Some(w.webhook_id.as_str()), + _ => None, + }, + ) + .collect() +} + +/// A rule with one condition holding an annotation action and a webhook +/// action for `wh-old`, each with its server-assigned id. +fn rule_with_webhook() -> Rule { + Rule { + rule_id: "rule-1".into(), + name: "overtemp".into(), + conditions: vec![RuleCondition { + rule_condition_id: "cond-1".into(), + actions: vec![ + RuleAction { + rule_action_id: "act-annotation".into(), + action_type: ActionKind::Annotation as i32, + configuration: Some(annotation_config()), + ..Default::default() + }, + RuleAction { + rule_action_id: "act-wh-old".into(), + action_type: ActionKind::Webhook as i32, + configuration: Some(webhook_config("wh-old")), + ..Default::default() + }, + ], + ..Default::default() + }], + ..Default::default() + } +} + +#[tokio::test] +async fn create_rule_attaches_webhooks_beside_the_annotation() { + let mut mock = MockRuleServiceImpl::new(); + mock.expect_create_rule() + .withf(|req| { + let update = req.get_ref().update.as_ref().unwrap(); + let actions = &update.conditions[0].actions; + actions.len() == 3 + && actions[0].action_type == ActionKind::Annotation as i32 + && webhook_ids(actions) == vec!["wh-1", "wh-2"] + && actions[1..].iter().all(|a| { + a.action_type == ActionKind::Webhook as i32 && a.rule_action_id.is_none() + }) + }) + .returning(|_| { + Ok(Response::new(CreateRuleResponse { + rule_id: "new-rule".into(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let rule_id = service + .create_rule( + UpdateRuleRequest { + name: "overtemp".into(), + conditions: vec![UpdateConditionRequest { + actions: vec![UpdateActionRequest { + action_type: ActionKind::Annotation as i32, + configuration: Some(annotation_config()), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + }, + // A repeated id must not attach twice; Sift rejects the rule if it does. + vec!["wh-1".into(), "wh-2".into(), "wh-1".into()], + ) + .await + .expect("create_rule failed"); + + assert_eq!(rule_id, "new-rule"); +} + +#[tokio::test] +async fn update_rule_keeps_existing_actions_when_webhooks_are_replaced() { + let mut mock = MockRuleServiceImpl::new(); + mock.expect_get_rule().returning(|_| { + Ok(Response::new(GetRuleResponse { + rule: Some(rule_with_webhook()), + })) + }); + + // Sift deletes any action left out of an update, so the existing annotation + // and the kept webhook must be resent with their ids, and `wh-old` must not repeat. + mock.expect_update_rule() + .withf(|req| { + let condition = &req.get_ref().conditions[0]; + let actions = &condition.actions; + condition.rule_condition_id.as_deref() == Some("cond-1") + && actions.len() == 3 + && actions[0].rule_action_id.as_deref() == Some("act-annotation") + && actions[1].rule_action_id.as_deref() == Some("act-wh-old") + && actions[2].rule_action_id.is_none() + && webhook_ids(actions) == vec!["wh-old", "wh-new"] + }) + .returning(|_| { + Ok(Response::new(UpdateRuleResponse { + rule_id: "rule-1".into(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let updated = service + .update_rule( + "rule-1".to_string(), + RuleUpdate { + webhook_ids: Some(vec!["wh-old".into(), "wh-new".into(), "wh-new".into()]), + ..Default::default() + }, + ) + .await + .expect("update_rule failed"); + + assert!(updated.detached_webhook_ids.is_empty()); +} + +#[tokio::test] +async fn update_rule_reports_the_webhooks_it_detaches() { + let mut mock = MockRuleServiceImpl::new(); + mock.expect_get_rule().returning(|_| { + Ok(Response::new(GetRuleResponse { + rule: Some(rule_with_webhook()), + })) + }); + + mock.expect_update_rule() + .withf(|req| { + let actions = &req.get_ref().conditions[0].actions; + actions.len() == 1 + && actions[0].rule_action_id.as_deref() == Some("act-annotation") + && webhook_ids(actions).is_empty() + }) + .returning(|_| { + Ok(Response::new(UpdateRuleResponse { + rule_id: "rule-1".into(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let updated = service + .update_rule( + "rule-1".to_string(), + RuleUpdate { + webhook_ids: Some(Vec::new()), + ..Default::default() + }, + ) + .await + .expect("update_rule failed"); + + assert_eq!(updated.detached_webhook_ids, vec!["wh-old".to_string()]); +} + +#[tokio::test] +async fn update_rule_carries_webhooks_onto_replaced_conditions() { + let mut mock = MockRuleServiceImpl::new(); + mock.expect_get_rule().returning(|_| { + Ok(Response::new(GetRuleResponse { + rule: Some(rule_with_webhook()), + })) + }); + + // The new conditions carry only an annotation. Webhooks are set through + // `webhook_ids`, so rewriting the logic must not silently drop `wh-old`. + mock.expect_update_rule() + .withf(|req| { + let actions = &req.get_ref().conditions[0].actions; + actions.len() == 2 + && actions[0].action_type == ActionKind::Annotation as i32 + && webhook_ids(actions) == vec!["wh-old"] + }) + .returning(|_| { + Ok(Response::new(UpdateRuleResponse { + rule_id: "rule-1".into(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let updated = service + .update_rule( + "rule-1".to_string(), + RuleUpdate { + conditions: Some(vec![UpdateConditionRequest { + actions: vec![UpdateActionRequest { + action_type: ActionKind::Annotation as i32, + configuration: Some(annotation_config()), + ..Default::default() + }], + ..Default::default() + }]), + ..Default::default() + }, + ) + .await + .expect("update_rule failed"); + + assert!(updated.detached_webhook_ids.is_empty()); +} + +#[test] +fn has_webhook_action_finds_only_webhooks() { + let annotation = UpdateConditionRequest { + actions: vec![UpdateActionRequest { + action_type: ActionKind::Annotation as i32, + configuration: Some(annotation_config()), + ..Default::default() + }], + ..Default::default() + }; + let webhook = UpdateConditionRequest { + actions: vec![UpdateActionRequest { + action_type: ActionKind::Webhook as i32, + configuration: Some(webhook_config("wh-1")), + ..Default::default() + }], + ..Default::default() + }; + + assert!(!super::has_webhook_action(std::slice::from_ref( + &annotation + ))); + assert!(super::has_webhook_action(&[annotation, webhook])); +} diff --git a/rust/crates/sift_mcp/src/service/webhooks/mod.rs b/rust/crates/sift_mcp/src/service/webhooks/mod.rs new file mode 100644 index 0000000000..f67e41951c --- /dev/null +++ b/rust/crates/sift_mcp/src/service/webhooks/mod.rs @@ -0,0 +1,165 @@ +use crate::policy::{RetryPolicy, with_retry}; +use crate::service::common; +use anyhow::{Context, Result, bail}; +use reqwest::Url; +use sift_rs::{ + SiftChannel, + webhooks::v1::{ + ListWebhooksRequest, ListWebhooksResponse, Webhook, + webhook_service_client::WebhookServiceClient, + }, +}; + +#[cfg(test)] +mod test; + +#[derive(Clone)] +pub struct WebhookService { + channel: SiftChannel, + policy: RetryPolicy, +} + +impl WebhookService { + pub fn new(channel: SiftChannel, policy: RetryPolicy) -> Self { + Self { channel, policy } + } + + /// Lists webhooks with their secrets withheld; see [`redact`]. + pub async fn list_webhooks( + &self, + filter: String, + order_by: Option, + limit: Option, + ) -> Result> { + let (page_size, record_limit) = common::paging(limit); + + let mut page_token = String::new(); + let mut results = Vec::new(); + let mut has_more = false; + + let order_by = order_by.unwrap_or_default(); + + loop { + let channel = self.channel.clone(); + let filter = filter.clone(); + let order_by = order_by.clone(); + let token = page_token.clone(); + + let resp = with_retry(&self.policy, move || { + let channel = channel.clone(); + let filter = filter.clone(); + let order_by = order_by.clone(); + let token = token.clone(); + async move { + let mut client = WebhookServiceClient::new(channel); + client + .list_webhooks(ListWebhooksRequest { + filter, + page_size, + page_token: token, + order_by, + }) + .await + .map(|resp| resp.into_inner()) + } + }) + .await + .context("failed to query webhooks")?; + + let ListWebhooksResponse { + webhooks, + next_page_token, + } = resp; + if webhooks.is_empty() { + break; + } + results.extend(webhooks.into_iter().map(redact)); + + if results.len() >= record_limit { + // The cap, not the end of the data: report that more exist so the + // caller does not read this page's size as the match total. + has_more = results.len() > record_limit || !next_page_token.is_empty(); + break; + } + if next_page_token.is_empty() { + break; + } + page_token = next_page_token; + } + + results.truncate(record_limit); + + Ok(common::Page { + items: results, + has_more, + }) + } + + /// Fetches the webhooks with the given ids, archived ones included. An id + /// with no webhook is absent from the result, so the caller can tell a + /// missing webhook from an archived one. + pub async fn get_webhooks(&self, webhook_ids: &[String]) -> Result> { + if webhook_ids.is_empty() { + return Ok(Vec::new()); + } + + let ids = webhook_ids + .iter() + .map(|id| format!("\"{}\"", common::cel_escape(id))) + .collect::>() + .join(", "); + + let page = self + .list_webhooks( + format!("webhook_id in [{ids}]"), + None, + Some(common::PAGE_SIZE), + ) + .await?; + + if page.has_more { + bail!( + "webhook lookup matched more than {} webhooks and is incomplete; \ + pass fewer webhook ids", + common::PAGE_SIZE, + ); + } + Ok(page.items) + } +} + +/// Withholds the credentials a webhook carries. `ListWebhooks` returns them in +/// plaintext: header values such as `Authorization`, target URLs that are +/// themselves the secret (a Slack or Teams path, a Pipedream or Lambda +/// function URL subdomain), and body templates that embed a key (PagerDuty's +/// `routing_key`). Headers keep only their names, the body template is +/// dropped, and the target is cut by [`redact_target`]. +fn redact(mut webhook: Webhook) -> Webhook { + webhook.target_url = redact_target(&webhook.target_url); + for header in &mut webhook.http_headers { + header.value.clear(); + } + webhook.payload = None; + webhook +} + +/// Cuts a target URL to its scheme and the last two labels of its host, such +/// as `https://slack.com`: enough to tell where a webhook sends, with no path, +/// query, port, user info, or per-endpoint subdomain left to replay. An IP +/// address host is kept whole. Empty when the URL has no host. +fn redact_target(target_url: &str) -> String { + let Ok(url) = Url::parse(target_url) else { + return String::new(); + }; + let host = match url.domain() { + Some(domain) => { + let labels: Vec<&str> = domain.split('.').collect(); + labels[labels.len().saturating_sub(2)..].join(".") + } + None => match url.host_str() { + Some(ip) => ip.to_string(), + None => return String::new(), + }, + }; + format!("{}://{host}", url.scheme()) +} diff --git a/rust/crates/sift_mcp/src/service/webhooks/test.rs b/rust/crates/sift_mcp/src/service/webhooks/test.rs new file mode 100644 index 0000000000..dc92d6dc5b --- /dev/null +++ b/rust/crates/sift_mcp/src/service/webhooks/test.rs @@ -0,0 +1,285 @@ +use sift_rs::webhooks::v1::{ + ListWebhooksResponse, Webhook, WebhookHttpHeader, webhook_service_server::WebhookServiceServer, +}; +use sift_test_util::{grpc::memory_sift_channel, mock::webhooks::v1::MockWebhookServiceImpl}; +use tokio::task::JoinHandle; +use tonic::{Response, Status, transport::Server}; + +use super::WebhookService; +use crate::service::common::{DEFAULT_LIMIT, PAGE_SIZE}; + +async fn service_with_mock(mock: MockWebhookServiceImpl) -> (WebhookService, JoinHandle<()>) { + let (client, server) = tokio::io::duplex(1024); + let channel = memory_sift_channel(client).await; + + let handle = tokio::spawn(async move { + Server::builder() + .add_service(WebhookServiceServer::new(mock)) + .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server))) + .await + .unwrap(); + }); + + ( + WebhookService::new(channel, crate::policy::RetryPolicy::default()), + handle, + ) +} + +fn webhook(id: &str) -> Webhook { + Webhook { + webhook_id: id.into(), + name: format!("{id} name"), + ..Default::default() + } +} + +#[tokio::test] +async fn list_webhooks_returns_single_page() { + let mut mock = MockWebhookServiceImpl::new(); + mock.expect_list_webhooks() + .withf(|req| { + let req = req.get_ref(); + req.filter == "is_archived == false" && req.order_by == "created_date desc" + }) + .returning(|_| { + Ok(Response::new(ListWebhooksResponse { + webhooks: vec![webhook("wh1")], + next_page_token: String::new(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let page = service + .list_webhooks( + "is_archived == false".to_string(), + Some("created_date desc".to_string()), + None, + ) + .await + .expect("list_webhooks failed"); + + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].webhook_id, "wh1"); + assert!(!page.has_more); +} + +#[tokio::test] +async fn list_webhooks_paginates_until_token_empty() { + let mut mock = MockWebhookServiceImpl::new(); + mock.expect_list_webhooks().times(2).returning(|req| { + let req = req.into_inner(); + assert_eq!(req.page_size, DEFAULT_LIMIT); + let (webhooks, next) = match req.page_token.as_str() { + "" => (vec![webhook("wh1")], "page-2".to_string()), + "page-2" => (vec![webhook("wh2")], String::new()), + other => return Err(Status::invalid_argument(format!("bad token: {other}"))), + }; + Ok(Response::new(ListWebhooksResponse { + webhooks, + next_page_token: next, + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let page = service + .list_webhooks(String::new(), None, None) + .await + .expect("list_webhooks failed"); + + let ids: Vec<&str> = page.items.iter().map(|w| w.webhook_id.as_str()).collect(); + assert_eq!(ids, vec!["wh1", "wh2"]); + assert!(!page.has_more); +} + +#[tokio::test] +async fn list_webhooks_truncates_to_limit_and_reports_more() { + let mut mock = MockWebhookServiceImpl::new(); + mock.expect_list_webhooks().times(1).returning(|req| { + assert_eq!(req.get_ref().page_size, 2); + Ok(Response::new(ListWebhooksResponse { + webhooks: vec![webhook("wh1"), webhook("wh2")], + next_page_token: "page-2".into(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let page = service + .list_webhooks(String::new(), None, Some(2)) + .await + .expect("list_webhooks failed"); + + assert_eq!(page.items.len(), 2); + assert!(page.has_more); +} + +#[tokio::test] +async fn list_webhooks_breaks_on_empty_page() { + let mut mock = MockWebhookServiceImpl::new(); + mock.expect_list_webhooks().times(1).returning(|_| { + Ok(Response::new(ListWebhooksResponse { + webhooks: vec![], + next_page_token: "ignored".into(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let page = service + .list_webhooks(String::new(), None, None) + .await + .expect("list_webhooks failed"); + + assert!(page.items.is_empty()); + assert!(!page.has_more); +} + +#[tokio::test] +async fn list_webhooks_propagates_grpc_error() { + let mut mock = MockWebhookServiceImpl::new(); + mock.expect_list_webhooks() + .returning(|_| Err(Status::permission_denied("no access"))); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .list_webhooks(String::new(), None, None) + .await + .expect_err("expected error"); + + assert!(err.to_string().contains("failed to query webhooks")); +} + +#[tokio::test] +async fn list_webhooks_withholds_credentials() { + let mut mock = MockWebhookServiceImpl::new(); + mock.expect_list_webhooks().returning(|_| { + Ok(Response::new(ListWebhooksResponse { + webhooks: vec![ + Webhook { + webhook_id: "slack".into(), + target_url: "https://hooks.slack.com/services/T000/B000/secret".into(), + http_headers: vec![WebhookHttpHeader { + name: "Authorization".into(), + value: "Bearer secret".into(), + }], + ..Default::default() + }, + Webhook { + webhook_id: "pagerduty".into(), + target_url: "https://events.pagerduty.com/v2/enqueue".into(), + payload: Some(r#"{"routing_key": "secret"}"#.into()), + ..Default::default() + }, + ], + next_page_token: String::new(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let page = service + .list_webhooks(String::new(), None, None) + .await + .expect("list_webhooks failed"); + + assert_eq!(page.items[0].target_url, "https://slack.com"); + assert_eq!(page.items[0].http_headers[0].name, "Authorization"); + assert_eq!(page.items[0].http_headers[0].value, ""); + assert_eq!(page.items[1].target_url, "https://pagerduty.com"); + assert_eq!(page.items[1].payload, None); + + let body = serde_json::to_string(&page.items).unwrap(); + assert!(!body.contains("secret"), "credentials leaked: {body}"); +} + +#[test] +fn redact_target_keeps_only_scheme_and_domain() { + let cases = [ + ( + "https://eo3k9x2secret.m.pipedream.net/", + "https://pipedream.net", + ), + ( + "https://secretid.lambda-url.us-east-1.on.aws/", + "https://on.aws", + ), + ( + "https://user:secret@example.com:8443/hook?token=secret", + "https://example.com", + ), + ("http://10.0.0.5:8080/hook/secret", "http://10.0.0.5"), + ("http://[::1]:8080/hook/secret", "http://[::1]"), + ("https://localhost/secret", "https://localhost"), + ("not a url secret", ""), + ("mailto:secret@example.com", ""), + ]; + for (input, expected) in cases { + assert_eq!(super::redact_target(input), expected, "{input}"); + } +} + +#[tokio::test] +async fn get_webhooks_filters_by_id_and_includes_archived() { + let mut mock = MockWebhookServiceImpl::new(); + mock.expect_list_webhooks() + .withf(|req| { + let req = req.get_ref(); + req.filter == r#"webhook_id in ["wh1", "wh\"2"]"# && req.page_size == PAGE_SIZE + }) + .returning(|_| { + Ok(Response::new(ListWebhooksResponse { + webhooks: vec![webhook("wh1")], + next_page_token: String::new(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let webhooks = service + .get_webhooks(&["wh1".to_string(), "wh\"2".to_string()]) + .await + .expect("get_webhooks failed"); + + assert_eq!(webhooks.len(), 1); + assert_eq!(webhooks[0].webhook_id, "wh1"); +} + +#[tokio::test] +async fn get_webhooks_skips_the_call_for_no_ids() { + let mut mock = MockWebhookServiceImpl::new(); + mock.expect_list_webhooks().times(0); + + let (service, _h) = service_with_mock(mock).await; + + let webhooks = service + .get_webhooks(&[]) + .await + .expect("get_webhooks failed"); + + assert!(webhooks.is_empty()); +} + +#[tokio::test] +async fn get_webhooks_fails_when_the_lookup_is_capped() { + let mut mock = MockWebhookServiceImpl::new(); + mock.expect_list_webhooks().times(1).returning(|req| { + let page_size = req.get_ref().page_size as usize; + Ok(Response::new(ListWebhooksResponse { + webhooks: (0..page_size).map(|i| webhook(&format!("wh{i}"))).collect(), + next_page_token: "more".into(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + let err = service + .get_webhooks(&["wh0".to_string()]) + .await + .expect_err("expected error"); + + assert!(err.to_string().contains("incomplete")); +} diff --git a/rust/crates/sift_mcp/src/tool/mod.rs b/rust/crates/sift_mcp/src/tool/mod.rs index 95e2a3296f..fe517047ac 100644 --- a/rust/crates/sift_mcp/src/tool/mod.rs +++ b/rust/crates/sift_mcp/src/tool/mod.rs @@ -17,3 +17,4 @@ pub mod test_reports; pub mod update; pub mod user_defined_functions; pub mod users; +pub mod webhooks; diff --git a/rust/crates/sift_mcp/src/tool/rules/mod.rs b/rust/crates/sift_mcp/src/tool/rules/mod.rs index a82f632490..911fea095a 100644 --- a/rust/crates/sift_mcp/src/tool/rules/mod.rs +++ b/rust/crates/sift_mcp/src/tool/rules/mod.rs @@ -6,12 +6,15 @@ use rmcp::{ tool, tool_router, }; use serde::Deserialize; -use sift_rs::rules::v1::{UpdateConditionRequest, UpdateRuleRequest}; +use sift_rs::{ + rules::v1::{UpdateConditionRequest, UpdateRuleRequest}, + webhooks::v1::Webhook, +}; use crate::{ error::{self, from_anyhow}, server::SiftMcpServer, - service::rules::RuleUpdate, + service::rules::{RuleUpdate, has_webhook_action}, tool::common::{ListParams, list_body, url_clause, with_urls}, }; @@ -21,6 +24,7 @@ mod test; #[derive(Debug, Deserialize, JsonSchema)] pub struct RuleDefinitionParams { rule_json: String, + webhook_ids: Option>, } #[derive(Debug, Deserialize, JsonSchema)] @@ -40,6 +44,7 @@ pub struct UpdateRuleParams { asset_ids: Option>, tag_ids: Option>, conditions_json: Option, + webhook_ids: Option>, } #[derive(Debug, Deserialize, JsonSchema)] @@ -219,24 +224,32 @@ impl SiftMcpServer { - `rule_json`: a JSON object matching `protos/sift/rules/v1/rules.proto::UpdateRuleRequest`. Use the proto field names. Omit `rule_id` (this is a create). Key fields: - `name` (required), `description` (required). - - `conditions` (required): array of `{ \"expression\": {...}, \"actions\": [...] }`. The - expression and action shapes are non-trivial; mirror an existing rule retrieved via + - `conditions` (required): exactly one `{ \"expression\": {...}, \"actions\": [...] }`; Sift + rejects more. The expression shape is non-trivial; mirror an existing rule retrieved via `list_rules` rather than authoring blind. + - `actions`: include one annotation action so violations are recorded: + `{ \"action_type\": \"ANNOTATION\", \"configuration\": { \"annotation\": { \"tag_ids\": [], + \"annotation_type\": \"ANNOTATION_TYPE_DATA_REVIEW\" } } }`. Put webhooks in + `webhook_ids`, not here. Sift rejects `NOTIFICATION` actions on rules. - `asset_configuration`: `{ \"asset_ids\": [...], \"tag_ids\": [...] }` — which assets/tags the rule applies to. - `metadata` (required): array of metadata values (may be empty: `[]`). - `client_key` (optional): a caller-defined identifier. Immutable after creation. - `is_live_evaluation_enabled` (optional): evaluate against live data. + - `webhook_ids`: optional. Webhooks from `list_webhooks` to call when the condition is violated and + again when it resolves. Use it when the user asks to be notified or alerted outside Sift. Errors: - - `INVALID_PARAMS` if `rule_json` is not valid JSON or does not match the rule schema. + - `INVALID_PARAMS` if `rule_json` is not valid JSON, does not match the rule schema, does not hold + exactly one condition, or holds a webhook action, or if a webhook is archived. + - `RESOURCE_NOT_FOUND` if a `webhook_ids` entry matches no webhook. - `INTERNAL_ERROR` for upstream gRPC failures (e.g. a referenced asset does not exist). Guidance: - - Rule definitions are complex. Retrieve a similar rule with `list_rules` first and adapt its shape; - do not invent `conditions`/`actions` structures. - - This creates a live resource. Confirm the rule's name, the target assets/tags, and the conditions - with the user before calling. Do not silently default them. + - Rule expressions are complex. Retrieve a similar rule with `list_rules` first and adapt its + expression; do not invent one. + - This creates a live resource. Confirm the rule's name, the target assets/tags, the conditions, and + any webhooks with the user before calling. Do not silently default them. ", annotations( title = "rules/create_rule", @@ -248,23 +261,30 @@ impl SiftMcpServer { pub async fn create_rule(&self, params: Parameters) -> error::McpResult { self.require_create()?; - let Parameters(RuleDefinitionParams { rule_json }) = params; + let Parameters(RuleDefinitionParams { + rule_json, + webhook_ids, + }) = params; let update = parse_rule_definition(&rule_json)?; + validate_conditions("rule_json", &update.conditions)?; + let webhook_ids = webhook_ids.unwrap_or_default(); + let webhooks = self.resolve_webhooks(&webhook_ids).await?; let rule_name = update.name.clone(); let rule_id = self .rule_service - .create_rule(update) + .create_rule(update, webhook_ids) .await .map_err(from_anyhow)?; let rule_url = self.url_service.build_rule_url(&rule_id).ok(); let next_step = format!( - "Created rule with id `{rule_id}`.{} Tell the user the new rule id. If they haven't \ + "Created rule with id `{rule_id}`.{}{} Tell the user the new rule id. If they haven't \ indicated a next step, offer to confirm it with `list_rules` \ (filter `rule_id == \"{rule_id}\"`).", url_clause("rule", Some(rule_name.as_str()), rule_url.as_deref()), + webhook_clause("It calls", &webhooks), ); let mut result = CallToolResult::structured(serde_json::json!({ @@ -297,19 +317,25 @@ impl SiftMcpServer { - `asset_ids`: optional. Replaces the rule's set of target asset ids. - `tag_ids`: optional. Replaces the rule's set of target tag ids. - `conditions_json`: optional. Replaces the rule's conditions. A JSON array matching the - `conditions` field of `create_rule`'s `rule_json`. Omit it to keep the existing conditions - unchanged — only provide it when you intend to rewrite the rule's logic. + `conditions` field of `create_rule`'s `rule_json`, without webhook actions. Omit it to keep the + existing conditions unchanged — only provide it when you intend to rewrite the rule's logic. + - `webhook_ids`: optional. Replaces the rule's set of webhooks, from `list_webhooks`, that it calls + when its condition is violated and again when it resolves. Include the ids it already calls + (`conditions[].actions[].configuration.webhook.webhook_id` in `list_rules`) to keep them; `[]` + removes them all. Omit it to keep the current webhooks, even when `conditions_json` is set. - At least one field besides `rule_id` must be set. Errors: - - `INVALID_PARAMS` if no field besides `rule_id` is set, or if `conditions_json` is not a valid - conditions array. - - `RESOURCE_NOT_FOUND` if no rule has the given `rule_id`. + - `INVALID_PARAMS` if no field besides `rule_id` is set, if `conditions_json` is not a valid + conditions array of exactly one condition or holds a webhook action, or if a webhook is + archived. + - `RESOURCE_NOT_FOUND` if no rule has the given `rule_id`, or a `webhook_ids` entry matches no + webhook. - `INTERNAL_ERROR` for upstream gRPC failures. Guidance: - Prefer this over recreating a rule. To change only the name, pass `rule_id` and `name`; the - conditions, metadata, and asset targeting are preserved automatically. + conditions, metadata, webhooks, and asset targeting are preserved automatically. - Confirm the intended changes with the user before calling. ", annotations( @@ -331,12 +357,16 @@ impl SiftMcpServer { asset_ids, tag_ids, conditions_json, + webhook_ids, }) = params; let conditions = conditions_json .as_deref() .map(parse_conditions) .transpose()?; + if let Some(conditions) = &conditions { + validate_conditions("conditions_json", conditions)?; + } let changes = RuleUpdate { name, @@ -346,6 +376,7 @@ impl SiftMcpServer { asset_ids, tag_ids, conditions, + webhook_ids, }; let nothing_to_change = changes.name.is_none() @@ -354,7 +385,8 @@ impl SiftMcpServer { && changes.is_live_evaluation_enabled.is_none() && changes.asset_ids.is_none() && changes.tag_ids.is_none() - && changes.conditions.is_none(); + && changes.conditions.is_none() + && changes.webhook_ids.is_none(); if nothing_to_change { return Err(ErrorData::invalid_params( "at least one field besides `rule_id` must be set", @@ -362,19 +394,26 @@ impl SiftMcpServer { )); } + let webhooks = match &changes.webhook_ids { + Some(webhook_ids) => self.resolve_webhooks(webhook_ids).await?, + None => Vec::new(), + }; let rule_name = changes.name.clone(); - let rule_id = self + let updated = self .rule_service .update_rule(rule_id, changes) .await .map_err(from_anyhow)?; + let rule_id = updated.rule_id; let rule_url = self.url_service.build_rule_url(&rule_id).ok(); let next_step = format!( - "Updated rule `{rule_id}`.{} Tell the user the update succeeded. If they haven't \ + "Updated rule `{rule_id}`.{}{}{} Tell the user the update succeeded. If they haven't \ indicated a next step, offer to confirm it with `list_rules` \ (filter `rule_id == \"{rule_id}\"`).", url_clause("rule", rule_name.as_deref(), rule_url.as_deref()), + webhook_clause("It now calls", &webhooks), + self.detached_clause(&updated.detached_webhook_ids).await, ); let mut result = CallToolResult::structured(serde_json::json!({ @@ -509,6 +548,134 @@ impl SiftMcpServer { } } +impl SiftMcpServer { + /// Confirms each webhook exists and is live before a rule is pointed at it. + /// Sift accepts an archived webhook on a rule and then never calls it, so + /// the check has to happen here. + async fn resolve_webhooks(&self, webhook_ids: &[String]) -> Result, ErrorData> { + let webhooks = self + .webhook_service + .get_webhooks(webhook_ids) + .await + .map_err(from_anyhow)?; + check_webhooks(webhook_ids, webhooks) + } + + /// A `next_step` sentence naming the webhooks an update detached, so a set + /// replaced without the rule's current ids does not drop one unnoticed. + /// The rule is already saved, so a failed name lookup falls back to ids + /// instead of failing the call. + async fn detached_clause(&self, webhook_ids: &[String]) -> String { + if webhook_ids.is_empty() { + return String::new(); + } + let webhooks = self + .webhook_service + .get_webhooks(webhook_ids) + .await + .unwrap_or_default(); + let names = webhook_ids + .iter() + .map(|id| match webhooks.iter().find(|w| &w.webhook_id == id) { + Some(webhook) => format!("`{}`", webhook.name), + None => format!("`{id}`"), + }) + .collect::>() + .join(", "); + format!( + " It no longer calls {names}. Tell the user, and if they did not ask to remove it, offer \ + to restore it by passing it in `webhook_ids`." + ) + } +} + +/// Refuses conditions Sift would reject or that would bypass the webhook check. +/// Sift takes exactly one condition per rule, and answers a rule with none +/// with an internal error that reads as a backend fault, so the count is +/// checked here. Webhooks reach a rule only through `webhook_ids`, where each +/// is checked before saving, so a webhook action written into the conditions +/// is refused with a pointer to the right parameter. +fn validate_conditions( + param: &str, + conditions: &[UpdateConditionRequest], +) -> Result<(), ErrorData> { + if conditions.len() != 1 { + return Err(ErrorData::invalid_params( + format!( + "`{param}` holds {} conditions; a Sift rule takes exactly one", + conditions.len() + ), + None, + )); + } + if has_webhook_action(conditions) { + return Err(ErrorData::invalid_params( + format!("`{param}` holds a webhook action; pass webhook ids in `webhook_ids` instead"), + None, + )); + } + Ok(()) +} + +/// Fails with `RESOURCE_NOT_FOUND` when an id matches no webhook and with +/// `INVALID_PARAMS` when a webhook is archived. Returns the webhooks otherwise. +fn check_webhooks( + webhook_ids: &[String], + webhooks: Vec, +) -> Result, ErrorData> { + let missing: Vec = webhook_ids + .iter() + .filter(|id| !webhooks.iter().any(|w| &w.webhook_id == *id)) + .map(|id| format!("`{id}`")) + .collect(); + if !missing.is_empty() { + return Err(ErrorData::resource_not_found( + format!( + "no webhook has id {}. Find webhook ids with `list_webhooks`.", + missing.join(", ") + ), + None, + )); + } + + let archived: Vec = webhooks + .iter() + .filter(|w| w.is_archived || w.archived_date.is_some()) + .map(|w| format!("`{}` ({})", w.name, w.webhook_id)) + .collect(); + if !archived.is_empty() { + return Err(ErrorData::invalid_params( + format!( + "webhook {} is archived, and Sift never calls an archived webhook. Ask the user \ + to pick a live one from `list_webhooks` with `is_archived == false`.", + archived.join(", ") + ), + None, + )); + } + + Ok(webhooks) +} + +/// A `next_step` sentence naming the webhooks a rule calls, so the agent can +/// tell the user where alerts go. Empty when there are none. +fn webhook_clause(lead: &str, webhooks: &[Webhook]) -> String { + if webhooks.is_empty() { + return String::new(); + } + let names = webhooks + .iter() + .map(|w| format!("`{}`", w.name)) + .collect::>() + .join(", "); + let noun = if webhooks.len() == 1 { + "webhook" + } else { + "webhooks" + }; + format!(" {lead} {noun} {names} when its condition is violated and again when it resolves.") +} + /// Build the web URL for a rule when its `rule_id` is known. Returns `None` when /// `rule_id` is empty (the caller targeted the rule by `client_key`, so there is /// no id to link) or when the host can't be derived. diff --git a/rust/crates/sift_mcp/src/tool/rules/test.rs b/rust/crates/sift_mcp/src/tool/rules/test.rs index c07f453a84..85870479ef 100644 --- a/rust/crates/sift_mcp/src/tool/rules/test.rs +++ b/rust/crates/sift_mcp/src/tool/rules/test.rs @@ -1,6 +1,107 @@ -use rmcp::model::ErrorCode; +use rmcp::{handler::server::wrapper::Parameters, model::ErrorCode}; +use sift_rs::{ + rules::v1::{ + ActionKind, CreateRuleResponse, GetRuleResponse, Rule, RuleAction, RuleActionConfiguration, + RuleCondition, UpdateRuleResponse, WebhookActionConfiguration, + rule_action_configuration::Configuration, rule_service_server::RuleServiceServer, + }, + webhooks::v1::{ListWebhooksResponse, Webhook, webhook_service_server::WebhookServiceServer}, +}; +use sift_test_util::{ + grpc::memory_sift_channel, + mock::{rules::v1::MockRuleServiceImpl, webhooks::v1::MockWebhookServiceImpl}, +}; +use tokio::task::JoinHandle; +use tonic::{Response, transport::Server}; -use super::{parse_rule_definition, rule_identifier}; +use super::{ + RuleDefinitionParams, UpdateRuleParams, check_webhooks, parse_rule_definition, rule_identifier, +}; +use crate::{server::SiftMcpServer, tool::common::test_support::structured}; + +/// A rule definition in the shape `create_rule` documents, with one annotation action. +const RULE_JSON: &str = r#"{ + "name": "overtemp", + "description": "engine over temperature", + "conditions": [{ + "expression": {}, + "actions": [{ + "action_type": "ANNOTATION", + "configuration": { + "annotation": { "tag_ids": [], "annotation_type": "ANNOTATION_TYPE_DATA_REVIEW" } + } + }] + }], + "metadata": [] +}"#; + +async fn server_with_mocks( + rule_mock: MockRuleServiceImpl, + webhook_mock: MockWebhookServiceImpl, +) -> (SiftMcpServer, JoinHandle<()>) { + let (client, server) = tokio::io::duplex(1024); + let channel = memory_sift_channel(client).await; + + let handle = tokio::spawn(async move { + Server::builder() + .add_service(RuleServiceServer::new(rule_mock)) + .add_service(WebhookServiceServer::new(webhook_mock)) + .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server))) + .await + .unwrap(); + }); + + ( + SiftMcpServer::new(channel, String::from("https://app.test.local"), true, true), + handle, + ) +} + +fn webhook(id: &str, name: &str) -> Webhook { + Webhook { + webhook_id: id.into(), + name: name.into(), + ..Default::default() + } +} + +fn webhook_mock_returning(webhooks: Vec) -> MockWebhookServiceImpl { + let mut mock = MockWebhookServiceImpl::new(); + mock.expect_list_webhooks().returning(move |_| { + Ok(Response::new(ListWebhooksResponse { + webhooks: webhooks.clone(), + next_page_token: String::new(), + })) + }); + mock +} + +fn create_params(webhook_ids: Option>) -> Parameters { + Parameters(RuleDefinitionParams { + rule_json: RULE_JSON.to_string(), + webhook_ids, + }) +} + +fn update_params(webhook_ids: Option>) -> Parameters { + Parameters(UpdateRuleParams { + rule_id: "rule-1".to_string(), + name: None, + description: None, + version_notes: None, + is_live_evaluation_enabled: None, + asset_ids: None, + tag_ids: None, + conditions_json: None, + webhook_ids, + }) +} + +/// A webhook action as `create_rule` would refuse to find it in `rule_json`. +const WEBHOOK_ACTION_JSON: &str = r#"{ + "action_type": "WEBHOOK", + "configuration": { "webhook": { "webhook_id": "wh1" } } +}"#; #[test] fn rule_identifier_accepts_rule_id_only() { @@ -44,3 +145,280 @@ fn parse_rule_definition_rejects_invalid_json() { let err = parse_rule_definition("not json").expect_err("should reject"); assert_eq!(err.code, ErrorCode::INVALID_PARAMS); } + +#[test] +fn parse_rule_definition_accepts_the_documented_annotation_action() { + let update = parse_rule_definition(RULE_JSON).expect("should parse"); + let action = &update.conditions[0].actions[0]; + assert_eq!(action.action_type, ActionKind::Annotation as i32); + assert!(matches!( + action.configuration.as_ref().unwrap().configuration, + Some(Configuration::Annotation(_)) + )); +} + +#[test] +fn check_webhooks_accepts_live_webhooks() { + let webhooks = check_webhooks(&["wh1".to_string()], vec![webhook("wh1", "alerts")]) + .expect("should accept"); + assert_eq!(webhooks[0].name, "alerts"); +} + +#[test] +fn check_webhooks_rejects_an_unknown_id() { + let err = check_webhooks( + &["wh1".to_string(), "wh-missing".to_string()], + vec![webhook("wh1", "alerts")], + ) + .expect_err("should reject"); + assert_eq!(err.code, ErrorCode::RESOURCE_NOT_FOUND); + assert!(err.message.contains("wh-missing")); +} + +#[test] +fn check_webhooks_rejects_an_archived_webhook() { + let archived = Webhook { + is_archived: true, + ..webhook("wh1", "old alerts") + }; + let err = check_webhooks(&["wh1".to_string()], vec![archived]).expect_err("should reject"); + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); + assert!(err.message.contains("old alerts")); +} + +#[tokio::test] +async fn create_rule_attaches_webhooks_and_names_them() { + let mut rule_mock = MockRuleServiceImpl::new(); + rule_mock + .expect_create_rule() + .withf(|req| { + let actions = &req.get_ref().update.as_ref().unwrap().conditions[0].actions; + actions.len() == 2 && actions[1].action_type == ActionKind::Webhook as i32 + }) + .returning(|_| { + Ok(Response::new(CreateRuleResponse { + rule_id: "new-rule".into(), + })) + }); + + let (server, _h) = server_with_mocks( + rule_mock, + webhook_mock_returning(vec![webhook("wh1", "slack alerts")]), + ) + .await; + + let result = server + .create_rule(create_params(Some(vec!["wh1".to_string()]))) + .await + .expect("create_rule failed"); + + let next_step = structured(result)["next_step"] + .as_str() + .unwrap() + .to_string(); + assert!(next_step.contains("`slack alerts`"), "{next_step}"); +} + +#[tokio::test] +async fn create_rule_refuses_an_archived_webhook_before_writing() { + let mut rule_mock = MockRuleServiceImpl::new(); + rule_mock.expect_create_rule().times(0); + + let archived = Webhook { + is_archived: true, + ..webhook("wh1", "old alerts") + }; + let (server, _h) = server_with_mocks(rule_mock, webhook_mock_returning(vec![archived])).await; + + let err = server + .create_rule(create_params(Some(vec!["wh1".to_string()]))) + .await + .expect_err("should refuse"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn create_rule_without_webhooks_skips_the_lookup() { + let mut rule_mock = MockRuleServiceImpl::new(); + rule_mock.expect_create_rule().times(1).returning(|_| { + Ok(Response::new(CreateRuleResponse { + rule_id: "new-rule".into(), + })) + }); + let mut webhook_mock = MockWebhookServiceImpl::new(); + webhook_mock.expect_list_webhooks().times(0); + + let (server, _h) = server_with_mocks(rule_mock, webhook_mock).await; + + server + .create_rule(create_params(None)) + .await + .expect("create_rule failed"); +} + +#[tokio::test] +async fn create_rule_refuses_a_webhook_action_in_rule_json() { + let (server, _h) = + server_with_mocks(MockRuleServiceImpl::new(), MockWebhookServiceImpl::new()).await; + + let rule_json = RULE_JSON.replacen( + r#""actions": ["#, + &format!(r#""actions": [{WEBHOOK_ACTION_JSON}, "#), + 1, + ); + let err = server + .create_rule(Parameters(RuleDefinitionParams { + rule_json, + webhook_ids: None, + })) + .await + .expect_err("should refuse"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); + assert!(err.message.contains("webhook_ids"), "{}", err.message); +} + +#[tokio::test] +async fn update_rule_refuses_a_webhook_action_in_conditions_json() { + let (server, _h) = + server_with_mocks(MockRuleServiceImpl::new(), MockWebhookServiceImpl::new()).await; + + let mut params = update_params(None); + params.0.conditions_json = Some(format!( + r#"[{{ "expression": {{}}, "actions": [{WEBHOOK_ACTION_JSON}] }}]"# + )); + let err = server.update_rule(params).await.expect_err("should refuse"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn update_rule_replaces_webhooks_and_names_the_detached_one() { + let mut rule_mock = MockRuleServiceImpl::new(); + rule_mock.expect_get_rule().returning(|_| { + Ok(Response::new(GetRuleResponse { + rule: Some(Rule { + rule_id: "rule-1".into(), + conditions: vec![RuleCondition { + actions: vec![RuleAction { + rule_action_id: "act-old".into(), + action_type: ActionKind::Webhook as i32, + configuration: Some(RuleActionConfiguration { + configuration: Some(Configuration::Webhook( + WebhookActionConfiguration { + webhook_id: "wh-old".into(), + }, + )), + }), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + }), + })) + }); + rule_mock + .expect_update_rule() + .withf(|req| { + let actions = &req.get_ref().conditions[0].actions; + actions.len() == 1 + && matches!( + actions[0].configuration.as_ref().unwrap().configuration.as_ref(), + Some(Configuration::Webhook(w)) if w.webhook_id == "wh-new" + ) + }) + .returning(|_| { + Ok(Response::new(UpdateRuleResponse { + rule_id: "rule-1".into(), + })) + }); + + // One lookup validates the new set; a second names the webhook detached. + let mut webhook_mock = MockWebhookServiceImpl::new(); + webhook_mock + .expect_list_webhooks() + .times(2) + .returning(|req| { + let filter = &req.get_ref().filter; + let webhooks = if filter.contains("wh-new") { + vec![webhook("wh-new", "new alerts")] + } else { + vec![webhook("wh-old", "old alerts")] + }; + Ok(Response::new(ListWebhooksResponse { + webhooks, + next_page_token: String::new(), + })) + }); + + let (server, _h) = server_with_mocks(rule_mock, webhook_mock).await; + + let result = server + .update_rule(update_params(Some(vec!["wh-new".to_string()]))) + .await + .expect("update_rule failed"); + + let next_step = structured(result)["next_step"] + .as_str() + .unwrap() + .to_string(); + assert!( + next_step.contains("now calls webhook `new alerts`"), + "{next_step}" + ); + assert!( + next_step.contains("no longer calls `old alerts`"), + "{next_step}" + ); +} + +#[tokio::test] +async fn update_rule_refuses_an_archived_webhook_before_writing() { + let mut rule_mock = MockRuleServiceImpl::new(); + rule_mock.expect_get_rule().times(0); + rule_mock.expect_update_rule().times(0); + + let archived = Webhook { + is_archived: true, + ..webhook("wh1", "old alerts") + }; + let (server, _h) = server_with_mocks(rule_mock, webhook_mock_returning(vec![archived])).await; + + let err = server + .update_rule(update_params(Some(vec!["wh1".to_string()]))) + .await + .expect_err("should refuse"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn create_rule_refuses_a_rule_without_exactly_one_condition() { + let (server, _h) = + server_with_mocks(MockRuleServiceImpl::new(), MockWebhookServiceImpl::new()).await; + + let err = server + .create_rule(Parameters(RuleDefinitionParams { + rule_json: r#"{ "name": "overtemp", "description": "d", "metadata": [] }"#.to_string(), + webhook_ids: None, + })) + .await + .expect_err("should refuse"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); + assert!(err.message.contains("exactly one"), "{}", err.message); +} + +#[tokio::test] +async fn update_rule_refuses_empty_conditions_json() { + let (server, _h) = + server_with_mocks(MockRuleServiceImpl::new(), MockWebhookServiceImpl::new()).await; + + let mut params = update_params(None); + params.0.conditions_json = Some("[]".to_string()); + let err = server.update_rule(params).await.expect_err("should refuse"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} diff --git a/rust/crates/sift_mcp/src/tool/webhooks/mod.rs b/rust/crates/sift_mcp/src/tool/webhooks/mod.rs new file mode 100644 index 0000000000..20e3893a2a --- /dev/null +++ b/rust/crates/sift_mcp/src/tool/webhooks/mod.rs @@ -0,0 +1,83 @@ +use rmcp::{handler::server::wrapper::Parameters, model::CallToolResult, tool, tool_router}; + +use crate::{ + error::{self, from_anyhow}, + server::SiftMcpServer, + tool::common::{ListParams, list_body, to_values}, +}; + +#[cfg(test)] +mod test; + +#[tool_router(router = webhooks_router, vis = "pub(crate)")] +impl SiftMcpServer { + #[tool( + name = "list_webhooks", + description = " + List the webhooks a Sift rule can call when it fires, optionally filtered by a CEL expression. Use it to + resolve a webhook the user names to the `webhook_id` that `create_rule` and `update_rule` take. + + Output: + - `{ \"webhooks\": [Webhook, ...] }`. Each item carries `webhook_id`, `name`, `event_type`, + `target_url`, `http_headers`, `is_archived`, `archived_date`, and creator, modifier, and + timestamp fields. + - Credentials are withheld. `target_url` is cut to its scheme and the last two labels of its host + (`https://hooks.slack.com/services/...` becomes `https://slack.com`) and is omitted when it can't + be parsed. Each `http_headers` entry carries only its `name`, and the request body template is + omitted. Attaching a webhook needs none of them, so do not ask the user for them. + - Fields at their proto3 default are OMITTED: a missing `is_archived` key means `false`. + - `count`: how many items THIS response carries, not how many match `filter`. + - `has_more`: `true` when the service hit `limit` with matches left over. Never report `count` as a + total while `has_more` is `true`. + + Parameters: + - `filter`: CEL expression. Pass an empty string to list everything, archived webhooks included. + Filterable fields: `webhook_id`, `name`, `event_type`, `is_archived`, `archived_date`. `event_type` + takes the enum name: `event_type == \"WEBHOOK_EVENT_TYPE_RULE_VIOLATION\"`. + When filtering or searching, use `name.matches(\"(?i)slack\")`, not `==`. Use `==` only for an + exact value from a prior result. `contains`/`startsWith`/`endsWith` are case-SENSITIVE: + `contains(\"Slack\")` silently misses `slack-alerts`. + - `order_by`: optional comma-separated `FIELD_NAME[ desc]` list. Orderable fields: `created_date`. + Default sort is `created_date desc` (newest first). Example: `\"created_date\"`. + - `limit`: max items to return. Start at 50 and only raise it if the result is capped + and you still need more. Values are clamped to `1..=200`; omitting it defaults to 50. + - `fields`: optional array of field names to keep on each item, e.g. + `[\"webhook_id\", \"name\"]`. Omit it for the full object. Names match case-insensitively + and ignore underscores and hyphens. Any name that matched nothing on any returned item + is listed in `unmatched_fields`. + + Errors: + - `INVALID_PARAMS` if `filter` is not a valid CEL expression or `order_by` references an unknown field. + - `INTERNAL_ERROR` for upstream gRPC failures. + + Guidance: + - Default add `is_archived == false` to the filter. Sift never calls an archived webhook, and the + rule tools refuse one. + - This server cannot create webhooks. When none fits, tell the user to create one in the Sift web app. + ", + annotations(title = "webhooks/list_webhooks", read_only_hint = true) + )] + pub async fn list_webhooks(&self, params: Parameters) -> error::McpResult { + let Parameters(ListParams { + filter, + order_by, + limit, + fields, + }) = params; + + let page = self + .webhook_service + .list_webhooks(filter, order_by, limit) + .await + .map_err(from_anyhow)?; + + let webhooks = to_values(&page.items)?; + + Ok(CallToolResult::structured(list_body( + "webhooks", + webhooks, + fields, + page.has_more, + ))) + } +} diff --git a/rust/crates/sift_mcp/src/tool/webhooks/test.rs b/rust/crates/sift_mcp/src/tool/webhooks/test.rs new file mode 100644 index 0000000000..53399b5d21 --- /dev/null +++ b/rust/crates/sift_mcp/src/tool/webhooks/test.rs @@ -0,0 +1,97 @@ +use sift_rs::webhooks::v1::{ + ListWebhooksResponse, Webhook, WebhookHttpHeader, webhook_service_server::WebhookServiceServer, +}; +use sift_test_util::{grpc::memory_sift_channel, mock::webhooks::v1::MockWebhookServiceImpl}; +use tokio::task::JoinHandle; +use tonic::{Response, transport::Server}; + +use crate::{ + server::SiftMcpServer, + tool::common::test_support::{list_params, structured}, +}; + +async fn server_with_mock(mock: MockWebhookServiceImpl) -> (SiftMcpServer, JoinHandle<()>) { + let (client, server) = tokio::io::duplex(1024); + let channel = memory_sift_channel(client).await; + + let handle = tokio::spawn(async move { + Server::builder() + .add_service(WebhookServiceServer::new(mock)) + .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server))) + .await + .unwrap(); + }); + + ( + SiftMcpServer::new( + channel, + String::from("https://app.test.local"), + false, + false, + ), + handle, + ) +} + +#[tokio::test] +async fn list_webhooks_returns_redacted_webhooks() { + let mut mock = MockWebhookServiceImpl::new(); + mock.expect_list_webhooks() + .withf(|req| req.get_ref().filter == "is_archived == false") + .returning(|_| { + Ok(Response::new(ListWebhooksResponse { + webhooks: vec![Webhook { + webhook_id: "wh1".into(), + name: "slack alerts".into(), + target_url: "https://hooks.slack.com/services/T000/B000/secret".into(), + http_headers: vec![WebhookHttpHeader { + name: "Authorization".into(), + value: "Bearer secret".into(), + }], + payload: Some(r#"{"token": "secret"}"#.into()), + ..Default::default() + }], + next_page_token: String::new(), + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let result = server + .list_webhooks(list_params("is_archived == false", None)) + .await + .expect("list_webhooks failed"); + let body = structured(result); + + assert_eq!(body["count"], 1); + assert_eq!(body["has_more"], false); + let webhook = &body["webhooks"][0]; + assert_eq!(webhook["webhookId"], "wh1"); + assert_eq!(webhook["targetUrl"], "https://slack.com"); + assert_eq!( + webhook["httpHeaders"], + serde_json::json!([{ "name": "Authorization" }]) + ); + assert!(webhook.get("payload").is_none()); + assert!(!body.to_string().contains("secret")); +} + +#[tokio::test] +async fn list_webhooks_is_available_read_only() { + let mut mock = MockWebhookServiceImpl::new(); + mock.expect_list_webhooks().times(1).returning(|_| { + Ok(Response::new(ListWebhooksResponse { + webhooks: vec![], + next_page_token: String::new(), + })) + }); + + let (server, _h) = server_with_mock(mock).await; + + let result = server + .list_webhooks(list_params("", None)) + .await + .expect("list_webhooks should not need a write flag"); + + assert_eq!(structured(result)["count"], 0); +} diff --git a/rust/crates/sift_mcp/src/tool_events.json b/rust/crates/sift_mcp/src/tool_events.json index 7ceb2c4d90..ed4c768ba9 100644 --- a/rust/crates/sift_mcp/src/tool_events.json +++ b/rust/crates/sift_mcp/src/tool_events.json @@ -37,6 +37,7 @@ "list_user_defined_function_versions": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_LIST_USER_DEFINED_FUNCTION_VERSIONS", "list_user_defined_functions": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_LIST_USER_DEFINED_FUNCTIONS", "list_users": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_LIST_USERS", + "list_webhooks": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_LIST_WEBHOOKS", "ping": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_PING", "preview_rule": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_PREVIEW_RULE", "search_docs": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_SEARCH_DOCS", diff --git a/rust/crates/sift_test_util/src/mock/mod.rs b/rust/crates/sift_test_util/src/mock/mod.rs index ae0053c504..f4995f9c7d 100644 --- a/rust/crates/sift_test_util/src/mock/mod.rs +++ b/rust/crates/sift_test_util/src/mock/mod.rs @@ -15,6 +15,7 @@ pub mod runs; pub mod test_reports; pub mod user_defined_functions; pub mod users; +pub mod webhooks; /// A test demonstrating a little bit of everything of how to leverage the mock API. #[cfg(test)] diff --git a/rust/crates/sift_test_util/src/mock/webhooks/mod.rs b/rust/crates/sift_test_util/src/mock/webhooks/mod.rs new file mode 100644 index 0000000000..a3a6d96c3f --- /dev/null +++ b/rust/crates/sift_test_util/src/mock/webhooks/mod.rs @@ -0,0 +1 @@ +pub mod v1; diff --git a/rust/crates/sift_test_util/src/mock/webhooks/v1.rs b/rust/crates/sift_test_util/src/mock/webhooks/v1.rs new file mode 100644 index 0000000000..af9b1e0212 --- /dev/null +++ b/rust/crates/sift_test_util/src/mock/webhooks/v1.rs @@ -0,0 +1,98 @@ +use async_trait::async_trait; +use mockall::mock; +use sift_rs::webhooks::v1::{ + BatchCreateWebhookLogsRequest, BatchCreateWebhookLogsResponse, CreateWebhookRequest, + CreateWebhookResponse, CreateWebhookSignatureKeyRequest, CreateWebhookSignatureKeyResponse, + GetWebhookRequest, GetWebhookResponse, GetWebhookSignatureKeyRequest, + GetWebhookSignatureKeyResponse, ListWebhookLogsRequest, ListWebhookLogsResponse, + ListWebhooksRequest, ListWebhooksResponse, RotateWebhookSignatureKeyRequest, + RotateWebhookSignatureKeyResponse, TestWebhookRequest, TestWebhookResponse, + ToggleWebhookSignatureKeyActivationRequest, ToggleWebhookSignatureKeyActivationResponse, + UpdateWebhookRequest, UpdateWebhookResponse, webhook_service_server::WebhookService, +}; +use tonic::{Request, Response, Status}; + +mock! { + pub WebhookServiceImpl {} + + #[async_trait] + impl WebhookService for WebhookServiceImpl { + async fn get_webhook( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn create_webhook( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn update_webhook( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn list_webhooks( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn test_webhook( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn create_webhook_signature_key( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn get_webhook_signature_key( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn toggle_webhook_signature_key_activation( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn rotate_webhook_signature_key( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn batch_create_webhook_logs( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + async fn list_webhook_logs( + &self, + request: Request, + ) -> std::result::Result< + Response, + Status, + >; + } +} From 1a5796e1f278ec554b8b9373aba75a8c6a379518 Mon Sep 17 00:00:00 2001 From: Bill Raj-Derouin Date: Thu, 15 Oct 2026 10:57:17 -0700 Subject: [PATCH 2/3] rust(fix): keep archived webhooks a rule already calls and catch typed-only webhook actions - An action typed WEBHOOK now counts as a webhook action with or without a configuration. It is refused in rule_json and conditions_json, and replacing the set with webhook_ids drops one that names no webhook. - update_rule checks only newly attached webhooks for archived status, so passing a rule's current set back is always safe. The rule is fetched and merged before the check, which adds no call. - The ListWebhooksRequest and ListWebhookLogsRequest order_by comments now say an empty order_by sorts newest first, matching the API. --- go/gen/sift/webhooks/v1/webhooks.pb.go | 4 +- protos/sift/webhooks/v1/webhooks.proto | 4 +- python/lib/sift/webhooks/v1/webhooks_pb2.pyi | 4 +- rust/crates/sift_mcp/src/service/rules/mod.rs | 99 +++++-- .../crates/sift_mcp/src/service/rules/test.rs | 255 +++++++++++++----- rust/crates/sift_mcp/src/tool/rules/mod.rs | 88 ++++-- rust/crates/sift_mcp/src/tool/rules/test.rs | 117 +++++++- 7 files changed, 450 insertions(+), 121 deletions(-) diff --git a/go/gen/sift/webhooks/v1/webhooks.pb.go b/go/gen/sift/webhooks/v1/webhooks.pb.go index 0009740529..56c7b93529 100644 --- a/go/gen/sift/webhooks/v1/webhooks.pb.go +++ b/go/gen/sift/webhooks/v1/webhooks.pb.go @@ -906,7 +906,7 @@ type ListWebhooksRequest struct { Filter string `protobuf:"bytes,3,opt,name=filter,proto3" json:"filter,omitempty"` // How to order the retrieved webhooks. Formatted as a comma-separated string i.e. "FIELD_NAME[ desc],...". // Available field to order_by is `created_date`. - // If left empty, items are ordered by `created_date` in ascending order (oldest-first). + // If left empty, items are ordered by `created_date` in descending order (newest-first). // For more information about the format of this field, read [this](https://google.aip.dev/132#ordering) // Example: "created_date desc" OrderBy string `protobuf:"bytes,4,opt,name=order_by,json=orderBy,proto3" json:"order_by,omitempty"` @@ -1576,7 +1576,7 @@ type ListWebhookLogsRequest struct { Filter string `protobuf:"bytes,3,opt,name=filter,proto3" json:"filter,omitempty"` // How to order the retrieved webhook logs. Formatted as a comma-separated string i.e. "FIELD_NAME[ desc],...". // Available field to order_by is `created_date`. - // If left empty, items are ordered by `created_date` in ascending order (oldest-first). + // If left empty, items are ordered by `created_date` in descending order (newest-first). // For more information about the format of this field, read [this](https://google.aip.dev/132#ordering) // Example: "created_date desc" OrderBy string `protobuf:"bytes,4,opt,name=order_by,json=orderBy,proto3" json:"order_by,omitempty"` diff --git a/protos/sift/webhooks/v1/webhooks.proto b/protos/sift/webhooks/v1/webhooks.proto index c37c7b468f..0668aa6495 100644 --- a/protos/sift/webhooks/v1/webhooks.proto +++ b/protos/sift/webhooks/v1/webhooks.proto @@ -259,7 +259,7 @@ message ListWebhooksRequest { // How to order the retrieved webhooks. Formatted as a comma-separated string i.e. "FIELD_NAME[ desc],...". // Available field to order_by is `created_date`. - // If left empty, items are ordered by `created_date` in ascending order (oldest-first). + // If left empty, items are ordered by `created_date` in descending order (newest-first). // For more information about the format of this field, read [this](https://google.aip.dev/132#ordering) // Example: "created_date desc" string order_by = 4 [(google.api.field_behavior) = OPTIONAL]; @@ -351,7 +351,7 @@ message ListWebhookLogsRequest { // How to order the retrieved webhook logs. Formatted as a comma-separated string i.e. "FIELD_NAME[ desc],...". // Available field to order_by is `created_date`. - // If left empty, items are ordered by `created_date` in ascending order (oldest-first). + // If left empty, items are ordered by `created_date` in descending order (newest-first). // For more information about the format of this field, read [this](https://google.aip.dev/132#ordering) // Example: "created_date desc" string order_by = 4 [(google.api.field_behavior) = OPTIONAL]; diff --git a/python/lib/sift/webhooks/v1/webhooks_pb2.pyi b/python/lib/sift/webhooks/v1/webhooks_pb2.pyi index 2db1ee87e4..f7d4081430 100644 --- a/python/lib/sift/webhooks/v1/webhooks_pb2.pyi +++ b/python/lib/sift/webhooks/v1/webhooks_pb2.pyi @@ -395,7 +395,7 @@ class ListWebhooksRequest(google.protobuf.message.Message): order_by: builtins.str """How to order the retrieved webhooks. Formatted as a comma-separated string i.e. "FIELD_NAME[ desc],...". Available field to order_by is `created_date`. - If left empty, items are ordered by `created_date` in ascending order (oldest-first). + If left empty, items are ordered by `created_date` in descending order (newest-first). For more information about the format of this field, read [this](https://google.aip.dev/132#ordering) Example: "created_date desc" """ @@ -658,7 +658,7 @@ class ListWebhookLogsRequest(google.protobuf.message.Message): order_by: builtins.str """How to order the retrieved webhook logs. Formatted as a comma-separated string i.e. "FIELD_NAME[ desc],...". Available field to order_by is `created_date`. - If left empty, items are ordered by `created_date` in ascending order (oldest-first). + If left empty, items are ordered by `created_date` in descending order (newest-first). For more information about the format of this field, read [this](https://google.aip.dev/132#ordering) Example: "created_date desc" """ diff --git a/rust/crates/sift_mcp/src/service/rules/mod.rs b/rust/crates/sift_mcp/src/service/rules/mod.rs index ddc16e5151..8919d72230 100644 --- a/rust/crates/sift_mcp/src/service/rules/mod.rs +++ b/rust/crates/sift_mcp/src/service/rules/mod.rs @@ -13,7 +13,7 @@ use sift_rs::{ }; /// A partial set of changes to apply to an existing rule. Every field is -/// optional; `None` means "leave unchanged". This is what makes `update_rule` +/// optional; `None` means "leave unchanged". This is what makes an update /// safe: only the fields set here are touched, everything else on the rule is /// preserved across the full-replace `UpdateRule` RPC. #[derive(Debug, Default)] @@ -30,11 +30,16 @@ pub struct RuleUpdate { pub webhook_ids: Option>, } -/// The outcome of `update_rule`. +/// An update merged over the current rule and ready to save with +/// [`RuleService::save_update`]. Splitting the fetch from the save lets the +/// caller check the new webhooks against the ones the rule already calls +/// without fetching the rule twice. #[derive(Debug)] -pub struct UpdatedRule { - pub rule_id: String, - /// Webhooks the rule called before the update and no longer calls. +pub struct PendingRuleUpdate { + update: UpdateRuleRequest, + /// Webhooks the rule calls before the update. + pub current_webhook_ids: Vec, + /// Webhooks the rule calls before the update and will no longer call. pub detached_webhook_ids: Vec, } @@ -129,7 +134,7 @@ impl RuleService { mut update: UpdateRuleRequest, webhook_ids: Vec, ) -> Result { - set_webhooks(&mut update.conditions, &webhook_ids); + attach_webhooks(&mut update.conditions, &webhook_ids); let channel = self.channel.clone(); @@ -152,11 +157,15 @@ impl RuleService { Ok(resp.rule_id) } - /// Safely updates an existing rule by `rule_id`. The underlying `UpdateRule` - /// RPC is a full replace, so this fetches the current rule, overlays only the - /// fields set in `changes`, and sends the merged definition back. Fields not - /// named in `changes` are preserved. - pub async fn update_rule(&self, rule_id: String, changes: RuleUpdate) -> Result { + /// Fetches the rule and overlays only the fields set in `changes`. The + /// `UpdateRule` RPC is a full replace, so fields not named in `changes` are + /// carried over from the current rule. Nothing is written until + /// [`Self::save_update`]. + pub async fn prepare_update( + &self, + rule_id: String, + changes: RuleUpdate, + ) -> Result { let current = self .get_rule(rule_id.clone()) .await? @@ -201,13 +210,33 @@ impl RuleService { if let Some(conditions) = conditions { update.conditions = conditions; } - let webhook_ids = webhook_ids.unwrap_or_else(|| current_webhook_ids.clone()); - set_webhooks(&mut update.conditions, &webhook_ids); - let detached_webhook_ids = current_webhook_ids - .into_iter() - .filter(|id| !webhook_ids.contains(id)) - .collect(); + let detached_webhook_ids = match webhook_ids { + Some(webhook_ids) => { + set_webhooks(&mut update.conditions, &webhook_ids); + current_webhook_ids + .iter() + .filter(|id| !webhook_ids.contains(id)) + .cloned() + .collect() + } + // Re-attaches the current webhooks when `conditions` replaced the + // actions they sat on, and changes nothing otherwise. + None => { + attach_webhooks(&mut update.conditions, ¤t_webhook_ids); + Vec::new() + } + }; + + Ok(PendingRuleUpdate { + update, + current_webhook_ids, + detached_webhook_ids, + }) + } + /// Saves an update from [`Self::prepare_update`]. Returns the `rule_id`. + pub async fn save_update(&self, pending: PendingRuleUpdate) -> Result { + let update = pending.update; let channel = self.channel.clone(); let resp = with_retry(&self.policy, move || { let channel = channel.clone(); @@ -223,10 +252,7 @@ impl RuleService { .await .context("failed to update rule")?; - Ok(UpdatedRule { - rule_id: resp.rule_id, - detached_webhook_ids, - }) + Ok(resp.rule_id) } /// Retrieves the latest version of a rule by id, or `None` if it does not exist. @@ -367,7 +393,7 @@ impl RuleService { /// Converts a fetched [`Rule`] into the [`UpdateRuleRequest`] shape so it can be /// resent through the full-replace `UpdateRule` RPC. Every meaningful field is -/// carried over verbatim; this is the base that `update_rule` overlays changes +/// carried over verbatim; this is the base that `prepare_update` overlays changes /// onto so unspecified fields survive the round-trip. fn rule_to_update_request(rule: Rule) -> UpdateRuleRequest { UpdateRuleRequest { @@ -417,16 +443,25 @@ fn action_to_update(action: RuleAction) -> UpdateActionRequest { /// Makes every condition call exactly `webhook_ids`. Sift fires a webhook from /// the condition that holds it and allows one condition per rule, so every -/// condition is that one. An action for a kept webhook keeps its -/// `rule_action_id`, since Sift deletes any action an update leaves out, and a -/// repeated id is attached once because Sift rejects the rule otherwise. +/// condition is that one. Every other webhook action is dropped, including one +/// typed `WEBHOOK` that names no webhook, which Sift saves but cannot call. /// Non-webhook actions are left alone. fn set_webhooks(conditions: &mut [UpdateConditionRequest], webhook_ids: &[String]) { - for condition in conditions { + for condition in conditions.iter_mut() { condition.actions.retain(|action| { - webhook_id(action).is_none_or(|id| webhook_ids.iter().any(|w| w == id)) + !is_webhook_action(action) + || webhook_id(action).is_some_and(|id| webhook_ids.iter().any(|w| w == id)) }); + } + attach_webhooks(conditions, webhook_ids); +} +/// Attaches each of `webhook_ids` that a condition does not already call. An +/// action for a webhook already attached keeps its `rule_action_id`, since +/// Sift deletes any action an update leaves out, and a repeated id is attached +/// once because Sift rejects the rule otherwise. +fn attach_webhooks(conditions: &mut [UpdateConditionRequest], webhook_ids: &[String]) { + for condition in conditions { for id in webhook_ids { let attached = condition .actions @@ -445,7 +480,15 @@ pub fn has_webhook_action(conditions: &[UpdateConditionRequest]) -> bool { conditions .iter() .flat_map(|condition| &condition.actions) - .any(|action| webhook_id(action).is_some()) + .any(is_webhook_action) +} + +/// Whether an action is a webhook action by its type or its configuration. +/// Sift checks a configuration only when one is set, so an action typed +/// `WEBHOOK` with no configuration still saves, as a webhook action with no +/// webhook. +fn is_webhook_action(action: &UpdateActionRequest) -> bool { + action.action_type == ActionKind::Webhook as i32 || webhook_id(action).is_some() } /// The webhooks the conditions call, in order, each once. diff --git a/rust/crates/sift_mcp/src/service/rules/test.rs b/rust/crates/sift_mcp/src/service/rules/test.rs index a1db896f40..238488ad25 100644 --- a/rust/crates/sift_mcp/src/service/rules/test.rs +++ b/rust/crates/sift_mcp/src/service/rules/test.rs @@ -33,6 +33,19 @@ async fn service_with_mock(mock: MockRuleServiceImpl) -> (RuleService, JoinHandl ) } +/// Prepares and saves an update the way `update_rule` does, returning the +/// saved rule id and the webhooks the update detached. +async fn update_rule( + service: &RuleService, + rule_id: &str, + changes: RuleUpdate, +) -> anyhow::Result<(String, Vec)> { + let pending = service.prepare_update(rule_id.to_string(), changes).await?; + let detached = pending.detached_webhook_ids.clone(); + let rule_id = service.save_update(pending).await?; + Ok((rule_id, detached)) +} + #[tokio::test] async fn list_rules_returns_single_page() { let mut mock = MockRuleServiceImpl::new(); @@ -326,19 +339,19 @@ async fn update_rule_merges_changes_over_current_definition() { let (service, _h) = service_with_mock(mock).await; - let updated = service - .update_rule( - "rule-1".to_string(), - RuleUpdate { - name: Some("new name".to_string()), - ..Default::default() - }, - ) - .await - .expect("update_rule failed"); + let updated = update_rule( + &service, + "rule-1", + RuleUpdate { + name: Some("new name".to_string()), + ..Default::default() + }, + ) + .await + .expect("update_rule failed"); - assert_eq!(updated.rule_id, "rule-1"); - assert!(updated.detached_webhook_ids.is_empty()); + assert_eq!(updated.0, "rule-1"); + assert!(updated.1.is_empty()); } #[tokio::test] @@ -349,16 +362,16 @@ async fn update_rule_errors_when_rule_missing() { let (service, _h) = service_with_mock(mock).await; - let err = service - .update_rule( - "missing".to_string(), - RuleUpdate { - name: Some("x".to_string()), - ..Default::default() - }, - ) - .await - .expect_err("expected error"); + let err = update_rule( + &service, + "missing", + RuleUpdate { + name: Some("x".to_string()), + ..Default::default() + }, + ) + .await + .expect_err("expected error"); assert!(err.to_string().contains("not found")); } @@ -568,18 +581,18 @@ async fn update_rule_keeps_existing_actions_when_webhooks_are_replaced() { let (service, _h) = service_with_mock(mock).await; - let updated = service - .update_rule( - "rule-1".to_string(), - RuleUpdate { - webhook_ids: Some(vec!["wh-old".into(), "wh-new".into(), "wh-new".into()]), - ..Default::default() - }, - ) - .await - .expect("update_rule failed"); + let updated = update_rule( + &service, + "rule-1", + RuleUpdate { + webhook_ids: Some(vec!["wh-old".into(), "wh-new".into(), "wh-new".into()]), + ..Default::default() + }, + ) + .await + .expect("update_rule failed"); - assert!(updated.detached_webhook_ids.is_empty()); + assert!(updated.1.is_empty()); } #[tokio::test] @@ -606,18 +619,18 @@ async fn update_rule_reports_the_webhooks_it_detaches() { let (service, _h) = service_with_mock(mock).await; - let updated = service - .update_rule( - "rule-1".to_string(), - RuleUpdate { - webhook_ids: Some(Vec::new()), - ..Default::default() - }, - ) - .await - .expect("update_rule failed"); + let updated = update_rule( + &service, + "rule-1", + RuleUpdate { + webhook_ids: Some(Vec::new()), + ..Default::default() + }, + ) + .await + .expect("update_rule failed"); - assert_eq!(updated.detached_webhook_ids, vec!["wh-old".to_string()]); + assert_eq!(updated.1, vec!["wh-old".to_string()]); } #[tokio::test] @@ -646,25 +659,25 @@ async fn update_rule_carries_webhooks_onto_replaced_conditions() { let (service, _h) = service_with_mock(mock).await; - let updated = service - .update_rule( - "rule-1".to_string(), - RuleUpdate { - conditions: Some(vec![UpdateConditionRequest { - actions: vec![UpdateActionRequest { - action_type: ActionKind::Annotation as i32, - configuration: Some(annotation_config()), - ..Default::default() - }], + let updated = update_rule( + &service, + "rule-1", + RuleUpdate { + conditions: Some(vec![UpdateConditionRequest { + actions: vec![UpdateActionRequest { + action_type: ActionKind::Annotation as i32, + configuration: Some(annotation_config()), ..Default::default() - }]), + }], ..Default::default() - }, - ) - .await - .expect("update_rule failed"); + }]), + ..Default::default() + }, + ) + .await + .expect("update_rule failed"); - assert!(updated.detached_webhook_ids.is_empty()); + assert!(updated.1.is_empty()); } #[test] @@ -691,3 +704,125 @@ fn has_webhook_action_finds_only_webhooks() { ))); assert!(super::has_webhook_action(&[annotation, webhook])); } + +/// `rule_with_webhook` plus a webhook action that has a type but no +/// configuration, which Sift saves as a webhook action with no webhook. +fn rule_with_unconfigured_webhook() -> Rule { + let mut rule = rule_with_webhook(); + rule.conditions[0].actions.push(RuleAction { + rule_action_id: "act-wh-empty".into(), + action_type: ActionKind::Webhook as i32, + ..Default::default() + }); + rule +} + +#[tokio::test] +async fn update_rule_replacing_webhooks_drops_one_with_no_webhook() { + let mut mock = MockRuleServiceImpl::new(); + mock.expect_get_rule().returning(|_| { + Ok(Response::new(GetRuleResponse { + rule: Some(rule_with_unconfigured_webhook()), + })) + }); + mock.expect_update_rule() + .withf(|req| { + let actions = &req.get_ref().conditions[0].actions; + actions.len() == 2 + && actions + .iter() + .all(|a| a.rule_action_id.as_deref() != Some("act-wh-empty")) + && webhook_ids(actions) == vec!["wh-old"] + }) + .returning(|_| { + Ok(Response::new(UpdateRuleResponse { + rule_id: "rule-1".into(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + update_rule( + &service, + "rule-1", + RuleUpdate { + webhook_ids: Some(vec!["wh-old".into()]), + ..Default::default() + }, + ) + .await + .expect("update_rule failed"); +} + +#[tokio::test] +async fn update_rule_without_webhook_ids_leaves_actions_alone() { + let mut mock = MockRuleServiceImpl::new(); + mock.expect_get_rule().returning(|_| { + Ok(Response::new(GetRuleResponse { + rule: Some(rule_with_unconfigured_webhook()), + })) + }); + mock.expect_update_rule() + .withf(|req| { + let u = req.get_ref(); + u.name == "renamed" && u.conditions[0].actions.len() == 3 + }) + .returning(|_| { + Ok(Response::new(UpdateRuleResponse { + rule_id: "rule-1".into(), + })) + }); + + let (service, _h) = service_with_mock(mock).await; + + update_rule( + &service, + "rule-1", + RuleUpdate { + name: Some("renamed".into()), + ..Default::default() + }, + ) + .await + .expect("update_rule failed"); +} + +#[tokio::test] +async fn prepare_update_reports_current_webhooks_without_writing() { + let mut mock = MockRuleServiceImpl::new(); + mock.expect_get_rule().returning(|_| { + Ok(Response::new(GetRuleResponse { + rule: Some(rule_with_webhook()), + })) + }); + mock.expect_update_rule().times(0); + + let (service, _h) = service_with_mock(mock).await; + + let pending = service + .prepare_update( + "rule-1".to_string(), + RuleUpdate { + webhook_ids: Some(vec!["wh-new".into()]), + ..Default::default() + }, + ) + .await + .expect("prepare_update failed"); + + assert_eq!(pending.current_webhook_ids, vec!["wh-old".to_string()]); + assert_eq!(pending.detached_webhook_ids, vec!["wh-old".to_string()]); +} + +#[test] +fn has_webhook_action_counts_a_type_with_no_configuration() { + let typed_only = UpdateConditionRequest { + actions: vec![UpdateActionRequest { + action_type: ActionKind::Webhook as i32, + ..Default::default() + }], + ..Default::default() + }; + + assert!(super::has_webhook_action(&[typed_only])); +} diff --git a/rust/crates/sift_mcp/src/tool/rules/mod.rs b/rust/crates/sift_mcp/src/tool/rules/mod.rs index 911fea095a..265ab2479f 100644 --- a/rust/crates/sift_mcp/src/tool/rules/mod.rs +++ b/rust/crates/sift_mcp/src/tool/rules/mod.rs @@ -269,7 +269,7 @@ impl SiftMcpServer { let update = parse_rule_definition(&rule_json)?; validate_conditions("rule_json", &update.conditions)?; let webhook_ids = webhook_ids.unwrap_or_default(); - let webhooks = self.resolve_webhooks(&webhook_ids).await?; + let webhooks = self.resolve_webhooks(&webhook_ids, &[]).await?; let rule_name = update.name.clone(); let rule_id = self @@ -321,14 +321,15 @@ impl SiftMcpServer { existing conditions unchanged — only provide it when you intend to rewrite the rule's logic. - `webhook_ids`: optional. Replaces the rule's set of webhooks, from `list_webhooks`, that it calls when its condition is violated and again when it resolves. Include the ids it already calls - (`conditions[].actions[].configuration.webhook.webhook_id` in `list_rules`) to keep them; `[]` - removes them all. Omit it to keep the current webhooks, even when `conditions_json` is set. + (`conditions[].actions[].configuration.webhook.webhook_id` in `list_rules`), archived ones too, to + keep them; `[]` removes them all. Omit it to keep the current webhooks, even when + `conditions_json` is set. - At least one field besides `rule_id` must be set. Errors: - `INVALID_PARAMS` if no field besides `rule_id` is set, if `conditions_json` is not a valid - conditions array of exactly one condition or holds a webhook action, or if a webhook is - archived. + conditions array of exactly one condition or holds a webhook action, or if a webhook the rule + does not already call is archived. - `RESOURCE_NOT_FOUND` if no rule has the given `rule_id`, or a `webhook_ids` entry matches no webhook. - `INTERNAL_ERROR` for upstream gRPC failures. @@ -394,17 +395,26 @@ impl SiftMcpServer { )); } - let webhooks = match &changes.webhook_ids { - Some(webhook_ids) => self.resolve_webhooks(webhook_ids).await?, + let webhook_ids = changes.webhook_ids.clone(); + let rule_name = changes.name.clone(); + let pending = self + .rule_service + .prepare_update(rule_id, changes) + .await + .map_err(from_anyhow)?; + let webhooks = match &webhook_ids { + Some(webhook_ids) => { + self.resolve_webhooks(webhook_ids, &pending.current_webhook_ids) + .await? + } None => Vec::new(), }; - let rule_name = changes.name.clone(); - let updated = self + let detached_webhook_ids = pending.detached_webhook_ids.clone(); + let rule_id = self .rule_service - .update_rule(rule_id, changes) + .save_update(pending) .await .map_err(from_anyhow)?; - let rule_id = updated.rule_id; let rule_url = self.url_service.build_rule_url(&rule_id).ok(); let next_step = format!( @@ -413,7 +423,7 @@ impl SiftMcpServer { (filter `rule_id == \"{rule_id}\"`).", url_clause("rule", rule_name.as_deref(), rule_url.as_deref()), webhook_clause("It now calls", &webhooks), - self.detached_clause(&updated.detached_webhook_ids).await, + self.detached_clause(&detached_webhook_ids).await, ); let mut result = CallToolResult::structured(serde_json::json!({ @@ -549,16 +559,22 @@ impl SiftMcpServer { } impl SiftMcpServer { - /// Confirms each webhook exists and is live before a rule is pointed at it. - /// Sift accepts an archived webhook on a rule and then never calls it, so - /// the check has to happen here. - async fn resolve_webhooks(&self, webhook_ids: &[String]) -> Result, ErrorData> { + /// Confirms each webhook exists, and that each one not in `already_attached` + /// is live, before a rule is pointed at it. Sift accepts an archived webhook + /// on a rule and then never calls it, so the check has to happen here. An + /// archived webhook the rule already calls may stay, so passing the rule's + /// current set back is always safe. + async fn resolve_webhooks( + &self, + webhook_ids: &[String], + already_attached: &[String], + ) -> Result, ErrorData> { let webhooks = self .webhook_service .get_webhooks(webhook_ids) .await .map_err(from_anyhow)?; - check_webhooks(webhook_ids, webhooks) + check_webhooks(webhook_ids, webhooks, already_attached) } /// A `next_step` sentence naming the webhooks an update detached, so a set @@ -618,10 +634,12 @@ fn validate_conditions( } /// Fails with `RESOURCE_NOT_FOUND` when an id matches no webhook and with -/// `INVALID_PARAMS` when a webhook is archived. Returns the webhooks otherwise. +/// `INVALID_PARAMS` when a webhook outside `already_attached` is archived. +/// Returns the webhooks otherwise. fn check_webhooks( webhook_ids: &[String], webhooks: Vec, + already_attached: &[String], ) -> Result, ErrorData> { let missing: Vec = webhook_ids .iter() @@ -640,7 +658,7 @@ fn check_webhooks( let archived: Vec = webhooks .iter() - .filter(|w| w.is_archived || w.archived_date.is_some()) + .filter(|w| is_archived(w) && !already_attached.contains(&w.webhook_id)) .map(|w| format!("`{}` ({})", w.name, w.webhook_id)) .collect(); if !archived.is_empty() { @@ -657,12 +675,34 @@ fn check_webhooks( Ok(webhooks) } -/// A `next_step` sentence naming the webhooks a rule calls, so the agent can -/// tell the user where alerts go. Empty when there are none. +fn is_archived(webhook: &Webhook) -> bool { + webhook.is_archived || webhook.archived_date.is_some() +} + +/// `next_step` sentences naming the webhooks a rule calls, so the agent can +/// tell the user where alerts go, and any archived ones it keeps, which Sift +/// never calls. Empty when there are none. fn webhook_clause(lead: &str, webhooks: &[Webhook]) -> String { - if webhooks.is_empty() { - return String::new(); + let (archived, live): (Vec<&Webhook>, Vec<&Webhook>) = + webhooks.iter().partition(|w| is_archived(w)); + let mut clause = String::new(); + if !live.is_empty() { + clause.push_str(&format!( + " {lead} {} when its condition is violated and again when it resolves.", + webhook_names(&live) + )); } + if !archived.is_empty() { + clause.push_str(&format!( + " It still lists archived {}, which Sift never calls.", + webhook_names(&archived) + )); + } + clause +} + +/// "webhook `a`" or "webhooks `a`, `b`". +fn webhook_names(webhooks: &[&Webhook]) -> String { let names = webhooks .iter() .map(|w| format!("`{}`", w.name)) @@ -673,7 +713,7 @@ fn webhook_clause(lead: &str, webhooks: &[Webhook]) -> String { } else { "webhooks" }; - format!(" {lead} {noun} {names} when its condition is violated and again when it resolves.") + format!("{noun} {names}") } /// Build the web URL for a rule when its `rule_id` is known. Returns `None` when diff --git a/rust/crates/sift_mcp/src/tool/rules/test.rs b/rust/crates/sift_mcp/src/tool/rules/test.rs index 85870479ef..4e230b2dda 100644 --- a/rust/crates/sift_mcp/src/tool/rules/test.rs +++ b/rust/crates/sift_mcp/src/tool/rules/test.rs @@ -159,7 +159,7 @@ fn parse_rule_definition_accepts_the_documented_annotation_action() { #[test] fn check_webhooks_accepts_live_webhooks() { - let webhooks = check_webhooks(&["wh1".to_string()], vec![webhook("wh1", "alerts")]) + let webhooks = check_webhooks(&["wh1".to_string()], vec![webhook("wh1", "alerts")], &[]) .expect("should accept"); assert_eq!(webhooks[0].name, "alerts"); } @@ -169,6 +169,7 @@ fn check_webhooks_rejects_an_unknown_id() { let err = check_webhooks( &["wh1".to_string(), "wh-missing".to_string()], vec![webhook("wh1", "alerts")], + &[], ) .expect_err("should reject"); assert_eq!(err.code, ErrorCode::RESOURCE_NOT_FOUND); @@ -181,7 +182,7 @@ fn check_webhooks_rejects_an_archived_webhook() { is_archived: true, ..webhook("wh1", "old alerts") }; - let err = check_webhooks(&["wh1".to_string()], vec![archived]).expect_err("should reject"); + let err = check_webhooks(&["wh1".to_string()], vec![archived], &[]).expect_err("should reject"); assert_eq!(err.code, ErrorCode::INVALID_PARAMS); assert!(err.message.contains("old alerts")); } @@ -377,7 +378,15 @@ async fn update_rule_replaces_webhooks_and_names_the_detached_one() { #[tokio::test] async fn update_rule_refuses_an_archived_webhook_before_writing() { let mut rule_mock = MockRuleServiceImpl::new(); - rule_mock.expect_get_rule().times(0); + rule_mock.expect_get_rule().returning(|_| { + Ok(Response::new(GetRuleResponse { + rule: Some(Rule { + rule_id: "rule-1".into(), + conditions: vec![RuleCondition::default()], + ..Default::default() + }), + })) + }); rule_mock.expect_update_rule().times(0); let archived = Webhook { @@ -422,3 +431,105 @@ async fn update_rule_refuses_empty_conditions_json() { assert_eq!(err.code, ErrorCode::INVALID_PARAMS); } + +#[test] +fn check_webhooks_keeps_an_archived_webhook_already_attached() { + let archived = Webhook { + is_archived: true, + ..webhook("wh1", "old alerts") + }; + let webhooks = check_webhooks(&["wh1".to_string()], vec![archived], &["wh1".to_string()]) + .expect("an archived webhook the rule already calls may stay"); + assert_eq!(webhooks.len(), 1); +} + +#[tokio::test] +async fn create_rule_refuses_a_webhook_action_with_no_configuration() { + let (server, _h) = + server_with_mocks(MockRuleServiceImpl::new(), MockWebhookServiceImpl::new()).await; + + let rule_json = RULE_JSON.replacen( + r#""actions": ["#, + r#""actions": [{ "action_type": "WEBHOOK" }, "#, + 1, + ); + let err = server + .create_rule(Parameters(RuleDefinitionParams { + rule_json, + webhook_ids: None, + })) + .await + .expect_err("should refuse"); + + assert_eq!(err.code, ErrorCode::INVALID_PARAMS); +} + +#[tokio::test] +async fn update_rule_keeps_an_archived_webhook_the_rule_already_calls() { + let mut rule_mock = MockRuleServiceImpl::new(); + rule_mock.expect_get_rule().returning(|_| { + Ok(Response::new(GetRuleResponse { + rule: Some(Rule { + rule_id: "rule-1".into(), + conditions: vec![RuleCondition { + actions: vec![RuleAction { + rule_action_id: "act-old".into(), + action_type: ActionKind::Webhook as i32, + configuration: Some(RuleActionConfiguration { + configuration: Some(Configuration::Webhook( + WebhookActionConfiguration { + webhook_id: "wh-old".into(), + }, + )), + }), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + }), + })) + }); + rule_mock + .expect_update_rule() + .withf(|req| { + let actions = &req.get_ref().conditions[0].actions; + actions.len() == 2 && actions[0].rule_action_id.as_deref() == Some("act-old") + }) + .returning(|_| { + Ok(Response::new(UpdateRuleResponse { + rule_id: "rule-1".into(), + })) + }); + + let archived = Webhook { + is_archived: true, + ..webhook("wh-old", "old alerts") + }; + let (server, _h) = server_with_mocks( + rule_mock, + webhook_mock_returning(vec![archived, webhook("wh-new", "new alerts")]), + ) + .await; + + let result = server + .update_rule(update_params(Some(vec![ + "wh-old".to_string(), + "wh-new".to_string(), + ]))) + .await + .expect("update_rule failed"); + + let next_step = structured(result)["next_step"] + .as_str() + .unwrap() + .to_string(); + assert!( + next_step.contains("now calls webhook `new alerts`"), + "{next_step}" + ); + assert!( + next_step.contains("still lists archived webhook `old alerts`"), + "{next_step}" + ); +} From 43dbad853b63d5680ab3aed7aaaa8e33fa5b98fd Mon Sep 17 00:00:00 2001 From: Bill Raj-Derouin Date: Thu, 15 Oct 2026 11:07:07 -0700 Subject: [PATCH 3/3] python(chore): add a chore fragment for the webhooks stub comment fix --- python/changelog.d/837.chore.md | 0 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 python/changelog.d/837.chore.md diff --git a/python/changelog.d/837.chore.md b/python/changelog.d/837.chore.md new file mode 100644 index 0000000000..e69de29bb2