/
opt
/
cloudlinux
/
venv
/
lib
/
python3.11
/
site-packages
/
ssa
/
internal
/
Upload Filee
HOME
# -*- coding: utf-8 -*- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT """ This module contains SSA Mailer class and helper tools """ import logging import os import smtplib import subprocess from configparser import ConfigParser, SectionProxy from email.message import EmailMessage from socket import gethostname from clcommon.lib.cledition import is_cl_solo_edition from jinja2 import Environment, FileSystemLoader from clcommon.lib.network import get_ip_addr from .constants import mail_template_location from .exceptions import SSAMailerError from .utils import duration_cast, format_date def _get_server_ip(hostname): return get_ip_addr(hostname) or 'UNKNOWN' def render_report_table(report_data: dict) -> tuple: """ Render an HTML table for e-mail report and a full e-mail message """ env = Environment( loader=FileSystemLoader(mail_template_location), trim_blocks=True, lstrip_blocks=True ) env.filters['duration_cast'] = duration_cast table_template = env.get_template('report_table.html') report_table = table_template.render(domains=report_data['domains'], # TODO: [unification] make mail template same across all editions is_solo=is_cl_solo_edition( skip_jwt_check=True)) mail_template = env.get_template('mail.html') hostname = gethostname() mail = mail_template.render(date=format_date(report_data['date']), report_table=report_table, hostname=hostname, ip_addr=_get_server_ip(hostname)) return report_table, mail class Mailer: """ Class contains SSA e-mail send logic """ def __init__(self): self.logger = logging.getLogger('mailer') self._sender = None @property def mail_server(self) -> tuple: """ Local mail server address """ return ('localhost',) @property def sender(self) -> str: """ 'From' mail address """ return f"slow-site-analyzer@{gethostname()}" @staticmethod def read_template(name: str) -> SectionProxy: """ Get preformatted data for e-mail by name of template """ tmpl = f'{mail_template_location}/{name}.ini' if os.path.exists(tmpl): config = ConfigParser(interpolation=None) config.read(tmpl) return config['data'] raise SSAMailerError( f'Failed to find template {name} in {mail_template_location}') def _smtp_send(self, message: EmailMessage) -> None: """ Send preformatted e-mail via localhost SMTP """ self.logger.info('Try to send via smtp') try: with smtplib.SMTP(*self.mail_server) as server: result = server.send_message(message) self.logger.info('Send result: %s', result) except smtplib.SMTPException as e: raise SSAMailerError(f'smtp mailing failed: {str(e)}') except (ConnectionError, OSError) as e: raise SSAMailerError(f'smtp connection failed: {str(e)}') def _console_send(self, message: EmailMessage) -> None: """ Send preformatted e-mail via sendmail utility """ self.logger.info('Try to send via sendmail utility') cmd = ["/usr/sbin/sendmail", "-t", "-oi"] try: subprocess.run(cmd, input=message.as_string(), capture_output=True, text=True, check=True) except (OSError, ValueError, subprocess.CalledProcessError) as e: raise SSAMailerError(f'sendmail utility failed with {str(e)}') def _send(self, mail: EmailMessage) -> None: """ Try to send mail via localhost smtp server, if fails -- try to use sendmail utility """ try: self._smtp_send(mail) except SSAMailerError as e: self.logger.error(str(e)) try: self._console_send(mail) except SSAMailerError as e: self.logger.error(str(e)) self.logger.critical( 'Both smtp and sendmail failed to send message to %s', mail['To']) def send_simple_mail(self, recipient: str, template: str = 'ssa_report', **kwargs) -> None: """ Create a message from given template, including additional kwargs, and send it to given recipient """ self._send(self._message(recipient, template, **kwargs)) def _message(self, recipient: str, template: str, **kwargs) -> EmailMessage: """ Form a message """ data = self.read_template(template) msg = EmailMessage() msg['Subject'] = data['subject'] % kwargs msg['From'] = self.sender msg['To'] = recipient msg.set_content(data['text'] % kwargs) msg.add_alternative(data['html'] % kwargs, subtype='html') self.logger.info('Generated mail --> %s', msg.as_string()) return msg