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 

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

17""" 

18 

19import html as _html 

20import logging 

21import os 

22import smtplib 

23from datetime import UTC, datetime 

24from email.mime.multipart import MIMEMultipart 

25from email.mime.text import MIMEText 

26from typing import Any 

27 

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

29 

30log = logging.getLogger(__name__) 

31 

32 

33class EmailNotConfiguredError(Exception): 

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

35 

36 

37class EmailSendError(Exception): 

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

39 

40 

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

42 return { 

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

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

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

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

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

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

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

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

51 } 

52 

53 

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

55 """ 

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

57 the Configuration UI. Passwords are never included. 

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

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

60 """ 

61 

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

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

64 return v or None 

65 

66 host = _env("OPENHANGAR_SMTP_HOST") 

67 from_address = _env("OPENHANGAR_SMTP_FROM_ADDRESS") 

68 return { 

69 "host": host, 

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

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

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

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

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

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

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

77 "from_address": from_address, 

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

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

80 } 

81 

82 

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

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

85 try: 

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

87 

88 if not has_app_context(): 

89 return 

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

91 

92 if success: 

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

94 for key, val in [ 

95 ("email_last_success_at", now), 

96 ("email_consecutive_failures", "0"), 

97 ]: 

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

99 if s: 

100 s.value = val 

101 else: 

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

103 else: 

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

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

106 if s: 

107 s.value = str(count) 

108 else: 

109 db.session.add( 

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

111 ) 

112 db.session.commit() 

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

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

115 

116 

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

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

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

120 return { 

121 "status": "unconfigured", 

122 "consecutive_failures": 0, 

123 "last_success_at": None, 

124 } 

125 try: 

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

127 

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

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

130 consecutive_failures = ( 

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

132 ) 

133 last_success_at = success_row.value if success_row else None 

134 

135 if consecutive_failures == 0: 

136 status = "ok" 

137 elif last_success_at: 

138 status = "degraded" 

139 else: 

140 status = "never_worked" 

141 

142 return { 

143 "status": status, 

144 "consecutive_failures": consecutive_failures, 

145 "last_success_at": last_success_at, 

146 } 

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

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

149 

150 

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

152 

153 

154def send_email( 

155 to: str, 

156 subject: str, 

157 text_body: str, 

158 html_body: str | None = None, 

159 locale: str = "en", 

160) -> None: 

161 """ 

162 Send an email. 

163 

164 Raises EmailNotConfiguredError if SMTP_HOST is unset. 

165 Raises EmailSendError on SMTP failure. 

166 Silently does nothing in demo mode. 

167 

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

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

170 """ 

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

172 return 

173 

174 s = _smtp_settings() 

175 if not s["host"]: 

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

177 if not s["from_address"]: 

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

179 

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

181 

182 quote = random_aviation_quote(locale) 

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

184 if html_body and _QUOTE_PLACEHOLDER in html_body: 

185 quote_html = ( 

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

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

188 ) 

189 html_body = html_body.replace(_QUOTE_PLACEHOLDER, quote_html) 

190 

191 from_header = ( 

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

193 if s["from_name"] 

194 else s["from_address"] 

195 ) 

196 

197 msg = MIMEMultipart("alternative") 

198 msg["Subject"] = subject 

199 msg["From"] = from_header 

200 msg["To"] = to 

201 

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

203 if html_body: 

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

205 

206 try: 

207 conn: smtplib.SMTP 

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

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

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

211 elif s["use_tls"]: 

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

213 conn.ehlo() 

214 conn.starttls() 

215 conn.ehlo() 

216 else: 

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

218 

219 if s["user"]: 

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

221 

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

223 conn.quit() 

224 _record_health(success=True) 

225 except smtplib.SMTPException as exc: 

226 _record_health(success=False) 

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

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

229 raise EmailSendError(str(exc)) from exc 

230 except OSError as exc: 

231 _record_health(success=False) 

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

233 log.warning( 

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

235 ) 

236 raise EmailSendError(str(exc)) from exc