## Copyright © 2026 Olaf Kolkman
## SPDX-License-Identifier: GPL-3.0-or-later
from email.message import EmailMessage
from html import escape
from pathlib import Path
from smtplib import SMTP
import json
from urllib.parse import urlparse
from backend.app.core.config import settings
from backend.app.database import get_connection
from backend.app.services.secret_store import decrypt_secret, encrypt_secret
LOGO_PATH = Path(__file__).resolve().parents[3] / 'frontend' / 'static' / 'logo.svg'
def _html_email(body_html: str) -> str:
public_url = settings.public_url
parsed_url = urlparse(public_url)
public_hostname = parsed_url.hostname or public_url
safe_public_url = escape(public_url, quote=True)
safe_public_hostname = escape(public_hostname)
return f'''
'''
def _add_html_body(message: EmailMessage, html_body: str) -> None:
message.add_alternative(_html_email(html_body), subtype='html')
html_part = message.get_payload()[-1]
try:
logo = LOGO_PATH.read_bytes()
except OSError:
return
html_part.add_related(logo, maintype='image', subtype='svg+xml', cid='')
def get_smtp_settings() -> dict:
values = {
'smtp_host': settings.smtp_host,
'smtp_port': settings.smtp_port,
'smtp_username': settings.smtp_username,
'smtp_password': settings.smtp_password,
'smtp_from': settings.smtp_from,
'smtp_use_tls': settings.smtp_use_tls,
}
with get_connection() as conn:
row = conn.execute('SELECT value FROM app_settings WHERE name = ?', ('smtp',)).fetchone()
if row:
values.update(json.loads(row['value']))
values['smtp_password'] = decrypt_secret(values['smtp_password'])
return values
def save_smtp_settings(values: dict) -> None:
with get_connection() as conn:
conn.execute(
'''INSERT INTO app_settings (name, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)
ON CONFLICT(name) DO UPDATE SET value = excluded.value, updated_at = CURRENT_TIMESTAMP''',
('smtp', json.dumps({**values, 'smtp_password': encrypt_secret(values['smtp_password'])})),
)
conn.commit()
def smtp_configured(smtp_values: dict | None = None) -> bool:
smtp = smtp_values or get_smtp_settings()
return bool(smtp['smtp_host'] and smtp['smtp_from'])
def send_message(email: str, subject: str, body: str, html_body: str | None = None,
smtp_values: dict | None = None) -> None:
smtp = smtp_values or get_smtp_settings()
if not smtp_configured(smtp):
raise RuntimeError('SMTP is not configured; set LINKLOG_SMTP_HOST and LINKLOG_SMTP_FROM')
message = EmailMessage()
message['Subject'] = subject
message['From'] = smtp['smtp_from']
message['To'] = email
message.set_content(body)
if html_body:
_add_html_body(message, html_body)
with SMTP(smtp['smtp_host'], smtp['smtp_port'], timeout=10) as connection:
if smtp['smtp_use_tls']:
connection.starttls()
if smtp['smtp_username']:
connection.login(smtp['smtp_username'], smtp['smtp_password'])
connection.send_message(message)
def send_verification_email(email: str, username: str, verification_url: str) -> None:
safe_username = escape(username)
safe_url = escape(verification_url, quote=True)
send_message(
email,
'Verify your LinkLog email address',
f'Hello {username},\n\n'
f'Verify your LinkLog email address by opening this link:\n{verification_url}\n\n'
f'This link expires in {settings.email_verification_expiry_hours} hours.\n',
f'Hello {safe_username},
Verify your LinkLog email address:
'
f'Verify email address
'
f'This link expires in {settings.email_verification_expiry_hours} hours.
',
)
def send_test_email(email: str, smtp_values: dict | None = None) -> None:
send_message(
email,
'LinkLog SMTP test',
'This is a test message from LinkLog. SMTP is configured correctly.\n',
'This is a test message from LinkLog.
SMTP is configured correctly.
',
smtp_values=smtp_values,
)
def send_password_reset_email(email: str, username: str, reset_url: str) -> None:
safe_username = escape(username)
safe_url = escape(reset_url, quote=True)
send_message(
email,
'Reset your LinkLog password',
f'Hello {username},\n\n'
f'Reset your LinkLog password by opening this link:\n{reset_url}\n\n'
f'This link expires in {settings.password_reset_expiry_hours} hours.\n',
f'Hello {safe_username},
Reset your LinkLog password:
'
f'Reset password
'
f'This link expires in {settings.password_reset_expiry_hours} hours.
',
)