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/changelog.d/837.chore.md b/python/changelog.d/837.chore.md new file mode 100644 index 0000000000..e69de29bb2 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_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..8919d72230 100644 --- a/rust/crates/sift_mcp/src/service/rules/mod.rs +++ b/rust/crates/sift_mcp/src/service/rules/mod.rs @@ -4,15 +4,16 @@ 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, }, }; /// 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)] @@ -24,6 +25,22 @@ 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>, +} + +/// 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 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, } #[cfg(test)] @@ -110,8 +127,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 { + attach_webhooks(&mut update.conditions, &webhook_ids); + let channel = self.channel.clone(); let resp = with_retry(&self.policy, move || { @@ -133,17 +157,22 @@ 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. Returns the `rule_id`. - 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? .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 +182,7 @@ impl RuleService { asset_ids, tag_ids, conditions, + webhook_ids, } = changes; if let Some(name) = name { @@ -180,7 +210,33 @@ impl RuleService { if let Some(conditions) = conditions { update.conditions = conditions; } + 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(); @@ -337,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 { @@ -384,3 +440,88 @@ 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. 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.iter_mut() { + condition.actions.retain(|action| { + !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 + .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(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. +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..238488ad25 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; @@ -30,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(); @@ -266,11 +282,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,18 +339,19 @@ async fn update_rule_merges_changes_over_current_definition() { let (service, _h) = service_with_mock(mock).await; - let rule_id = 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!(rule_id, "rule-1"); + assert_eq!(updated.0, "rule-1"); + assert!(updated.1.is_empty()); } #[tokio::test] @@ -342,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")); } @@ -422,9 +442,387 @@ 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 = 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.1.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 = update_rule( + &service, + "rule-1", + RuleUpdate { + webhook_ids: Some(Vec::new()), + ..Default::default() + }, + ) + .await + .expect("update_rule failed"); + + assert_eq!(updated.1, 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 = 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() + }]), + ..Default::default() + }, + ) + .await + .expect("update_rule failed"); + + assert!(updated.1.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])); +} + +/// `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/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..265ab2479f 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,26 @@ 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`), 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, 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 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. 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 +358,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 +377,7 @@ impl SiftMcpServer { asset_ids, tag_ids, conditions, + webhook_ids, }; let nothing_to_change = changes.name.is_none() @@ -354,7 +386,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 +395,35 @@ impl SiftMcpServer { )); } + 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 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_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(&detached_webhook_ids).await, ); let mut result = CallToolResult::structured(serde_json::json!({ @@ -509,6 +558,164 @@ impl SiftMcpServer { } } +impl SiftMcpServer { + /// 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, already_attached) + } + + /// 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 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() + .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| is_archived(w) && !already_attached.contains(&w.webhook_id)) + .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) +} + +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 { + 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)) + .collect::>() + .join(", "); + let noun = if webhooks.len() == 1 { + "webhook" + } else { + "webhooks" + }; + format!("{noun} {names}") +} + /// 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..4e230b2dda 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,391 @@ 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().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 { + 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); +} + +#[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}" + ); +} 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, + >; + } +}