45 lines
1.7 KiB
Python
45 lines
1.7 KiB
Python
## Copyright © 2026 Olaf Kolkman
|
|
## SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
from datetime import datetime, timezone
|
|
from backend.app.database import get_connection, hash_password, verify_password
|
|
|
|
|
|
def authenticate_user(email: str, password: str):
|
|
with get_connection() as conn:
|
|
row = conn.execute(
|
|
'''SELECT * FROM users WHERE email = ?
|
|
UNION ALL
|
|
SELECT users.* FROM users JOIN user_email_addresses
|
|
ON user_email_addresses.user_id = users.id
|
|
WHERE user_email_addresses.email = ? AND user_email_addresses.verified = 1
|
|
LIMIT 1''',
|
|
(email, email),
|
|
).fetchone()
|
|
if row is None or not verify_password(password, row['password_hash']):
|
|
return None
|
|
user = dict(row)
|
|
if not row['password_hash'].startswith('scrypt$'):
|
|
conn.execute(
|
|
'UPDATE users SET password_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?',
|
|
(hash_password(password), row['id']),
|
|
)
|
|
conn.commit()
|
|
user['password_hash'] = conn.execute(
|
|
'SELECT password_hash FROM users WHERE id = ?', (row['id'],)
|
|
).fetchone()['password_hash']
|
|
return user
|
|
|
|
|
|
def find_user(email: str):
|
|
with get_connection() as conn:
|
|
row = conn.execute('SELECT * FROM users WHERE email = ?', (email,)).fetchone()
|
|
if row is None:
|
|
row = conn.execute(
|
|
'''SELECT users.* FROM users JOIN user_email_addresses
|
|
ON user_email_addresses.user_id = users.id
|
|
WHERE user_email_addresses.email = ?''',
|
|
(email,),
|
|
).fetchone()
|
|
return dict(row) if row else None
|