Coverage for app/services/email_service.py: 100%

1""" 

2Outbound email service. 

3 

4Configuration is read entirely from environment variables so operators 

5manage it via their Docker Compose / .env file — no DB row needed. 

6 

7Required env vars (if OPENHANGAR_SMTP_HOST is unset, all sends are skipped): 

8 OPENHANGAR_SMTP_HOST — e.g. smtp.example.com 

9 OPENHANGAR_SMTP_PORT — default 587 

10 OPENHANGAR_SMTP_USER — SMTP login username 

11 OPENHANGAR_SMTP_PASSWORD — SMTP login password 

12 OPENHANGAR_SMTP_USE_TLS — "true" (default) uses STARTTLS; "false" for plain SMTP 

13 OPENHANGAR_SMTP_FROM_ADDRESS— e.g. no-reply@example.com 

14 OPENHANGAR_SMTP_FROM_NAME — display name, e.g. "OpenHangar" 

15 

16Optional: 

17 OPENHANGAR_EMAIL_SUBJECT_PREFIX — "true" (default) prefixes every subject 

18 with a small-airplane emoji so OpenHangar's notifications are easy to 

19 pick out in an inbox; "false" disables it; any other value (emoji or 

20 plain text) is used as a custom prefix instead of the default emoji. 

21 

22Demo mode (OPENHANGAR_ENV=demo): all sends are silently skipped. 

23""" 

24 

25import html as _html 

26import logging 

27import os 

28import smtplib 

29from datetime import UTC, datetime 

30from email.mime.multipart import MIMEMultipart 

31from email.mime.text import MIMEText 

32from typing import Any 

33 

34from init import _env_or_file # pyright: ignore[reportMissingImports] 

35 

36log = logging.getLogger(__name__) 

37 

38 

39class EmailNotConfiguredError(Exception): 

40 """Raised when OPENHANGAR_SMTP_HOST is not set.""" 

41 

42 

43class EmailSendError(Exception): 

44 """Raised when the SMTP transaction fails.""" 

45 

46 

47def _smtp_settings() -> dict[str, Any]: 

48 return { 

49 "host": os.environ.get("OPENHANGAR_SMTP_HOST", "").strip(), 

50 "port": int(os.environ.get("OPENHANGAR_SMTP_PORT", "587")), 

51 "user": os.environ.get("OPENHANGAR_SMTP_USER", "").strip(), 

52 "password": _env_or_file("SMTP_PASSWORD"), 

53 "use_tls": os.environ.get("OPENHANGAR_SMTP_USE_TLS", "true").lower() 

54 not in ("false", "0", "no"), 

55 "from_address": os.environ.get("OPENHANGAR_SMTP_FROM_ADDRESS", "").strip(), 

56 "from_name": os.environ.get("OPENHANGAR_SMTP_FROM_NAME", "OpenHangar").strip(), 

57 } 

58 

59 

60def get_smtp_status() -> dict[str, Any]: 

61 """ 

62 Return a dict describing the current SMTP configuration for display in 

63 the Configuration UI. Passwords are never included. 

64 Each value is the env var's value if explicitly set, or None if absent 

65 (so the UI can distinguish "not set" from a default). 

66 """ 

67 

68 def _env(key: str) -> str | None: 

69 v = os.environ.get(key, "").strip() 

70 return v or None 

71 

72 host = _env("OPENHANGAR_SMTP_HOST") 

73 from_address = _env("OPENHANGAR_SMTP_FROM_ADDRESS") 

74 return { 

75 "host": host, 

76 "port": int(os.environ.get("OPENHANGAR_SMTP_PORT", "587")), 

77 "port_is_default": "OPENHANGAR_SMTP_PORT" not in os.environ, 

78 "user": _env("OPENHANGAR_SMTP_USER"), 

79 "password_set": bool(_env_or_file("SMTP_PASSWORD").strip()), 

80 "use_tls": os.environ.get("OPENHANGAR_SMTP_USE_TLS", "true").lower() 

81 not in ("false", "0", "no"), 

82 "use_tls_is_default": "OPENHANGAR_SMTP_USE_TLS" not in os.environ, 

83 "from_address": from_address, 

84 "from_name": _env("OPENHANGAR_SMTP_FROM_NAME"), 

85 "configured": bool(host and from_address), 

86 } 

87 

88 

89def _record_health(success: bool) -> None: 

90 """Update email delivery health counters in AppSetting. Silently no-ops outside app context.""" 

91 try: 

92 from flask import has_app_context # pyright: ignore[reportMissingImports] 

93 

94 if not has_app_context(): 

95 return 

96 from models import AppSetting, db # pyright: ignore[reportMissingImports] 

97 

98 if success: 

99 now = datetime.now(UTC).isoformat() 

100 for key, val in [ 

101 ("email_last_success_at", now), 

102 ("email_consecutive_failures", "0"), 

103 ]: 

104 s = db.session.get(AppSetting, key) 

105 if s: 

106 s.value = val 

107 else: 

108 db.session.add(AppSetting(key=key, value=val)) 

109 else: 

110 s = db.session.get(AppSetting, "email_consecutive_failures") 

111 count = (int(s.value) + 1) if s and s.value else 1 

112 if s: 

113 s.value = str(count) 

114 else: 

115 db.session.add( 

116 AppSetting(key="email_consecutive_failures", value=str(count)) 

117 ) 

118 db.session.commit() 

119 except Exception as exc: # noqa: BLE001 -- non-fatal health tracking, must not break the send path 

120 log.debug("email health tracking failed (non-fatal): %s", exc) 

121 

122 

123def get_email_health() -> dict[str, Any]: 

124 """Return email delivery health dict. Must be called within an app context.""" 

125 if not os.environ.get("OPENHANGAR_SMTP_HOST", "").strip(): 

126 return { 

127 "status": "unconfigured", 

128 "consecutive_failures": 0, 

129 "last_success_at": None, 

130 } 

131 try: 

132 from models import AppSetting, db # pyright: ignore[reportMissingImports] 

133 

134 failures_row = db.session.get(AppSetting, "email_consecutive_failures") 

135 success_row = db.session.get(AppSetting, "email_last_success_at") 

136 consecutive_failures = ( 

137 int(failures_row.value) if failures_row and failures_row.value else 0 

138 ) 

139 last_success_at = success_row.value if success_row else None 

140 

141 if consecutive_failures == 0: 

142 status = "ok" 

143 elif last_success_at: 

144 status = "degraded" 

145 else: 

146 status = "never_worked" 

147 

148 return { 

149 "status": status, 

150 "consecutive_failures": consecutive_failures, 

151 "last_success_at": last_success_at, 

152 } 

153 except Exception: # noqa: BLE001 -- cosmetic health widget, degrade to "ok" rather than 500 

154 return {"status": "ok", "consecutive_failures": 0, "last_success_at": None} 

155 

156 

157_QUOTE_PLACEHOLDER = "<!-- QUOTE_PLACEHOLDER -->" 

158_DEFAULT_SUBJECT_PREFIX = "🛩️" 

159 

160 

161def _subject_prefix() -> str | None: 

162 """Return the text to prefix the subject with, or None if disabled. 

163 

164 OPENHANGAR_EMAIL_SUBJECT_PREFIX is true/false-like ("true"/"1"/"yes", 

165 "false"/"0"/"no") to toggle the default emoji, or any other short value 

166 (emoji or plain text) to use as a custom prefix instead. 

167 """ 

168 raw = os.environ.get("OPENHANGAR_EMAIL_SUBJECT_PREFIX", "true").strip() 

169 if raw.lower() in ("false", "0", "no"): 

170 return None 

171 if raw == "" or raw.lower() in ("true", "1", "yes"): 

172 return _DEFAULT_SUBJECT_PREFIX 

173 return raw 

174 

175 

176def send_email( 

177 to: str, 

178 subject: str, 

179 text_body: str, 

180 html_body: str | None = None, 

181 locale: str = "en", 

182) -> None: 

183 """ 

184 Send an email. 

185 

186 Raises EmailNotConfiguredError if SMTP_HOST is unset. 

187 Raises EmailSendError on SMTP failure. 

188 Silently does nothing in demo mode. 

189 

190 A randomly chosen aviation quote (locale-aware) is appended to the plain-text 

191 body and injected into the HTML body at the <!-- QUOTE_PLACEHOLDER --> anchor. 

192 """ 

193 if os.environ.get("OPENHANGAR_ENV") == "demo": 

194 return 

195 

196 s = _smtp_settings() 

197 if not s["host"]: 

198 raise EmailNotConfiguredError("SMTP_HOST is not configured.") 

199 if not s["from_address"]: 

200 raise EmailNotConfiguredError("SMTP_FROM_ADDRESS is not configured.") 

201 

202 from quotes import random_aviation_quote # pyright: ignore[reportMissingImports] 

203 

204 quote = random_aviation_quote(locale) 

205 text_body = text_body + f"\n\n—\n{quote}" 

206 if html_body and _QUOTE_PLACEHOLDER in html_body: 

207 quote_html = ( 

208 f'<p style="font-style:italic;color:#9ca3af;' 

209 f'margin:12px 0 0;font-size:11px;">{_html.escape(quote)}</p>' 

210 ) 

211 html_body = html_body.replace(_QUOTE_PLACEHOLDER, quote_html) 

212 

213 from_header = ( 

214 f"{s['from_name']} <{s['from_address']}>" 

215 if s["from_name"] 

216 else s["from_address"] 

217 ) 

218 

219 prefix = _subject_prefix() 

220 if prefix: 

221 subject = f"{prefix} {subject}" 

222 

223 msg = MIMEMultipart("alternative") 

224 msg["Subject"] = subject 

225 msg["From"] = from_header 

226 msg["To"] = to 

227 

228 msg.attach(MIMEText(text_body, "plain", "utf-8")) 

229 if html_body: 

230 msg.attach(MIMEText(html_body, "html", "utf-8")) 

231 

232 try: 

233 conn: smtplib.SMTP 

234 if s["use_tls"] and s["port"] == 465: 

235 # Port 465 = implicit SSL (SMTPS) — must use SMTP_SSL, not STARTTLS 

236 conn = smtplib.SMTP_SSL(s["host"], s["port"], timeout=10) 

237 elif s["use_tls"]: 

238 conn = smtplib.SMTP(s["host"], s["port"], timeout=10) 

239 conn.ehlo() 

240 conn.starttls() 

241 conn.ehlo() 

242 else: 

243 conn = smtplib.SMTP(s["host"], s["port"], timeout=10) 

244 

245 if s["user"]: 

246 conn.login(s["user"], s["password"]) 

247 

248 conn.sendmail(s["from_address"], [to], msg.as_bytes()) 

249 conn.quit() 

250 _record_health(success=True) 

251 except smtplib.SMTPException as exc: 

252 _record_health(success=False) 

253 _safe_to = to.replace("\n", " ").replace("\r", " ") 

254 log.warning("SMTP error sending to %s: %s", _safe_to, str(exc).splitlines()[0]) 

255 raise EmailSendError(str(exc)) from exc 

256 except OSError as exc: 

257 _record_health(success=False) 

258 _safe_to = to.replace("\n", " ").replace("\r", " ") 

259 log.warning( 

260 "OS error sending email to %s: %s", _safe_to, str(exc).splitlines()[0] 

261 ) 

262 raise EmailSendError(str(exc)) from exc