diff --git a/modules/maintenance.py b/modules/maintenance.py index 56194a3..c2f7423 100644 --- a/modules/maintenance.py +++ b/modules/maintenance.py @@ -8,6 +8,7 @@ import json import os import sqlite3 import time +from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Any, Callable @@ -109,6 +110,67 @@ def _count_log_errors_last_24h(log_path: Path) -> tuple[int | str, int | str]: return 'n/a', 'n/a' + +@dataclass(frozen=True) +class SmtpSettings: + """The [notif.*] SMTP settings shared by the nightly digest and the radio alerts.""" + + host: str + security: str + user: str + password: str + from_name: str + from_email: str + port: int + allow_local: bool + + @property + def from_header(self) -> str: + return f'{self.from_name} <{self.from_email}>' + + +def read_smtp_settings(get_notif: Callable[[str], str]) -> SmtpSettings: + """Read the SMTP settings; port defaults to 465 for ssl, else 587 (587 when unparseable).""" + security = get_notif('smtp_security') or 'starttls' + try: + port = int(get_notif('smtp_port') or (465 if security == 'ssl' else 587)) + except ValueError: + port = 587 + return SmtpSettings( + host=get_notif('smtp_host'), + security=security, + user=get_notif('smtp_user'), + password=get_notif('smtp_password'), + from_name=get_notif('from_name') or 'MeshCore Bot', + from_email=get_notif('from_email'), + port=port, + allow_local=get_notif('allow_local_smtp').lower() == 'true', + ) + + +def send_smtp_message(settings: SmtpSettings, msg: Any) -> None: + """Deliver *msg* over SSL, STARTTLS or plain SMTP per the settings; raises on failure.""" + import smtplib + import ssl as _ssl + + context = _ssl.create_default_context() + smtp_timeout = 30 + if settings.security == 'ssl': + with smtplib.SMTP_SSL(settings.host, settings.port, context=context, timeout=smtp_timeout) as s: + if settings.user and settings.password: + s.login(settings.user, settings.password) + s.send_message(msg) + else: + with smtplib.SMTP(settings.host, settings.port, timeout=smtp_timeout) as s: + if settings.security == 'starttls': + s.ehlo() + s.starttls(context=context) + s.ehlo() + if settings.user and settings.password: + s.login(settings.user, settings.password) + s.send_message(msg) + + class MaintenanceRunner: """Runs data retention, nightly email, log rotation hot-apply, and DB backups.""" @@ -415,41 +477,28 @@ class MaintenanceRunner: def send_nightly_email(self) -> None: """Build and dispatch the nightly maintenance digest if enabled.""" - import smtplib - import ssl as _ssl from email.message import EmailMessage if self.get_notif('nightly_enabled') != 'true': return - smtp_host = self.get_notif('smtp_host') - smtp_security = self.get_notif('smtp_security') or 'starttls' - smtp_user = self.get_notif('smtp_user') - smtp_password = self.get_notif('smtp_password') - from_name = self.get_notif('from_name') or 'MeshCore Bot' - from_email = self.get_notif('from_email') + smtp = read_smtp_settings(self.get_notif) recipients = [r.strip() for r in self.get_notif('recipients').split(',') if r.strip()] - if not smtp_host or not from_email or not recipients: + if not smtp.host or not smtp.from_email or not recipients: self.logger.warning( "Nightly email enabled but SMTP settings incomplete " - f"(host={smtp_host!r}, from={from_email!r}, recipients={recipients})" + f"(host={smtp.host!r}, from={smtp.from_email!r}, recipients={recipients})" ) return - allow_local = self.get_notif('allow_local_smtp').lower() == 'true' - if not validate_external_url(f'http://{smtp_host}', allow_private=allow_local): + if not validate_external_url(f'http://{smtp.host}', allow_private=smtp.allow_local): self.logger.error( "Nightly email aborted: SMTP host %r resolves to a private or reserved address", - smtp_host, + smtp.host, ) return - try: - smtp_port = int(self.get_notif('smtp_port') or (465 if smtp_security == 'ssl' else 587)) - except ValueError: - smtp_port = 587 - now_utc = _utc_now() yesterday = now_utc - datetime.timedelta(days=1) period_start = yesterday.strftime('%Y-%m-%d %H:%M UTC') @@ -461,7 +510,7 @@ class MaintenanceRunner: msg = EmailMessage() msg['Subject'] = f'MeshCore Bot — Nightly Report {now_utc.strftime("%Y-%m-%d")}' - msg['From'] = f'{from_name} <{from_email}>' + msg['From'] = smtp.from_header msg['To'] = ', '.join(recipients) msg.set_content(body) @@ -478,22 +527,7 @@ class MaintenanceRunner: except Exception as attach_err: self.logger.warning(f"Could not attach log file to nightly email: {attach_err}") - context = _ssl.create_default_context() - _smtp_timeout = 30 - if smtp_security == 'ssl': - with smtplib.SMTP_SSL(smtp_host, smtp_port, context=context, timeout=_smtp_timeout) as s: - if smtp_user and smtp_password: - s.login(smtp_user, smtp_password) - s.send_message(msg) - else: - with smtplib.SMTP(smtp_host, smtp_port, timeout=_smtp_timeout) as s: - if smtp_security == 'starttls': - s.ehlo() - s.starttls(context=context) - s.ehlo() - if smtp_user and smtp_password: - s.login(smtp_user, smtp_password) - s.send_message(msg) + send_smtp_message(smtp, msg) self.logger.info( f"Nightly maintenance email sent to {recipients} " diff --git a/modules/scheduler.py b/modules/scheduler.py index e90c147..f6d0779 100644 --- a/modules/scheduler.py +++ b/modules/scheduler.py @@ -23,7 +23,7 @@ from apscheduler.triggers.date import DateTrigger from meshcore.events import EventType from .flood_scope import scope_key_hex -from .maintenance import MaintenanceRunner +from .maintenance import MaintenanceRunner, read_smtp_settings, send_smtp_message from .models import CHANNEL_REGIONAL_FLOOD_SCOPE_BODY_OVERHEAD, channel_body_limit, self_info_name from .scheduled_message_cron import ( is_valid_legacy_hhmm, @@ -1744,8 +1744,6 @@ class MessageScheduler: This method is intentionally synchronous so it can be run in a thread executor from the async event loop without blocking it. """ - import smtplib - import ssl as _ssl from email.message import EmailMessage zombie_alert_enabled = self.bot.config.getboolean( @@ -1770,12 +1768,7 @@ class MessageScheduler: if not zombie_alert_enabled: return - smtp_host = self._get_notif('smtp_host') - smtp_security = self._get_notif('smtp_security') or 'starttls' - smtp_user = self._get_notif('smtp_user') - smtp_password = self._get_notif('smtp_password') - from_name = self._get_notif('from_name') or 'MeshCore Bot' - from_email = self._get_notif('from_email') + smtp = read_smtp_settings(self._get_notif) # Alert recipients: dedicated config key, falls back to nightly recipients alert_email_cfg = zombie_alert_email_cfg or self.bot.config.get( @@ -1788,27 +1781,21 @@ class MessageScheduler: else: recipients = [r.strip() for r in self._get_notif('recipients').split(',') if r.strip()] - if not smtp_host or not from_email or not recipients: + if not smtp.host or not smtp.from_email or not recipients: self.bot.logger.warning( "Zombie alert email enabled but SMTP settings incomplete " - f"(host={smtp_host!r}, from={from_email!r}, recipients={recipients}) " + f"(host={smtp.host!r}, from={smtp.from_email!r}, recipients={recipients}) " "— alert email not sent" ) return - allow_local = self._get_notif('allow_local_smtp').lower() == 'true' - if not validate_external_url(f'http://{smtp_host}', allow_private=allow_local): + if not validate_external_url(f'http://{smtp.host}', allow_private=smtp.allow_local): self.bot.logger.error( "Zombie alert email aborted: SMTP host %r resolves to a private or reserved address", - smtp_host, + smtp.host, ) return - try: - smtp_port = int(self._get_notif('smtp_port') or (465 if smtp_security == 'ssl' else 587)) - except ValueError: - smtp_port = 587 - now_utc = datetime.datetime.now(datetime.timezone.utc) connection_type = self.bot.config.get('Connection', 'connection_type', fallback='unknown') serial_port = self.bot.config.get('Connection', 'serial_port', fallback='n/a') @@ -1850,27 +1837,11 @@ class MessageScheduler: try: msg = EmailMessage() msg['Subject'] = subject - msg['From'] = f'{from_name} <{from_email}>' + msg['From'] = smtp.from_header msg['To'] = ', '.join(recipients) msg.set_content(body) - context = _ssl.create_default_context() - _smtp_timeout = 30 - - if smtp_security == 'ssl': - with smtplib.SMTP_SSL(smtp_host, smtp_port, context=context, timeout=_smtp_timeout) as s: - if smtp_user and smtp_password: - s.login(smtp_user, smtp_password) - s.send_message(msg) - else: - with smtplib.SMTP(smtp_host, smtp_port, timeout=_smtp_timeout) as s: - if smtp_security == 'starttls': - s.ehlo() - s.starttls(context=context) - s.ehlo() - if smtp_user and smtp_password: - s.login(smtp_user, smtp_password) - s.send_message(msg) + send_smtp_message(smtp, msg) self.bot.logger.info( f"Zombie radio alert email sent to {recipients}" @@ -1903,8 +1874,6 @@ class MessageScheduler: Intentionally synchronous — intended to be run in a daemon thread. """ - import smtplib - import ssl as _ssl from email.message import EmailMessage alert_enabled = self.bot.config.getboolean( @@ -1915,12 +1884,7 @@ class MessageScheduler: if not alert_enabled: return - smtp_host = self._get_notif('smtp_host') - smtp_security = self._get_notif('smtp_security') or 'starttls' - smtp_user = self._get_notif('smtp_user') - smtp_password = self._get_notif('smtp_password') - from_name = self._get_notif('from_name') or 'MeshCore Bot' - from_email = self._get_notif('from_email') + smtp = read_smtp_settings(self._get_notif) alert_email_cfg = self.bot.config.get( 'Connection', @@ -1932,27 +1896,21 @@ class MessageScheduler: else: recipients = [r.strip() for r in self._get_notif('recipients').split(',') if r.strip()] - if not smtp_host or not from_email or not recipients: + if not smtp.host or not smtp.from_email or not recipients: self.bot.logger.warning( "Radio-offline alert email enabled but SMTP settings incomplete " - f"(host={smtp_host!r}, from={from_email!r}, recipients={recipients}) " + f"(host={smtp.host!r}, from={smtp.from_email!r}, recipients={recipients}) " "— alert email not sent" ) return - allow_local = self._get_notif('allow_local_smtp').lower() == 'true' - if not validate_external_url(f'http://{smtp_host}', allow_private=allow_local): + if not validate_external_url(f'http://{smtp.host}', allow_private=smtp.allow_local): self.bot.logger.error( "Radio-offline alert email aborted: SMTP host %r resolves to a private or reserved address", - smtp_host, + smtp.host, ) return - try: - smtp_port = int(self._get_notif('smtp_port') or (465 if smtp_security == 'ssl' else 587)) - except ValueError: - smtp_port = 587 - now_utc = datetime.datetime.now(datetime.timezone.utc) connection_type = self.bot.config.get('Connection', 'connection_type', fallback='unknown') serial_port = self.bot.config.get('Connection', 'serial_port', fallback='n/a') @@ -1992,27 +1950,11 @@ class MessageScheduler: try: msg = EmailMessage() msg['Subject'] = subject - msg['From'] = f'{from_name} <{from_email}>' + msg['From'] = smtp.from_header msg['To'] = ', '.join(recipients) msg.set_content(body) - context = _ssl.create_default_context() - _smtp_timeout = 30 - - if smtp_security == 'ssl': - with smtplib.SMTP_SSL(smtp_host, smtp_port, context=context, timeout=_smtp_timeout) as s: - if smtp_user and smtp_password: - s.login(smtp_user, smtp_password) - s.send_message(msg) - else: - with smtplib.SMTP(smtp_host, smtp_port, timeout=_smtp_timeout) as s: - if smtp_security == 'starttls': - s.ehlo() - s.starttls(context=context) - s.ehlo() - if smtp_user and smtp_password: - s.login(smtp_user, smtp_password) - s.send_message(msg) + send_smtp_message(smtp, msg) self.bot.logger.info(f"Radio-offline alert email sent to {recipients}") except Exception as e: