138 lines
4.8 KiB
Python
138 lines
4.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 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(days=settings.token_expiry_days)
|
|
|
|
|
|
def _persist_token(conn, user_id: str, token: str, token_type: str, expires_at: datetime, device_id: str | None, family_id: str | None) -> 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:
|
|
if device_id is None or not device_id.strip():
|
|
device_id = f'device-{uuid4().hex}'
|
|
device_id = device_id.strip()
|
|
token_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 = access_expires_at + timedelta(days=30)
|
|
|
|
with get_connection() as conn:
|
|
_persist_token(conn, user_id, access_token, 'access', access_expires_at, device_id, token_family_id)
|
|
_persist_token(conn, user_id, refresh_token, 'refresh', refresh_expires_at, device_id, token_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': token_family_id,
|
|
}
|
|
|
|
|
|
def validate_token(token: str) -> dict | None:
|
|
token_hash = hash_token(token)
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
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, now),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
return dict(row)
|
|
|
|
|
|
def validate_refresh_token(token: str, device_id: str | None = None) -> dict | None:
|
|
token_hash = hash_token(token)
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
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 = ?)
|
|
''',
|
|
(token_hash, now, device_id, device_id),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
return dict(row)
|
|
|
|
|
|
def rotate_refresh_token(refresh_token: str, device_id: str | None = None) -> dict | None:
|
|
current = validate_refresh_token(refresh_token, device_id)
|
|
if current is None:
|
|
return None
|
|
|
|
family_id = current.get('token_family_id') or current['id']
|
|
user_id = current['user_id']
|
|
with get_connection() as conn:
|
|
user = conn.execute('SELECT username FROM users WHERE id = ?', (user_id,)).fetchone()
|
|
if user is None:
|
|
return None
|
|
|
|
conn.execute(
|
|
'UPDATE tokens SET revoked = 1 WHERE token_family_id = ? AND token_type = ? AND revoked = 0',
|
|
(family_id, 'refresh'),
|
|
)
|
|
conn.execute(
|
|
'UPDATE tokens SET revoked = 1 WHERE id = ?',
|
|
(current['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 = access_expires_at + timedelta(days=30)
|
|
|
|
_persist_token(conn, user_id, new_access, 'access', access_expires_at, device_id, family_id)
|
|
_persist_token(conn, user_id, new_refresh, 'refresh', refresh_expires_at, 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': device_id,
|
|
'user_id': user_id,
|
|
'username': user['username'],
|
|
}
|
|
|
|
|
|
def revoke_token(token: str, token_type: str = 'access') -> bool:
|
|
token_hash = hash_token(token)
|
|
with get_connection() as conn:
|
|
cursor = conn.execute(
|
|
'UPDATE tokens SET revoked = 1 WHERE token_hash = ? AND token_type = ?',
|
|
(token_hash, token_type),
|
|
)
|
|
conn.commit()
|
|
return cursor.rowcount > 0
|