73 lines
3.1 KiB
Python
73 lines
3.1 KiB
Python
## 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
|
|
from backend.app.services.email_service import send_verification_email
|
|
|
|
|
|
def list_user_email_addresses(user_id: str) -> list[dict]:
|
|
with get_connection() as conn:
|
|
rows = conn.execute(
|
|
'SELECT id, email, verified, created_at FROM user_email_addresses WHERE user_id = ? ORDER BY created_at',
|
|
(user_id,),
|
|
).fetchall()
|
|
return [dict(row) | {'verified': bool(row['verified']), 'can_be_primary': bool(row['verified'])} for row in rows]
|
|
|
|
|
|
def add_user_email_address(user_id: str, email: str) -> dict:
|
|
email = email.strip().lower()
|
|
if not email:
|
|
raise ValueError('Email address is required')
|
|
with get_connection() as conn:
|
|
try:
|
|
row = conn.execute(
|
|
'INSERT INTO user_email_addresses (id, user_id, email) VALUES (?, ?, ?) RETURNING id, email, verified, created_at',
|
|
(str(uuid4()), user_id, email),
|
|
).fetchone()
|
|
conn.commit()
|
|
except Exception as error:
|
|
if 'UNIQUE constraint failed' in str(error):
|
|
raise ValueError('Email address already exists') from error
|
|
raise
|
|
return dict(row) | {'verified': bool(row['verified']), 'can_be_primary': bool(row['verified'])}
|
|
|
|
|
|
def create_email_verification(user_id: str, address_id: str) -> tuple[str, str]:
|
|
token = token_urlsafe(32)
|
|
expires_at = datetime.now(timezone.utc) + timedelta(hours=settings.email_verification_expiry_hours)
|
|
with get_connection() as conn:
|
|
address = conn.execute(
|
|
'SELECT email FROM user_email_addresses WHERE id = ? AND user_id = ?',
|
|
(address_id, user_id),
|
|
).fetchone()
|
|
if address is None:
|
|
raise ValueError('Email address not found')
|
|
conn.execute('DELETE FROM email_address_verification_tokens WHERE email_address_id = ?', (address_id,))
|
|
conn.execute(
|
|
'INSERT INTO email_address_verification_tokens (id, email_address_id, token_hash, expires_at) VALUES (?, ?, ?, ?)',
|
|
(str(uuid4()), address_id, sha256(token.encode()).hexdigest(), expires_at.isoformat()),
|
|
)
|
|
conn.commit()
|
|
return address['email'], token
|
|
|
|
|
|
def verify_user_email_address(token: str) -> bool:
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
with get_connection() as conn:
|
|
row = conn.execute(
|
|
'''SELECT email_address_id FROM email_address_verification_tokens
|
|
WHERE token_hash = ? AND expires_at > ?''',
|
|
(sha256(token.encode()).hexdigest(), now),
|
|
).fetchone()
|
|
if row is None:
|
|
return False
|
|
conn.execute('UPDATE user_email_addresses SET verified = 1, updated_at = CURRENT_TIMESTAMP WHERE id = ?', (row['email_address_id'],))
|
|
conn.execute('DELETE FROM email_address_verification_tokens WHERE email_address_id = ?', (row['email_address_id'],))
|
|
conn.commit()
|
|
return True |