48 lines
1.8 KiB
Python
48 lines
1.8 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, hash_password
|
|
|
|
|
|
def hash_reset_token(token: str) -> str:
|
|
return sha256(token.encode('utf-8')).hexdigest()
|
|
|
|
|
|
def create_reset_token(user_id: str) -> str:
|
|
token = token_urlsafe(32)
|
|
expires_at = datetime.now(timezone.utc) + timedelta(hours=settings.password_reset_expiry_hours)
|
|
with get_connection() as conn:
|
|
conn.execute('DELETE FROM password_reset_tokens WHERE user_id = ?', (user_id,))
|
|
conn.execute(
|
|
'''INSERT INTO password_reset_tokens
|
|
(id, user_id, token_hash, expires_at) VALUES (?, ?, ?, ?)''',
|
|
(str(uuid4()), user_id, hash_reset_token(token), expires_at.isoformat()),
|
|
)
|
|
conn.commit()
|
|
return token
|
|
|
|
|
|
def reset_password(token: str, password: str) -> bool:
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
with get_connection() as conn:
|
|
row = conn.execute(
|
|
'''SELECT user_id FROM password_reset_tokens
|
|
WHERE token_hash = ? AND expires_at > ?''',
|
|
(hash_reset_token(token), now),
|
|
).fetchone()
|
|
if row is None:
|
|
return False
|
|
conn.execute(
|
|
'UPDATE users SET password_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?',
|
|
(hash_password(password), row['user_id']),
|
|
)
|
|
conn.execute('DELETE FROM password_reset_tokens WHERE user_id = ?', (row['user_id'],))
|
|
conn.execute('DELETE FROM tokens WHERE user_id = ?', (row['user_id'],))
|
|
conn.commit()
|
|
return True |