""" Token service for access token generation and validation. Implements opaque token strategy per ADR-004: - Tokens are cryptographically random strings - Tokens are stored as SHA-256 hashes in database - Tokens contain no user information (opaque) - Tokens are validated via database lookup """ import hashlib import logging import secrets from datetime import datetime, timedelta from typing import Optional from sqlalchemy import text from gondulf.database.connection import Database logger = logging.getLogger("gondulf.token_service") class TokenService: """ Service for access token generation and validation. Implements opaque token strategy per ADR-004: - Tokens are cryptographically random strings - Tokens are stored as SHA-256 hashes in database - Tokens contain no user information (opaque) - Tokens are validated via database lookup """ def __init__( self, database: Database, token_length: int = 32, # 32 bytes = 256 bits token_ttl: int = 3600 # 1 hour in seconds ): """ Initialize token service. Args: database: Database instance from Phase 1 token_length: Token length in bytes (default: 32 = 256 bits) token_ttl: Token time-to-live in seconds (default: 3600 = 1 hour) """ self.database = database self.token_length = token_length self.token_ttl = token_ttl logger.debug( f"TokenService initialized with token_length={token_length}, " f"token_ttl={token_ttl}s" ) def generate_token( self, me: str, client_id: str, scope: str = "" ) -> str: """ Generate opaque access token and store in database. Token generation: 1. Generate cryptographically secure random string (256 bits) 2. Hash token with SHA-256 for storage 3. Store hash + metadata in database 4. Return plaintext token to caller (only time it exists in plaintext) Args: me: User's domain URL (e.g., "https://example.com") client_id: Client application URL scope: Requested scopes (empty string for v1.0.0 authentication) Returns: Opaque access token (43-character base64url string) Raises: DatabaseError: If database operations fail """ # SECURITY: Generate cryptographically secure token (256 bits) token = secrets.token_urlsafe(self.token_length) # 32 bytes = 43-char base64url # SECURITY: Hash token for storage (prevent recovery from database) token_hash = hashlib.sha256(token.encode('utf-8')).hexdigest() # Calculate expiration timestamp issued_at = datetime.utcnow() expires_at = issued_at + timedelta(seconds=self.token_ttl) # Store token metadata in database engine = self.database.get_engine() with engine.begin() as conn: conn.execute( text(""" INSERT INTO tokens (token_hash, me, client_id, scope, issued_at, expires_at, revoked) VALUES (:token_hash, :me, :client_id, :scope, :issued_at, :expires_at, 0) """), { "token_hash": token_hash, "me": me, "client_id": client_id, "scope": scope, "issued_at": issued_at, "expires_at": expires_at } ) # PRIVACY: Log token generation without revealing full token logger.info( f"Token generated for {me} (client: {client_id}, " f"prefix: {token[:8]}..., expires: {expires_at.isoformat()})" ) return token # Return plaintext token (only time it exists in plaintext) def validate_token(self, provided_token: str) -> Optional[dict[str, str]]: """ Validate access token and return metadata. Validation steps: 1. Hash provided token with SHA-256 2. Lookup hash in database (constant-time comparison) 3. Check expiration (database timestamp vs current time) 4. Check revocation flag 5. Return metadata if valid, None if invalid Args: provided_token: Access token from Authorization header Returns: Token metadata dict if valid: {me, client_id, scope} None if invalid (not found, expired, or revoked) Raises: No exceptions raised - returns None for all error cases """ try: # SECURITY: Hash provided token for constant-time comparison token_hash = hashlib.sha256(provided_token.encode('utf-8')).hexdigest() # Lookup token in database engine = self.database.get_engine() with engine.connect() as conn: result = conn.execute( text(""" SELECT me, client_id, scope, expires_at, revoked FROM tokens WHERE token_hash = :token_hash """), {"token_hash": token_hash} ).fetchone() # Token not found if not result: logger.warning(f"Token validation failed: not found (prefix: {provided_token[:8]}...)") return None # Convert Row to dict token_data = dict(result._mapping) # Check expiration expires_at = token_data['expires_at'] if isinstance(expires_at, str): # SQLite returns timestamps as strings, parse them expires_at = datetime.fromisoformat(expires_at) if datetime.utcnow() > expires_at: logger.info( f"Token validation failed: expired " f"(me: {token_data['me']}, expired: {expires_at.isoformat()})" ) return None # Check revocation if token_data['revoked']: logger.warning( f"Token validation failed: revoked " f"(me: {token_data['me']}, client: {token_data['client_id']})" ) return None # Valid token - return metadata logger.debug(f"Token validated successfully (me: {token_data['me']})") return { 'me': token_data['me'], 'client_id': token_data['client_id'], 'scope': token_data['scope'] } except Exception as e: logger.error(f"Token validation error: {e}") return None def revoke_token(self, provided_token: str) -> bool: """ Revoke access token. Note: Not used in v1.0.0 (no revocation endpoint). Included for Phase 3 completeness and future use. Args: provided_token: Access token to revoke Returns: True if token revoked successfully False if token not found Raises: No exceptions raised """ try: # Hash token for lookup token_hash = hashlib.sha256(provided_token.encode('utf-8')).hexdigest() # Update revoked flag engine = self.database.get_engine() with engine.begin() as conn: result = conn.execute( text(""" UPDATE tokens SET revoked = 1 WHERE token_hash = :token_hash """), {"token_hash": token_hash} ) rows_affected = result.rowcount if rows_affected > 0: logger.info(f"Token revoked (prefix: {provided_token[:8]}...)") return True else: logger.warning(f"Token revocation failed: not found (prefix: {provided_token[:8]}...)") return False except Exception as e: logger.error(f"Token revocation error: {e}") return False def cleanup_expired_tokens(self) -> int: """ Delete expired tokens from database. Note: Can be called periodically (e.g., hourly) to prevent database growth. Not critical for v1.0.0 (small scale). Returns: Number of tokens deleted Raises: DatabaseError: If database operations fail """ current_time = datetime.utcnow() engine = self.database.get_engine() with engine.begin() as conn: result = conn.execute( text(""" DELETE FROM tokens WHERE expires_at < :current_time """), {"current_time": current_time} ) deleted_count = result.rowcount if deleted_count > 0: logger.info(f"Cleaned up {deleted_count} expired tokens") else: logger.debug("No expired tokens to clean up") return deleted_count