131 lines
5.1 KiB
Python
131 lines
5.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 uuid import uuid4
|
|
|
|
from backend.app.core.config import settings
|
|
from backend.app.database import get_connection
|
|
|
|
|
|
def hash_token(token: str) -> str:
|
|
return sha256(token.encode('utf-8')).hexdigest()
|
|
|
|
|
|
def _token_expiry() -> datetime:
|
|
return datetime.now(timezone.utc) + timedelta(minutes=settings.token_expiry_minutes)
|
|
|
|
|
|
def _persist_token(conn, user_id: str, token: str, token_type: str, expires_at: datetime,
|
|
device_id: str, family_id: str) -> None:
|
|
conn.execute(
|
|
'''
|
|
INSERT INTO tokens
|
|
(id, user_id, token_hash, token_type, expires_at, created_at, revoked, device_id, token_family_id)
|
|
VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP, 0, ?, ?)
|
|
''',
|
|
(str(uuid4()), user_id, hash_token(token), token_type, expires_at.isoformat(), device_id, family_id),
|
|
)
|
|
|
|
|
|
def issue_token(user_id: str, username: str, device_id: str | None = None) -> dict:
|
|
device_id = device_id.strip() if device_id and device_id.strip() else f'device-{uuid4().hex}'
|
|
family_id = str(uuid4())
|
|
access_token = f'token-{username}-{uuid4().hex}'
|
|
refresh_token = f'refresh-{username}-{uuid4().hex}'
|
|
access_expires_at = _token_expiry()
|
|
refresh_expires_at = datetime.now(timezone.utc) + timedelta(days=settings.refresh_token_expiry_days)
|
|
|
|
with get_connection() as conn:
|
|
_persist_token(conn, user_id, access_token, 'access', access_expires_at, device_id, family_id)
|
|
_persist_token(conn, user_id, refresh_token, 'refresh', refresh_expires_at, device_id, family_id)
|
|
conn.commit()
|
|
|
|
return {
|
|
'access_token': access_token,
|
|
'token_type': 'bearer',
|
|
'expires_at': access_expires_at.isoformat(),
|
|
'refresh_token': refresh_token,
|
|
'device_id': device_id,
|
|
'token_family_id': family_id,
|
|
}
|
|
|
|
|
|
def validate_token(token: str) -> dict | None:
|
|
token_hash = hash_token(token)
|
|
with get_connection() as conn:
|
|
row = conn.execute(
|
|
'''
|
|
SELECT * FROM tokens
|
|
WHERE token_hash = ? AND token_type = 'access' AND revoked = 0 AND expires_at > ?
|
|
''',
|
|
(token_hash, datetime.now(timezone.utc).isoformat()),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
return dict(row)
|
|
|
|
|
|
def validate_refresh_token(token: str, device_id: str | None = None) -> dict | None:
|
|
with get_connection() as conn:
|
|
row = conn.execute(
|
|
'''
|
|
SELECT * FROM tokens
|
|
WHERE token_hash = ? AND token_type = 'refresh' AND revoked = 0 AND expires_at > ?
|
|
AND (? IS NULL OR device_id = ?)
|
|
''',
|
|
(hash_token(token), datetime.now(timezone.utc).isoformat(), device_id, device_id),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def rotate_refresh_token(refresh_token: str, device_id: str | None = None) -> dict | None:
|
|
current = validate_refresh_token(refresh_token, device_id)
|
|
with get_connection() as conn:
|
|
if current is None:
|
|
row = conn.execute(
|
|
'SELECT token_family_id FROM tokens WHERE token_hash = ? AND token_type = ? AND token_family_id IS NOT NULL',
|
|
(hash_token(refresh_token), 'refresh'),
|
|
).fetchone()
|
|
if row:
|
|
conn.execute('UPDATE tokens SET revoked = 1 WHERE token_family_id = ?', (row['token_family_id'],))
|
|
conn.commit()
|
|
return None
|
|
|
|
user = conn.execute('SELECT username FROM users WHERE id = ?', (current['user_id'],)).fetchone()
|
|
if user is None:
|
|
return None
|
|
family_id = current['token_family_id']
|
|
conn.execute('UPDATE tokens SET revoked = 1 WHERE token_family_id = ?', (family_id,))
|
|
new_access = f'token-{user["username"]}-{uuid4().hex}'
|
|
new_refresh = f'refresh-{user["username"]}-{uuid4().hex}'
|
|
access_expires_at = _token_expiry()
|
|
refresh_expires_at = datetime.now(timezone.utc) + timedelta(days=settings.refresh_token_expiry_days)
|
|
_persist_token(conn, current['user_id'], new_access, 'access', access_expires_at, current['device_id'], family_id)
|
|
_persist_token(conn, current['user_id'], new_refresh, 'refresh', refresh_expires_at, current['device_id'], family_id)
|
|
conn.commit()
|
|
return {
|
|
'access_token': new_access,
|
|
'token_type': 'bearer',
|
|
'expires_at': access_expires_at.isoformat(),
|
|
'refresh_token': new_refresh,
|
|
'device_id': current['device_id'],
|
|
'user_id': current['user_id'],
|
|
'username': user['username'],
|
|
}
|
|
|
|
|
|
def revoke_token(token: str) -> bool:
|
|
token_hash = hash_token(token)
|
|
with get_connection() as conn:
|
|
cursor = conn.execute(
|
|
'''UPDATE tokens SET revoked = 1
|
|
WHERE token_hash = ? OR token_family_id = (
|
|
SELECT token_family_id FROM tokens WHERE token_hash = ?
|
|
)''',
|
|
(token_hash, token_hash),
|
|
)
|
|
conn.commit()
|
|
return cursor.rowcount > 0
|