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 datetime, timezone 

24from typing import Any 

25from email.mime.multipart import MIMEMultipart 

26from email.mime.text import MIMEText 

27 

28log = logging.getLogger(__name__) 

29 

30 

31class EmailNotConfiguredError(Exception): 

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

33 

34 

35class EmailSendError(Exception): 

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

37 

38 

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

40 return { 

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

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

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

44 "password": os.environ.get("OPENHANGAR_SMTP_PASSWORD", ""), 

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

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

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

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

49 } 

50 

51 

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

53 """ 

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

55 the Configuration UI. Passwords are never included. 

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

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

58 """ 

59 

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

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

62 return v or None 

63 

64 host = _env("OPENHANGAR_SMTP_HOST") 

65 from_address = _env("OPENHANGAR_SMTP_FROM_ADDRESS") 

66 return { 

67 "host": host, 

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

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

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

71 "password_set": bool(os.environ.get("OPENHANGAR_SMTP_PASSWORD", "").strip()), 

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

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

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

75 "from_address": from_address, 

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

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

78 } 

79 

80 

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

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

83 try: 

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

85 

86 if not has_app_context(): 

87 return 

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

89 

90 if success: 

91 now = datetime.now(timezone.utc).isoformat() 

92 for key, val in [ 

93 ("email_last_success_at", now), 

94 ("email_consecutive_failures", "0"), 

95 ]: 

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

97 if s: 

98 s.value = val 

99 else: 

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

101 else: 

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

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

104 if s: 

105 s.value = str(count) 

106 else: 

107 db.session.add( 

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

109 ) 

110 db.session.commit() 

111 except Exception as exc: 

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

113 

114 

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

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

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

118 return { 

119 "status": "unconfigured", 

120 "consecutive_failures": 0, 

121 "last_success_at": None, 

122 } 

123 try: 

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

125 

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

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

128 consecutive_failures = ( 

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

130 ) 

131 last_success_at = success_row.value if success_row else None 

132 

133 if consecutive_failures == 0: 

134 status = "ok" 

135 elif last_success_at: 

136 status = "degraded" 

137 else: 

138 status = "never_worked" 

139 

140 return { 

141 "status": status, 

142 "consecutive_failures": consecutive_failures, 

143 "last_success_at": last_success_at, 

144 } 

145 except Exception: 

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

147 

148 

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

150 

151 

152def send_email( 

153 to: str, 

154 subject: str, 

155 text_body: str, 

156 html_body: str | None = None, 

157 locale: str = "en", 

158) -> None: 

159 """ 

160 Send an email. 

161 

162 Raises EmailNotConfiguredError if SMTP_HOST is unset. 

163 Raises EmailSendError on SMTP failure. 

164 Silently does nothing in demo mode. 

165 

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

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

168 """ 

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

170 return 

171 

172 s = _smtp_settings() 

173 if not s["host"]: 

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

175 if not s["from_address"]: 

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

177 

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

179 

180 quote = random_aviation_quote(locale) 

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

182 if html_body and _QUOTE_PLACEHOLDER in html_body: 

183 quote_html = ( 

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

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

186 ) 

187 html_body = html_body.replace(_QUOTE_PLACEHOLDER, quote_html) 

188 

189 from_header = ( 

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

191 if s["from_name"] 

192 else s["from_address"] 

193 ) 

194 

195 msg = MIMEMultipart("alternative") 

196 msg["Subject"] = subject 

197 msg["From"] = from_header 

198 msg["To"] = to 

199 

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

201 if html_body: 

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

203 

204 try: 

205 conn: smtplib.SMTP 

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

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

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

209 elif s["use_tls"]: 

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

211 conn.ehlo() 

212 conn.starttls() 

213 conn.ehlo() 

214 else: 

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

216 

217 if s["user"]: 

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

219 

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

221 conn.quit() 

222 _record_health(success=True) 

223 except smtplib.SMTPException as exc: 

224 _record_health(success=False) 

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

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

227 raise EmailSendError(str(exc)) from exc 

228 except OSError as exc: 

229 _record_health(success=False) 

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

231 log.warning( 

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

233 ) 

234 raise EmailSendError(str(exc)) from exc