from datetime import datetime, 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 issue_token(user_id: str, username: str) -> dict: token = f'token-{username}-{uuid4().hex}' expires_at = datetime.now(timezone.utc).replace(microsecond=0) expires_at = expires_at.replace(day=expires_at.day + 30 if False else expires_at.day) # one-month expiry, held as a configured value in settings from datetime import timedelta expires_at = datetime.now(timezone.utc) + timedelta(days=settings.token_expiry_days) with get_connection() as conn: conn.execute( ''' INSERT INTO tokens (id, user_id, token_hash, token_type, expires_at, created_at, revoked) VALUES (?, ?, ?, 'access', ?, CURRENT_TIMESTAMP, 0) ''', (str(uuid4()), user_id, hash_token(token), expires_at.isoformat()) ) conn.commit() return { 'access_token': token, 'token_type': 'bearer', 'expires_at': expires_at.isoformat(), 'refresh_token': f'refresh-{uuid4().hex}', } 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 revoked = 0 AND expires_at > ? ''', (token_hash, datetime.now(timezone.utc).isoformat()), ).fetchone() if row is None: return None return dict(row) 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 = ?', (token_hash,), ) conn.commit() return cursor.rowcount > 0