Coverage for custom_components/supernotify/transports/email.py: 95%
329 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-01 18:25 +0000
1from __future__ import annotations
3import datetime as dt
4import email.utils
5import logging
6import os
7import os.path
8import smtplib
9import time
10from contextlib import suppress
11from email.mime.application import MIMEApplication
12from email.mime.image import MIMEImage
13from email.mime.multipart import MIMEMultipart
14from email.mime.text import MIMEText
15from traceback import format_exception
16from typing import TYPE_CHECKING, Any, TypedDict
18import aiofiles
19from anyio import Path
20from homeassistant.components.notify.const import ATTR_DATA, ATTR_MESSAGE, ATTR_TARGET, ATTR_TITLE
21from homeassistant.components.smtp.const import CONF_SENDER_NAME, CONF_SERVER
22from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_SENDER, CONF_TIMEOUT, CONF_USERNAME, CONF_VERIFY_SSL
23from homeassistant.helpers.template import Template, TemplateError
24from homeassistant.util import dt as dt_util
25from homeassistant.util.ssl import create_client_context
27import custom_components.supernotify
28from custom_components.supernotify import const
29from custom_components.supernotify.common import CallRecord
30from custom_components.supernotify.const import (
31 ATTR_ACTION_URL,
32 ATTR_ACTION_URL_TITLE,
33 ATTR_EMAIL,
34 ATTR_MEDIA,
35 ATTR_MEDIA_SNAPSHOT_URL,
36 CONF_CONNECTION,
37 CONF_DELIVERY_DEFAULTS,
38 CONF_ENCRYPTION,
39 CONF_OPTIONS,
40 CONF_TEMPLATE,
41 EMAIL_OPTION_MODE_DIRECT,
42 EMAIL_OPTION_MODE_HA_SMTP,
43 OPTION_DEFAULT_TITLE,
44 OPTION_JPEG,
45 OPTION_MESSAGE_USAGE,
46 OPTION_MODE,
47 OPTION_PNG,
48 OPTION_SENDER,
49 OPTION_SENDER_NAME,
50 OPTION_SIMPLIFY_TEXT,
51 OPTION_STRICT_TEMPLATE,
52 OPTION_STRIP_URLS,
53 OPTION_TARGET_CATEGORIES,
54 TRANSPORT_EMAIL,
55)
56from custom_components.supernotify.model import (
57 DebugTrace,
58 DeliveryConfig,
59 MessageOnlyPolicy,
60 SuppressionReason,
61 TransportConfig,
62 TransportFeature,
63)
64from custom_components.supernotify.transport import Transport
66if TYPE_CHECKING:
67 from ssl import SSLContext
69 from homeassistant.helpers.typing import ConfigType
71 from custom_components.supernotify.context import Context
72 from custom_components.supernotify.envelope import Envelope
73 from custom_components.supernotify.hass_api import HomeAssistantAPI
75RE_VALID_EMAIL = (
76 r"^[a-zA-Z0-9.+/=?^_-]+@[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)+$"
77)
78OPTION_PREHEADER_BLANK = "preheader_blank"
79OPTION_PREHEADER_LENGTH = "preheader_length"
81DEFAULT_SMTP_PORT = 587
82DEFAULT_SMTP_ENCRYPTION = "starttls"
83DEFAULT_SMTP_TIMEOUT = 5
84NULL_RETURN_PATH = "<>"
86# Keys used in the HA core smtp integration's config entry data, for reuse when no
87# connection is configured here. "server" is smtp-specific; the rest match generic
88# homeassistant.const keys already imported above.
89HA_SMTP_DOMAIN = "smtp"
91IMPORTANCE_HEADER_MAP: dict[str, str] = {
92 const.PRIORITY_CRITICAL: "high",
93 const.PRIORITY_HIGH: "high",
94 const.PRIORITY_MEDIUM: "normal",
95 const.PRIORITY_LOW: "low",
96 const.PRIORITY_MINIMUM: "low",
97}
98PRIORITY_HEADER_MAP: dict[str, str] = {
99 const.PRIORITY_CRITICAL: "urgent",
100 const.PRIORITY_HIGH: "urgent",
101 const.PRIORITY_MEDIUM: "normal",
102 const.PRIORITY_LOW: "non-urgent",
103 const.PRIORITY_MINIMUM: "non-urgent",
104}
105X_MSMAIL_PRIORITY_HEADER_MAP: dict[str, str] = {
106 const.PRIORITY_CRITICAL: "High",
107 const.PRIORITY_HIGH: "High",
108 const.PRIORITY_MEDIUM: "Normal",
109 const.PRIORITY_LOW: "Low",
110 const.PRIORITY_MINIMUM: "Low",
111}
112X_PRIORITY_HEADER_MAP: dict[str, str] = {
113 const.PRIORITY_CRITICAL: "1",
114 const.PRIORITY_HIGH: "2",
115 const.PRIORITY_MEDIUM: "3",
116 const.PRIORITY_LOW: "4",
117 const.PRIORITY_MINIMUM: "5",
118}
120_LOGGER = logging.getLogger(__name__)
123class AlertServer(TypedDict):
124 name: str
125 internal_url: str
126 external_url: str
127 language: str
130class AlertImage(TypedDict):
131 url: str
132 desc: str
135class Alert(TypedDict):
136 message: str | None
137 title: str | None
138 preheader: str | None
139 priority: str
140 envelope: Envelope
141 action_url: str | None
142 action_url_title: str | None
143 subheading: str
144 server: AlertServer
145 preformatted_html: str | None
146 img: AlertImage | None
149class EmailTransport(Transport):
150 name = TRANSPORT_EMAIL
152 def __init__(self, context: Context, transport_config: ConfigType | None = None) -> None:
153 super().__init__(context, transport_config)
154 self.default_template_path: Path = Path(os.path.join(custom_components.supernotify.__path__[0], "default_templates"))
155 self.custom_template_path: Path | None = context.custom_template_path
156 self.custom_email_template_path: Path | None = None
157 self.template_cache: dict[str, str] = {}
159 # Connection details for sending via a direct SMTP connection - only used for
160 # deliveries with the OPTION_MODE option set to direct, rather than the default of
161 # calling an HA notify action, but always read here since a delivery can request
162 # direct sending independently of how this transport itself was configured.
163 connection: ConfigType = (transport_config or {}).get(CONF_CONNECTION, {})
164 self.host: str | None = connection.get(CONF_HOST)
165 self.port: int = connection.get(CONF_PORT, DEFAULT_SMTP_PORT)
166 self.encryption: str = connection.get(CONF_ENCRYPTION, DEFAULT_SMTP_ENCRYPTION)
167 self.username: str | None = connection.get(CONF_USERNAME)
168 self.password: str | None = connection.get(CONF_PASSWORD)
169 self.timeout: int = connection.get(CONF_TIMEOUT, DEFAULT_SMTP_TIMEOUT)
170 self.verify_ssl: bool = connection.get(CONF_VERIFY_SSL, True)
171 options: dict[str, Any] = (transport_config or {}).get(CONF_DELIVERY_DEFAULTS, {}).get(CONF_OPTIONS, {})
172 self.sender: str | None = options.get(OPTION_SENDER)
173 self.sender_name: str | None = options.get(OPTION_SENDER_NAME)
174 self.default_title: str | None = options.get(OPTION_DEFAULT_TITLE)
176 if not self.host:
177 self._reuse_ha_smtp_connection()
179 def _reuse_ha_smtp_connection(self) -> None:
180 """No direct SMTP connection configured here; fall back to a configured HA smtp
181 integration entry, if any."""
182 entry_data = self.hass_api.find_config_entry_data(HA_SMTP_DOMAIN)
183 if not entry_data:
184 _LOGGER.debug("SUPERNOTIFY No home assistant official smtp configuration to reuse")
185 return
186 _LOGGER.info("SUPERNOTIFY Email transport reusing connection from HA smtp integration for direct SMTP sends")
187 self.host = entry_data.get(CONF_SERVER)
188 self.port = entry_data.get(CONF_PORT, self.port)
189 self.encryption = entry_data.get(CONF_ENCRYPTION, self.encryption)
190 self.username = entry_data.get(CONF_USERNAME, self.username)
191 self.password = entry_data.get(CONF_PASSWORD, self.password)
192 self.verify_ssl = entry_data.get(CONF_VERIFY_SSL, self.verify_ssl)
193 if not self.sender:
194 self.sender = entry_data.get(CONF_SENDER)
195 if not self.sender_name:
196 self.sender_name = entry_data.get(CONF_SENDER_NAME)
198 async def initialize(self) -> None:
199 try:
200 if self.custom_template_path is not None:
201 if await self.custom_template_path.exists():
202 if await (self.custom_template_path / "email").exists():
203 _LOGGER.debug("SUPERNOTIFY Using email specific custom templates at %s", self.custom_template_path)
204 self.custom_email_template_path = Path(self.custom_template_path / "email")
205 else:
206 _LOGGER.debug("SUPERNOTIFY Email specific custom templates not configured")
207 else:
208 _LOGGER.info("SUPERNOTIFY Custom email template directory not present at %s", self.custom_template_path)
209 self.custom_template_path = None
210 else:
211 _LOGGER.info("SUPERNOTIFY Custom email templates not configured")
212 except Exception as e:
213 _LOGGER.error("SUPERNOTIFY Failed to verify custom template path %s: %s", self.custom_template_path, e)
215 def validate_action(self, action: str | None) -> bool:
216 """Valid either with an HA notify action, or a usable direct SMTP connection for
217 deliveries that set OPTION_MODE to 'direct'."""
218 return action is not None or bool(self.host and self.sender)
220 def auto_configure(self, hass_api: HomeAssistantAPI) -> DeliveryConfig | None:
221 action: str | None = hass_api.find_service("notify", "homeassistant.components.smtp.notify")
222 if action:
223 delivery_config: DeliveryConfig = self.delivery_defaults
224 delivery_config.action = action
225 return delivery_config
226 return None
228 @property
229 def supported_features(self) -> TransportFeature:
230 return (
231 TransportFeature.MESSAGE
232 | TransportFeature.TITLE
233 | TransportFeature.ACTIONS
234 | TransportFeature.IMAGES
235 | TransportFeature.TEMPLATE_FILE
236 | TransportFeature.SNAPSHOT_IMAGE
237 )
239 def extra_attributes(self) -> dict[str, Any]:
240 return {
241 "cached_templates": list(self.template_cache.keys()),
242 "custom_templates": str(self.custom_template_path) if self.custom_template_path else None,
243 "custom_email_templates": str(self.custom_email_template_path) if self.custom_email_template_path else None,
244 }
246 @property
247 def default_config(self) -> TransportConfig:
248 config = TransportConfig()
249 config.delivery_defaults.options = {
250 OPTION_SIMPLIFY_TEXT: False,
251 OPTION_STRIP_URLS: False,
252 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD,
253 OPTION_TARGET_CATEGORIES: [ATTR_EMAIL],
254 # use sensible defaults for image attachments
255 OPTION_JPEG: {"progressive": "true", "optimize": "true"},
256 OPTION_PNG: {"optimize": "true"},
257 OPTION_STRICT_TEMPLATE: False,
258 OPTION_PREHEADER_BLANK: "͏‌ ",
259 OPTION_PREHEADER_LENGTH: 100,
260 # only used for deliveries with OPTION_MODE set to 'direct'
261 OPTION_SENDER_NAME: "Home Assistant",
262 OPTION_DEFAULT_TITLE: "Home Assistant Notification",
263 }
264 return config
266 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:
267 _LOGGER.debug("SUPERNOTIFY notify_email: %s %s", envelope.delivery_name, envelope.target.email)
269 data: dict[str, Any] = envelope.data or {}
270 html: str | None = data.get("html")
271 template_name: str | None = data.get(CONF_TEMPLATE, envelope.delivery.template)
272 strict_template: bool = envelope.delivery.options.get(OPTION_STRICT_TEMPLATE, False)
273 addresses: list[str] = envelope.target.email or []
274 snapshot_url: str | None = data.get(ATTR_MEDIA, {}).get(ATTR_MEDIA_SNAPSHOT_URL)
275 if snapshot_url is None:
276 # older location for backward compatibility
277 snapshot_url = data.get(ATTR_MEDIA_SNAPSHOT_URL)
278 # TODO: centralize in config
279 footer_template = data.get("footer")
280 footer = None
281 if footer_template:
282 try:
283 footer = footer_template.format(e=envelope)
284 except (KeyError, ValueError, AttributeError) as ex: # py3.13 compat
285 _LOGGER.warning("SUPERNOTIFY email: failed to render footer template: %s", ex)
287 action_data: dict[str, Any] = envelope.core_action_data()
288 extra_data: dict[str, Any] = {k: v for k, v in data.items() if k not in action_data}
290 if len(addresses) > 0:
291 action_data[ATTR_TARGET] = addresses
292 # default to SMTP platform default recipients if no explicit addresses
294 if data and data.get("data"):
295 action_data[ATTR_DATA] = data.get("data")
297 image_path: Path | None = await envelope.grab_image()
298 if image_path:
299 action_data.setdefault("data", {})
300 action_data["data"]["images"] = [str(image_path)]
302 if not template_name:
303 if footer and action_data.get(ATTR_MESSAGE):
304 action_data[ATTR_MESSAGE] = f"{action_data[ATTR_MESSAGE]}\n\n{footer}"
306 if envelope.message_html:
307 action_data.setdefault("data", {})
308 html = envelope.message_html
309 if image_path:
310 image_name = image_path.name
311 if html and not html.rstrip().endswith("</html>"):
312 if snapshot_url:
313 html += f'<div><p><a href="{snapshot_url}">'
314 html += f'<img src="cid:{image_name}"/></a>'
315 html += "</p></div>"
316 else:
317 html += f'<div><p><img src="cid:{image_name}"></p></div>'
319 action_data["data"]["html"] = html
320 else:
321 html = await self.render_template(
322 template_name,
323 envelope,
324 action_data,
325 debug_trace,
326 image_path=image_path,
327 snapshot_url=snapshot_url,
328 extra_data=extra_data,
329 strict_template=strict_template,
330 )
331 if html:
332 action_data.setdefault("data", {})
333 action_data["data"]["html"] = html
334 return await self._send(envelope, action_data)
336 async def _send(self, envelope: Envelope, action_data: dict[str, Any]) -> bool:
337 """Send the built action_data, either via an HA notify action, or by owning the SMTP
338 connection directly - for deliveries with the OPTION_MODE option set, so
339 email can be sent to arbitrary addresses without every recipient needing to be
340 pre-registered as a notify entity, and isn't limited to whatever a given HA notify
341 action exposes."""
342 if envelope.delivery.options.get(OPTION_MODE, EMAIL_OPTION_MODE_HA_SMTP) == EMAIL_OPTION_MODE_DIRECT:
343 return await self._send_direct_smtp(envelope, action_data)
344 return await self.call_action(envelope, action_data=action_data)
346 async def _send_direct_smtp(self, envelope: Envelope, action_data: dict[str, Any]) -> bool:
347 addresses: list[str] = action_data.get(ATTR_TARGET) or []
348 start_time = time.time()
349 timestamp = dt.datetime.now(tz=dt_util.get_default_time_zone())
350 if not self.host or not self.sender:
351 _LOGGER.debug("SUPERNOTIFY Skipping direct smtp delivery %s, no connection configured", envelope.delivery.name)
352 envelope.skipped = 1
353 envelope.skip_reason = SuppressionReason.NO_ACTION
354 return False
355 if not addresses:
356 _LOGGER.debug("SUPERNOTIFY Skipping direct smtp delivery %s, no target addresses", envelope.delivery.name)
357 envelope.skipped = 1
358 envelope.skip_reason = SuppressionReason.NO_TARGET
359 return False
361 try:
362 msg = await self._build_message(action_data, addresses, envelope.priority, envelope.id)
363 await self.hass_api.create_job(self._send_smtp, msg, addresses)
364 envelope.calls.append(
365 CallRecord(
366 timestamp,
367 time.time() - start_time,
368 domain="smtp",
369 action="send_message",
370 debug=envelope.delivery.debug,
371 action_data=dict(action_data),
372 target_data={ATTR_TARGET: addresses},
373 )
374 )
375 envelope.delivered = 1
376 self.log_delivery_recovered()
377 return True
378 except Exception as e:
379 self.record_error(str(e), method="_send_direct_smtp")
380 envelope.failed_calls.append(
381 CallRecord(
382 timestamp,
383 time.time() - start_time,
384 domain="smtp",
385 action="send_message",
386 action_data=dict(action_data),
387 target_data={ATTR_TARGET: addresses},
388 exception=str(e),
389 )
390 )
391 self.log_delivery_failure(e, "SUPERNOTIFY Failed to send smtp email for %s", envelope.delivery.name)
392 envelope.error_count += 1
393 envelope.delivery_error = format_exception(e)
394 return False
396 async def _build_message(
397 self, action_data: dict[str, Any], addresses: list[str], priority: str | None, id: str | None
398 ) -> MIMEMultipart | MIMEText:
399 title: str | None = action_data.get(ATTR_TITLE)
400 message: str = action_data.get(ATTR_MESSAGE) or ""
401 data: dict[str, Any] = action_data.get(ATTR_DATA) or {}
402 html: str | None = data.get("html")
403 images: list[str] = data.get("images") or []
405 msg: MIMEMultipart | MIMEText
406 if html or images:
407 msg = MIMEMultipart("related")
408 alternative = MIMEMultipart("alternative")
409 alternative.attach(MIMEText(message, _charset="utf-8"))
410 if html:
411 alternative.attach(MIMEText(html, "html", _charset="utf-8"))
412 msg.attach(alternative)
413 for image_path in images:
414 attachment = await self._attach_file(image_path)
415 if attachment:
416 msg.attach(attachment)
417 else:
418 msg = MIMEText(message)
420 msg["Subject"] = title or self.default_title or ""
421 msg["To"] = ", ".join(addresses)
422 if self.sender_name or self.sender:
423 sender: str = email.utils.formataddr((self.sender_name or "", self.sender or ""))
424 else:
425 sender = NULL_RETURN_PATH
427 msg["From"] = sender
428 msg["X-Mailer"] = "Home Assistant Supernotify"
429 msg["Date"] = email.utils.format_datetime(dt_util.now())
430 msg["Message-Id"] = email.utils.make_msgid(idstring=id)
431 if priority:
432 msg["Importance"] = IMPORTANCE_HEADER_MAP.get(priority, "Normal")
433 msg["Priority"] = PRIORITY_HEADER_MAP.get(priority, "normal")
434 msg["X-Priority"] = X_PRIORITY_HEADER_MAP.get(priority, "3")
435 msg["X-MSMail-Priority"] = X_MSMAIL_PRIORITY_HEADER_MAP.get(priority, "Normal")
436 return msg
438 async def _attach_file(self, image_path: str) -> MIMEImage | MIMEApplication | None:
439 try:
440 async with aiofiles.open(image_path, "rb") as attachment_file:
441 file_bytes = await attachment_file.read()
442 except OSError:
443 _LOGGER.warning("SUPERNOTIFY SMTP attachment %s not found, skipping", image_path)
444 return None
446 content_id: str = os.path.basename(image_path)
447 attachment: MIMEImage | MIMEApplication
448 try:
449 attachment = MIMEImage(file_bytes)
450 except TypeError:
451 attachment = MIMEApplication(file_bytes, Name=content_id)
452 attachment["Content-Disposition"] = f'attachment; filename="{content_id}"'
453 attachment.add_header("Content-ID", f"<{content_id}>")
454 return attachment
456 def _send_smtp(self, msg: MIMEMultipart | MIMEText, addresses: list[str]) -> None:
457 if not self.host or not self.port:
458 _LOGGER.warning("SUPERNOTIFY Direct SMTP connection not configured")
459 return
461 ssl_context: SSLContext | None = create_client_context() if self.verify_ssl else None
462 client: smtplib.SMTP | smtplib.SMTP_SSL
463 if self.encryption == "tls":
464 client = smtplib.SMTP_SSL(self.host, self.port, timeout=self.timeout, context=ssl_context)
465 else:
466 client = smtplib.SMTP(self.host, self.port, timeout=self.timeout)
467 try:
468 client.ehlo_or_helo_if_needed()
469 if self.encryption == "starttls":
470 client.starttls(context=ssl_context)
471 client.ehlo()
472 if self.username and self.password:
473 client.login(self.username, self.password)
474 client.sendmail(self.sender or NULL_RETURN_PATH, addresses, msg.as_string())
475 finally:
476 with suppress(smtplib.SMTPException):
477 client.quit()
479 async def load_template(self, template_name: str) -> str | None:
480 if template_name in self.template_cache:
481 return self.template_cache[template_name]
483 for root_path in (
484 self.custom_email_template_path,
485 self.custom_template_path,
486 self.default_template_path / "email",
487 self.default_template_path,
488 ):
489 if root_path is not None:
490 template_path: Path = root_path / template_name
491 if await template_path.exists():
492 template: str
493 async with aiofiles.open(template_path) as file:
494 template = os.linesep.join(await file.readlines())
495 self.template_cache[template_name] = template
496 return template
497 return None
499 async def render_template(
500 self,
501 template_name: str,
502 envelope: Envelope,
503 action_data: dict[str, Any],
504 debug_trace: DebugTrace | None = None,
505 image_path: Path | None = None,
506 snapshot_url: str | None = None,
507 extra_data: dict[str, Any] | None = None,
508 strict_template: bool = False,
509 ) -> str | None:
510 extra_data = extra_data or {}
511 alert: Alert
513 try:
514 title: str | None = action_data.get(ATTR_TITLE)
515 message: str | None = action_data.get(ATTR_MESSAGE)
516 preheader: str = f"{title or ''}{' ' if title else ''}{message}"
517 preheader = preheader or "Home Assistant Notification"
518 alert = Alert(
519 message=message,
520 title=title,
521 preheader=self.pack_preheader(preheader, envelope.delivery.options),
522 priority=envelope.priority,
523 action_url=extra_data.get(ATTR_ACTION_URL),
524 action_url_title=extra_data.get(ATTR_ACTION_URL_TITLE),
525 envelope=envelope,
526 subheading="Home Assistant Notification",
527 server=AlertServer(
528 name=self.hass_api.hass_name,
529 internal_url=self.hass_api.internal_url,
530 external_url=self.hass_api.external_url,
531 language=self.hass_api.language,
532 ),
533 preformatted_html=envelope.message_html,
534 img=None,
535 )
537 if snapshot_url:
538 alert["img"] = AlertImage(url=snapshot_url, desc="Snapshot Image")
539 elif image_path:
540 alert["img"] = AlertImage(url=f"cid:{image_path.name}", desc=image_path.name)
542 template_content: str | None = await self.load_template(template_name)
544 if template_content is None:
545 _LOGGER.error("SUPERNOTIFY No template found for %s", template_name)
546 return None
548 template_obj: Template = self.context.hass_api.template(template_content)
549 template_obj.ensure_valid()
551 if debug_trace:
552 debug_trace.record_delivery_artefact(envelope.delivery.name, "alert", alert)
554 html: str = template_obj.async_render(variables={"alert": alert}, parse_result=False, strict=strict_template)
555 if not html:
556 _LOGGER.error("SUPERNOTIFY Empty result from template %s", template_name)
557 else:
558 return html
559 except TemplateError as te:
560 _LOGGER.exception("SUPERNOTIFY Failed to render template html mail")
561 if debug_trace:
562 debug_trace.record_delivery_exception(envelope.delivery.name, "html_template", te)
563 except Exception as e:
564 _LOGGER.exception("SUPERNOTIFY Failed to generate html mail")
565 if debug_trace:
566 debug_trace.record_delivery_exception(envelope.delivery.name, "html_template", e)
567 return None
569 def pack_preheader(self, preheader: str, options: dict[str, Any]) -> str:
570 preheader = preheader or ""
571 phchars: str = options.get(OPTION_PREHEADER_BLANK, "")
572 phlength: int = options.get(OPTION_PREHEADER_LENGTH, 0)
573 if phlength and phchars:
574 return f"{preheader}{phchars * (phlength - len(preheader))}"
575 return preheader