autumn_plugin_slack/
plugin.rs

1//! `SlackPlugin` and the started runtime.
2
3use std::borrow::Cow;
4use std::future::Future;
5use std::sync::{Arc, Mutex, OnceLock};
6use std::time::Duration;
7
8use autumn_web::actuator::{HealthIndicator, MetricsSource};
9use autumn_web::app::AppBuilder;
10use autumn_web::config::AutumnConfig;
11use autumn_web::plugin::Plugin;
12use autumn_web::plugin_contract::PluginContract;
13use autumn_web::reexports::axum::Router;
14use autumn_web::time::{ClockSource, SystemClock};
15use autumn_web::webhook::{InMemoryWebhookReplayStore, WebhookReplayStore};
16use autumn_web::{AppState, AutumnError, Role};
17use tokio_util::task::TaskTracker;
18
19use crate::client::SlackClient;
20use crate::config::{SECTION, SlackConfig, env_var};
21use crate::engine::Engine;
22use crate::error::{HandlerResult, SlackError};
23use crate::handlers::{Registration, Registry, Runner, SlackContext, boxed};
24use crate::health::SlackHealth;
25use crate::metrics::SlackMetrics;
26use crate::payload::{
27    BlockAction, CommandResponse, EventCallback, Message, Shortcut, SlashCommand, ViewPayload,
28    ViewResponse,
29};
30use crate::routes::{self, COMMANDS, EVENTS, EngineCell, INTERACTIONS};
31use crate::transport::{HttpTransport, ReqwestTransport};
32use crate::verify::Verifier;
33
34/// Plugin name for duplicate detection.
35pub const PLUGIN_NAME: &str = "autumn-plugin-slack";
36/// Default base path of the Slack routes.
37pub const DEFAULT_BASE_PATH: &str = "/slack";
38
39/// Slack plugin.
40///
41/// ```rust,ignore
42/// autumn_web::app()
43///     .plugin(
44///         SlackPlugin::new()
45///             .on_event("app_mention", on_mention)
46///             .command("/deploy", deploy),
47///     )
48///     .run()
49///     .await;
50/// ```
51pub struct SlackPlugin {
52    config: Option<SlackConfig>,
53    base_path: String,
54    readiness: bool,
55    signing_secret: Option<String>,
56    previous_secrets: Vec<String>,
57    bot_token: Option<String>,
58    transport: Option<Arc<dyn HttpTransport>>,
59    clock: Option<Arc<dyn ClockSource>>,
60    replay: Option<Arc<dyn WebhookReplayStore>>,
61    registrations: Vec<Registration>,
62    metrics: Arc<SlackMetrics>,
63}
64
65impl std::fmt::Debug for SlackPlugin {
66    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
67        f.debug_struct("SlackPlugin")
68            .field("base_path", &self.base_path)
69            .field(
70                "handlers",
71                &self
72                    .registrations
73                    .iter()
74                    .map(Registration::describe)
75                    .collect::<Vec<_>>(),
76            )
77            .finish_non_exhaustive()
78    }
79}
80
81impl Default for SlackPlugin {
82    fn default() -> Self {
83        Self::new()
84    }
85}
86
87impl SlackPlugin {
88    /// Makes the plugin. It reads `[slack]` at startup.
89    #[must_use]
90    pub fn new() -> Self {
91        Self {
92            config: None,
93            base_path: DEFAULT_BASE_PATH.to_owned(),
94            readiness: false,
95            signing_secret: None,
96            previous_secrets: Vec::new(),
97            bot_token: None,
98            transport: None,
99            clock: None,
100            replay: None,
101            registrations: Vec::new(),
102            metrics: Arc::new(SlackMetrics::new()),
103        }
104    }
105
106    /// Makes the plugin with this config. It reads no files.
107    #[must_use]
108    pub fn with_config(config: SlackConfig) -> Self {
109        Self {
110            config: Some(config),
111            ..Self::new()
112        }
113    }
114
115    /// Sets the base path of the routes. Default: `/slack`.
116    ///
117    /// autumn mounts routes before it loads config. So this is a builder
118    /// setting, not a config key.
119    #[must_use]
120    pub fn base_path(mut self, path: &str) -> Self {
121        self.base_path = normalize_base(path);
122        self
123    }
124
125    /// Puts the health indicator in `/ready` too. Default: only in
126    /// `/actuator/health`.
127    #[must_use]
128    pub const fn readiness(mut self, on: bool) -> Self {
129        self.readiness = on;
130        self
131    }
132
133    /// Uses this signing secret, not the env var.
134    #[must_use]
135    pub fn with_signing_secret(mut self, secret: impl Into<String>) -> Self {
136        self.signing_secret = Some(secret.into());
137        self
138    }
139
140    /// Also accepts this old signing secret. Use during rotation.
141    #[must_use]
142    pub fn with_previous_signing_secret(mut self, secret: impl Into<String>) -> Self {
143        self.previous_secrets.push(secret.into());
144        self
145    }
146
147    /// Uses this bot token, not the env var.
148    #[must_use]
149    pub fn with_bot_token(mut self, token: impl Into<String>) -> Self {
150        self.bot_token = Some(token.into());
151        self
152    }
153
154    /// Uses this transport. Default: [`crate::transport::ReqwestTransport`].
155    #[must_use]
156    pub fn with_transport(self, transport: impl HttpTransport) -> Self {
157        self.with_transport_arc(Arc::new(transport))
158    }
159
160    /// Uses this shared transport.
161    #[must_use]
162    pub fn with_transport_arc(mut self, transport: Arc<dyn HttpTransport>) -> Self {
163        self.transport = Some(transport);
164        self
165    }
166
167    /// Uses this clock for the timestamp check. Default: the system clock.
168    #[must_use]
169    pub fn with_clock(mut self, clock: Arc<dyn ClockSource>) -> Self {
170        self.clock = Some(clock);
171        self
172    }
173
174    /// Uses this store to drop event retries. Default: in memory, per
175    /// process. Use a Redis store for many replicas.
176    #[must_use]
177    pub fn with_replay_store(mut self, store: Arc<dyn WebhookReplayStore>) -> Self {
178        self.replay = Some(store);
179        self
180    }
181
182    /// Runs `handler` for Events API events of `event_type`, after the ack.
183    #[must_use]
184    pub fn on_event<F, Fut>(mut self, event_type: &str, handler: F) -> Self
185    where
186        F: Fn(SlackContext, EventCallback) -> Fut + Send + Sync + 'static,
187        Fut: Future<Output = HandlerResult<()>> + Send + 'static,
188    {
189        self.registrations
190            .push(Registration::Event(event_type.to_owned(), boxed(handler)));
191        self
192    }
193
194    /// Runs `handler` for the slash command `name`, for example `/deploy`.
195    #[must_use]
196    pub fn command<F, Fut>(mut self, name: &str, handler: F) -> Self
197    where
198        F: Fn(SlackContext, SlashCommand) -> Fut + Send + Sync + 'static,
199        Fut: Future<Output = HandlerResult<CommandResponse>> + Send + 'static,
200    {
201        self.registrations
202            .push(Registration::Command(name.to_owned(), boxed(handler)));
203        self
204    }
205
206    /// Runs `handler` for a block action with this `action_id`, after the
207    /// ack. A returned message goes to the `response_url`.
208    #[must_use]
209    pub fn action<F, Fut>(mut self, action_id: &str, handler: F) -> Self
210    where
211        F: Fn(SlackContext, BlockAction) -> Fut + Send + Sync + 'static,
212        Fut: Future<Output = HandlerResult<Option<Message>>> + Send + 'static,
213    {
214        self.registrations
215            .push(Registration::Action(action_id.to_owned(), boxed(handler)));
216        self
217    }
218
219    /// Runs `handler` for a modal submit with this `callback_id`.
220    #[must_use]
221    pub fn view_submission<F, Fut>(mut self, callback_id: &str, handler: F) -> Self
222    where
223        F: Fn(SlackContext, ViewPayload) -> Fut + Send + Sync + 'static,
224        Fut: Future<Output = HandlerResult<ViewResponse>> + Send + 'static,
225    {
226        self.registrations.push(Registration::ViewSubmission(
227            callback_id.to_owned(),
228            boxed(handler),
229        ));
230        self
231    }
232
233    /// Runs `handler` when a user closes a modal with this `callback_id`.
234    /// The modal must set `notify_on_close`.
235    #[must_use]
236    pub fn view_closed<F, Fut>(mut self, callback_id: &str, handler: F) -> Self
237    where
238        F: Fn(SlackContext, ViewPayload) -> Fut + Send + Sync + 'static,
239        Fut: Future<Output = HandlerResult<()>> + Send + 'static,
240    {
241        self.registrations.push(Registration::ViewClosed(
242            callback_id.to_owned(),
243            boxed(handler),
244        ));
245        self
246    }
247
248    /// Runs `handler` for a global or message shortcut with this
249    /// `callback_id`, after the ack.
250    #[must_use]
251    pub fn shortcut<F, Fut>(mut self, callback_id: &str, handler: F) -> Self
252    where
253        F: Fn(SlackContext, Shortcut) -> Fut + Send + Sync + 'static,
254        Fut: Future<Output = HandlerResult<()>> + Send + 'static,
255    {
256        self.registrations.push(Registration::Shortcut(
257            callback_id.to_owned(),
258            boxed(handler),
259        ));
260        self
261    }
262
263    /// The metrics. They are shared with the started runtime.
264    #[must_use]
265    pub fn metrics(&self) -> Arc<SlackMetrics> {
266        Arc::clone(&self.metrics)
267    }
268
269    /// Starts the plugin outside an app build. Use it in tests, with
270    /// [`SlackRuntime::router`]. [`Plugin::build`] does the same at startup.
271    ///
272    /// # Errors
273    /// Returns a config error.
274    pub fn start(self, state: &AppState) -> Result<SlackRuntime, SlackError> {
275        let role = state.role().clone();
276        self.start_with_role(state, role)
277    }
278
279    /// Starts the plugin as if the process has `role`. A role that serves no
280    /// HTTP (`worker`, or a preset without HTTP) needs no signing secret and
281    /// no CSRF exemption. A [`ProcessRole`](autumn_web::ProcessRole) converts into a [`Role`].
282    ///
283    /// # Errors
284    /// Same as [`Self::start`].
285    pub fn start_with_role(
286        self,
287        state: &AppState,
288        role: impl Into<Role>,
289    ) -> Result<SlackRuntime, SlackError> {
290        let health = Arc::new(SlackHealth::new(self.readiness));
291        self.start_inner(state, &role.into(), health)
292    }
293
294    #[allow(clippy::too_many_lines)] // One linear setup sequence.
295    fn start_inner(
296        self,
297        state: &AppState,
298        role: &Role,
299        health: Arc<SlackHealth>,
300    ) -> Result<SlackRuntime, SlackError> {
301        let config = match self.config {
302            Some(c) => c,
303            // autumn sets the state profile from the config; "default" means none.
304            None => SlackConfig::load(Some(state.profile()).filter(|p| *p != "default"))?,
305        };
306        config.validate()?;
307        // A worker serves no Slack route, so it needs no inbound checks.
308        let inbound = role.serves_http();
309        if inbound {
310            check_security(&state.config(), &self.base_path)?;
311        }
312        let registry = Registry::build(self.registrations)?;
313        let secret = self
314            .signing_secret
315            .or_else(|| env_var(&config.signing_secret_env))
316            .filter(|s| !s.trim().is_empty());
317        let secret = match secret {
318            Some(s) => s,
319            None if !inbound => String::new(),
320            None => {
321                return Err(SlackError::Config(format!(
322                    "no signing secret; set the env var {}",
323                    config.signing_secret_env
324                )));
325            }
326        };
327        // An empty secret would let anyone sign. `Verifier` drops it too.
328        let mut previous: Vec<String> = self
329            .previous_secrets
330            .into_iter()
331            .filter(|s| !s.trim().is_empty())
332            .collect();
333        for name in &config.previous_signing_secret_envs {
334            if let Some(v) = env_var(name).filter(|s| !s.trim().is_empty()) {
335                previous.push(v);
336            } else {
337                tracing::warn!(env = %name, "slack: previous signing secret env var is not set");
338            }
339        }
340        let token = self
341            .bot_token
342            .or_else(|| env_var(&config.bot_token_env))
343            .filter(|t| !t.trim().is_empty());
344        if token.is_none() {
345            tracing::warn!(env = %config.bot_token_env, "slack: no bot token; Web API calls fail");
346        }
347        let transport: Arc<dyn HttpTransport> = match self.transport {
348            Some(t) => t,
349            None => Arc::new(
350                ReqwestTransport::new(Duration::from_millis(config.api.timeout_ms))
351                    .map_err(|e| SlackError::Config(e.0))?,
352            ),
353        };
354        let client = SlackClient::new(transport, config.clone(), token, Arc::clone(&self.metrics));
355        health.set(
356            client.clone(),
357            Duration::from_secs(config.health.cache_secs),
358        );
359        state.insert_extension(client.clone());
360        tracing::info!(
361            base_path = %self.base_path,
362            handlers = registry.len(),
363            "slack started"
364        );
365        let engine = Engine {
366            verifier: Verifier::new(&secret, &previous, config.timestamp_tolerance_secs),
367            registry,
368            client,
369            metrics: Arc::clone(&self.metrics),
370            clock: self.clock.unwrap_or_else(|| Arc::new(SystemClock)),
371            replay: self
372                .replay
373                .unwrap_or_else(|| Arc::new(InMemoryWebhookReplayStore::default())),
374            runner: Runner {
375                tracker: TaskTracker::new(),
376                metrics: Arc::clone(&self.metrics),
377            },
378            state: state.clone(),
379            config,
380        };
381        Ok(SlackRuntime {
382            engine: Arc::new(engine),
383            health,
384        })
385    }
386}
387
388/// `"/slack/"` and `"slack"` become `"/slack"`. `"/"` becomes `""` (the root).
389fn normalize_base(path: &str) -> String {
390    let trimmed = path.trim().trim_matches('/');
391    if trimmed.is_empty() {
392        String::new()
393    } else {
394        format!("/{trimmed}")
395    }
396}
397
398/// The full paths of the three routes.
399fn route_paths(base: &str) -> [String; 3] {
400    [EVENTS, COMMANDS, INTERACTIONS].map(|p| format!("{base}{p}"))
401}
402
403/// Same rule as autumn: an exact match, or a `/`-delimited subtree.
404fn is_exempt(path: &str, exempt: &[String]) -> bool {
405    exempt.iter().any(|p| {
406        let p = p.as_str();
407        path == p
408            || (path.starts_with(p)
409                && (p.ends_with('/') || path.as_bytes().get(p.len()) == Some(&b'/')))
410    })
411}
412
413/// Fails when CSRF or CAPTCHA would block the Slack routes. A plugin cannot
414/// add an exemption, so the error gives the config fix.
415fn check_security(cfg: &AutumnConfig, base: &str) -> Result<(), SlackError> {
416    // At the root, name the exact paths: "/" would exempt the whole app.
417    let fix = if base.is_empty() {
418        format!("{:?}", route_paths(base))
419    } else {
420        format!("[\"{base}\"]")
421    };
422    let mut csrf_exempt = cfg.security.csrf.exempt_paths.clone();
423    csrf_exempt.extend(
424        cfg.security
425            .webhooks
426            .endpoints
427            .iter()
428            .map(|e| e.path.clone()),
429    );
430    let [events, commands, interactions] = route_paths(base);
431    for path in [&events, &commands, &interactions] {
432        if cfg.security.csrf.enabled && !is_exempt(path, &csrf_exempt) {
433            return Err(SlackError::Config(format!(
434                "CSRF protection blocks {path}. Slack sends no CSRF token. \
435                 Add {fix} to [security.csrf] exempt_paths. \
436                 The Slack signature check protects these routes."
437            )));
438        }
439    }
440    // autumn checks CAPTCHA only on form bodies, and not with `dev_bypass`.
441    let bot = &cfg.bot_protection;
442    let mut captcha_exempt = cfg.security.captcha_exempt_paths.clone();
443    captcha_exempt.extend(
444        cfg.security
445            .webhooks
446            .endpoints
447            .iter()
448            .map(|e| e.path.clone()),
449    );
450    for path in [&commands, &interactions] {
451        if bot.enabled && !bot.dev_bypass && !is_exempt(path, &captcha_exempt) {
452            return Err(SlackError::Config(format!(
453                "Bot protection blocks {path}. Slack sends no CAPTCHA token. \
454                 Add {fix} to [security] captcha_exempt_paths."
455            )));
456        }
457    }
458    Ok(())
459}
460
461/// A started plugin.
462pub struct SlackRuntime {
463    engine: Arc<Engine>,
464    health: Arc<SlackHealth>,
465}
466
467impl std::fmt::Debug for SlackRuntime {
468    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
469        f.debug_struct("SlackRuntime")
470            .field("in_flight", &self.in_flight())
471            .finish_non_exhaustive()
472    }
473}
474
475impl SlackRuntime {
476    /// The Web API client.
477    #[must_use]
478    pub fn client(&self) -> SlackClient {
479        self.engine.client.clone()
480    }
481
482    /// The metrics.
483    #[must_use]
484    pub fn metrics(&self) -> Arc<SlackMetrics> {
485        Arc::clone(&self.engine.metrics)
486    }
487
488    /// The health indicator.
489    #[must_use]
490    pub fn health(&self) -> Arc<SlackHealth> {
491        Arc::clone(&self.health)
492    }
493
494    /// Handlers that run now. Reply tasks do not count.
495    #[must_use]
496    pub fn in_flight(&self) -> u64 {
497        self.engine.metrics.in_flight()
498    }
499
500    /// The three routes, relative to the base path. Use it in tests, or to
501    /// mount the routes yourself.
502    pub fn router(&self) -> Router {
503        let cell: EngineCell = Arc::new(OnceLock::new());
504        let _ = cell.set(Arc::clone(&self.engine));
505        routes::router(&cell)
506    }
507
508    /// Stops new handler runs and waits for running handlers, up to
509    /// `drain_timeout_secs`. After this, the routes reply 503.
510    ///
511    /// In an app, autumn calls this at shutdown. autumn gives all shutdown
512    /// hooks `[server] shutdown_timeout_secs` in total, so the wait can be
513    /// shorter than `drain_timeout_secs`.
514    pub async fn shutdown(&self) {
515        drain(&self.engine).await;
516    }
517}
518
519/// Closes the tracker and waits for its tasks, up to the drain timeout.
520async fn drain(engine: &Engine) {
521    let tracker = &engine.runner.tracker;
522    tracker.close();
523    let limit = Duration::from_secs(engine.config.drain_timeout_secs);
524    if tokio::time::timeout(limit, tracker.wait()).await.is_err() {
525        tracing::warn!(
526            left = tracker.len(),
527            "slack: handlers still run after the drain timeout"
528        );
529    }
530}
531
532impl Plugin for SlackPlugin {
533    fn name(&self) -> Cow<'static, str> {
534        Cow::Borrowed(PLUGIN_NAME)
535    }
536
537    fn contract(&self) -> Option<PluginContract> {
538        Some(
539            PluginContract::new(env!("CARGO_PKG_NAME"))
540                .plugin_version(env!("CARGO_PKG_VERSION"))
541                .autumn_web("0.10"),
542        )
543    }
544
545    fn build(self, app: AppBuilder) -> AppBuilder {
546        let base = self.base_path.clone();
547        let cell: EngineCell = Arc::new(OnceLock::new());
548        let health = Arc::new(SlackHealth::new(self.readiness));
549        let metrics = Arc::clone(&self.metrics);
550        let failure: Arc<Mutex<Option<SlackError>>> = Arc::default();
551        let routes = routes::autumn_routes(&cell, &base);
552        let (init_cell, init_health, init_failure) =
553            (Arc::clone(&cell), Arc::clone(&health), Arc::clone(&failure));
554        let (check_cell, stop_cell) = (Arc::clone(&cell), Arc::clone(&cell));
555        app.config_section(SECTION)
556            .metrics_source("slack", metrics as Arc<dyn MetricsSource>)
557            .health_indicator("slack", health as Arc<dyn HealthIndicator>)
558            .routes(routes)
559            // Start before job workers, so a `#[job]` can use `SlackClient::from_state`.
560            .state_initializer(move |state| {
561                match self.start_inner(state, state.role(), init_health) {
562                    Ok(rt) => {
563                        let _ = init_cell.set(rt.engine);
564                    }
565                    Err(e) => {
566                        // Also log: some autumn modes run no startup hook.
567                        tracing::error!(error = %e, "slack plugin did not start");
568                        *init_failure
569                            .lock()
570                            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(e);
571                    }
572                }
573            })
574            // An initializer cannot fail the boot. This hook reports its error.
575            .on_startup(move |_state| {
576                let failed = failure
577                    .lock()
578                    .unwrap_or_else(std::sync::PoisonError::into_inner)
579                    .take();
580                let started = check_cell.get().is_some();
581                async move {
582                    match failed {
583                        Some(e) => Err(AutumnError::internal_server_error(e)),
584                        None if !started => Err(AutumnError::internal_server_error(
585                            SlackError::Config("the slack plugin did not start".to_owned()),
586                        )),
587                        None => Ok(()),
588                    }
589                }
590            })
591            .on_shutdown(move || shutdown_hook(&stop_cell))
592    }
593}
594
595/// The autumn shutdown hook: drain the started engine, if any.
596fn shutdown_hook(cell: &EngineCell) -> impl Future<Output = ()> + Send + use<> {
597    let engine = cell.get().cloned();
598    async move {
599        if let Some(engine) = engine {
600            drain(&engine).await;
601        }
602    }
603}
604
605#[cfg(test)]
606mod tests {
607    use super::*;
608
609    #[tokio::test]
610    async fn shutdown_hook_drains_the_started_engine() {
611        use std::sync::atomic::{AtomicBool, Ordering};
612        // No engine: the hook ends at once.
613        shutdown_hook(&Arc::new(OnceLock::new())).await;
614        let rt = SlackPlugin::with_config(SlackConfig::default())
615            .with_signing_secret("s")
616            .with_transport(crate::transport::MemoryTransport::new())
617            .start(&AppState::for_test())
618            .unwrap();
619        let gate = Arc::new(tokio::sync::Notify::new());
620        let done = Arc::new(AtomicBool::new(false));
621        let (g, d) = (Arc::clone(&gate), Arc::clone(&done));
622        rt.engine.runner.spawn_task(async move {
623            g.notified().await;
624            d.store(true, Ordering::SeqCst);
625        });
626        let cell: EngineCell = Arc::new(OnceLock::new());
627        let _ = cell.set(Arc::clone(&rt.engine));
628        tokio::spawn(async move {
629            tokio::time::sleep(Duration::from_millis(100)).await;
630            gate.notify_one();
631        });
632        shutdown_hook(&cell).await;
633        assert!(
634            done.load(Ordering::SeqCst),
635            "the hook did not wait for the task"
636        );
637        assert!(rt.engine.runner.tracker.is_closed());
638    }
639
640    #[test]
641    fn base_path_normalizes() {
642        assert_eq!(normalize_base("/slack/"), "/slack");
643        assert_eq!(normalize_base("hooks/slack"), "/hooks/slack");
644        assert_eq!(normalize_base("/"), "");
645        assert_eq!(route_paths("")[0], "/events");
646    }
647
648    #[test]
649    fn exempt_rule_matches_autumn() {
650        let ex = |list: &[&str]| list.iter().map(|s| (*s).to_owned()).collect::<Vec<_>>();
651        assert!(is_exempt("/slack/events", &ex(&["/slack"])));
652        assert!(is_exempt("/slack/events", &ex(&["/slack/"])));
653        assert!(is_exempt("/slack/events", &ex(&["/slack/events"])));
654        assert!(!is_exempt("/slack/events", &ex(&["/sla"])));
655        assert!(!is_exempt("/slack/events", &ex(&["/slack/e"])));
656        // Like autumn: an empty entry exempts all paths.
657        assert!(is_exempt("/slack/events", &ex(&[""])));
658    }
659
660    #[test]
661    fn webhook_endpoint_paths_count_as_csrf_exempt() {
662        let mut cfg = AutumnConfig::default();
663        cfg.security.csrf.enabled = true;
664        assert!(check_security(&cfg, "/slack").is_err());
665        for p in route_paths("/slack") {
666            cfg.security.webhooks.endpoints.push(
667                autumn_web::webhook::WebhookEndpointConfig::slack("s", p, "S"),
668            );
669        }
670        check_security(&cfg, "/slack").unwrap();
671    }
672}
autumn_plugin_slack/plugin.rs

autumn_plugin_slack/
plugin.rs
