"""Email notification functionality.""" import logging import smtplib from . import datetime_str from email.mime.text import MIMEText from email.utils import formatdate from jinja2 import Environment, BaseLoader def load_template(path): """ Load template from file. Args: path: Path to template. Returns: Template if path was not None, None otherwise. """ if path is None: return with open(path) as fh: data = fh.read() return Environment( loader=BaseLoader(), trim_blocks=True ).from_string(data) class EmailService: """Email service for sending notifications.""" def __init__(self, config): """ Initialize email service. Args: config: Application configuration dictionary. """ self.config = config.get("email", {}) self.enabled = self.config.get("enabled", False) self.change_template = load_template(self.config.get("change_notification_template", None)) self.expiry_template = load_template(self.config.get("expiry_notification_template", None)) if not self.change_template and not self.expiry_template and self.enabled: logging.warning("No templates configured, disabling Email") self.enabled = False def send(self, to, subject, body): """ Send email using configured SMTP server. Args: to: Recipient email address. subject: Email subject. body: Email body text. Returns: True if sent successfully, False otherwise. """ if not self.enabled: logging.debug("Email disabled, skipping") return False try: msg = MIMEText(body) msg["Subject"] = subject msg["From"] = self.config["from_address"] msg["To"] = to msg["Date"] = formatdate(localtime=True) smtp_host = self.config["smtp_host"] smtp_port = self.config["smtp_port"] with smtplib.SMTP(smtp_host, smtp_port) as server: if self.config.get("smtp_starttls", False): server.starttls() if self.config.get("smtp_user"): server.login( self.config["smtp_user"], self.config["smtp_password"] ) server.sendmail(msg["From"], [to], msg.as_string()) logging.info(f"Email sent: to={to} subject={subject}") return True except Exception as e: logging.error(f"Email send failed: to={to} error={e}") return False def send_change_notification( self, email, hostname, ipv4_changed, ipv6_changed ): """ Send hostname changed notification email. Args: email: Recipient email. hostname: Changed hostname. ipv4_changed: Wheter IPv4 address changed. ipv6_changed: Wheter IPv6 address changed. Returns: True if sent successfully. """ if not self.enabled: logging.debug("Email disabled, skipping") return False if not self.change_template: logging.debug("No change notification template configured, skipping") return False fqdn = f"{hostname.hostname}.{hostname.zone}" template_variables = { "hostname": hostname.hostname, "zone": hostname.zone, "fqdn": fqdn, "ipv4_changed": ipv4_changed, "ipv4": hostname.last_ipv4, "last_ipv4_update": datetime_str(hostname.last_ipv4_update), "ipv6_changed": ipv6_changed, "ipv6": hostname.last_ipv6, "last_ipv6_update": datetime_str(hostname.last_ipv6_update), "expiry_ttl": hostname.expiry_ttl, } subject = f"DDNS hostname changed: {fqdn}" body = self.change_template.render(**template_variables) return self.send(email, subject, body) def send_expiry_notification( self, email, hostname, ipv4_expired, ipv6_expired ): """ Send hostname expiry notification email. Args: email: Recipient email. hostname: Expired hostname. ipv4_expired: Wheter IPv4 address expired. ipv6_expired: Wheter IPv6 address expired. Returns: True if sent successfully. """ if not self.enabled: logging.debug("Email disabled, skipping") return False if not self.expiry_template: logging.debug("No expiry notification template configured, skipping") return False fqdn = f"{hostname.hostname}.{hostname.zone}" template_variables = { "hostname": hostname.hostname, "zone": hostname.zone, "fqdn": fqdn, "ipv4_expired": ipv4_expired, "last_ipv4": hostname.last_ipv4, "last_ipv4_update": datetime_str(hostname.last_ipv4_update), "ipv6_expired": ipv6_expired, "last_ipv6": hostname.last_ipv6, "last_ipv6_update": datetime_str(hostname.last_ipv6_update), "expiry_ttl": hostname.expiry_ttl, } subject = f"DDNS hostname expired: {fqdn}" body = self.expiry_template.render(**template_variables) return self.send(email, subject, body)