From a87d9ddc2940ec6bec9cf88f090fba3ea6f10146 Mon Sep 17 00:00:00 2001 From: Miles Feldstein Date: Sat, 25 Jul 2026 02:56:36 -0400 Subject: [PATCH] feat(integrations): add Slack Connect channel bridge Signed-off-by: Miles Feldstein --- Cargo.lock | 22 + Cargo.toml | 1 + README.md | 5 + examples/slack-connect-bridge/Cargo.toml | 27 + examples/slack-connect-bridge/README.md | 194 ++++ .../slack-connect-bridge/bridge.example.json | 13 + .../slack-app-manifest.yaml | 31 + examples/slack-connect-bridge/src/bridge.rs | 959 ++++++++++++++++++ examples/slack-connect-bridge/src/config.rs | 230 +++++ examples/slack-connect-bridge/src/main.rs | 65 ++ .../slack-connect-bridge/src/slack/api.rs | 242 +++++ .../slack-connect-bridge/src/slack/mod.rs | 113 +++ .../slack-connect-bridge/src/slack/webhook.rs | 543 ++++++++++ examples/slack-connect-bridge/src/state.rs | 417 ++++++++ 14 files changed, 2862 insertions(+) create mode 100644 examples/slack-connect-bridge/Cargo.toml create mode 100644 examples/slack-connect-bridge/README.md create mode 100644 examples/slack-connect-bridge/bridge.example.json create mode 100644 examples/slack-connect-bridge/slack-app-manifest.yaml create mode 100644 examples/slack-connect-bridge/src/bridge.rs create mode 100644 examples/slack-connect-bridge/src/config.rs create mode 100644 examples/slack-connect-bridge/src/main.rs create mode 100644 examples/slack-connect-bridge/src/slack/api.rs create mode 100644 examples/slack-connect-bridge/src/slack/mod.rs create mode 100644 examples/slack-connect-bridge/src/slack/webhook.rs create mode 100644 examples/slack-connect-bridge/src/state.rs diff --git a/Cargo.lock b/Cargo.lock index 9d0190868d..6752ae899a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1230,6 +1230,28 @@ dependencies = [ "uuid", ] +[[package]] +name = "buzz-slack-connect-bridge" +version = "0.1.0" +dependencies = [ + "anyhow", + "axum", + "buzz-sdk", + "buzz-ws-client", + "hex", + "hmac 0.13.0", + "nostr", + "reqwest 0.13.4", + "serde", + "serde_json", + "sha2 0.11.0", + "tempfile", + "tokio", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "buzz-test-client" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 3499285f91..7ea4952690 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,6 +27,7 @@ members = [ "crates/buzz-relay-mesh", "crates/buzz-dev-mcp", "examples/countdown-bot", + "examples/slack-connect-bridge", ] exclude = ["desktop/src-tauri"] resolver = "2" diff --git a/README.md b/README.md index 89a131e73e..906f9d1784 100644 --- a/README.md +++ b/README.md @@ -158,6 +158,11 @@ Want a single-node / VPS relay instead of the local-dev stack? Use the productio For agents, set `BUZZ_PRIVATE_KEY` and use [`buzz-cli`](crates/buzz-cli) — JSON in, JSON out, designed for LLM tool calls. +Need to keep a customer or partner channel in Slack Connect during adoption? +The operator-run [Slack Connect bridge](examples/slack-connect-bridge) mirrors +live text messages and threads across explicit channel pairs without putting +Slack credentials in the relay. + --- ## Windows prerequisites diff --git a/examples/slack-connect-bridge/Cargo.toml b/examples/slack-connect-bridge/Cargo.toml new file mode 100644 index 0000000000..33771d7549 --- /dev/null +++ b/examples/slack-connect-bridge/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "buzz-slack-connect-bridge" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +publish = false + +[dependencies] +anyhow.workspace = true +axum.workspace = true +buzz-sdk.workspace = true +buzz-ws-client.workspace = true +hex.workspace = true +hmac.workspace = true +nostr.workspace = true +reqwest.workspace = true +serde.workspace = true +serde_json.workspace = true +sha2.workspace = true +tokio.workspace = true +tracing.workspace = true +tracing-subscriber.workspace = true +uuid.workspace = true + +[dev-dependencies] +tempfile = "3" diff --git a/examples/slack-connect-bridge/README.md b/examples/slack-connect-bridge/README.md new file mode 100644 index 0000000000..03a2dde045 --- /dev/null +++ b/examples/slack-connect-bridge/README.md @@ -0,0 +1,194 @@ +# Slack Connect Bridge + +An operator-run reference bridge for teams adopting Buzz while customers or +partners still collaborate in Slack Connect. + +It mirrors live text messages and threads between explicitly mapped channel +pairs: + +```text +Slack Connect channel ←→ bridge bot identities ←→ Buzz channel +``` + +The bridge is deliberately outside `buzz-relay`. Slack credentials remain in a +separate process, Buzz keeps using signed events, and operators decide exactly +which channel pairs cross the organizational boundary. + +## Safety model + +- Every route is an explicit `(Slack team ID, Slack channel ID, Buzz channel + UUID)` mapping. Wildcard or name-based routes do not exist. +- The Slack bot token must belong to the configured Slack workspace. +- Each Slack channel must report `is_ext_shared=true` at startup. Local Slack + channels are rejected unless `allow_non_shared_channels` is explicitly set. +- `channel_unshared` pauses the corresponding route. `channel_shared` resumes + it, and `channel_id_changed` follows Slack's private-to-shared ID migration. +- Slack callback signatures use HMAC-SHA256 and a five-minute replay window. +- Slack-origin messages are signed by the bridge's Buzz key and labeled + `Name · Slack`; Buzz-origin messages are posted by the Slack app and labeled + `Name · Buzz`. The bridge never impersonates users. +- Buzz text is escaped before entering Slack, so strings such as `` + cannot become cross-organization mass mentions. +- The listener defaults to loopback. Put it behind a TLS reverse proxy; never + expose plaintext webhook traffic to the internet. +- Secrets come from environment variables, not the JSON mapping file. + +## Supported behavior + +| Behavior | Support | +| --- | --- | +| New text messages | Two-way | +| Replies whose root was bridged | Two-way, preserved as threads | +| Reply with a pre-bridge root | Mirrored at channel level with a visible warning | +| Slack Connect channel ID changes | Followed durably | +| Slack unshare/reshare | Route paused/resumed | +| Slack webhook retries | Idempotent through durable timestamp ↔ event-ID mappings | +| Buzz reconnect/replay | Bounded replay with durable deduplication | +| Files, edits, deletes, reactions, DMs, huddles | Not in this focused reference slice | + +This is a live coexistence bridge, not a history importer. For workspace +migration, see the separate Slack import work in +[`block/buzz#2704`](https://github.com/block/buzz/pull/2704). + +## 1. Create the Slack app + +1. Copy [`slack-app-manifest.yaml`](slack-app-manifest.yaml). +2. Replace `YOUR_BRIDGE_HOST` with the public TLS hostname that forwards to the + bridge listener. +3. Create an app from the manifest in the Slack API dashboard. +4. Install it in the workspace from whose perspective you access the Slack + Connect channels. +5. Invite the app to every mapped shared channel. + +The requested scopes are intentionally narrow: + +- `chat:write` — mirror Buzz messages; +- `channels:history` / `groups:history` — receive new messages in public and + private channels the app can access; +- `channels:read` / `groups:read` — validate public/private shared channels and + receive share lifecycle events; +- `users:read` — display external members' names. Email access is not requested. + +Copy the app's bot token and signing secret into your secret manager. Do not put +them in the bridge JSON. + +## 2. Prepare the Buzz identity + +Generate a dedicated Nostr key for the bridge. On an open relay, the bridge +best-effort self-adds to mapped public channels as a bot. On a private channel, +an owner/admin must add the bridge public key before startup. + +For a closed relay, either: + +- admit the bridge public key as a standalone relay member; or +- enable NIP-OA (`BUZZ_ALLOW_NIP_OA_AUTH=true`) and provide a `BUZZ_AUTH_TAG` + valid for the bridge key. + +The bridge publishes a kind `0` profile named **Slack Connect Bridge** and uses +ordinary kind `9` channel messages. It adds provenance tags for loop prevention +and auditability; it does not add a new Buzz event kind. + +## 3. Configure channel pairs + +Copy [`bridge.example.json`](bridge.example.json) and replace every placeholder. +Use IDs, never names: Slack Connect channel names can differ between connected +workspaces. + +```json +{ + "listen_addr": "127.0.0.1:3100", + "state_path": "./slack-connect-bridge-state.json", + "allow_non_shared_channels": false, + "replay_lookback_secs": 86400, + "channels": [ + { + "slack_team_id": "T0123456789", + "slack_channel_id": "C0123456789", + "buzz_channel_id": "018f2f7d-44f4-7df1-a2b5-001122334455" + } + ] +} +``` + +One Buzz channel may map to only one Slack channel. This prevents accidental +fan-out of a message into multiple external organizations. + +`replay_lookback_secs` controls how far a restarted process re-reads Buzz +events, up to 30 days. A brand-new bridge starts at the current time and never +backfills old Buzz messages into Slack. On restart, durable mappings suppress +duplicates within the replay window. + +Back up `state_path`. It contains no credentials, but it is the durable mapping +between Slack timestamps and Buzz event IDs. If you intentionally remap a Buzz +channel to a different Slack channel, start with a reviewed, separate state +file. + +## 4. Run + +Activate Buzz's pinned toolchain: + +```bash +. ./bin/activate-hermit +``` + +Then start the bridge: + +```bash +BUZZ_SLACK_BRIDGE_CONFIG=/etc/buzz/slack-connect.json \ +BUZZ_RELAY_URL=wss://buzz.example.com \ +BUZZ_SLACK_BRIDGE_PRIVATE_KEY='nsec1…' \ +BUZZ_SLACK_BOT_TOKEN='xoxb-…' \ +BUZZ_SLACK_SIGNING_SECRET='…' \ +BUZZ_AUTH_TAG='["auth","…"]' \ +cargo run --release -p buzz-slack-connect-bridge +``` + +`BUZZ_AUTH_TAG` is optional on relays where the bridge key can authenticate +directly. + +Set `RUST_LOG=buzz_slack_connect_bridge=debug` for additional routing logs. +Logs contain IDs and error codes, but never tokens, signing secrets, or full +message bodies. + +## 5. Expose the webhook safely + +Terminate TLS at a reverse proxy and forward only `/slack/events` to the +configured listener. `/healthz` returns: + +- `200` when Slack validation and the Buzz subscription are ready; +- `503` while the bridge is starting or reconnecting. + +Slack callbacks are acknowledged only after the event has been handled. If the +bridge is unavailable, its queue is full, or processing exceeds Slack's +three-second window, it returns `503` so Slack retries. A late successful +attempt remains safe because Slack timestamps and deterministic Buzz events +deduplicate the retry. + +Run exactly one bridge process for a given state file. The reference binary +does not provide distributed leader election. + +## Manual verification + +1. Start with one dedicated test channel pair. +2. Confirm `/healthz` returns `200`. +3. Send a Slack root message and reply; both should appear in one Buzz thread. +4. Send a Buzz root message and reply; both should appear in one Slack thread. +5. Restart the bridge and confirm neither side receives duplicate messages. +6. Temporarily use a non-shared Slack test channel and confirm startup rejects + it while `allow_non_shared_channels` is `false`. +7. Review the first customer/partner channel's disclosure policy before + enabling its route. + +## Operational notes + +- Slack may connect a shared channel to many organizations. Everyone in that + channel can see bridged Buzz content. +- Slash commands and message actions are workspace-local and are intentionally + not part of this bridge. +- External Slack profiles may expose less metadata than local profiles. The + bridge falls back to the Slack user ID if `users.info` cannot resolve a name. +- Slack thread roots created before the bridge has no cross-system ID mapping. + Their replies are mirrored at channel level with a visible warning instead of + being silently dropped or attached to the wrong thread. +- File-only Slack messages are ignored in this slice. Text accompanying a file + is bridged, but the file is not downloaded or re-hosted. diff --git a/examples/slack-connect-bridge/bridge.example.json b/examples/slack-connect-bridge/bridge.example.json new file mode 100644 index 0000000000..9fedbe2828 --- /dev/null +++ b/examples/slack-connect-bridge/bridge.example.json @@ -0,0 +1,13 @@ +{ + "listen_addr": "127.0.0.1:3100", + "state_path": "./slack-connect-bridge-state.json", + "allow_non_shared_channels": false, + "replay_lookback_secs": 86400, + "channels": [ + { + "slack_team_id": "T0123456789", + "slack_channel_id": "C0123456789", + "buzz_channel_id": "00000000-0000-0000-0000-000000000000" + } + ] +} diff --git a/examples/slack-connect-bridge/slack-app-manifest.yaml b/examples/slack-connect-bridge/slack-app-manifest.yaml new file mode 100644 index 0000000000..7d06ee90c2 --- /dev/null +++ b/examples/slack-connect-bridge/slack-app-manifest.yaml @@ -0,0 +1,31 @@ +display_information: + name: Buzz Slack Connect Bridge + description: Bridges explicitly mapped Slack Connect and Buzz channels. + background_color: "#131622" +features: + bot_user: + display_name: Buzz Bridge + always_online: true +oauth_config: + scopes: + bot: + - channels:history + - channels:read + - groups:history + - groups:read + - chat:write + - users:read +settings: + event_subscriptions: + request_url: https://YOUR_BRIDGE_HOST/slack/events + bot_events: + - channel_id_changed + - channel_shared + - channel_unshared + - message.channels + - message.groups + interactivity: + is_enabled: false + org_deploy_enabled: false + socket_mode_enabled: false + token_rotation_enabled: false diff --git a/examples/slack-connect-bridge/src/bridge.rs b/examples/slack-connect-bridge/src/bridge.rs new file mode 100644 index 0000000000..aa38bd1191 --- /dev/null +++ b/examples/slack-connect-bridge/src/bridge.rs @@ -0,0 +1,959 @@ +//! Live Slack Connect ↔ Buzz message bridge. + +use std::{ + collections::HashMap, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; + +use anyhow::{bail, Context, Result}; +use buzz_ws_client::{NostrWsConnection, RelayMessage, WsClientError}; +use nostr::{Event, EventBuilder, EventId, Keys, Kind, Tag, Timestamp}; +use serde_json::{json, Value}; +use tokio::sync::mpsc; +use tracing::{error, info, warn}; +use uuid::Uuid; + +use crate::{ + config::{ChannelMapping, Config}, + slack::{ + escape_markdown_label, slack_mrkdwn_to_markdown, SlackClient, SlackDelivery, SlackEvent, + WebhookControl, + }, + state::{SlackMessageRef, StateStore}, +}; + +const SUBSCRIPTION_ID: &str = "slack-connect-bridge"; +const BRIDGE_NAME: &str = "slack-connect-bridge"; +const BRIDGE_DISPLAY_NAME: &str = "Slack Connect Bridge"; +const BRIDGE_ABOUT: &str = + "Bridges explicitly mapped Buzz channels and Slack Connect channels without impersonating users."; +const BRIDGE_ICON_DATA_URL: &str = "data:image/svg+xml,%3Csvg xmlns='http://www.w3.org/2000/svg' viewBox='0 0 128 128'%3E%3Crect width='128' height='128' rx='28' fill='%23131622'/%3E%3Cpath d='M36 64h56M64 36v56' stroke='%237dd3fc' stroke-width='13' stroke-linecap='round'/%3E%3Ccircle cx='36' cy='64' r='13' fill='%23facc15'/%3E%3Ccircle cx='92' cy='64' r='13' fill='%23a78bfa'/%3E%3C/svg%3E"; +const RECONNECT_MAX_SECS: u64 = 30; +const RELAY_POLL_TIMEOUT: Duration = Duration::from_secs(1); +const PROFILE_QUERY_TIMEOUT: Duration = Duration::from_secs(5); + +pub(crate) struct Bridge { + config: Config, + state: StateStore, + slack: SlackClient, + slack_bot_user_id: String, + delivery_rx: mpsc::Receiver, + webhook: WebhookControl, + slack_user_names: HashMap, + buzz_user_names: HashMap, + profile_subscription_sequence: AtomicU64, +} + +enum SessionOutcome { + Reconnect(anyhow::Error), + Shutdown, +} + +struct SlackMessageInput<'a> { + event_id: &'a str, + team_id: &'a str, + channel_id: &'a str, + user_id: &'a str, + text: &'a str, + ts: &'a str, + thread_ts: Option<&'a str>, + is_ext_shared: Option, +} + +struct SlackOriginInput<'a> { + buzz_channel_id: Uuid, + content: &'a str, + team_id: &'a str, + channel_id: &'a str, + slack_ts: &'a str, + user_id: &'a str, + thread_ts: Option<&'a str>, + reply_to: Option, +} + +impl Bridge { + pub(crate) async fn initialize( + config: Config, + delivery_rx: mpsc::Receiver, + webhook: WebhookControl, + ) -> Result { + let state = StateStore::load(config.state_path.clone())?; + let slack = SlackClient::new(config.slack_bot_token.clone())?; + let identity = slack.auth_test().await?; + validate_installation(&config, &identity.team_id)?; + + let mut bridge = Self { + config, + state, + slack, + slack_bot_user_id: identity.user_id, + delivery_rx, + webhook, + slack_user_names: HashMap::new(), + buzz_user_names: HashMap::new(), + profile_subscription_sequence: AtomicU64::new(0), + }; + bridge.validate_slack_routes().await?; + Ok(bridge) + } + + pub(crate) async fn run(mut self) -> Result<()> { + let mut reconnect_delay = 1_u64; + loop { + let session_started = tokio::time::Instant::now(); + let outcome = tokio::select! { + result = self.run_session() => match result { + Ok(()) => SessionOutcome::Reconnect(anyhow::anyhow!("Buzz relay session ended")), + Err(error) => SessionOutcome::Reconnect(error), + }, + signal = tokio::signal::ctrl_c() => { + signal.context("failed to listen for shutdown signal")?; + SessionOutcome::Shutdown + } + }; + self.webhook.set_ready(false); + if session_started.elapsed() >= Duration::from_secs(RECONNECT_MAX_SECS) { + reconnect_delay = 1; + } + + match outcome { + SessionOutcome::Shutdown => { + info!("Slack Connect bridge shutting down"); + return Ok(()); + } + SessionOutcome::Reconnect(error) => { + error!(reason = %error, reconnect_delay, "Buzz relay session failed"); + } + } + + tokio::select! { + _ = tokio::time::sleep(Duration::from_secs(reconnect_delay)) => {} + signal = tokio::signal::ctrl_c() => { + signal.context("failed to listen for shutdown signal")?; + info!("Slack Connect bridge shutting down"); + return Ok(()); + } + } + reconnect_delay = (reconnect_delay * 2).min(RECONNECT_MAX_SECS); + } + } + + async fn run_session(&mut self) -> Result<()> { + let mut connection = self.connect_buzz().await?; + self.webhook.set_ready(true); + info!( + routes = self.config.channels.len(), + "Slack Connect bridge is ready" + ); + + loop { + tokio::select! { + delivery = self.delivery_rx.recv() => { + let Some(delivery) = delivery else { + bail!("Slack delivery queue closed"); + }; + let result = self + .process_slack_event(&mut connection, delivery.event) + .await; + let completion = result + .as_ref() + .map(|_| ()) + .map_err(|error| error.to_string()); + let _ = delivery.completion.send(completion); + if let Err(error) = result { + warn!(reason = %error, "Slack event was not bridged"); + } + } + relay_message = connection.next_event(RELAY_POLL_TIMEOUT) => { + match relay_message { + Ok(RelayMessage::Event { subscription_id, event }) + if subscription_id == SUBSCRIPTION_ID => + { + self.process_buzz_event(&event).await?; + self.state.record_buzz_cursor(event.created_at.as_secs())?; + } + Ok(RelayMessage::Closed { subscription_id, message }) + if subscription_id == SUBSCRIPTION_ID => + { + bail!("Buzz relay closed bridge subscription: {message}"); + } + Ok(RelayMessage::Notice { message }) => { + warn!(%message, "Buzz relay notice"); + } + Ok(_) => {} + Err(WsClientError::Timeout) => {} + Err(error) => return Err(error).context("Buzz relay receive failed"), + } + } + } + } + } + + async fn connect_buzz(&mut self) -> Result { + let mut connection = NostrWsConnection::connect_authenticated( + &self.config.relay_url, + &self.config.bridge_keys, + self.config.owner_auth_tag.as_ref(), + ) + .await + .context("failed to connect and authenticate to Buzz relay")?; + + self.publish_bridge_profile(&mut connection).await?; + for route in &self.config.channels { + self.announce_channel_membership(&mut connection, route) + .await; + } + + let channels: Vec = self + .config + .channels + .iter() + .map(|route| route.buzz_channel_id.to_string()) + .collect(); + let since = self + .state + .subscription_since(Timestamp::now().as_secs(), self.config.replay_lookback_secs)?; + connection + .send_raw(&json!([ + "REQ", + SUBSCRIPTION_ID, + { + "kinds": [ + buzz_sdk::kind::KIND_STREAM_MESSAGE, + buzz_sdk::kind::KIND_STREAM_MESSAGE_V2 + ], + "#h": channels, + "since": since + } + ])) + .await + .context("failed to subscribe to mapped Buzz channels")?; + Ok(connection) + } + + async fn publish_bridge_profile(&self, connection: &mut NostrWsConnection) -> Result<()> { + let event = buzz_sdk::build_profile( + Some(BRIDGE_DISPLAY_NAME), + Some(BRIDGE_NAME), + Some(BRIDGE_ICON_DATA_URL), + Some(BRIDGE_ABOUT), + None, + )? + .sign_with_keys(&self.config.bridge_keys)?; + send_event_checked(connection, event) + .await + .context("failed to publish bridge profile") + } + + async fn announce_channel_membership( + &self, + connection: &mut NostrWsConnection, + route: &ChannelMapping, + ) { + let event = build_membership_event(route.buzz_channel_id, &self.config.bridge_keys); + let result = match event { + Ok(event) => send_event_checked(connection, event).await, + Err(error) => Err(error), + }; + if let Err(error) = result { + warn!( + buzz_channel_id = %route.buzz_channel_id, + reason = %error, + "bridge could not self-add as channel bot; private channels require an owner/admin to add the bridge pubkey" + ); + } + } + + async fn process_slack_event( + &mut self, + connection: &mut NostrWsConnection, + event: SlackEvent, + ) -> Result<()> { + match event { + SlackEvent::Message { + event_id, + team_id, + channel_id, + user_id, + text, + ts, + thread_ts, + is_ext_shared, + } => { + self.bridge_slack_message( + connection, + SlackMessageInput { + event_id: &event_id, + team_id: &team_id, + channel_id: &channel_id, + user_id: &user_id, + text: &text, + ts: &ts, + thread_ts: thread_ts.as_deref(), + is_ext_shared, + }, + ) + .await + } + SlackEvent::ChannelIdChanged { + event_id, + team_id, + old_channel_id, + new_channel_id, + } => { + if self + .route_for_slack(&team_id, &old_channel_id) + .or_else(|| self.route_for_slack(&team_id, &new_channel_id)) + .is_some() + { + self.state.record_channel_id_change( + &team_id, + &old_channel_id, + &new_channel_id, + )?; + info!( + %event_id, + %team_id, + %old_channel_id, + %new_channel_id, + "updated mapped Slack channel ID" + ); + } + Ok(()) + } + SlackEvent::ChannelShared { + event_id, + team_id, + channel_id, + } => { + if let Some(route) = self.route_for_slack(&team_id, &channel_id).cloned() { + self.state.set_route_paused(route.buzz_channel_id, false)?; + info!( + %event_id, + %team_id, + %channel_id, + buzz_channel_id = %route.buzz_channel_id, + "resumed shared-channel route" + ); + } + Ok(()) + } + SlackEvent::ChannelUnshared { + event_id, + team_id, + channel_id, + is_ext_shared, + } => { + if let Some(route) = self.route_for_slack(&team_id, &channel_id).cloned() { + if is_ext_shared { + info!( + %event_id, + %team_id, + %channel_id, + buzz_channel_id = %route.buzz_channel_id, + "one organization left the Slack Connect channel; route remains shared" + ); + } else { + self.state.set_route_paused(route.buzz_channel_id, true)?; + warn!( + %event_id, + %team_id, + %channel_id, + buzz_channel_id = %route.buzz_channel_id, + "paused route because Slack reported channel_unshared" + ); + } + } + Ok(()) + } + } + } + + async fn bridge_slack_message( + &mut self, + connection: &mut NostrWsConnection, + input: SlackMessageInput<'_>, + ) -> Result<()> { + let SlackMessageInput { + event_id, + team_id, + channel_id, + user_id, + text, + ts, + thread_ts, + is_ext_shared, + } = input; + if user_id == self.slack_bot_user_id { + return Ok(()); + } + if is_ext_shared == Some(false) && !self.config.allow_non_shared_channels { + warn!( + %event_id, + %team_id, + %channel_id, + "ignored message explicitly marked as non-shared" + ); + return Ok(()); + } + + let Some(route) = self.route_for_slack(team_id, channel_id).cloned() else { + return Ok(()); + }; + if self.state.route_is_paused(route.buzz_channel_id) { + warn!( + %event_id, + buzz_channel_id = %route.buzz_channel_id, + "ignored message for paused Slack Connect route" + ); + return Ok(()); + } + if self + .state + .buzz_event_for_slack(route.buzz_channel_id, ts) + .is_some() + { + return Ok(()); + } + if text.trim().is_empty() { + info!(%event_id, "ignored Slack message without text"); + return Ok(()); + } + + let author = self.slack_display_name(user_id).await?; + let reply_to = thread_ts + .filter(|root_ts| *root_ts != ts) + .and_then(|root_ts| { + self.state + .buzz_event_for_slack(route.buzz_channel_id, root_ts) + }) + .map(EventId::from_hex) + .transpose() + .context("bridge state contains an invalid Buzz event ID")?; + let thread_fallback = thread_ts + .filter(|root_ts| *root_ts != ts) + .is_some_and(|_| reply_to.is_none()); + let fallback_label = if thread_fallback { + "↳ _Slack thread root was not bridged; showing this reply at channel level._\n\n" + } else { + "" + }; + let content = format!( + "**{} · Slack**\n{}{}", + escape_markdown_label(&author), + fallback_label, + slack_mrkdwn_to_markdown(text) + ); + let event = build_slack_origin_event( + &self.config.bridge_keys, + SlackOriginInput { + buzz_channel_id: route.buzz_channel_id, + content: &content, + team_id, + channel_id, + slack_ts: ts, + user_id, + thread_ts, + reply_to, + }, + )?; + let buzz_event_id = event.id.to_hex(); + send_event_checked(connection, event).await?; + + let canonical_channel = self.state.canonical_channel_id(team_id, channel_id); + self.state.record_message_pair( + route.buzz_channel_id, + &buzz_event_id, + SlackMessageRef { + team_id: team_id.to_owned(), + channel_id: canonical_channel, + ts: ts.to_owned(), + thread_ts: thread_ts.map(str::to_owned), + }, + )?; + info!( + %event_id, + %buzz_event_id, + buzz_channel_id = %route.buzz_channel_id, + "bridged Slack message to Buzz" + ); + Ok(()) + } + + async fn process_buzz_event(&mut self, event: &Event) -> Result<()> { + if event.pubkey == self.config.bridge_keys.public_key() + || has_slack_origin(event) + || self + .state + .slack_message_for_buzz(&event.id.to_hex()) + .is_some() + { + return Ok(()); + } + let Some(buzz_channel_id) = event_channel_id(event) else { + return Ok(()); + }; + let Some(route) = self.route_for_buzz(buzz_channel_id).cloned() else { + return Ok(()); + }; + if self.state.route_is_paused(route.buzz_channel_id) { + return Ok(()); + } + + let thread_root = event_thread_root(event); + let thread_ts = thread_root.as_deref().and_then(|root_id| { + self.state.slack_message_for_buzz(root_id).map(|reference| { + reference + .thread_ts + .as_deref() + .unwrap_or(&reference.ts) + .to_owned() + }) + }); + let author = self.buzz_display_name(&event.pubkey.to_hex()).await; + let fallback_label = if thread_root.is_some() && thread_ts.is_none() { + "↳ Buzz thread root was not bridged; showing this reply at channel level.\n\n" + } else { + "" + }; + let text = format!( + "*{} · Buzz*\n{}{}", + escape_slack_label(&author), + fallback_label, + escape_slack_message_body(&event.content) + ); + let channel_id = self + .state + .canonical_channel_id(&route.slack_team_id, &route.slack_channel_id); + let posted = self + .slack + .post_message(&channel_id, &text, thread_ts.as_deref(), &event.id.to_hex()) + .await + .with_context(|| { + format!( + "failed to post Buzz event {} to mapped Slack channel", + event.id.to_hex() + ) + })?; + + self.state.record_message_pair( + route.buzz_channel_id, + &event.id.to_hex(), + SlackMessageRef { + team_id: route.slack_team_id, + channel_id, + ts: posted.ts, + thread_ts, + }, + )?; + info!( + buzz_event_id = %event.id.to_hex(), + buzz_channel_id = %route.buzz_channel_id, + "bridged Buzz message to Slack" + ); + Ok(()) + } + + async fn slack_display_name(&mut self, user_id: &str) -> Result { + if let Some(name) = self.slack_user_names.get(user_id) { + return Ok(name.clone()); + } + if let Some(name) = self.state.slack_user_name(user_id) { + let name = name.to_owned(); + self.slack_user_names + .insert(user_id.to_owned(), name.clone()); + return Ok(name); + } + let name = match self.slack.user_display_name(user_id).await { + Ok(name) => name, + Err(error) => { + warn!(%user_id, reason = %error, "could not resolve Slack display name"); + user_id.to_owned() + } + }; + self.state.record_slack_user_name(user_id, &name)?; + self.slack_user_names + .insert(user_id.to_owned(), name.clone()); + Ok(name) + } + + async fn buzz_display_name(&mut self, pubkey: &str) -> String { + if let Some(name) = self.buzz_user_names.get(pubkey) { + return name.clone(); + } + let fallback = abbreviated_pubkey(pubkey); + let name = match self.query_buzz_profile(pubkey).await { + Ok(Some(name)) => name, + Ok(None) => fallback, + Err(error) => { + warn!(pubkey = %abbreviated_pubkey(pubkey), reason = %error, "could not resolve Buzz profile"); + fallback + } + }; + self.buzz_user_names.insert(pubkey.to_owned(), name.clone()); + name + } + + async fn query_buzz_profile(&self, pubkey: &str) -> Result> { + let mut connection = NostrWsConnection::connect_authenticated( + &self.config.relay_url, + &self.config.bridge_keys, + self.config.owner_auth_tag.as_ref(), + ) + .await?; + let sequence = self + .profile_subscription_sequence + .fetch_add(1, Ordering::Relaxed); + let subscription_id = format!("slack-profile-{sequence}"); + connection + .send_raw(&json!([ + "REQ", + subscription_id, + { "kinds": [0], "authors": [pubkey], "limit": 1 } + ])) + .await?; + + let deadline = tokio::time::Instant::now() + PROFILE_QUERY_TIMEOUT; + let mut result = None; + while let Some(remaining) = deadline.checked_duration_since(tokio::time::Instant::now()) { + match connection.next_event(remaining).await { + Ok(RelayMessage::Event { + subscription_id: response_id, + event, + }) if response_id == subscription_id => { + result = profile_display_name(&event.content); + } + Ok(RelayMessage::Eose { + subscription_id: response_id, + }) if response_id == subscription_id => break, + Ok(RelayMessage::Closed { + subscription_id: response_id, + message, + }) if response_id == subscription_id => { + bail!("Buzz profile query closed: {message}"); + } + Ok(_) => {} + Err(WsClientError::Timeout) => break, + Err(error) => return Err(error.into()), + } + } + let _ = connection.disconnect().await; + Ok(result) + } + + async fn validate_slack_routes(&mut self) -> Result<()> { + for route in self.config.channels.clone() { + let channel_id = self + .state + .canonical_channel_id(&route.slack_team_id, &route.slack_channel_id); + let conversation = self.slack.conversation_info(&channel_id).await?; + if conversation.is_archived { + bail!( + "Slack channel {channel_id} ({}) is archived", + conversation.name + ); + } + if !conversation.is_ext_shared && !self.config.allow_non_shared_channels { + bail!( + "Slack channel {channel_id} ({}) is not a Slack Connect channel; set allow_non_shared_channels only after reviewing the disclosure boundary", + conversation.name + ); + } + if conversation.is_ext_shared { + self.state.set_route_paused(route.buzz_channel_id, false)?; + } + info!( + %channel_id, + channel_name = %conversation.name, + is_private = conversation.is_private, + buzz_channel_id = %route.buzz_channel_id, + "validated Slack channel route" + ); + } + Ok(()) + } + + fn route_for_slack(&self, team_id: &str, channel_id: &str) -> Option<&ChannelMapping> { + let incoming = self.state.canonical_channel_id(team_id, channel_id); + self.config.channels.iter().find(|route| { + route.slack_team_id == team_id + && self + .state + .canonical_channel_id(team_id, &route.slack_channel_id) + == incoming + }) + } + + fn route_for_buzz(&self, channel_id: Uuid) -> Option<&ChannelMapping> { + self.config + .channels + .iter() + .find(|route| route.buzz_channel_id == channel_id) + } +} + +fn validate_installation(config: &Config, installed_team_id: &str) -> Result<()> { + for route in &config.channels { + if route.slack_team_id != installed_team_id { + bail!( + "Slack bot token is installed in {installed_team_id}, but a route uses {}; run one bridge process per Slack installation", + route.slack_team_id + ); + } + } + Ok(()) +} + +fn build_membership_event(channel_id: Uuid, keys: &Keys) -> Result { + let channel_id = channel_id.to_string(); + let pubkey = keys.public_key().to_hex(); + Ok( + EventBuilder::new(Kind::Custom(buzz_sdk::kind::KIND_NIP29_PUT_USER as u16), "") + .tags([ + Tag::parse(["h", channel_id.as_str()])?, + Tag::parse(["p", pubkey.as_str()])?, + Tag::parse(["role", "bot"])?, + ]) + .sign_with_keys(keys)?, + ) +} + +fn build_slack_origin_event(keys: &Keys, input: SlackOriginInput<'_>) -> Result { + let SlackOriginInput { + buzz_channel_id, + content, + team_id, + channel_id, + slack_ts, + user_id, + thread_ts, + reply_to, + } = input; + if content.len() > 64 * 1024 { + bail!("Slack message exceeds Buzz's 64 KiB message limit"); + } + let external_id = format!("slack:{team_id}:{channel_id}:{slack_ts}"); + let mut tags = vec![ + Tag::parse(["h", buzz_channel_id.to_string().as_str()])?, + Tag::parse(["i", external_id.as_str()])?, + Tag::parse(["proxy", "slack", team_id, channel_id, slack_ts, user_id])?, + Tag::parse(["client", BRIDGE_NAME])?, + ]; + if let Some(thread_ts) = thread_ts { + tags.push(Tag::parse(["slack_thread_ts", thread_ts])?); + } + if let Some(reply_to) = reply_to { + let reply_to = reply_to.to_hex(); + tags.push(Tag::parse(["e", reply_to.as_str(), "", "reply"])?); + } + let created_at = slack_timestamp(slack_ts)?; + Ok(EventBuilder::new( + Kind::Custom(buzz_sdk::kind::KIND_STREAM_MESSAGE as u16), + content, + ) + .tags(tags) + .custom_created_at(created_at) + .sign_with_keys(keys)?) +} + +async fn send_event_checked(connection: &mut NostrWsConnection, event: Event) -> Result<()> { + let event_id = event.id.to_hex(); + let response = connection.send_event(event).await?; + if !response.accepted { + bail!("Buzz relay rejected event {event_id}: {}", response.message); + } + Ok(()) +} + +fn slack_timestamp(ts: &str) -> Result { + let seconds = ts + .split_once('.') + .map_or(ts, |(seconds, _)| seconds) + .parse::() + .context("Slack message has an invalid ts")?; + Ok(Timestamp::from(seconds)) +} + +fn has_slack_origin(event: &Event) -> bool { + event.tags.iter().any(|tag| { + let parts = tag.as_slice(); + parts.first().map(String::as_str) == Some("proxy") + && parts.get(1).map(String::as_str) == Some("slack") + }) +} + +fn event_channel_id(event: &Event) -> Option { + event.tags.iter().find_map(|tag| { + let parts = tag.as_slice(); + (parts.first().map(String::as_str) == Some("h")) + .then(|| parts.get(1)) + .flatten() + .and_then(|value| Uuid::parse_str(value).ok()) + }) +} + +fn event_thread_root(event: &Event) -> Option { + let mut reply = None; + for tag in event.tags.iter() { + let parts = tag.as_slice(); + if parts.first().map(String::as_str) != Some("e") { + continue; + } + let id = parts.get(1).filter(|id| { + id.len() == 64 && id.chars().all(|character| character.is_ascii_hexdigit()) + }); + match (parts.get(3).map(String::as_str), id) { + (Some("root"), Some(id)) => return Some(id.clone()), + (Some("reply"), Some(id)) => reply = Some(id.clone()), + _ => {} + } + } + reply +} + +fn profile_display_name(content: &str) -> Option { + let value: Value = serde_json::from_str(content).ok()?; + for field in ["display_name", "name"] { + if let Some(name) = value.get(field).and_then(Value::as_str) { + if !name.trim().is_empty() { + return Some(name.trim().to_owned()); + } + } + } + None +} + +fn abbreviated_pubkey(pubkey: &str) -> String { + if pubkey.len() <= 16 { + return pubkey.to_owned(); + } + format!("{}…{}", &pubkey[..8], &pubkey[pubkey.len() - 6..]) +} + +fn escape_slack_label(input: &str) -> String { + input + .replace('&', "&") + .replace('<', "<") + .replace('>', ">") + .replace(['\r', '\n', '*', '_', '`'], " ") + .chars() + .take(160) + .collect() +} + +/// Slack interprets angle-bracket control sequences as mentions and links. +/// Escape Buzz-authored content before posting it into an externally shared +/// channel so a string such as `` cannot become a mass mention. +fn escape_slack_message_body(input: &str) -> String { + input + .replace('&', "&") + .replace('<', "<") + .replace('>', ">") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn slack_origin_is_deterministic_and_threaded() { + let keys = Keys::generate(); + let channel = Uuid::new_v4(); + let root = EventId::from_hex(&"a".repeat(64)).unwrap(); + let first = build_slack_origin_event( + &keys, + SlackOriginInput { + buzz_channel_id: channel, + content: "hello", + team_id: "T12345678", + channel_id: "C12345678", + slack_ts: "1700000000.000001", + user_id: "U12345678", + thread_ts: Some("1699999999.000001"), + reply_to: Some(root), + }, + ) + .unwrap(); + let second = build_slack_origin_event( + &keys, + SlackOriginInput { + buzz_channel_id: channel, + content: "hello", + team_id: "T12345678", + channel_id: "C12345678", + slack_ts: "1700000000.000001", + user_id: "U12345678", + thread_ts: Some("1699999999.000001"), + reply_to: Some(root), + }, + ) + .unwrap(); + assert_eq!(first.id, second.id); + assert!(has_slack_origin(&first)); + assert_eq!(event_thread_root(&first), Some("a".repeat(64))); + assert_eq!(event_channel_id(&first), Some(channel)); + } + + #[test] + fn direct_and_nested_replies_resolve_to_root() { + let keys = Keys::generate(); + let root = "a".repeat(64); + let parent = "b".repeat(64); + let direct = EventBuilder::new(Kind::Custom(9), "direct") + .tags([Tag::parse(["e", root.as_str(), "", "reply"]).unwrap()]) + .sign_with_keys(&keys) + .unwrap(); + assert_eq!(event_thread_root(&direct), Some(root.clone())); + + let nested = EventBuilder::new(Kind::Custom(9), "nested") + .tags([ + Tag::parse(["e", root.as_str(), "", "root"]).unwrap(), + Tag::parse(["e", parent.as_str(), "", "reply"]).unwrap(), + ]) + .sign_with_keys(&keys) + .unwrap(); + assert_eq!(event_thread_root(&nested), Some(root)); + } + + #[test] + fn profile_name_prefers_display_name() { + assert_eq!( + profile_display_name(r#"{"name":"alice","display_name":"Alice A."}"#), + Some("Alice A.".into()) + ); + assert_eq!(profile_display_name("{}"), None); + } + + #[test] + fn install_token_cannot_cross_team_boundaries() { + let config = Config { + relay_url: "ws://localhost:3000".into(), + bridge_keys: Keys::generate(), + owner_auth_tag: None, + slack_signing_secret: "secret".into(), + slack_bot_token: "token".into(), + listen_addr: "127.0.0.1:3100".parse().unwrap(), + state_path: "state.json".into(), + allow_non_shared_channels: false, + replay_lookback_secs: 60, + channels: vec![ChannelMapping { + slack_team_id: "T12345678".into(), + slack_channel_id: "C12345678".into(), + buzz_channel_id: Uuid::new_v4(), + }], + }; + let error = validate_installation(&config, "T87654321") + .unwrap_err() + .to_string(); + assert!(error.contains("one bridge process per Slack installation")); + } + + #[test] + fn buzz_content_cannot_create_slack_control_mentions() { + assert_eq!( + escape_slack_message_body("deploy and <@U12345678>"), + "deploy <!channel> and <@U12345678>" + ); + } +} diff --git a/examples/slack-connect-bridge/src/config.rs b/examples/slack-connect-bridge/src/config.rs new file mode 100644 index 0000000000..7c444ae9d4 --- /dev/null +++ b/examples/slack-connect-bridge/src/config.rs @@ -0,0 +1,230 @@ +//! Bridge configuration and validation. + +use std::{ + collections::HashSet, + net::SocketAddr, + path::{Path, PathBuf}, +}; + +use anyhow::{bail, Context, Result}; +use buzz_sdk::nip_oa; +use nostr::{Keys, Tag}; +use serde::Deserialize; +use uuid::Uuid; + +const DEFAULT_LISTEN_ADDR: &str = "127.0.0.1:3100"; +const DEFAULT_REPLAY_LOOKBACK_SECS: u64 = 86_400; +const MAX_REPLAY_LOOKBACK_SECS: u64 = 30 * 86_400; + +/// One explicit Slack channel to Buzz channel route. +#[derive(Clone, Debug, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub(crate) struct ChannelMapping { + pub(crate) slack_team_id: String, + pub(crate) slack_channel_id: String, + pub(crate) buzz_channel_id: Uuid, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct FileConfig { + #[serde(default = "default_listen_addr")] + listen_addr: String, + #[serde(default = "default_state_path")] + state_path: PathBuf, + #[serde(default)] + allow_non_shared_channels: bool, + #[serde(default = "default_replay_lookback_secs")] + replay_lookback_secs: u64, + channels: Vec, +} + +fn default_listen_addr() -> String { + DEFAULT_LISTEN_ADDR.to_owned() +} + +fn default_state_path() -> PathBuf { + PathBuf::from("slack-connect-bridge-state.json") +} + +const fn default_replay_lookback_secs() -> u64 { + DEFAULT_REPLAY_LOOKBACK_SECS +} + +/// Fully validated runtime configuration. +pub(crate) struct Config { + pub(crate) relay_url: String, + pub(crate) bridge_keys: Keys, + pub(crate) owner_auth_tag: Option, + pub(crate) slack_signing_secret: String, + pub(crate) slack_bot_token: String, + pub(crate) listen_addr: SocketAddr, + pub(crate) state_path: PathBuf, + pub(crate) allow_non_shared_channels: bool, + pub(crate) replay_lookback_secs: u64, + pub(crate) channels: Vec, +} + +impl Config { + pub(crate) fn from_env() -> Result { + let config_path = PathBuf::from(required_env("BUZZ_SLACK_BRIDGE_CONFIG")?); + let raw = std::fs::read_to_string(&config_path) + .with_context(|| format!("failed to read {}", config_path.display()))?; + let file: FileConfig = serde_json::from_str(&raw) + .with_context(|| format!("failed to parse {}", config_path.display()))?; + validate_file_config(&file)?; + + let relay_url = + std::env::var("BUZZ_RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".to_owned()); + let bridge_keys = Keys::parse(&required_env("BUZZ_SLACK_BRIDGE_PRIVATE_KEY")?) + .context("BUZZ_SLACK_BRIDGE_PRIVATE_KEY must be an nsec or hex private key")?; + let owner_auth_tag = parse_owner_auth_tag(&bridge_keys)?; + let listen_addr = file + .listen_addr + .parse() + .with_context(|| format!("invalid listen_addr {:?}", file.listen_addr))?; + let state_path = resolve_relative_path(&config_path, &file.state_path); + + Ok(Self { + relay_url, + bridge_keys, + owner_auth_tag, + slack_signing_secret: required_env("BUZZ_SLACK_SIGNING_SECRET")?, + slack_bot_token: required_env("BUZZ_SLACK_BOT_TOKEN")?, + listen_addr, + state_path, + allow_non_shared_channels: file.allow_non_shared_channels, + replay_lookback_secs: file.replay_lookback_secs, + channels: file.channels, + }) + } +} + +fn parse_owner_auth_tag(bridge_keys: &Keys) -> Result> { + let Ok(raw) = std::env::var("BUZZ_AUTH_TAG") else { + return Ok(None); + }; + if raw.trim().is_empty() { + return Ok(None); + } + nip_oa::verify_auth_tag(&raw, &bridge_keys.public_key()) + .context("BUZZ_AUTH_TAG is not valid for BUZZ_SLACK_BRIDGE_PRIVATE_KEY")?; + Ok(Some(nip_oa::parse_auth_tag(&raw)?)) +} + +fn resolve_relative_path(config_path: &Path, path: &Path) -> PathBuf { + if path.is_absolute() { + return path.to_owned(); + } + config_path + .parent() + .unwrap_or_else(|| Path::new(".")) + .join(path) +} + +fn validate_file_config(config: &FileConfig) -> Result<()> { + if config.channels.is_empty() { + bail!("config must contain at least one channel mapping"); + } + if config.replay_lookback_secs == 0 || config.replay_lookback_secs > MAX_REPLAY_LOOKBACK_SECS { + bail!("replay_lookback_secs must be between 1 and {MAX_REPLAY_LOOKBACK_SECS}"); + } + + let mut slack_routes = HashSet::new(); + let mut buzz_routes = HashSet::new(); + for route in &config.channels { + validate_slack_id("slack_team_id", &route.slack_team_id, &['T'])?; + validate_slack_id("slack_channel_id", &route.slack_channel_id, &['C', 'G'])?; + if !slack_routes.insert((route.slack_team_id.clone(), route.slack_channel_id.clone())) { + bail!( + "duplicate Slack route {}:{}", + route.slack_team_id, + route.slack_channel_id + ); + } + if !buzz_routes.insert(route.buzz_channel_id) { + bail!( + "Buzz channel {} is mapped more than once; one-to-many fan-out must be explicit in a future bridge", + route.buzz_channel_id + ); + } + } + Ok(()) +} + +fn validate_slack_id(field: &str, value: &str, prefixes: &[char]) -> Result<()> { + let mut chars = value.chars(); + let prefix = chars.next(); + if value.len() < 9 + || value.len() > 32 + || !prefix.is_some_and(|p| prefixes.contains(&p)) + || !chars.all(|c| c.is_ascii_uppercase() || c.is_ascii_digit()) + { + bail!("{field} has an invalid Slack ID"); + } + Ok(()) +} + +fn required_env(name: &str) -> Result { + let value = std::env::var(name).with_context(|| format!("{name} is required"))?; + if value.trim().is_empty() { + bail!("{name} must not be empty"); + } + Ok(value) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn file_config() -> FileConfig { + FileConfig { + listen_addr: default_listen_addr(), + state_path: default_state_path(), + allow_non_shared_channels: false, + replay_lookback_secs: DEFAULT_REPLAY_LOOKBACK_SECS, + channels: vec![ChannelMapping { + slack_team_id: "T12345678".into(), + slack_channel_id: "C12345678".into(), + buzz_channel_id: Uuid::new_v4(), + }], + } + } + + #[test] + fn accepts_one_to_one_routes() { + validate_file_config(&file_config()).unwrap(); + } + + #[test] + fn rejects_duplicate_buzz_destinations() { + let mut config = file_config(); + config.channels.push(ChannelMapping { + slack_team_id: "T12345678".into(), + slack_channel_id: "C87654321".into(), + buzz_channel_id: config.channels[0].buzz_channel_id, + }); + let error = validate_file_config(&config).unwrap_err().to_string(); + assert!(error.contains("mapped more than once"), "{error}"); + } + + #[test] + fn rejects_invalid_slack_ids_without_echoing_secrets() { + let mut config = file_config(); + config.channels[0].slack_channel_id = "not-a-channel".into(); + let error = validate_file_config(&config).unwrap_err().to_string(); + assert_eq!(error, "slack_channel_id has an invalid Slack ID"); + assert!(!error.contains("not-a-channel")); + } + + #[test] + fn relative_state_path_is_relative_to_config() { + assert_eq!( + resolve_relative_path( + Path::new("/opt/buzz/slack-bridge.json"), + Path::new("state.json") + ), + PathBuf::from("/opt/buzz/state.json") + ); + } +} diff --git a/examples/slack-connect-bridge/src/main.rs b/examples/slack-connect-bridge/src/main.rs new file mode 100644 index 0000000000..699af27987 --- /dev/null +++ b/examples/slack-connect-bridge/src/main.rs @@ -0,0 +1,65 @@ +#![deny(unsafe_code)] +//! Operator-run reference bridge between Buzz channels and Slack Connect. + +mod bridge; +mod config; +mod slack; +mod state; + +use anyhow::{Context, Result}; +use tokio::sync::{mpsc, watch}; +use tracing_subscriber::EnvFilter; + +use crate::{ + bridge::Bridge, + config::Config, + slack::{run_webhook_server, WebhookServerState}, +}; + +const DELIVERY_QUEUE_CAPACITY: usize = 256; + +#[tokio::main] +async fn main() -> Result<()> { + init_tracing()?; + run().await +} + +async fn run() -> Result<()> { + let config = Config::from_env()?; + let listen_addr = config.listen_addr; + let listener = tokio::net::TcpListener::bind(listen_addr) + .await + .with_context(|| format!("failed to bind Slack webhook listener on {listen_addr}"))?; + let (delivery_tx, delivery_rx) = mpsc::channel(DELIVERY_QUEUE_CAPACITY); + let (webhook_state, webhook_control) = + WebhookServerState::new(config.slack_signing_secret.clone(), delivery_tx); + let (shutdown_tx, shutdown_rx) = watch::channel(false); + let mut server = tokio::spawn(run_webhook_server(listener, webhook_state, shutdown_rx)); + + let bridge = Bridge::initialize(config, delivery_rx, webhook_control).await?; + tokio::select! { + bridge_result = bridge.run() => { + let _ = shutdown_tx.send(true); + let server_result = server.await.context("Slack webhook server task panicked")?; + bridge_result?; + server_result + } + server_result = &mut server => { + let _ = shutdown_tx.send(true); + server_result.context("Slack webhook server task panicked")??; + anyhow::bail!("Slack webhook server stopped unexpectedly") + } + } +} + +fn init_tracing() -> Result<()> { + let filter = match EnvFilter::try_from_default_env() { + Ok(filter) => filter, + Err(_) => EnvFilter::new("buzz_slack_connect_bridge=info"), + }; + tracing_subscriber::fmt() + .with_env_filter(filter) + .with_target(false) + .try_init() + .map_err(|error| anyhow::anyhow!("failed to initialize tracing: {error}")) +} diff --git a/examples/slack-connect-bridge/src/slack/api.rs b/examples/slack-connect-bridge/src/slack/api.rs new file mode 100644 index 0000000000..69fa548554 --- /dev/null +++ b/examples/slack-connect-bridge/src/slack/api.rs @@ -0,0 +1,242 @@ +use std::{collections::HashMap, time::Duration}; + +use anyhow::{bail, Context, Result}; +use reqwest::{Method, StatusCode}; +use serde::Deserialize; +use serde_json::{json, Value}; + +const SLACK_API_ORIGIN: &str = "https://slack.com/api"; +const MAX_API_ATTEMPTS: usize = 3; +const MAX_RETRY_AFTER_SECS: u64 = 60; +const MAX_SLACK_TEXT_CHARS: usize = 39_000; + +pub(crate) struct SlackClient { + http: reqwest::Client, + bot_token: String, + api_origin: String, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct SlackIdentity { + pub(crate) team_id: String, + pub(crate) user_id: String, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct SlackConversation { + #[serde(default)] + pub(crate) is_ext_shared: bool, + #[serde(default)] + pub(crate) is_private: bool, + #[serde(default)] + pub(crate) is_archived: bool, + #[serde(default)] + pub(crate) name: String, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct SlackPostedMessage { + pub(crate) ts: String, +} + +#[derive(Deserialize)] +struct ApiEnvelope { + ok: bool, + #[serde(default)] + error: Option, + #[serde(flatten)] + rest: HashMap, +} + +impl SlackClient { + pub(crate) fn new(bot_token: String) -> Result { + let http = reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .user_agent("buzz-slack-connect-bridge/0.1") + .build() + .context("failed to build Slack HTTP client")?; + Ok(Self { + http, + bot_token, + api_origin: SLACK_API_ORIGIN.to_owned(), + }) + } + + pub(crate) async fn auth_test(&self) -> Result { + let value = self.call(Method::POST, "auth.test", None).await?; + Ok(SlackIdentity { + team_id: required_string(&value, "team_id", "auth.test")?, + user_id: required_string(&value, "user_id", "auth.test")?, + }) + } + + pub(crate) async fn conversation_info(&self, channel_id: &str) -> Result { + let value = self + .call( + Method::POST, + "conversations.info", + Some(json!({ "channel": channel_id })), + ) + .await?; + serde_json::from_value( + value + .get("channel") + .cloned() + .context("Slack conversations.info response omitted channel")?, + ) + .context("invalid channel in Slack conversations.info response") + } + + pub(crate) async fn user_display_name(&self, user_id: &str) -> Result { + let value = self + .call(Method::POST, "users.info", Some(json!({ "user": user_id }))) + .await?; + let user = value + .get("user") + .context("Slack users.info response omitted user")?; + let profile = user + .get("profile") + .context("Slack users.info response omitted profile")?; + for field in ["display_name", "real_name"] { + if let Some(name) = profile.get(field).and_then(Value::as_str) { + if !name.trim().is_empty() { + return Ok(name.trim().to_owned()); + } + } + } + Ok(user + .get("name") + .and_then(Value::as_str) + .filter(|name| !name.trim().is_empty()) + .unwrap_or(user_id) + .to_owned()) + } + + pub(crate) async fn post_message( + &self, + channel_id: &str, + text: &str, + thread_ts: Option<&str>, + client_msg_id: &str, + ) -> Result { + let mut payload = json!({ + "channel": channel_id, + "text": truncate_slack_text(text), + "client_msg_id": client_msg_id, + "unfurl_links": false, + "unfurl_media": false + }); + if let Some(thread_ts) = thread_ts { + payload["thread_ts"] = Value::String(thread_ts.to_owned()); + } + let value = self + .call(Method::POST, "chat.postMessage", Some(payload)) + .await?; + Ok(SlackPostedMessage { + ts: required_string(&value, "ts", "chat.postMessage")?, + }) + } + + async fn call(&self, method: Method, endpoint: &str, payload: Option) -> Result { + let url = format!("{}/{endpoint}", self.api_origin); + let mut last_error = None; + + for attempt in 0..MAX_API_ATTEMPTS { + let mut request = self + .http + .request(method.clone(), &url) + .bearer_auth(&self.bot_token); + if let Some(payload) = &payload { + request = request.json(payload); + } + + let response = match request.send().await { + Ok(response) => response, + Err(error) => { + last_error = Some( + anyhow::Error::new(error) + .context(format!("Slack {endpoint} request failed")), + ); + retry_transport(attempt).await; + continue; + } + }; + + if response.status() == StatusCode::TOO_MANY_REQUESTS { + let retry_after = response + .headers() + .get(reqwest::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()) + .unwrap_or(1) + .min(MAX_RETRY_AFTER_SECS); + last_error = Some(anyhow::anyhow!("Slack {endpoint} rate limited the bridge")); + tokio::time::sleep(Duration::from_secs(retry_after)).await; + continue; + } + + let status = response.status(); + let body: Value = response + .json() + .await + .with_context(|| format!("Slack {endpoint} returned non-JSON HTTP {status}"))?; + if !status.is_success() { + last_error = Some(anyhow::anyhow!("Slack {endpoint} returned HTTP {status}")); + if status.is_server_error() { + retry_transport(attempt).await; + continue; + } + break; + } + + let envelope: ApiEnvelope = serde_json::from_value(body) + .with_context(|| format!("invalid Slack {endpoint} response"))?; + if !envelope.ok { + let code = envelope.error.unwrap_or_else(|| "unknown_error".to_owned()); + bail!("Slack {endpoint} failed: {code}"); + } + return Ok(Value::Object(envelope.rest.into_iter().collect())); + } + + Err(last_error.unwrap_or_else(|| anyhow::anyhow!("Slack {endpoint} request failed"))) + } +} + +fn required_string(value: &Value, field: &str, endpoint: &str) -> Result { + value + .get(field) + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + .with_context(|| format!("Slack {endpoint} response omitted {field}")) +} + +async fn retry_transport(attempt: usize) { + if attempt + 1 < MAX_API_ATTEMPTS { + let delay_ms = 250_u64.saturating_mul(1_u64 << attempt.min(4)); + tokio::time::sleep(Duration::from_millis(delay_ms)).await; + } +} + +fn truncate_slack_text(text: &str) -> String { + if text.chars().count() <= MAX_SLACK_TEXT_CHARS { + return text.to_owned(); + } + let mut truncated: String = text.chars().take(MAX_SLACK_TEXT_CHARS - 16).collect(); + truncated.push_str("\n… _(truncated)_"); + truncated +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn truncation_preserves_utf8_boundaries() { + let text = "🐝".repeat(MAX_SLACK_TEXT_CHARS + 10); + let truncated = truncate_slack_text(&text); + assert!(truncated.is_char_boundary(truncated.len())); + assert!(truncated.chars().count() <= MAX_SLACK_TEXT_CHARS); + assert!(truncated.ends_with("… _(truncated)_")); + } +} diff --git a/examples/slack-connect-bridge/src/slack/mod.rs b/examples/slack-connect-bridge/src/slack/mod.rs new file mode 100644 index 0000000000..041d584f72 --- /dev/null +++ b/examples/slack-connect-bridge/src/slack/mod.rs @@ -0,0 +1,113 @@ +//! Minimal Slack Events/Web API client for the bridge. + +mod api; +mod webhook; + +pub(crate) use api::SlackClient; +pub(crate) use webhook::{ + run_webhook_server, SlackDelivery, SlackEvent, WebhookControl, WebhookServerState, +}; + +/// Convert the subset of Slack mrkdwn that would otherwise be unreadable in +/// Buzz. Unknown control tokens stay visible instead of being discarded. +pub(crate) fn slack_mrkdwn_to_markdown(input: &str) -> String { + let decoded = input + .replace("&", "&") + .replace("<", "<") + .replace(">", ">"); + let mut output = String::with_capacity(decoded.len()); + let mut rest = decoded.as_str(); + + while let Some(open) = rest.find('<') { + output.push_str(&rest[..open]); + let after_open = &rest[open + 1..]; + let Some(close) = after_open.find('>') else { + output.push_str(&rest[open..]); + return output; + }; + let token = &after_open[..close]; + output.push_str(&convert_control_token(token)); + rest = &after_open[close + 1..]; + } + output.push_str(rest); + output +} + +fn convert_control_token(token: &str) -> String { + if let Some(user_id) = token.strip_prefix('@') { + return format!("@{user_id}"); + } + if let Some(channel) = token.strip_prefix('#') { + let label = channel.split_once('|').map_or(channel, |(_, label)| label); + return format!("#{label}"); + } + if let Some(command) = token.strip_prefix('!') { + let label = command + .split_once('|') + .map_or(command, |(_, label)| label.trim_start_matches('@')); + return format!("@{label}"); + } + if let Some((url, label)) = token.split_once('|') { + if url.starts_with("http://") || url.starts_with("https://") { + return format!("[{label}]({url})"); + } + if let Some(address) = url.strip_prefix("mailto:") { + return format!("[{label}](mailto:{address})"); + } + } + if token.starts_with("http://") || token.starts_with("https://") { + return token.to_owned(); + } + if let Some(address) = token.strip_prefix("mailto:") { + return address.to_owned(); + } + format!("<{token}>") +} + +/// Make an untrusted display name safe in a Buzz Markdown bold span. +pub(crate) fn escape_markdown_label(input: &str) -> String { + input + .chars() + .flat_map(|character| match character { + '\\' | '*' | '_' | '[' | ']' | '`' => vec!['\\', character], + '\r' | '\n' => vec![' '], + other => vec![other], + }) + .take(160) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn converts_slack_links_channels_and_mentions() { + assert_eq!( + slack_mrkdwn_to_markdown( + "See in <#C12345678|project-x> with <@U12345678> & " + ), + "See [the docs](https://example.com) in #project-x with @U12345678 & @here" + ); + } + + #[test] + fn malformed_control_token_stays_visible() { + assert_eq!( + slack_mrkdwn_to_markdown("before after"), + "before after" + ); + } + + #[test] + fn escapes_untrusted_display_names() { + assert_eq!( + escape_markdown_label("*Mallory*\n`admin`"), + "\\*Mallory\\* \\`admin\\`" + ); + } +} diff --git a/examples/slack-connect-bridge/src/slack/webhook.rs b/examples/slack-connect-bridge/src/slack/webhook.rs new file mode 100644 index 0000000000..77eec1761d --- /dev/null +++ b/examples/slack-connect-bridge/src/slack/webhook.rs @@ -0,0 +1,543 @@ +use std::{ + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use anyhow::{Context, Result}; +use axum::{ + body::Bytes, + extract::{DefaultBodyLimit, State}, + http::{HeaderMap, StatusCode}, + response::{IntoResponse, Response}, + routing::{get, post}, + Json, Router, +}; +use hmac::{Hmac, KeyInit, Mac}; +use serde_json::{json, Value}; +use sha2::Sha256; +use tokio::sync::{mpsc, oneshot, watch}; +use tracing::{info, warn}; + +const SIGNATURE_VERSION: &str = "v0"; +const SIGNATURE_MAX_AGE: Duration = Duration::from_secs(5 * 60); +const CALLBACK_COMPLETION_TIMEOUT: Duration = Duration::from_millis(2_500); +const MAX_WEBHOOK_BYTES: usize = 1024 * 1024; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum SlackEvent { + Message { + event_id: String, + team_id: String, + channel_id: String, + user_id: String, + text: String, + ts: String, + thread_ts: Option, + is_ext_shared: Option, + }, + ChannelIdChanged { + event_id: String, + team_id: String, + old_channel_id: String, + new_channel_id: String, + }, + ChannelShared { + event_id: String, + team_id: String, + channel_id: String, + }, + ChannelUnshared { + event_id: String, + team_id: String, + channel_id: String, + is_ext_shared: bool, + }, +} + +pub(crate) struct SlackDelivery { + pub(crate) event: SlackEvent, + pub(crate) completion: oneshot::Sender>, +} + +#[derive(Clone)] +pub(crate) struct WebhookControl { + ready: Arc, +} + +impl WebhookControl { + pub(crate) fn set_ready(&self, ready: bool) { + self.ready.store(ready, Ordering::Release); + } +} + +#[derive(Clone)] +pub(crate) struct WebhookServerState { + signing_secret: Arc<[u8]>, + delivery_tx: mpsc::Sender, + ready: Arc, +} + +impl WebhookServerState { + pub(crate) fn new( + signing_secret: String, + delivery_tx: mpsc::Sender, + ) -> (Self, WebhookControl) { + let ready = Arc::new(AtomicBool::new(false)); + ( + Self { + signing_secret: Arc::from(signing_secret.into_bytes()), + delivery_tx, + ready: Arc::clone(&ready), + }, + WebhookControl { ready }, + ) + } +} + +pub(crate) async fn run_webhook_server( + listener: tokio::net::TcpListener, + state: WebhookServerState, + mut shutdown: watch::Receiver, +) -> Result<()> { + let app = Router::new() + .route("/healthz", get(healthz)) + .route("/slack/events", post(slack_events)) + .layer(DefaultBodyLimit::max(MAX_WEBHOOK_BYTES)) + .with_state(state); + let listen_addr = listener + .local_addr() + .context("failed to read Slack webhook listener address")?; + info!(%listen_addr, "Slack webhook listener ready"); + axum::serve(listener, app) + .with_graceful_shutdown(async move { + while !*shutdown.borrow() { + if shutdown.changed().await.is_err() { + break; + } + } + }) + .await + .context("Slack webhook server failed") +} + +async fn healthz(State(state): State) -> StatusCode { + if state.ready.load(Ordering::Acquire) { + StatusCode::OK + } else { + StatusCode::SERVICE_UNAVAILABLE + } +} + +async fn slack_events( + State(state): State, + headers: HeaderMap, + body: Bytes, +) -> Response { + if let Err(error) = verify_signature(&state.signing_secret, &headers, &body, SystemTime::now()) + { + warn!(reason = %error, "rejected Slack webhook signature"); + return StatusCode::UNAUTHORIZED.into_response(); + } + + let payload: Value = match serde_json::from_slice(&body) { + Ok(payload) => payload, + Err(_) => return StatusCode::BAD_REQUEST.into_response(), + }; + + match payload.get("type").and_then(Value::as_str) { + Some("url_verification") => { + let Some(challenge) = payload.get("challenge").and_then(Value::as_str) else { + return StatusCode::BAD_REQUEST.into_response(); + }; + return Json(json!({ "challenge": challenge })).into_response(); + } + Some("event_callback") => {} + _ => return StatusCode::OK.into_response(), + } + + let event = match parse_callback(&payload) { + Ok(Some(event)) => event, + Ok(None) => return StatusCode::OK.into_response(), + Err(error) => { + warn!(reason = %error, "rejected malformed Slack event callback"); + return StatusCode::BAD_REQUEST.into_response(); + } + }; + + if !state.ready.load(Ordering::Acquire) { + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + } + + let (completion, done) = oneshot::channel(); + match state + .delivery_tx + .try_send(SlackDelivery { event, completion }) + { + Ok(()) => {} + Err(mpsc::error::TrySendError::Full(_)) => { + warn!("Slack delivery queue is full; asking Slack to retry"); + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + } + Err(mpsc::error::TrySendError::Closed(_)) => { + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + } + } + + match tokio::time::timeout(CALLBACK_COMPLETION_TIMEOUT, done).await { + Ok(Ok(Ok(()))) => StatusCode::OK.into_response(), + Ok(Ok(Err(reason))) => { + warn!(%reason, "Slack callback processing failed; asking Slack to retry"); + StatusCode::SERVICE_UNAVAILABLE.into_response() + } + Ok(Err(_)) | Err(_) => { + warn!("Slack callback processing did not complete in time; asking Slack to retry"); + StatusCode::SERVICE_UNAVAILABLE.into_response() + } + } +} + +fn verify_signature( + signing_secret: &[u8], + headers: &HeaderMap, + body: &[u8], + now: SystemTime, +) -> Result<()> { + let timestamp = headers + .get("x-slack-request-timestamp") + .and_then(|value| value.to_str().ok()) + .context("missing Slack request timestamp")?; + let timestamp_secs = timestamp + .parse::() + .context("invalid Slack request timestamp")?; + let now_secs = now + .duration_since(UNIX_EPOCH) + .context("system clock is before the Unix epoch")? + .as_secs(); + if now_secs.abs_diff(timestamp_secs) > SIGNATURE_MAX_AGE.as_secs() { + anyhow::bail!("Slack request timestamp is outside the replay window"); + } + + let signature = headers + .get("x-slack-signature") + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.strip_prefix("v0=")) + .context("missing Slack v0 signature")?; + let signature = hex::decode(signature).context("invalid Slack signature encoding")?; + + let mut mac = Hmac::::new_from_slice(signing_secret) + .context("failed to initialize Slack signature verifier")?; + mac.update(SIGNATURE_VERSION.as_bytes()); + mac.update(b":"); + mac.update(timestamp.as_bytes()); + mac.update(b":"); + mac.update(body); + mac.verify_slice(&signature) + .context("Slack signature mismatch") +} + +fn parse_callback(payload: &Value) -> Result> { + let event_id = required(payload, "event_id")?; + let team_id = required(payload, "team_id")?; + let event = payload + .get("event") + .context("event callback omitted event")?; + let event_type = required(event, "type")?; + + let parsed = match event_type.as_str() { + "message" => parse_message(payload, event_id, team_id, event)?, + "channel_id_changed" => Some(SlackEvent::ChannelIdChanged { + event_id, + team_id, + old_channel_id: required(event, "old_channel_id")?, + new_channel_id: required(event, "new_channel_id")?, + }), + "channel_shared" => Some(SlackEvent::ChannelShared { + event_id, + team_id, + channel_id: required(event, "channel")?, + }), + "channel_unshared" => Some(SlackEvent::ChannelUnshared { + event_id, + team_id, + channel_id: required(event, "channel")?, + is_ext_shared: required_bool(event, "is_ext_shared")?, + }), + _ => None, + }; + Ok(parsed) +} + +fn parse_message( + payload: &Value, + event_id: String, + team_id: String, + event: &Value, +) -> Result> { + if event.get("bot_id").is_some() || event.get("hidden").and_then(Value::as_bool) == Some(true) { + return Ok(None); + } + let subtype = event.get("subtype").and_then(Value::as_str); + if subtype.is_some_and(|value| value != "thread_broadcast") { + return Ok(None); + } + + Ok(Some(SlackEvent::Message { + event_id, + team_id, + channel_id: required(event, "channel")?, + user_id: required(event, "user")?, + text: required_allow_empty(event, "text")?, + ts: required(event, "ts")?, + thread_ts: event + .get("thread_ts") + .and_then(Value::as_str) + .map(str::to_owned), + is_ext_shared: payload + .get("is_ext_shared_channel") + .and_then(Value::as_bool), + })) +} + +fn required(value: &Value, field: &str) -> Result { + value + .get(field) + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + .with_context(|| format!("Slack callback omitted {field}")) +} + +fn required_allow_empty(value: &Value, field: &str) -> Result { + value + .get(field) + .and_then(Value::as_str) + .map(str::to_owned) + .with_context(|| format!("Slack callback omitted {field}")) +} + +fn required_bool(value: &Value, field: &str) -> Result { + value + .get(field) + .and_then(Value::as_bool) + .with_context(|| format!("Slack callback omitted {field}")) +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::body::to_bytes; + use axum::http::HeaderValue; + + fn signed_headers(secret: &[u8], timestamp: u64, body: &[u8]) -> HeaderMap { + let timestamp = timestamp.to_string(); + let mut mac = Hmac::::new_from_slice(secret).unwrap(); + mac.update(format!("v0:{timestamp}:").as_bytes()); + mac.update(body); + let signature = format!("v0={}", hex::encode(mac.finalize().into_bytes())); + let mut headers = HeaderMap::new(); + headers.insert( + "x-slack-request-timestamp", + HeaderValue::from_str(×tamp).unwrap(), + ); + headers.insert( + "x-slack-signature", + HeaderValue::from_str(&signature).unwrap(), + ); + headers + } + + fn current_signed_headers(secret: &[u8], body: &[u8]) -> HeaderMap { + let timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs(); + signed_headers(secret, timestamp, body) + } + + #[test] + fn accepts_valid_signature_and_rejects_tampering() { + let now = UNIX_EPOCH + Duration::from_secs(1_700_000_000); + let body = br#"{"type":"event_callback"}"#; + let headers = signed_headers(b"secret", 1_700_000_000, body); + verify_signature(b"secret", &headers, body, now).unwrap(); + assert!(verify_signature(b"secret", &headers, b"tampered", now).is_err()); + } + + #[test] + fn rejects_replayed_signature() { + let body = br#"{}"#; + let headers = signed_headers(b"secret", 1_700_000_000, body); + let now = UNIX_EPOCH + Duration::from_secs(1_700_000_301); + let error = verify_signature(b"secret", &headers, body, now) + .unwrap_err() + .to_string(); + assert!(error.contains("replay window"), "{error}"); + } + + #[test] + fn parses_external_thread_message() { + let payload = json!({ + "type": "event_callback", + "event_id": "Ev123", + "team_id": "T12345678", + "is_ext_shared_channel": true, + "event": { + "type": "message", + "channel": "C12345678", + "user": "U12345678", + "text": "hello", + "ts": "1700000000.000001", + "thread_ts": "1699999999.000001" + } + }); + assert_eq!( + parse_callback(&payload).unwrap(), + Some(SlackEvent::Message { + event_id: "Ev123".into(), + team_id: "T12345678".into(), + channel_id: "C12345678".into(), + user_id: "U12345678".into(), + text: "hello".into(), + ts: "1700000000.000001".into(), + thread_ts: Some("1699999999.000001".into()), + is_ext_shared: Some(true), + }) + ); + } + + #[test] + fn parses_shared_channel_lifecycle_events() { + let unshared = json!({ + "event_id": "Ev-unshared", + "team_id": "T12345678", + "event": { + "type": "channel_unshared", + "channel": "C12345678", + "is_ext_shared": true + } + }); + assert_eq!( + parse_callback(&unshared).unwrap(), + Some(SlackEvent::ChannelUnshared { + event_id: "Ev-unshared".into(), + team_id: "T12345678".into(), + channel_id: "C12345678".into(), + is_ext_shared: true, + }) + ); + + let changed = json!({ + "event_id": "Ev-changed", + "team_id": "T12345678", + "event": { + "type": "channel_id_changed", + "old_channel_id": "C12345678", + "new_channel_id": "C87654321" + } + }); + assert_eq!( + parse_callback(&changed).unwrap(), + Some(SlackEvent::ChannelIdChanged { + event_id: "Ev-changed".into(), + team_id: "T12345678".into(), + old_channel_id: "C12345678".into(), + new_channel_id: "C87654321".into(), + }) + ); + } + + #[tokio::test] + async fn handles_signed_url_verification_before_ready() { + let (delivery_tx, _delivery_rx) = mpsc::channel(1); + let (state, _control) = WebhookServerState::new("secret".into(), delivery_tx); + let body = + Bytes::from_static(br#"{"type":"url_verification","challenge":"slack-challenge"}"#); + let response = + slack_events(State(state), current_signed_headers(b"secret", &body), body).await; + assert_eq!(response.status(), StatusCode::OK); + let body = to_bytes(response.into_body(), 1024).await.unwrap(); + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + json!({"challenge": "slack-challenge"}) + ); + } + + #[tokio::test] + async fn asks_slack_to_retry_when_bridge_is_not_ready() { + let (delivery_tx, _delivery_rx) = mpsc::channel(1); + let (state, _control) = WebhookServerState::new("secret".into(), delivery_tx); + let body = Bytes::from_static( + br#"{ + "type":"event_callback", + "event_id":"Ev123", + "team_id":"T12345678", + "event":{ + "type":"message", + "channel":"C12345678", + "user":"U12345678", + "text":"hello", + "ts":"1700000000.000001" + } + }"#, + ); + let response = + slack_events(State(state), current_signed_headers(b"secret", &body), body).await; + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + } + + #[tokio::test] + async fn acknowledges_callback_after_durable_processing() { + let (delivery_tx, mut delivery_rx) = mpsc::channel(1); + let (state, control) = WebhookServerState::new("secret".into(), delivery_tx); + control.set_ready(true); + let body = Bytes::from_static( + br#"{ + "type":"event_callback", + "event_id":"Ev123", + "team_id":"T12345678", + "event":{ + "type":"message", + "channel":"C12345678", + "user":"U12345678", + "text":"hello", + "ts":"1700000000.000001" + } + }"#, + ); + + let processing = tokio::spawn(async move { + let delivery = delivery_rx.recv().await.unwrap(); + assert!(matches!(delivery.event, SlackEvent::Message { .. })); + delivery.completion.send(Ok(())).unwrap(); + }); + let response = + slack_events(State(state), current_signed_headers(b"secret", &body), body).await; + processing.await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + } + + #[test] + fn ignores_bot_and_edit_events() { + let base = json!({ + "event_id": "Ev123", + "team_id": "T12345678", + "event": { + "type": "message", + "channel": "C12345678", + "user": "U12345678", + "text": "hello", + "ts": "1700000000.000001", + "bot_id": "B123" + } + }); + assert_eq!(parse_callback(&base).unwrap(), None); + + let mut edit = base; + edit["event"]["bot_id"] = Value::Null; + edit["event"]["subtype"] = Value::String("message_changed".into()); + assert_eq!(parse_callback(&edit).unwrap(), None); + } +} diff --git a/examples/slack-connect-bridge/src/state.rs b/examples/slack-connect-bridge/src/state.rs new file mode 100644 index 0000000000..4b3d29c9a5 --- /dev/null +++ b/examples/slack-connect-bridge/src/state.rs @@ -0,0 +1,417 @@ +//! Durable bridge identity mappings. +//! +//! Slack timestamps and Buzz event IDs are the cross-system idempotency keys. +//! State is written through an atomic same-directory rename so a process crash +//! cannot leave a partially written JSON document. + +use std::{ + collections::{BTreeMap, BTreeSet, HashSet}, + fs::OpenOptions, + io::Write, + path::{Path, PathBuf}, +}; + +use anyhow::{bail, Context, Result}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +const STATE_VERSION: u32 = 1; +const MAX_ALIAS_HOPS: usize = 16; + +/// Slack-side identity for one bridged message. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub(crate) struct SlackMessageRef { + pub(crate) team_id: String, + pub(crate) channel_id: String, + pub(crate) ts: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) thread_ts: Option, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +struct PersistedState { + version: u32, + #[serde(default)] + slack_to_buzz: BTreeMap, + #[serde(default)] + buzz_to_slack: BTreeMap, + #[serde(default)] + channel_aliases: BTreeMap, + #[serde(default)] + paused_buzz_channels: BTreeSet, + #[serde(default)] + last_buzz_created_at: Option, + #[serde(default)] + slack_user_names: BTreeMap, +} + +impl Default for PersistedState { + fn default() -> Self { + Self { + version: STATE_VERSION, + slack_to_buzz: BTreeMap::new(), + buzz_to_slack: BTreeMap::new(), + channel_aliases: BTreeMap::new(), + paused_buzz_channels: BTreeSet::new(), + last_buzz_created_at: None, + slack_user_names: BTreeMap::new(), + } + } +} + +#[derive(Debug)] +pub(crate) struct StateStore { + path: PathBuf, + data: PersistedState, +} + +impl StateStore { + pub(crate) fn load(path: PathBuf) -> Result { + let data = match std::fs::read(&path) { + Ok(bytes) => { + let parsed: PersistedState = serde_json::from_slice(&bytes) + .with_context(|| format!("failed to parse {}", path.display()))?; + if parsed.version != STATE_VERSION { + bail!( + "unsupported bridge state version {} in {} (expected {STATE_VERSION})", + parsed.version, + path.display() + ); + } + validate_aliases(&parsed.channel_aliases)?; + parsed + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => PersistedState::default(), + Err(error) => { + return Err(error).with_context(|| format!("failed to read {}", path.display())); + } + }; + Ok(Self { path, data }) + } + + pub(crate) fn buzz_event_for_slack( + &self, + buzz_channel_id: Uuid, + slack_ts: &str, + ) -> Option<&str> { + self.data + .slack_to_buzz + .get(&slack_key(buzz_channel_id, slack_ts)) + .map(String::as_str) + } + + pub(crate) fn slack_message_for_buzz(&self, buzz_event_id: &str) -> Option<&SlackMessageRef> { + self.data.buzz_to_slack.get(buzz_event_id) + } + + pub(crate) fn record_message_pair( + &mut self, + buzz_channel_id: Uuid, + buzz_event_id: &str, + slack: SlackMessageRef, + ) -> Result<()> { + let event_id = buzz_event_id.to_owned(); + let slack_ts = slack.ts.clone(); + self.commit(move |next| { + next.slack_to_buzz + .insert(slack_key(buzz_channel_id, &slack_ts), event_id.clone()); + next.buzz_to_slack.insert(event_id, slack); + Ok(()) + }) + } + + pub(crate) fn canonical_channel_id(&self, team_id: &str, channel_id: &str) -> String { + canonical_channel_id(&self.data.channel_aliases, team_id, channel_id) + } + + pub(crate) fn record_channel_id_change( + &mut self, + team_id: &str, + old_channel_id: &str, + new_channel_id: &str, + ) -> Result<()> { + let team_id = team_id.to_owned(); + let old_channel_id = old_channel_id.to_owned(); + let new_channel_id = new_channel_id.to_owned(); + self.commit(move |next| { + let old_canonical = + canonical_channel_id(&next.channel_aliases, &team_id, &old_channel_id); + let new_canonical = + canonical_channel_id(&next.channel_aliases, &team_id, &new_channel_id); + if old_canonical == new_canonical { + return Ok(()); + } + if new_canonical == old_channel_id { + bail!("Slack channel ID change would create an alias cycle"); + } + next.channel_aliases + .insert(channel_alias_key(&team_id, &old_canonical), new_canonical); + validate_aliases(&next.channel_aliases) + }) + } + + pub(crate) fn set_route_paused(&mut self, buzz_channel_id: Uuid, paused: bool) -> Result<()> { + self.commit(move |next| { + if paused { + next.paused_buzz_channels.insert(buzz_channel_id); + } else { + next.paused_buzz_channels.remove(&buzz_channel_id); + } + Ok(()) + }) + } + + pub(crate) fn route_is_paused(&self, buzz_channel_id: Uuid) -> bool { + self.data.paused_buzz_channels.contains(&buzz_channel_id) + } + + /// Return a safe replay cursor. + /// + /// A brand-new bridge begins at `now` so enabling it cannot backfill a + /// channel into Slack. Restarts replay a bounded window before the last + /// successfully handled event; durable message mappings suppress + /// duplicates inside that window. + pub(crate) fn subscription_since(&mut self, now: u64, lookback_secs: u64) -> Result { + if let Some(cursor) = self.data.last_buzz_created_at { + return Ok(cursor.saturating_sub(lookback_secs)); + } + self.commit(move |next| { + next.last_buzz_created_at = Some(now); + Ok(()) + })?; + Ok(now) + } + + pub(crate) fn record_buzz_cursor(&mut self, created_at: u64) -> Result<()> { + if self + .data + .last_buzz_created_at + .is_some_and(|current| current >= created_at) + { + return Ok(()); + } + self.commit(move |next| { + next.last_buzz_created_at = Some(created_at); + Ok(()) + }) + } + + pub(crate) fn slack_user_name(&self, user_id: &str) -> Option<&str> { + self.data.slack_user_names.get(user_id).map(String::as_str) + } + + pub(crate) fn record_slack_user_name(&mut self, user_id: &str, name: &str) -> Result<()> { + if self.data.slack_user_names.get(user_id).map(String::as_str) == Some(name) { + return Ok(()); + } + let user_id = user_id.to_owned(); + let name = name.to_owned(); + self.commit(move |next| { + next.slack_user_names.insert(user_id, name); + Ok(()) + }) + } + + fn commit(&mut self, update: impl FnOnce(&mut PersistedState) -> Result<()>) -> Result<()> { + let mut next = self.data.clone(); + update(&mut next)?; + write_atomic(&self.path, &next)?; + self.data = next; + Ok(()) + } +} + +fn slack_key(buzz_channel_id: Uuid, slack_ts: &str) -> String { + format!("{buzz_channel_id}:{slack_ts}") +} + +fn channel_alias_key(team_id: &str, channel_id: &str) -> String { + format!("{team_id}:{channel_id}") +} + +fn canonical_channel_id( + aliases: &BTreeMap, + team_id: &str, + channel_id: &str, +) -> String { + let mut current = channel_id.to_owned(); + let mut seen = HashSet::new(); + for _ in 0..MAX_ALIAS_HOPS { + if !seen.insert(current.clone()) { + break; + } + let Some(next) = aliases.get(&channel_alias_key(team_id, ¤t)) else { + break; + }; + current.clone_from(next); + } + current +} + +fn validate_aliases(aliases: &BTreeMap) -> Result<()> { + for key in aliases.keys() { + let Some((team_id, channel_id)) = key.split_once(':') else { + bail!("invalid Slack channel alias key in bridge state"); + }; + let mut current = channel_id.to_owned(); + let mut seen = HashSet::new(); + for _ in 0..=MAX_ALIAS_HOPS { + if !seen.insert(current.clone()) { + bail!("Slack channel alias cycle in bridge state"); + } + let Some(next) = aliases.get(&channel_alias_key(team_id, ¤t)) else { + break; + }; + current.clone_from(next); + } + if seen.len() > MAX_ALIAS_HOPS { + bail!("Slack channel alias chain is too deep in bridge state"); + } + } + Ok(()) +} + +fn write_atomic(path: &Path, data: &PersistedState) -> Result<()> { + let parent = path.parent().unwrap_or_else(|| Path::new(".")); + std::fs::create_dir_all(parent) + .with_context(|| format!("failed to create {}", parent.display()))?; + + let file_name = path + .file_name() + .and_then(|name| name.to_str()) + .context("bridge state path must have a UTF-8 file name")?; + let temp_path = parent.join(format!(".{file_name}.{}.tmp", std::process::id())); + let bytes = serde_json::to_vec_pretty(data).context("failed to serialize bridge state")?; + + let mut options = OpenOptions::new(); + options.create(true).truncate(true).write(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let mut file = options + .open(&temp_path) + .with_context(|| format!("failed to open {}", temp_path.display()))?; + file.write_all(&bytes) + .with_context(|| format!("failed to write {}", temp_path.display()))?; + file.write_all(b"\n") + .with_context(|| format!("failed to write {}", temp_path.display()))?; + file.sync_all() + .with_context(|| format!("failed to sync {}", temp_path.display()))?; + std::fs::rename(&temp_path, path) + .with_context(|| format!("failed to replace {}", path.display()))?; + #[cfg(unix)] + { + std::fs::File::open(parent) + .with_context(|| format!("failed to open {} for sync", parent.display()))? + .sync_all() + .with_context(|| format!("failed to sync {}", parent.display()))?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn message_pairs_survive_restart() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("state.json"); + let channel = Uuid::new_v4(); + let slack = SlackMessageRef { + team_id: "T12345678".into(), + channel_id: "C12345678".into(), + ts: "123.456".into(), + thread_ts: None, + }; + + let mut state = StateStore::load(path.clone()).unwrap(); + state + .record_message_pair(channel, &"a".repeat(64), slack.clone()) + .unwrap(); + + let reloaded = StateStore::load(path).unwrap(); + assert_eq!( + reloaded.buzz_event_for_slack(channel, "123.456"), + Some("a".repeat(64).as_str()) + ); + assert_eq!( + reloaded.slack_message_for_buzz(&"a".repeat(64)), + Some(&slack) + ); + } + + #[test] + fn channel_aliases_follow_multiple_id_changes() { + let dir = tempfile::tempdir().unwrap(); + let mut state = StateStore::load(dir.path().join("state.json")).unwrap(); + state + .record_channel_id_change("T12345678", "G12345678", "C12345678") + .unwrap(); + state + .record_channel_id_change("T12345678", "C12345678", "C87654321") + .unwrap(); + assert_eq!( + state.canonical_channel_id("T12345678", "G12345678"), + "C87654321" + ); + } + + #[test] + fn pause_state_is_durable() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("state.json"); + let channel = Uuid::new_v4(); + let mut state = StateStore::load(path.clone()).unwrap(); + state.set_route_paused(channel, true).unwrap(); + assert!(StateStore::load(path).unwrap().route_is_paused(channel)); + } + + #[test] + fn rejects_alias_cycles_on_load() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("state.json"); + std::fs::write( + &path, + r#"{ + "version": 1, + "channel_aliases": { + "T12345678:C12345678": "C87654321", + "T12345678:C87654321": "C12345678" + } + }"#, + ) + .unwrap(); + let error = StateStore::load(path).unwrap_err().to_string(); + assert!(error.contains("alias cycle"), "{error}"); + } + + #[test] + fn first_subscription_starts_now_and_restart_replays_window() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("state.json"); + let mut state = StateStore::load(path.clone()).unwrap(); + assert_eq!(state.subscription_since(10_000, 3_600).unwrap(), 10_000); + state.record_buzz_cursor(12_000).unwrap(); + + let mut reloaded = StateStore::load(path).unwrap(); + assert_eq!(reloaded.subscription_since(20_000, 3_600).unwrap(), 8_400); + } + + #[test] + fn slack_user_names_are_durable_for_deterministic_retries() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("state.json"); + let mut state = StateStore::load(path.clone()).unwrap(); + state + .record_slack_user_name("U12345678", "External Partner") + .unwrap(); + assert_eq!( + StateStore::load(path).unwrap().slack_user_name("U12345678"), + Some("External Partner") + ); + } +}