""" OIDC Service - Business Logic for OpenID Connect Flow Handles authorization, token exchange, and userinfo """ from typing import Optional, Dict, Any from datetime import datetime, timedelta from app.core.database import db from models import User, Client, AuthorizationCode, AccessToken, AuditLog import secrets import jwt from config import get_config import os class OIDCService: """ OIDC service - contains business logic for OpenID Connect flows. Following Python Quick Start Guide: - Service layer contains business rules - Orchestrates token generation and validation """ def __init__(self, db_session=None): """Initialize OIDC service.""" self.db = db_session or db.session # Load config env = os.environ.get('FLASK_ENV', 'development') config = get_config(env)() self.config = config def validate_authorization_request( self, client_id: str, redirect_uri: str, response_type: str, scope: str = '', state: str = '' ) -> Dict[str, Any]: """ Validate authorization request parameters. Business Rules: 1. Client must exist and be valid 2. Response type must be 'code' 3. Redirect URI must be in client's allowed list Args: client_id: OIDC client ID redirect_uri: Redirect URI from request response_type: OAuth response type scope: Requested scopes state: Optional state parameter Returns: Dict with 'success' (bool) and 'auth_request' data or 'error' """ # Business Rule 1: Validate client client = Client.query.filter_by(client_id=client_id).first() if not client: return {'success': False, 'error': 'Invalid client_id'} # Business Rule 2: Check response type if response_type != 'code': return {'success': False, 'error': "Unsupported response_type. Use 'code'"} # Business Rule 3: Validate redirect_uri allowed_uris = client.get_redirect_uris() if not redirect_uri or redirect_uri not in allowed_uris: return {'success': False, 'error': 'Invalid or missing redirect_uri'} return { 'success': True, 'auth_request': { 'client_id': client_id, 'redirect_uri': redirect_uri, 'scope': scope, 'state': state } } def authorize_with_credentials( self, username: str, password: str, client_id: str, redirect_uri: str, scope: str, state: Optional[str] = None ) -> Dict[str, Any]: """ Authenticate user and create authorization code. Business Rules: 1. User must exist and be active 2. Password must be correct 3. Create authorization code for valid user 4. Build redirect URL with code Args: username: User's username password: User's password client_id: OIDC client ID redirect_uri: Redirect URI scope: Requested scopes state: Optional state parameter Returns: Dict with 'success', 'redirect_url' or 'error' """ # Business Rule 1 & 2: Authenticate user user = User.query.filter_by(username=username, is_active=True).first() if not user or not user.check_password(password): return {'success': False, 'error': 'Invalid credentials'} # Business Rule 3: Create authorization code result = self.create_authorization_code( client_id=client_id, user_id=user.id, redirect_uri=redirect_uri, scope=scope, state=state ) if not result['success']: return result # Business Rule 4: Build redirect URL separator = '&' if '?' in redirect_uri else '?' redirect_url = f"{redirect_uri}{separator}code={result['code']}" if state: redirect_url += f"&state={state}" return { 'success': True, 'redirect_url': redirect_url } def create_authorization_code( self, client_id: str, user_id: int, redirect_uri: str, scope: str, state: Optional[str] = None ) -> Dict[str, Any]: """ Create an authorization code for OIDC flow. Business Rules: 1. Client must exist and be valid 2. Redirect URI must be in client's allowed list 3. User must exist 4. Code expires after configured lifetime Args: client_id: OIDC client ID user_id: Authenticated user ID redirect_uri: Redirect URI from request scope: Requested scopes state: Optional state parameter Returns: Dict with 'success', 'code', 'redirect_uri', 'state' """ # Business Rule 1: Validate client client = Client.query.filter_by(client_id=client_id).first() if not client: return {'success': False, 'error': 'invalid_client'} # Business Rule 2: Validate redirect_uri allowed_uris = client.get_redirect_uris() if redirect_uri not in allowed_uris: return {'success': False, 'error': 'invalid_redirect_uri'} # Business Rule 3: Validate user user = User.query.get(user_id) if not user: return {'success': False, 'error': 'invalid_user'} # Create authorization code code = secrets.token_urlsafe(32) auth_code = AuthorizationCode( code=code, client_id=client_id, user_id=user_id, redirect_uri=redirect_uri, scope=scope, expires_at=datetime.utcnow() + timedelta(seconds=self.config.AUTHORIZATION_CODE_LIFETIME) ) try: self.db.add(auth_code) self.db.commit() return { 'success': True, 'code': code, 'redirect_uri': redirect_uri, 'state': state } except Exception as e: self.db.rollback() return {'success': False, 'error': str(e)} def exchange_code_for_token( self, grant_type: str, code: str, redirect_uri: str, client_id: str, client_secret: str ) -> Dict[str, Any]: """ Exchange authorization code for access token. Business Rules: 1. Grant type must be 'authorization_code' 2. Client must be authenticated 3. Code must be valid and not expired 4. Redirect URI must match 5. Code can only be used once Args: grant_type: OAuth grant type code: Authorization code redirect_uri: Redirect URI from initial request client_id: Client ID client_secret: Client secret Returns: Dict with token response or error """ # Business Rule 1: Check grant type if grant_type != 'authorization_code': return {'error': 'unsupported_grant_type'} # Business Rule 2: Authenticate client client = Client.query.filter_by(client_id=client_id).first() if not client or not client.check_client_secret(client_secret): return {'error': 'invalid_client'} # Business Rule 3: Validate code auth_code = AuthorizationCode.query.filter_by(code=code).first() if not auth_code or not auth_code.is_valid(): return {'error': 'invalid_grant'} # Business Rule 4: Check redirect URI if auth_code.redirect_uri != redirect_uri: return {'error': 'invalid_grant'} # Business Rule 5: Mark code as used auth_code.used = True # Get user user = User.query.get(auth_code.user_id) if not user: return {'error': 'invalid_grant'} # Generate tokens access_token = secrets.token_urlsafe(32) id_token = self._generate_id_token(user, client_id) # Store access token token_record = AccessToken( token=access_token, client_id=client_id, user_id=user.id, scope=auth_code.scope, expires_at=datetime.utcnow() + timedelta(seconds=self.config.ACCESS_TOKEN_LIFETIME) ) try: self.db.add(token_record) self.db.commit() return { 'access_token': access_token, 'token_type': 'Bearer', 'expires_in': self.config.ACCESS_TOKEN_LIFETIME, 'id_token': id_token, 'scope': auth_code.scope } except Exception as e: self.db.rollback() return {'error': str(e)} def get_userinfo(self, access_token: str) -> Dict[str, Any]: """ Get user information from access token. Business Rules: 1. Token must be valid 2. Token must not be expired or revoked Args: access_token: Bearer access token Returns: User information dict or error """ # Extract token from Bearer header if needed if access_token.startswith('Bearer '): access_token = access_token[7:] # Business Rule 1 & 2: Validate token token = AccessToken.query.filter_by(token=access_token).first() if not token or token.is_expired() or token.revoked: return {'error': 'invalid_token'} # Get user info user = token.user return user.to_dict() def _generate_id_token(self, user: User, client_id: str) -> str: """ Generate JWT ID token for user. Args: user: User object client_id: Client ID Returns: Signed JWT ID token """ now = datetime.utcnow() payload = { 'iss': self.config.OIDC_ISSUER, 'sub': str(user.id), 'aud': client_id, 'exp': now + timedelta(seconds=self.config.ID_TOKEN_LIFETIME), 'iat': now, 'name': user.name, 'email': user.email, 'preferred_username': user.preferred_username, 'role': user.role } # Sign with private key private_key = self.config.OIDC_JWT_PRIVATE_KEY return jwt.encode(payload, private_key, algorithm='RS256')