90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
## Copyright © 2026 Olaf Kolkman
|
|
## SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
from email.message import EmailMessage
|
|
from smtplib import SMTP
|
|
import json
|
|
|
|
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
|
|
|
|
|
|
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, 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)
|
|
|
|
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:
|
|
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',
|
|
)
|
|
|
|
|
|
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',
|
|
smtp_values,
|
|
)
|
|
|
|
|
|
def send_password_reset_email(email: str, username: str, reset_url: str) -> None:
|
|
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',
|
|
) |