opssignal_core/providers/
email.rs

1use crate::config::{EmailConfig, EmailTransport, SesConfig, SmtpConfig, SmtpTls};
2use crate::delivery::Provider;
3use crate::signal::{Signal, SignalError};
4use crate::template::{self, Templates};
5use lettre::message::{header::ContentType, Mailbox, MessageBuilder};
6use lettre::transport::smtp::authentication::Credentials;
7use lettre::{AsyncSmtpTransport, AsyncTransport, Message, Tokio1Executor};
8use sha2::{Digest, Sha256};
9use std::time::Duration;
10
11/// Covers connect, TLS, auth and the whole SMTP exchange of one attempt,
12/// so a server that hangs costs one attempt rather than the retry loop.
13const SMTP_TIMEOUT: Duration = Duration::from_secs(30);
14
15/// Sends a plain-text email through an SMTP server or, in a build with
16/// AWS support, the Amazon SES API. Repeats of the same
17/// signal keep the same subject and `References` header, so mail
18/// clients thread them into one conversation instead of filling the
19/// inbox.
20pub struct EmailProvider {
21    config: EmailConfig,
22    templates: Option<Templates>,
23}
24
25impl EmailProvider {
26    pub fn new(config: EmailConfig) -> Self {
27        let templates = Templates::load("email", &config.template);
28        EmailProvider { config, templates }
29    }
30
31    /// `subject_prefix` goes before a templated subject too, so mail
32    /// filters keep working. A subject is one line.
33    fn subject_for(&self, signal: &Signal) -> String {
34        let subject = match template::render(&self.templates, "subject", signal, &[]) {
35            Some(text) => text.split_whitespace().collect::<Vec<_>>().join(" "),
36            None => {
37                let severity = signal.severity.to_string().to_uppercase();
38                format!("[{severity}] {}", signal.title)
39            }
40        };
41        match &self.config.subject_prefix {
42            Some(prefix) if !prefix.is_empty() => format!("{prefix} {subject}"),
43            _ => subject,
44        }
45    }
46
47    fn body_for(&self, signal: &Signal) -> String {
48        template::render(&self.templates, "body", signal, &[])
49            .unwrap_or_else(|| Self::default_body_for(signal))
50    }
51
52    fn default_body_for(signal: &Signal) -> String {
53        let mut body = format!("{}\n", signal.title);
54        if let Some(message) = &signal.message {
55            body.push('\n');
56            body.push_str(message);
57            body.push('\n');
58        }
59        body.push('\n');
60        let severity = signal.severity.to_string();
61        let mut fields: Vec<(&str, &str)> = vec![
62            ("Severity", &severity),
63            ("Source", &signal.source),
64            ("Event", &signal.event_type),
65        ];
66        let optional = [
67            ("Environment", &signal.environment),
68            ("Service", &signal.service),
69            ("Component", &signal.component),
70            ("Team", &signal.team),
71            ("Run ID", &signal.run_id),
72            ("Correlation ID", &signal.correlation_id),
73            ("Trace ID", &signal.trace_id),
74            ("Logs", &signal.log_url),
75        ];
76        for (label, value) in optional {
77            if let Some(value) = value {
78                fields.push((label, value));
79            }
80        }
81        for (label, value) in fields {
82            body.push_str(&format!("{:<16}{value}\n", format!("{label}:")));
83        }
84        if signal.metadata.as_object().is_some_and(|m| !m.is_empty()) {
85            let metadata = serde_json::to_string_pretty(&signal.metadata)
86                .unwrap_or_else(|_| signal.metadata.to_string());
87            body.push_str(&format!("\nMetadata:\n{metadata}\n"));
88        }
89        body.push_str(&format!(
90            "\n--\nSignal {} at {}\nSent by OpsSignal\n",
91            signal.id,
92            signal.timestamp.to_rfc3339()
93        ));
94        body
95    }
96
97    fn message_for(&self, signal: &Signal) -> Result<Message, SignalError> {
98        let mut builder: MessageBuilder = Message::builder()
99            .from(mailbox(&self.config.from)?)
100            .subject(self.subject_for(signal))
101            .header(ContentType::TEXT_PLAIN);
102        for to in &self.config.to {
103            builder = builder.to(mailbox(to)?);
104        }
105        for cc in &self.config.cc {
106            builder = builder.cc(mailbox(cc)?);
107        }
108        if let Some(key) = &signal.dedup_key {
109            let thread = thread_id(key);
110            builder = builder.in_reply_to(thread.clone()).references(thread);
111        }
112        builder
113            .body(self.body_for(signal))
114            .map_err(|e| SignalError::Delivery(format!("cannot build the email: {e}")))
115    }
116
117    fn transport(smtp: &SmtpConfig) -> Result<AsyncSmtpTransport<Tokio1Executor>, SignalError> {
118        let builder = match smtp.tls {
119            SmtpTls::Starttls => AsyncSmtpTransport::<Tokio1Executor>::starttls_relay(&smtp.host),
120            SmtpTls::Tls => AsyncSmtpTransport::<Tokio1Executor>::relay(&smtp.host),
121            SmtpTls::None => Ok(AsyncSmtpTransport::<Tokio1Executor>::builder_dangerous(
122                &smtp.host,
123            )),
124        }
125        .map_err(|e| SignalError::Delivery(format!("smtp: {e}")))?;
126        let mut builder = builder.port(smtp.port()).timeout(Some(SMTP_TIMEOUT));
127        if let (Some(username), Some(password)) = (&smtp.username, &smtp.password) {
128            builder = builder.credentials(Credentials::new(username.clone(), password.clone()));
129        }
130        Ok(builder.build())
131    }
132}
133
134fn mailbox(address: &str) -> Result<Mailbox, SignalError> {
135    address
136        .parse()
137        .map_err(|e| SignalError::Delivery(format!("invalid email address {address:?}: {e}")))
138}
139
140/// A message id that every email for the same dedup key refers to.
141/// Hashed because caller-supplied keys may hold characters a header
142/// can't carry.
143fn thread_id(dedup_key: &str) -> String {
144    let digest = format!("{:x}", Sha256::digest(dedup_key.as_bytes()));
145    format!("<{}@opssignal>", &digest[..32])
146}
147
148#[async_trait::async_trait]
149impl Provider for EmailProvider {
150    fn name(&self) -> &'static str {
151        "email"
152    }
153
154    fn preview(&self, signal: &Signal) -> Option<serde_json::Value> {
155        Some(serde_json::json!({
156            "subject": self.subject_for(signal),
157            "body": self.body_for(signal),
158        }))
159    }
160
161    async fn send(&self, signal: &Signal) -> Result<(), SignalError> {
162        match self.config.transport() {
163            Some(EmailTransport::Smtp(smtp)) => {
164                let message = self.message_for(signal)?;
165                Self::transport(smtp)?
166                    .send(message)
167                    .await
168                    .map_err(|e| SignalError::Delivery(format!("smtp: {e}")))?;
169                Ok(())
170            }
171            Some(EmailTransport::Ses(ses)) => self.send_ses(ses, signal).await,
172            None => Err(SignalError::Delivery(
173                "email needs exactly one of smtp and ses".into(),
174            )),
175        }
176    }
177}
178
179impl EmailProvider {
180    /// The same MIME message as over SMTP, so threading headers and
181    /// formatting don't depend on the transport.
182    #[cfg(feature = "aws")]
183    async fn send_ses(&self, ses: &SesConfig, signal: &Signal) -> Result<(), SignalError> {
184        let raw = self.message_for(signal)?.formatted();
185        crate::cloud::aws::send_ses_raw(
186            ses.region.as_deref(),
187            ses.configuration_set.as_deref(),
188            &raw,
189        )
190        .await
191    }
192
193    /// No AWS code in this build: the `opssignal[aws]` add-on sends it.
194    #[cfg(not(feature = "aws"))]
195    async fn send_ses(&self, _ses: &SesConfig, signal: &Signal) -> Result<(), SignalError> {
196        let config_json = serde_json::to_string(&self.config)
197            .map_err(|e| SignalError::Delivery(e.to_string()))?;
198        super::external::send_external("aws", "email", config_json, signal).await
199    }
200}
201
202#[cfg(test)]
203mod tests {
204    use super::*;
205    use crate::signal::{Severity, SignalInput};
206    use std::sync::{Arc, Mutex};
207    use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
208    use tokio::net::TcpListener;
209
210    fn signal(severity: Severity) -> Signal {
211        Signal::from_input(SignalInput {
212            source: "airflow".into(),
213            service: Some("orders".into()),
214            event_type: "task_failed".into(),
215            title: "dbt model failed".into(),
216            message: Some("model orders timed out".into()),
217            severity,
218            environment: Some("prod".into()),
219            log_url: Some("https://logs.example.test/run/1".into()),
220            metadata: Some(serde_json::json!({ "dag": "daily" })),
221            ..Default::default()
222        })
223    }
224
225    fn config(port: u16) -> EmailConfig {
226        EmailConfig {
227            from: "OpsSignal <alerts@example.test>".into(),
228            to: vec!["oncall@example.test".into()],
229            cc: vec!["data@example.test".into()],
230            subject_prefix: Some("[opssignal]".into()),
231            smtp: Some(SmtpConfig {
232                host: "127.0.0.1".into(),
233                port: Some(port),
234                username: Some("user".into()),
235                password: Some("pass".into()),
236                tls: SmtpTls::None,
237            }),
238            ses: None,
239            template: Default::default(),
240        }
241    }
242
243    /// A one-connection SMTP server that accepts everything except
244    /// recipients in `reject_rcpt`, and records what the client sent.
245    async fn fake_smtp(reject_rcpt: bool) -> (u16, Arc<Mutex<String>>) {
246        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
247        let port = listener.local_addr().expect("addr").port();
248        let transcript = Arc::new(Mutex::new(String::new()));
249        let log = Arc::clone(&transcript);
250        tokio::spawn(async move {
251            let Ok((socket, _)) = listener.accept().await else {
252                return;
253            };
254            let (read, mut write) = socket.into_split();
255            let mut lines = BufReader::new(read).lines();
256            let _ = write.write_all(b"220 fake ESMTP\r\n").await;
257            let mut in_data = false;
258            while let Ok(Some(line)) = lines.next_line().await {
259                log.lock().expect("log").push_str(&format!("{line}\n"));
260                let reply: &[u8] = if in_data {
261                    if line == "." {
262                        in_data = false;
263                        b"250 queued\r\n"
264                    } else {
265                        continue;
266                    }
267                } else {
268                    let verb = line.split(' ').next().unwrap_or("").to_uppercase();
269                    match verb.as_str() {
270                        "EHLO" | "HELO" => b"250-fake\r\n250 AUTH PLAIN LOGIN\r\n",
271                        "AUTH" => b"235 ok\r\n",
272                        "RCPT" if reject_rcpt => b"550 no such user\r\n",
273                        "DATA" => {
274                            in_data = true;
275                            b"354 go ahead\r\n"
276                        }
277                        "QUIT" => {
278                            let _ = write.write_all(b"221 bye\r\n").await;
279                            return;
280                        }
281                        _ => b"250 ok\r\n",
282                    }
283                };
284                let _ = write.write_all(reply).await;
285            }
286        });
287        (port, transcript)
288    }
289
290    #[test]
291    fn subject_is_stable_and_carries_the_severity() {
292        let provider = EmailProvider::new(config(25));
293        assert_eq!(
294            provider.subject_for(&signal(Severity::Error)),
295            "[opssignal] [ERROR] dbt model failed"
296        );
297        let mut plain = config(25);
298        plain.subject_prefix = None;
299        assert_eq!(
300            EmailProvider::new(plain).subject_for(&signal(Severity::Critical)),
301            "[CRITICAL] dbt model failed"
302        );
303    }
304
305    #[test]
306    fn body_lists_the_signal_fields() {
307        let signal = signal(Severity::Error);
308        let body = EmailProvider::default_body_for(&signal);
309        assert!(body.starts_with("dbt model failed\n\nmodel orders timed out\n"));
310        assert!(body.contains("Severity:       error\n"));
311        assert!(body.contains("Environment:    prod\n"));
312        assert!(body.contains("Logs:           https://logs.example.test/run/1\n"));
313        assert!(body.contains("\"dag\": \"daily\""));
314        assert!(body.contains(&signal.id.to_string()));
315        assert!(!body.contains("Team:"));
316    }
317
318    #[test]
319    fn repeats_of_a_signal_share_a_thread() {
320        let provider = EmailProvider::new(config(25));
321        let first = provider
322            .message_for(&signal(Severity::Error))
323            .expect("first");
324        let second = provider
325            .message_for(&signal(Severity::Error))
326            .expect("second");
327        let refs = |m: &Message| {
328            let text = String::from_utf8(m.formatted()).expect("utf8");
329            text.lines()
330                .find(|l| l.starts_with("References:"))
331                .map(str::to_string)
332        };
333        assert!(refs(&first).is_some());
334        assert_eq!(refs(&first), refs(&second));
335    }
336
337    #[test]
338    fn odd_dedup_keys_still_make_a_valid_thread_id() {
339        let id = thread_id("a key with <brackets> and spaces");
340        assert!(id.starts_with('<') && id.ends_with("@opssignal>"));
341        assert!(!id[1..].contains(' '));
342    }
343
344    #[test]
345    fn default_ports_follow_the_tls_mode() {
346        let smtp = |tls| SmtpConfig {
347            host: "h".into(),
348            port: None,
349            username: None,
350            password: None,
351            tls,
352        };
353        assert_eq!(smtp(SmtpTls::Starttls).port(), 587);
354        assert_eq!(smtp(SmtpTls::Tls).port(), 465);
355        assert_eq!(smtp(SmtpTls::None).port(), 25);
356    }
357
358    #[tokio::test]
359    async fn sends_the_email_over_smtp() {
360        let (port, transcript) = fake_smtp(false).await;
361        EmailProvider::new(config(port))
362            .send(&signal(Severity::Error))
363            .await
364            .expect("accepted");
365
366        let sent = transcript.lock().expect("log").clone();
367        assert!(sent.contains("AUTH PLAIN"), "{sent}");
368        assert!(sent.contains("MAIL FROM:<alerts@example.test>"), "{sent}");
369        assert!(sent.contains("RCPT TO:<oncall@example.test>"), "{sent}");
370        assert!(sent.contains("RCPT TO:<data@example.test>"), "{sent}");
371        assert!(
372            sent.contains("Subject: [opssignal] [ERROR] dbt model failed"),
373            "{sent}"
374        );
375        assert!(sent.contains("model orders timed out"), "{sent}");
376    }
377
378    #[tokio::test]
379    async fn rejected_recipient_is_a_delivery_error() {
380        let (port, _) = fake_smtp(true).await;
381        let err = EmailProvider::new(config(port))
382            .send(&signal(Severity::Error))
383            .await
384            .expect_err("550 must fail");
385        assert!(matches!(err, SignalError::Delivery(_)));
386        assert!(!err.to_string().contains("pass"), "{err}");
387    }
388
389    #[tokio::test]
390    async fn unreachable_server_is_a_delivery_error() {
391        // Bind then drop, so the port is very likely closed.
392        let port = TcpListener::bind("127.0.0.1:0")
393            .await
394            .expect("bind")
395            .local_addr()
396            .expect("addr")
397            .port();
398        let err = EmailProvider::new(config(port))
399            .send(&signal(Severity::Error))
400            .await
401            .expect_err("nothing listening");
402        assert!(matches!(err, SignalError::Delivery(_)));
403    }
404
405    /// One test: the sender registry is process-wide, so "not installed"
406    /// has to be checked before the fake add-on registers.
407    #[cfg(not(feature = "aws"))]
408    #[tokio::test]
409    async fn ses_without_aws_goes_to_the_registered_add_on() {
410        use super::super::external::{register_cloud_sender, ExternalSender};
411
412        let ses_yaml = "providers:\n  email:\n    from: a@example.test\n    to: [b@example.test]\n    ses:\n      region: eu-west-1\n";
413        let err = crate::config::Config::from_yaml_str(ses_yaml).expect_err("no add-on yet");
414        assert!(err.to_string().contains("opssignal[aws]"), "{err}");
415
416        type Calls = Arc<Mutex<Vec<(String, String, String)>>>;
417        struct FakeAddon(Calls);
418        impl ExternalSender for FakeAddon {
419            fn send(&self, provider: &str, config: &str, signal: &str) -> Result<(), String> {
420                self.0
421                    .lock()
422                    .expect("calls")
423                    .push((provider.into(), config.into(), signal.into()));
424                Ok(())
425            }
426        }
427        let calls: Calls = Default::default();
428        register_cloud_sender("aws", Arc::new(FakeAddon(Arc::clone(&calls))));
429
430        let config = crate::config::Config::from_yaml_str(ses_yaml).expect("add-on registered");
431        let email = config.providers.email.expect("email");
432        let signal = signal(Severity::Error);
433        EmailProvider::new(email).send(&signal).await.expect("sent");
434
435        let calls = calls.lock().expect("calls");
436        let (provider, config_json, signal_json) = &calls[0];
437        assert_eq!(provider, "email");
438        let config: EmailConfig = serde_json::from_str(config_json).expect("config json");
439        assert_eq!(
440            config.ses.expect("ses").region.as_deref(),
441            Some("eu-west-1")
442        );
443        let sent: Signal = serde_json::from_str(signal_json).expect("signal json");
444        assert_eq!(sent.id, signal.id);
445    }
446
447    #[test]
448    fn subject_and_body_templates_keep_the_prefix() {
449        let mut config = config(25);
450        config.template = [
451            (
452                "subject",
453                "{{ environment|upper }} {{ title }}\n(ignored line break)",
454            ),
455            ("body", "{{ title }}\nRun: {{ metadata.dag }}"),
456        ]
457        .iter()
458        .map(|(k, v)| (k.to_string(), v.to_string()))
459        .collect();
460        let provider = EmailProvider::new(config);
461        let signal = signal(Severity::Error);
462        assert_eq!(
463            provider.subject_for(&signal),
464            "[opssignal] PROD dbt model failed (ignored line break)"
465        );
466        assert_eq!(provider.body_for(&signal), "dbt model failed\nRun: daily");
467    }
468}
opssignal_core/providers/email.rs

opssignal_core/providers/
email.rs
