44 lines
1.6 KiB
Python
44 lines
1.6 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
|
|
|
|
|
|
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 |