from flask import Flask, request, jsonify from prometheus_flask_exporter import PrometheusMetrics import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.mime.application import MIMEApplication import os import time import html import re import logging class FilterRemoveDate(logging.Filter): # '192.168.0.102 - - [30/Jun/2024 01:14:03] "%s" %s %s' -> '192.168.0.102 - "%s" %s %s' pattern: re.Pattern = re.compile(r' - \[.+?]') def filter(self, record: logging.LogRecord) -> bool: record.msg = self.pattern.sub('', record.msg) return True class FilterReplaceWerkzeug(logging.Filter): # 'werkzeug:' -> 'app:flask:' pattern: re.Pattern = re.compile(r'werkzeug:') def filter(self, record: logging.LogRecord) -> bool: record.msg = self.pattern.sub('app:flask:', record.msg) return True class FilterReplaceLowercaseI(logging.Filter): # 'I:' -> 'i:' pattern: re.Pattern = re.compile(r'I:') def filter(self, record: logging.LogRecord) -> bool: record.msg = self.pattern.sub('i:', record.msg) return True # Setup logger logging.basicConfig( level=logging.DEBUG, format='%(asctime)s.%(msecs)03dZ %(name)s:%(levelname).1s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger('app') logger_werkzeug = logging.getLogger('werkzeug') logger_werkzeug.addFilter(FilterRemoveDate()) logger_werkzeug.addFilter(FilterReplaceWerkzeug()) logger_werkzeug.addFilter(FilterReplaceLowercaseI()) app = Flask(__name__, static_folder=None) # Wrap the Flask app with PrometheusMetrics metrics = PrometheusMetrics(app, defaults_prefix="mail") email_sent_counter = metrics.counter( 'mail_text_sent_total', 'Total number of text messages sent', labels={'endpoint': lambda: request.endpoint} ) pdf_sent_counter = metrics.counter( 'mail_pdf_sent_total', 'Total number of pdfs sent', labels={'endpoint': lambda: request.endpoint} ) @app.route('/v1') @metrics.do_not_track() def info(): today = time.strftime("%Y-%m-%d") url_map = app.url_map return f"

mail v1.0.0 - {today}

" + html.escape(str(url_map), False) + "
" @app.route('/v1/send/message', methods=['POST']) @email_sent_counter def send_message(): logger = logging.getLogger('app:send-message') data = request.json subject = data.get('subject') text_content = data.get('message') to_email = data.get('to_email') html_content = data.get('html') logger.info(f"Sending text email to {to_email}") logger.debug(f"Subject: {subject}") # SMTP Configuration smtp_server = os.environ.get('SMTP_RELAY_HOST') smtp_port = int(os.environ.get('SMTP_RELAY_PORT')) smtp_username = os.environ.get('SMTP_RELAY_USERNAME') smtp_password = os.environ.get('SMTP_RELAY_PASSWORD') try: # Create a message msg = MIMEMultipart('mixed') msg['From'] = smtp_username msg['To'] = to_email msg['Subject'] = subject # Create the multipart/alternative part msgAlternative = MIMEMultipart('alternative') msg.attach(msgAlternative) # Attach plain text message first (lower priority) msgAlternative.attach(MIMEText(text_content, 'plain', 'utf-8')) # Attach HTML version if provided (higher priority) if html_content: logger.debug("HTML content included in email") msgAlternative.attach(MIMEText(html_content, 'html', 'utf-8')) # Send email logger.debug(f"Connecting to SMTP server {smtp_server}:{smtp_port}") with smtplib.SMTP_SSL(smtp_server, smtp_port) as smtp: smtp.login(smtp_username, smtp_password) smtp.sendmail(smtp_username, to_email, msg.as_string()) smtp.quit() logger.info("Email sent successfully") return jsonify({"message": "Email sent successfully"}), 200 except Exception as e: logger.error(f"Error sending email: {str(e)}") return jsonify({"error": str(e)}), 500 @app.route('/v1/send/pdf', methods=['POST']) @pdf_sent_counter def send_pdf(): logger = logging.getLogger('app:send-pdf') data = request.form subject = data.get('subject') text_content = data.get('message') html_content = data.get('html') to_email = data.get('to_email') pdf_blob = request.files.get('pdf_blob') logger.info(f"Sending PDF email to {to_email}") logger.debug(f"Subject: {subject}") # SMTP Configuration smtp_server = os.environ.get('SMTP_RELAY_HOST') smtp_port = int(os.environ.get('SMTP_RELAY_PORT')) smtp_username = os.environ.get('SMTP_RELAY_USERNAME') smtp_password = os.environ.get('SMTP_RELAY_PASSWORD') try: # Create a message with mixed content (attachments + text) msg = MIMEMultipart('mixed') msg['From'] = smtp_username msg['To'] = to_email msg['Subject'] = subject # Create the multipart/alternative part for text and HTML msgAlternative = MIMEMultipart('alternative') msg.attach(msgAlternative) # Attach plain text message first (lower priority) msgAlternative.attach(MIMEText(text_content, 'plain', 'utf-8')) # Attach HTML version if provided (higher priority) if html_content: logger.debug("HTML content included in email") msgAlternative.attach(MIMEText(html_content, 'html', 'utf-8')) # Attach PDF if provided if pdf_blob: logger.debug("Attaching PDF to email") pdf_attachment = MIMEApplication(pdf_blob.read(), _subtype="pdf") pdf_attachment.add_header('Content-Disposition', 'attachment', filename='document.pdf') msg.attach(pdf_attachment) else: logger.warning("No PDF attachment provided") # Send email logger.debug(f"Connecting to SMTP server {smtp_server}:{smtp_port}") with smtplib.SMTP(smtp_server, smtp_port) as smtp: logger.debug("SMTP connection established") try: smtp.starttls() logger.debug("STARTTLS successful") except Exception as e: logger.error(f"STARTTLS failed: {str(e)}") smtp.login(smtp_username, smtp_password) smtp.sendmail(smtp_username, to_email, msg.as_string()) logger.debug("Email sent via SMTP") logger.info("PDF email sent successfully") return jsonify({"message": "Email sent successfully"}), 200 except Exception as e: logger.error(f"Error sending PDF email: {str(e)}") return jsonify({"error": str(e)}), 500 @app.route('/health', methods=['GET']) def health_check(): logger = logging.getLogger('app:health-check') health_status = {'status': 'ok'} # Check SMTP configuration smtp_server = os.environ.get('SMTP_RELAY_HOST') smtp_port = os.environ.get('SMTP_RELAY_PORT') smtp_username = os.environ.get('SMTP_RELAY_USERNAME') if not all([smtp_server, smtp_port, smtp_username]): logger.error("SMTP configuration incomplete") health_status['smtp_config'] = False health_status['status'] = 'error' else: health_status['smtp_config'] = True # Try to connect to SMTP server try: smtp_port_int = int(smtp_port) use_ssl = smtp_port_int == 465 if use_ssl: with smtplib.SMTP_SSL(smtp_server, smtp_port_int, timeout=5) as smtp: smtp.ehlo() health_status['smtp_responding'] = True else: with smtplib.SMTP(smtp_server, smtp_port_int, timeout=5) as smtp: smtp.ehlo() health_status['smtp_responding'] = True except Exception as e: logger.error(f"SMTP connection test failed: {str(e)}") health_status['smtp_responding'] = False health_status['status'] = 'error' logger.debug(f"Health check performed: {health_status}") return jsonify(health_status) if __name__ == '__main__': app.run(host='0.0.0.0', port=2222, debug=False)