autumn_plugin_slack/
client.rs

1//! Slack Web API client and `response_url` replies.
2
3use std::sync::Arc;
4
5use autumn_web::AppState;
6use serde::Deserialize;
7use serde_json::{Map, Value};
8
9use crate::config::SlackConfig;
10use crate::error::SlackError;
11use crate::metrics::SlackMetrics;
12use crate::payload::Message;
13use crate::policy::{Outcome, RetryRule, Step, decide, retry_after_ms};
14use crate::transport::{HttpReply, HttpRequest, HttpTransport, TransportError};
15
16/// Maximum 429 wait for a call with a `trigger_id`. A trigger is valid for
17/// 3 s only, so a longer wait gives `expired_trigger_id`.
18const TRIGGER_MAX_WAIT_MS: u64 = 1_000;
19/// Maximum attempts for a call with a `trigger_id`: one retry fits in 3 s.
20const TRIGGER_MAX_ATTEMPTS: u32 = 2;
21
22/// Metrics label for `response_url` posts.
23const RESPONSE_URL_LABEL: &str = "response_url";
24/// Fields that only a reply can have. API calls drop them.
25const REPLY_ONLY: [&str; 3] = ["response_type", "replace_original", "delete_original"];
26
27/// Where a message is: channel and timestamp.
28#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
29#[non_exhaustive]
30pub struct MessageRef {
31    /// Channel ID.
32    pub channel: String,
33    /// Message timestamp (its ID).
34    pub ts: String,
35}
36
37/// Result of `auth.test`.
38#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
39#[non_exhaustive]
40pub struct AuthTest {
41    /// Workspace ID.
42    pub team_id: String,
43    /// Bot user ID.
44    pub user_id: String,
45    /// Bot ID.
46    #[serde(default)]
47    pub bot_id: Option<String>,
48    /// Workspace name.
49    #[serde(default)]
50    pub team: Option<String>,
51    /// Workspace URL.
52    #[serde(default)]
53    pub url: Option<String>,
54}
55
56struct Inner {
57    transport: Arc<dyn HttpTransport>,
58    config: SlackConfig,
59    token: Option<String>,
60    metrics: Arc<SlackMetrics>,
61}
62
63/// Slack Web API client. A clone shares one connection pool.
64///
65/// Get it from a handler with [`crate::SlackContext::client`], or from the
66/// app state with [`SlackClient::from_state`].
67#[derive(Clone)]
68pub struct SlackClient {
69    inner: Arc<Inner>,
70}
71
72impl std::fmt::Debug for SlackClient {
73    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
74        f.debug_struct("SlackClient")
75            .field("api_base_url", &self.inner.config.api_base_url)
76            .field("has_token", &self.inner.token.is_some())
77            .finish_non_exhaustive()
78    }
79}
80
81impl SlackClient {
82    pub(crate) fn new(
83        transport: Arc<dyn HttpTransport>,
84        config: SlackConfig,
85        token: Option<String>,
86        metrics: Arc<SlackMetrics>,
87    ) -> Self {
88        Self {
89            inner: Arc::new(Inner {
90                transport,
91                config,
92                token,
93                metrics,
94            }),
95        }
96    }
97
98    /// Returns the client that the plugin put in the app state at startup.
99    #[must_use]
100    pub fn from_state(state: &AppState) -> Option<Self> {
101        state.extension::<Self>().map(|c| (*c).clone())
102    }
103
104    /// Returns `true` when a bot token is set.
105    #[must_use]
106    pub fn has_token(&self) -> bool {
107        self.inner.token.is_some()
108    }
109
110    /// Calls a write method. It retries only on 429.
111    ///
112    /// `params` is a JSON object. The client sends strings unchanged and other
113    /// values as JSON text. It does not send `null` values.
114    ///
115    /// # Errors
116    /// Returns [`SlackError::Api`] for `ok: false`, or a transport, HTTP,
117    /// rate-limit, decode, or config error.
118    pub async fn call(&self, method: &str, params: &Value) -> Result<Value, SlackError> {
119        self.call_with(method, params, false, self.rule()).await
120    }
121
122    /// Calls a read method. It also retries on 5xx and network errors.
123    ///
124    /// # Errors
125    /// Same as [`Self::call`].
126    pub async fn call_read(&self, method: &str, params: &Value) -> Result<Value, SlackError> {
127        self.call_with(method, params, true, self.rule()).await
128    }
129
130    /// The retry rule from the config.
131    fn rule(&self) -> RetryRule {
132        self.inner.config.api.rule()
133    }
134
135    /// The rule for calls with a `trigger_id`: one short retry at most.
136    fn trigger_rule(&self) -> RetryRule {
137        let mut r = self.rule();
138        r.max_wait_ms = r.max_wait_ms.min(TRIGGER_MAX_WAIT_MS);
139        r.max_attempts = r.max_attempts.min(TRIGGER_MAX_ATTEMPTS);
140        r
141    }
142
143    async fn call_with(
144        &self,
145        method: &str,
146        params: &Value,
147        idempotent: bool,
148        rule: RetryRule,
149    ) -> Result<Value, SlackError> {
150        if !valid_method(method) {
151            return Err(SlackError::Config(format!(
152                "not a Web API method name: {method:?}"
153            )));
154        }
155        let Value::Object(map) = params else {
156            return Err(SlackError::Config(format!(
157                "{method}: params must be a JSON object"
158            )));
159        };
160        let Some(token) = self.inner.token.as_deref() else {
161            return Err(SlackError::Config(format!(
162                "{method}: no bot token; set the env var {}",
163                self.inner.config.bot_token_env
164            )));
165        };
166        let url = format!("{}{method}", self.inner.config.api_base_url);
167        let body = form_body(map).into_bytes();
168        let mut attempt: u32 = 1;
169        loop {
170            let req = HttpRequest::new(url.as_str())
171                .header("authorization", format!("Bearer {token}"))
172                .header(
173                    "content-type",
174                    "application/x-www-form-urlencoded; charset=utf-8",
175                )
176                .body(body.clone());
177            let result = self.inner.transport.post(req).await;
178            let (outcome, retry_after) = classify(&result);
179            match decide(
180                outcome,
181                attempt,
182                idempotent,
183                retry_after.map(retry_after_ms),
184                &rule,
185            ) {
186                Step::Retry(wait_ms) => {
187                    self.inner.metrics.api_retry(method);
188                    tracing::debug!(method, attempt, wait_ms, "slack api retry");
189                    tokio::time::sleep(std::time::Duration::from_millis(wait_ms)).await;
190                    attempt += 1;
191                }
192                Step::Done | Step::GiveUp => return self.finish(method, result),
193            }
194        }
195    }
196
197    /// Turns the last attempt into a result. Counts it.
198    fn finish(
199        &self,
200        method: &str,
201        result: Result<HttpReply, TransportError>,
202    ) -> Result<Value, SlackError> {
203        let m = &self.inner.metrics;
204        let reply = match result {
205            Ok(r) => r,
206            Err(e) => {
207                m.api_call(method, "transport");
208                return Err(SlackError::Transport(e.0));
209            }
210        };
211        if reply.status == 429 {
212            m.api_call(method, "rate_limited");
213            return Err(SlackError::RateLimited {
214                retry_after_secs: reply.retry_after_secs,
215            });
216        }
217        if !(200..300).contains(&reply.status) {
218            m.api_call(method, "http_error");
219            return Err(http_error(&reply));
220        }
221        let v: Value = serde_json::from_slice(&reply.body).map_err(|e| {
222            m.api_call(method, "decode");
223            SlackError::Decode(format!("{method}: {e}"))
224        })?;
225        if v.get("ok").and_then(Value::as_bool) == Some(true) {
226            m.api_call(method, "ok");
227            return Ok(v);
228        }
229        m.api_call(method, "api_error");
230        Err(SlackError::Api {
231            method: method.to_owned(),
232            code: v
233                .get("error")
234                .and_then(Value::as_str)
235                .unwrap_or("unknown_error")
236                .to_owned(),
237            needed: v.get("needed").and_then(Value::as_str).map(str::to_owned),
238        })
239    }
240
241    /// Calls `method` and decodes the reply as `T`.
242    async fn call_as<T: serde::de::DeserializeOwned>(
243        &self,
244        method: &str,
245        params: Value,
246        idempotent: bool,
247    ) -> Result<T, SlackError> {
248        let v = self
249            .call_with(method, &params, idempotent, self.rule())
250            .await?;
251        T::deserialize(&v).map_err(|e| SlackError::Decode(format!("{method}: {e}")))
252    }
253
254    /// `chat.postMessage`.
255    ///
256    /// # Errors
257    /// Same as [`Self::call`].
258    pub async fn chat_post_message(
259        &self,
260        channel: &str,
261        msg: &Message,
262    ) -> Result<MessageRef, SlackError> {
263        let mut p = api_fields(msg);
264        p.insert("channel".to_owned(), channel.into());
265        self.call_as("chat.postMessage", Value::Object(p), false)
266            .await
267    }
268
269    /// `chat.postEphemeral`. Returns the `message_ts`.
270    ///
271    /// # Errors
272    /// Same as [`Self::call`].
273    pub async fn chat_post_ephemeral(
274        &self,
275        channel: &str,
276        user: &str,
277        msg: &Message,
278    ) -> Result<String, SlackError> {
279        #[derive(Deserialize)]
280        struct Reply {
281            message_ts: String,
282        }
283        let mut p = api_fields(msg);
284        p.insert("channel".to_owned(), channel.into());
285        p.insert("user".to_owned(), user.into());
286        let r: Reply = self
287            .call_as("chat.postEphemeral", Value::Object(p), false)
288            .await?;
289        Ok(r.message_ts)
290    }
291
292    /// `chat.update`.
293    ///
294    /// # Errors
295    /// Same as [`Self::call`].
296    pub async fn chat_update(
297        &self,
298        channel: &str,
299        ts: &str,
300        msg: &Message,
301    ) -> Result<MessageRef, SlackError> {
302        let mut p = api_fields(msg);
303        // `chat.update` cannot move a message to a thread.
304        p.remove("thread_ts");
305        p.insert("channel".to_owned(), channel.into());
306        p.insert("ts".to_owned(), ts.into());
307        self.call_as("chat.update", Value::Object(p), false).await
308    }
309
310    /// `chat.delete`.
311    ///
312    /// # Errors
313    /// Same as [`Self::call`].
314    pub async fn chat_delete(&self, channel: &str, ts: &str) -> Result<(), SlackError> {
315        self.call(
316            "chat.delete",
317            &serde_json::json!({"channel": channel, "ts": ts}),
318        )
319        .await
320        .map(drop)
321    }
322
323    /// `reactions.add`. `name` has no colons, for example `thumbsup`.
324    ///
325    /// # Errors
326    /// Same as [`Self::call`].
327    pub async fn reactions_add(
328        &self,
329        channel: &str,
330        ts: &str,
331        name: &str,
332    ) -> Result<(), SlackError> {
333        let p = serde_json::json!({"channel": channel, "timestamp": ts, "name": name.trim_matches(':')});
334        self.call("reactions.add", &p).await.map(drop)
335    }
336
337    /// `views.open`. Returns the view.
338    ///
339    /// # Errors
340    /// Same as [`Self::call`].
341    pub async fn views_open(&self, trigger_id: &str, view: &Value) -> Result<Value, SlackError> {
342        let p = serde_json::json!({"trigger_id": trigger_id, "view": view});
343        self.view_call("views.open", p, self.trigger_rule()).await
344    }
345
346    /// `views.update`. `hash` stops an update that would overwrite a newer
347    /// view. Returns the view.
348    ///
349    /// # Errors
350    /// Same as [`Self::call`].
351    pub async fn views_update(
352        &self,
353        view_id: &str,
354        view: &Value,
355        hash: Option<&str>,
356    ) -> Result<Value, SlackError> {
357        let p = serde_json::json!({"view_id": view_id, "view": view, "hash": hash});
358        self.view_call("views.update", p, self.rule()).await
359    }
360
361    /// `views.push`. Returns the view.
362    ///
363    /// # Errors
364    /// Same as [`Self::call`].
365    pub async fn views_push(&self, trigger_id: &str, view: &Value) -> Result<Value, SlackError> {
366        let p = serde_json::json!({"trigger_id": trigger_id, "view": view});
367        self.view_call("views.push", p, self.trigger_rule()).await
368    }
369
370    /// `views.publish` for the Home tab. Returns the view.
371    ///
372    /// # Errors
373    /// Same as [`Self::call`].
374    pub async fn views_publish(&self, user_id: &str, view: &Value) -> Result<Value, SlackError> {
375        let p = serde_json::json!({"user_id": user_id, "view": view});
376        self.view_call("views.publish", p, self.rule()).await
377    }
378
379    /// Calls a `views.*` write method. Returns the `view` field.
380    async fn view_call(
381        &self,
382        method: &str,
383        params: Value,
384        rule: RetryRule,
385    ) -> Result<Value, SlackError> {
386        let mut v = self.call_with(method, &params, false, rule).await?;
387        Ok(v.get_mut("view").map(Value::take).unwrap_or_default())
388    }
389
390    /// `users.info`. Returns the user.
391    ///
392    /// # Errors
393    /// Same as [`Self::call`].
394    pub async fn users_info(&self, user: &str) -> Result<Value, SlackError> {
395        let mut v = self
396            .call_read("users.info", &serde_json::json!({"user": user}))
397            .await?;
398        Ok(v.get_mut("user").map(Value::take).unwrap_or_default())
399    }
400
401    /// `auth.test`.
402    ///
403    /// # Errors
404    /// Same as [`Self::call`].
405    pub async fn auth_test(&self) -> Result<AuthTest, SlackError> {
406        self.call_as("auth.test", Value::Object(Map::new()), true)
407            .await
408    }
409
410    /// `auth.test` with one attempt, for the health check. A 429 gives
411    /// `RateLimited` at once, not a long wait.
412    pub(crate) async fn auth_test_once(&self) -> Result<(), SlackError> {
413        let mut rule = self.rule();
414        rule.max_attempts = 1;
415        self.call_with("auth.test", &Value::Object(Map::new()), true, rule)
416            .await
417            .map(drop)
418    }
419
420    /// Calls a read method once per page and joins the `key` arrays.
421    /// It follows `response_metadata.next_cursor` for at most `max_pages`.
422    ///
423    /// # Errors
424    /// Same as [`Self::call`].
425    pub async fn paginate(
426        &self,
427        method: &str,
428        params: &Value,
429        key: &str,
430        max_pages: usize,
431    ) -> Result<Vec<Value>, SlackError> {
432        let Value::Object(base) = params else {
433            return Err(SlackError::Config(format!(
434                "{method}: params must be a JSON object"
435            )));
436        };
437        let mut out = Vec::new();
438        let mut cursor: Option<String> = None;
439        for _ in 0..max_pages {
440            let mut p = base.clone();
441            if let Some(c) = cursor.take() {
442                p.insert("cursor".to_owned(), Value::String(c));
443            }
444            let mut v = self.call_read(method, &Value::Object(p)).await?;
445            if let Some(Value::Array(items)) = v.get_mut(key).map(Value::take) {
446                out.extend(items);
447            }
448            cursor = v
449                .pointer("/response_metadata/next_cursor")
450                .and_then(Value::as_str)
451                .filter(|c| !c.is_empty())
452                .map(str::to_owned);
453            if cursor.is_none() {
454                break;
455            }
456        }
457        Ok(out)
458    }
459
460    /// Posts `msg` as JSON to a `response_url`. It sends no token.
461    ///
462    /// # Errors
463    /// Returns [`SlackError::BadResponseUrl`] for a URL that is not HTTPS on
464    /// an allowed host. Returns an HTTP or transport error.
465    pub async fn respond(&self, response_url: &str, msg: &Message) -> Result<(), SlackError> {
466        let m = &self.inner.metrics;
467        if !allowed_response_url(response_url, &self.inner.config.response_url_hosts) {
468            m.api_call(RESPONSE_URL_LABEL, "rejected");
469            return Err(SlackError::BadResponseUrl);
470        }
471        let body = serde_json::to_vec(msg).map_err(|e| SlackError::Decode(e.to_string()))?;
472        let req = HttpRequest::new(response_url)
473            .header("content-type", "application/json; charset=utf-8")
474            .body(body);
475        // One attempt: a `response_url` allows only 5 uses.
476        match self.inner.transport.post(req).await {
477            Err(e) => {
478                m.api_call(RESPONSE_URL_LABEL, "transport");
479                Err(SlackError::Transport(e.0))
480            }
481            Ok(r) if (200..300).contains(&r.status) => {
482                m.api_call(RESPONSE_URL_LABEL, "ok");
483                Ok(())
484            }
485            Ok(r) => {
486                m.api_call(RESPONSE_URL_LABEL, "http_error");
487                Err(http_error(&r))
488            }
489        }
490    }
491}
492
493/// Classifies one attempt for the retry policy.
494fn classify(result: &Result<HttpReply, TransportError>) -> (Outcome, Option<u64>) {
495    result
496        .as_ref()
497        .map_or((Outcome::NetworkError, None), |r| match r.status {
498            429 => (Outcome::RateLimited, r.retry_after_secs),
499            200..=299 => (Outcome::Success, None),
500            500..=599 => (Outcome::ServerError, None),
501            _ => (Outcome::ClientError, None),
502        })
503}
504
505/// An HTTP error with the first 200 bytes of the body.
506fn http_error(reply: &HttpReply) -> SlackError {
507    let end = reply.body.len().min(200);
508    SlackError::Http {
509        status: reply.status,
510        body: String::from_utf8_lossy(&reply.body[..end]).into_owned(),
511    }
512}
513
514/// The message fields for an API call. Reply-only fields are dropped.
515fn api_fields(msg: &Message) -> Map<String, Value> {
516    let mut p = msg.to_map();
517    for k in REPLY_ONLY {
518        p.remove(k);
519    }
520    p
521}
522
523/// Returns `true` for an `https` URL on an allowed host, with no user info
524/// and no port.
525pub(crate) fn allowed_response_url(url: &str, hosts: &[String]) -> bool {
526    let Ok(u) = reqwest::Url::parse(url) else {
527        return false;
528    };
529    u.scheme() == "https"
530        && u.username().is_empty()
531        && u.password().is_none()
532        && u.port().is_none()
533        && u.host_str()
534            .is_some_and(|h| hosts.iter().any(|a| a.eq_ignore_ascii_case(h)))
535}
536
537/// Returns `true` for a method name such as `chat.postMessage`.
538pub(crate) fn valid_method(method: &str) -> bool {
539    let mut parts = 0;
540    for part in method.split('.') {
541        if part.is_empty() || !part.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_') {
542            return false;
543        }
544        parts += 1;
545    }
546    parts >= 2
547}
548
549/// Encodes JSON object params as a form body.
550pub(crate) fn form_body(params: &Map<String, Value>) -> String {
551    let pairs: Vec<(&str, String)> = params
552        .iter()
553        .filter_map(|(k, v)| {
554            let text = match v {
555                Value::Null => return None,
556                Value::String(s) => s.clone(),
557                other => other.to_string(),
558            };
559            Some((k.as_str(), text))
560        })
561        .collect();
562    // String pairs cannot fail to encode.
563    serde_urlencoded::to_string(pairs).unwrap_or_default()
564}
565
566#[cfg(test)]
567mod tests {
568    use super::*;
569    use proptest::prelude::*;
570    use serde_json::json;
571
572    fn hosts() -> Vec<String> {
573        vec!["hooks.slack.com".to_owned()]
574    }
575
576    #[test]
577    fn response_url_rules() {
578        assert!(allowed_response_url(
579            "https://hooks.slack.com/commands/T/1/x",
580            &hosts()
581        ));
582        assert!(allowed_response_url(
583            "https://HOOKS.slack.com/actions/T/1/x",
584            &hosts()
585        ));
586        for bad in [
587            "http://hooks.slack.com/x",
588            "https://hooks.slack.com.evil/x",
589            "https://evil/hooks.slack.com/x",
590            "https://a@hooks.slack.com/x",
591            "https://hooks.slack.com:444/x",
592            "javascript:alert(1)",
593            "",
594        ] {
595            assert!(!allowed_response_url(bad, &hosts()), "{bad}");
596        }
597    }
598
599    #[test]
600    fn method_names() {
601        assert!(valid_method("chat.postMessage"));
602        assert!(valid_method("admin.conversations.setTeams"));
603        for bad in [
604            "chat", "chat.", ".x", "a..b", "../x", "a/b", "a.b?c", "a.b c", "",
605        ] {
606            assert!(!valid_method(bad), "{bad}");
607        }
608    }
609
610    #[test]
611    fn form_encoding() {
612        let p = json!({"a": "x y&z", "n": 1, "b": false, "o": {"k": [1]}, "z": null});
613        let body = form_body(p.as_object().unwrap());
614        let back: std::collections::BTreeMap<String, String> =
615            serde_urlencoded::from_str(&body).unwrap();
616        assert_eq!(back["a"], "x y&z");
617        assert_eq!(back["n"], "1");
618        assert_eq!(back["b"], "false");
619        assert_eq!(back["o"], r#"{"k":[1]}"#);
620        assert!(!back.contains_key("z"));
621    }
622
623    proptest! {
624        #[test]
625        fn form_round_trips_strings(k in "[a-z_]{1,10}", v in ".{0,40}") {
626            let mut m = Map::new();
627            m.insert(k.clone(), Value::String(v.clone()));
628            let back: std::collections::BTreeMap<String, String> = serde_urlencoded::from_str(&form_body(&m)).unwrap();
629            prop_assert_eq!(&back[&k], &v);
630        }
631
632        #[test]
633        fn only_listed_hosts_pass(host in "[a-z]{1,12}(\\.[a-z]{2,6}){1,2}", path in "[a-z/]{0,20}") {
634            let url = format!("https://{host}/{path}");
635            prop_assert_eq!(allowed_response_url(&url, &hosts()), host == "hooks.slack.com");
636        }
637    }
638}
autumn_plugin_slack/client.rs

autumn_plugin_slack/
client.rs
