feat: extract mail from mp/mail — initial libreshop/mail

Source moved verbatim from mp/mail/ on 2026-04-29; mp was the first concrete adapter consuming the libreshop toolkit. Builds and publishes git.librete.ch/libreshop/mail on every main / v* push via the standard .gitea/workflows/build.yml shared across libreshop components.
2026-04-29 17:48:32 +02:00
commit c238d77562
11 changed files with 515 additions and 0 deletions
--- a/src/app.py
+++ b/src/app.py
@@ -0,0 +1,239 @@
+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"<h1>mail v1.0.0 - {today}</h1><pre>" + html.escape(str(url_map), False) + "</pre>"
+
+
+@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)
--- a/src/tests/test_app.py
+++ b/src/tests/test_app.py
@@ -0,0 +1,92 @@
+import unittest
+from flask import Flask
+from flask.testing import FlaskClient
+from unittest.mock import patch, MagicMock
+from app import app
+
+class TestApp(unittest.TestCase):
+    def setUp(self):
+        app.testing = True
+        self.app = app.test_client()
+
+    def test_info_endpoint(self):
+        response = self.app.get('/v1')
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"mail v1.0.0", response.data)
+
+    @patch('smtplib.SMTP')
+    def test_send_message_endpoint_success(self, mock_smtp):
+        mock_smtp_instance = MagicMock()
+        mock_smtp.return_value = mock_smtp_instance
+
+        with app.app_context():
+            response = self.app.post('/v1/send/message', json={
+                'subject': 'Test Subject',
+                'message': 'Test Message',
+                'to_email': 'test@example.com'
+            })
+
+        self.assertEqual(response.status_code, 200)
+        mock_smtp_instance.starttls.assert_called_once()
+        mock_smtp_instance.login.assert_called_once()
+        mock_smtp_instance.sendmail.assert_called_once()
+        mock_smtp_instance.quit.assert_called_once()
+
+    @patch('smtplib.SMTP')
+    def test_send_message_endpoint_failure(self, mock_smtp):
+        mock_smtp_instance = MagicMock()
+        mock_smtp_instance.login.side_effect = smtplib.SMTPAuthenticationError("Authentication failed")
+        mock_smtp.return_value = mock_smtp_instance
+
+        with app.app_context():
+            response = self.app.post('/v1/send/message', json={
+                'subject': 'Test Subject',
+                'message': 'Test Message',
+                'to_email': 'test@example.com'
+            })
+
+        self.assertEqual(response.status_code, 500)
+        mock_smtp_instance.starttls.assert_called_once()
+        mock_smtp_instance.login.assert_called_once()
+        mock_smtp_instance.sendmail.assert_not_called()
+        mock_smtp_instance.quit.assert_called_once()
+
+    @patch('smtplib.SMTP')
+    def test_send_pdf_endpoint_success(self, mock_smtp):
+        mock_smtp_instance = MagicMock()
+        mock_smtp.return_value = mock_smtp_instance
+
+        with app.app_context():
+            response = self.app.post('/v1/send/pdf', data={
+                'subject': 'Test Subject',
+                'message': 'Test Message',
+                'to_email': 'test@example.com',
+            }, content_type='multipart/form-data', files={'pdf_blob': ('document.pdf', b'PDF_CONTENT')})
+
+        self.assertEqual(response.status_code, 200)
+        mock_smtp_instance.starttls.assert_called_once()
+        mock_smtp_instance.login.assert_called_once()
+        mock_smtp_instance.sendmail.assert_called_once()
+        mock_smtp_instance.quit.assert_called_once()
+
+    @patch('smtplib.SMTP')
+    def test_send_pdf_endpoint_failure(self, mock_smtp):
+        mock_smtp_instance = MagicMock()
+        mock_smtp_instance.login.side_effect = smtplib.SMTPAuthenticationError("Authentication failed")
+        mock_smtp.return_value = mock_smtp_instance
+
+        with app.app_context():
+            response = self.app.post('/v1/send/pdf', data={
+                'subject': 'Test Subject',
+                'message': 'Test Message',
+                'to_email': 'test@example.com',
+            }, content_type='multipart/form-data', files={'pdf_blob': ('document.pdf', b'PDF_CONTENT')})
+
+        self.assertEqual(response.status_code, 500)
+        mock_smtp_instance.starttls.assert_called_once()
+        mock_smtp_instance.login.assert_called_once()
+        mock_smtp_instance.sendmail.assert_not_called()
+        mock_smtp_instance.quit.assert_called_once()
+
+if __name__ == '__main__':
+    unittest.main()
--- a/src/tests/test_starttls.py
+++ b/src/tests/test_starttls.py
@@ -0,0 +1,44 @@
+import smtplib
+import os
+
+# Get SMTP settings from environment variables
+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')
+
+print(f"Testing connection to {smtp_server}:{smtp_port}")
+
+# Create connection
+server = smtplib.SMTP(smtp_server, smtp_port)
+
+# Enable debug output
+server.set_debuglevel(1)
+
+# Test EHLO and STARTTLS
+try:
+	print("Sending EHLO...")
+	server.ehlo()
+	print("EHLO response received. Checking for STARTTLS support...")
+
+	server.starttls()
+	print("STARTTLS successful")
+
+	print("Attempting login...")
+	server.login(smtp_username, smtp_password)
+	print("Login successful")
+
+	server.quit()
+	print("Connection test completed successfully")
+except Exception as e:
+	print(f"Error: {e}")
+	print("Testing SSL connection instead...")
+
+	try:
+		ssl_server = smtplib.SMTP_SSL(smtp_server, smtp_port)
+		ssl_server.set_debuglevel(1)
+		ssl_server.login(smtp_username, smtp_password)
+		print("SSL connection successful")
+		ssl_server.quit()
+	except Exception as ssl_e:
+		print(f"SSL Error: {ssl_e}")