Added email functionality
Build LinkLog Development Image / development-image (push) Successful in 10s
Build LinkLog Development Image / development-image (push) Successful in 10s
This commit is contained in:
@@ -0,0 +1,33 @@
|
||||
## Copyright © 2026 Olaf Kolkman
|
||||
## SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
from email.message import EmailMessage
|
||||
from smtplib import SMTP
|
||||
|
||||
from backend.app.core.config import settings
|
||||
|
||||
|
||||
def smtp_configured() -> bool:
|
||||
return bool(settings.smtp_host and settings.smtp_from)
|
||||
|
||||
|
||||
def send_verification_email(email: str, username: str, verification_url: str) -> None:
|
||||
if not smtp_configured():
|
||||
raise RuntimeError('SMTP is not configured; set LINKLOG_SMTP_HOST and LINKLOG_SMTP_FROM')
|
||||
|
||||
message = EmailMessage()
|
||||
message['Subject'] = 'Verify your LinkLog email address'
|
||||
message['From'] = settings.smtp_from
|
||||
message['To'] = email
|
||||
message.set_content(
|
||||
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'
|
||||
)
|
||||
|
||||
with SMTP(settings.smtp_host, settings.smtp_port, timeout=10) as smtp:
|
||||
if settings.smtp_use_tls:
|
||||
smtp.starttls()
|
||||
if settings.smtp_username:
|
||||
smtp.login(settings.smtp_username, settings.smtp_password)
|
||||
smtp.send_message(message)
|
||||
@@ -0,0 +1,44 @@
|
||||
## Copyright © 2026 Olaf Kolkman
|
||||
## SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from hashlib import sha256
|
||||
from secrets import token_urlsafe
|
||||
from uuid import uuid4
|
||||
|
||||
from backend.app.core.config import settings
|
||||
from backend.app.database import get_connection
|
||||
|
||||
|
||||
def hash_verification_token(token: str) -> str:
|
||||
return sha256(token.encode('utf-8')).hexdigest()
|
||||
|
||||
|
||||
def create_verification_token(user_id: str) -> str:
|
||||
token = token_urlsafe(32)
|
||||
expires_at = datetime.now(timezone.utc) + timedelta(hours=settings.email_verification_expiry_hours)
|
||||
with get_connection() as conn:
|
||||
conn.execute('DELETE FROM email_verification_tokens WHERE user_id = ?', (user_id,))
|
||||
conn.execute(
|
||||
'''INSERT INTO email_verification_tokens
|
||||
(id, user_id, token_hash, expires_at) VALUES (?, ?, ?, ?)''',
|
||||
(str(uuid4()), user_id, hash_verification_token(token), expires_at.isoformat()),
|
||||
)
|
||||
conn.commit()
|
||||
return token
|
||||
|
||||
|
||||
def verify_email(token: str) -> bool:
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
with get_connection() as conn:
|
||||
row = conn.execute(
|
||||
'''SELECT user_id FROM email_verification_tokens
|
||||
WHERE token_hash = ? AND expires_at > ?''',
|
||||
(hash_verification_token(token), now),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
return False
|
||||
conn.execute('UPDATE users SET email_verified = 1, updated_at = CURRENT_TIMESTAMP WHERE id = ?', (row['user_id'],))
|
||||
conn.execute('DELETE FROM email_verification_tokens WHERE user_id = ?', (row['user_id'],))
|
||||
conn.commit()
|
||||
return True
|
||||
Reference in New Issue
Block a user