# ============================================================================== # Formeleon - Multi-tenant Form Generator # Author: Björn Runden # Module: routes/auth.py # Description: Authentication routes for Zitadel OIDC integration. # Provides login, logout, and callback handlers with dev mode bypass. # ============================================================================== """ Authentication routes for Formeleon. This module provides: - OIDC authentication via Zitadel (production) - Development mode bypass with fake user injection - Login/logout routes - Session management Routes: GET /auth/login - Initiate OIDC login flow or redirect to admin GET /auth/callback - OIDC callback handler GET /auth/logout - Logout and clear session GET /auth/status - Check current authentication status In development mode (DEV_MODE=true), the login route automatically creates a fake authenticated session without redirecting to Zitadel. """ import uuid import logging from typing import Optional from urllib.parse import urlencode from flask import Blueprint, request, redirect, url_for, session, current_app, jsonify from flask import render_template_string from models import db, User, UserRole, Tenant, AuditLog from services.auth_service import AuthService logger = logging.getLogger(__name__) # Create auth blueprint auth_bp = Blueprint('auth', __name__, template_folder='../templates') @auth_bp.before_request def check_dev_mode_auth(): """ Middleware to handle development mode and mock authentication. If DEV_MODE is enabled and user is not authenticated: - If MOCK_AUTH is enabled: create mock user session - Otherwise: create dev user session """ # Skip for auth routes themselves if request.path in ['/auth/login', '/auth/callback', '/auth/logout', '/auth/status']: return # Skip if already authenticated if 'user_id' in session and session.get('authenticated'): return # In dev mode, auto-authenticate if AuthService.is_dev_mode(): if AuthService.is_mock_auth_enabled(): # Use mock authentication mock_user = AuthService.create_mock_user_session() session['user_id'] = mock_user['id'] session['user'] = mock_user session['authenticated'] = True session.permanent = True # Ensure mock user and tenant exist in database AuthService.ensure_mock_user_exists() AuthService.ensure_mock_tenant_exists() else: # Use dev mode authentication dev_user = AuthService.create_dev_user_session() session['user_id'] = dev_user['id'] session['user'] = dev_user session['authenticated'] = True session.permanent = True # Ensure user and tenant exist in database AuthService.ensure_dev_user_exists() AuthService.ensure_dev_tenant_exists() @auth_bp.route('/login') def login(): """ Login route - initiates OIDC flow or handles dev/mock mode. In production: Redirects to Zitadel for OIDC authentication In development with MOCK_AUTH: Creates mock session and redirects to admin In development without MOCK_AUTH: Creates fake session and redirects to admin Query parameters: next: URL to redirect to after successful login """ next_url = request.args.get('next', url_for('admin.dashboard')) # Mock authentication mode: create mock session and redirect if AuthService.is_mock_auth_enabled(): mock_user = AuthService.create_mock_user_session() session.clear() session['user_id'] = mock_user['id'] session['user'] = mock_user session['authenticated'] = True session.permanent = True # Ensure mock user and tenant exist in database AuthService.ensure_mock_user_exists() AuthService.ensure_mock_tenant_exists() logger.info(f'Mock auth login: {mock_user["email"]}') AuditLog.log_action( action='login', resource_type='auth', user=User.query.get(mock_user['id']), tenant=Tenant.query.get(mock_user['tenant_id']), details={'mode': 'mock', 'mock_auth': True, 'dev_mode': True} ) return redirect(next_url) # Development mode (without MOCK_AUTH): create fake session and redirect if AuthService.is_dev_mode(): dev_user = AuthService.create_dev_user_session() session.clear() session['user_id'] = dev_user['id'] session['user'] = dev_user session['authenticated'] = True session.permanent = True # Ensure user and tenant exist in database AuthService.ensure_dev_user_exists() AuthService.ensure_dev_tenant_exists() logger.info(f'Dev mode login: {dev_user["email"]}') AuditLog.log_action( action='login', resource_type='auth', user=User.query.get(dev_user['id']), tenant=Tenant.query.get(dev_user['tenant_id']), details={'mode': 'dev', 'dev_mode': True} ) return redirect(next_url) # Production mode: redirect to Zitadel OIDC if not current_app.config['ZITADEL_ISSUER']: return jsonify({'error': 'OIDC not configured'}), 500 # Build authorization URL from authlib.integrations.flask_client import OAuth from flask import current_app oauth = OAuth(current_app) # Configure Zitadel if not already configured if 'zitadel' not in oauth._clients: issuer = current_app.config.get('ZITADEL_ISSUER', '').rstrip('/') client_secret = current_app.config.get('ZITADEL_CLIENT_SECRET') # Scopes aus Config lesen oder auf OIDC-Standard-Defaults zurückfallen scopes = current_app.config.get('ZITADEL_SCOPES') if not scopes: scopes = ['openid', 'profile', 'email'] scope_string = ' '.join(scopes) if isinstance(scopes, list) else scopes zitadel_config = { 'client_id': current_app.config.get('ZITADEL_CLIENT_ID'), 'client_secret': client_secret, # Automatische Konfiguration über Zitadels Discovery Endpoint 'server_metadata_url': f"{issuer}/.well-known/openid-configuration", 'client_kwargs': { 'scope': scope_string, 'code_challenge_method': 'S256', # PKCE Aktivierung # Falls KEIN Client Secret vorhanden ist (Pure PKCE Public App): # 'token_endpoint_auth_method': 'none' if not client_secret else 'client_secret_post' } } # Nur hinzufügen, wenn kein Secret da ist if not client_secret: zitadel_config['client_kwargs']['token_endpoint_auth_method'] = 'none' oauth.register('zitadel', **zitadel_config) # Generate state for CSRF protection state = str(uuid.uuid4()) session['oauth_state'] = state session['oauth_next'] = next_url # Build authorization URL redirect_uri = current_app.config.get('ZITADEL_REDIRECT_URI', url_for('auth.callback', _external=True)) authorization_url = f"{current_app.config.get('ZITADEL_ISSUER')}/v2/oauth/authorize" params = { 'client_id': current_app.config.get('ZITADEL_CLIENT_ID'), 'redirect_uri': redirect_uri, 'response_type': 'code', 'scope': ' '.join(current_app.config.get('ZITADEL_SCOPES', [])), 'state': state, 'nonce': str(uuid.uuid4()) } auth_url = f"{authorization_url}?{urlencode(params)}" logger.debug(f'Redirecting to OIDC provider: {auth_url}') return redirect(auth_url) @auth_bp.route('/callback') def callback(): """ OIDC callback handler. Handles the callback from Zitadel after successful authentication. Exchanges authorization code for tokens and retrieves user info. In mock auth mode, this route shouldn't be reached, but redirects gracefully. """ if AuthService.is_mock_auth_enabled(): # In mock auth mode, shouldn't reach here, but handle gracefully logger.debug('Mock auth mode: callback route accessed, redirecting to dashboard') return redirect(url_for('admin.dashboard')) if AuthService.is_dev_mode(): # In dev mode, shouldn't reach here, but handle gracefully return redirect(url_for('admin.dashboard')) # Verify state for CSRF protection state = request.args.get('state') stored_state = session.pop('oauth_state', None) if not state or state != stored_state: logger.error('OIDC callback: Invalid state parameter') return jsonify({'error': 'Invalid state parameter'}), 403 # Get authorization code code = request.args.get('code') if not code: error = request.args.get('error', 'No authorization code received') logger.error(f'OIDC callback error: {error}') return jsonify({'error': error}), 400 # Exchange code for tokens try: from authlib.integrations.flask_client import OAuth oauth = OAuth(current_app) redirect_uri = current_app.config.get('ZITADEL_REDIRECT_URI', url_for('auth.callback', _external=True)) zitadel = oauth.create_client('zitadel') # Get access token token = zitadel.authorize_access_token( redirect_uri=redirect_uri, code=code ) # Get user info user_info = zitadel.userinfo(token=token) logger.debug(f'OIDC userinfo: {user_info}') # Create or update user user, error = AuthService.process_oidc_callback(user_info) if error: return jsonify({'error': error}), 400 # Store in session session.clear() session['user_id'] = user.id session['user'] = { 'id': user.id, 'email': user.email, 'name': user.name, 'role': user.role.value, 'sub': user.sub, 'authenticated': True, 'dev_mode': False } session['authenticated'] = True session['access_token'] = token['access_token'] session['refresh_token'] = token.get('refresh_token') session['token_expires'] = token.get('expires_at') session.permanent = True # Log the login AuditLog.log_action( action='login', resource_type='auth', user=user, details={'mode': 'oidc', 'provider': 'zitadel'} ) # Redirect to next URL or dashboard next_url = session.pop('oauth_next', url_for('admin.dashboard')) return redirect(next_url) except Exception as e: logger.error(f'OIDC callback failed: {e}') return jsonify({'error': 'Authentication failed'}), 500 @auth_bp.route('/logout') def logout(): """ Logout route - clears session and redirects to login. In production: Also revokes tokens with Zitadel if configured In development/mock: Simply clears the session """ # Get current user for audit logging current_user = None if 'user_id' in session: current_user = User.query.get(session['user_id']) # Determine auth mode for logging auth_mode = 'mock' if AuthService.is_mock_auth_enabled() else ('dev' if AuthService.is_dev_mode() else 'oidc') # Clear session session.clear() # In production, revoke token with Zitadel if not AuthService.is_dev_mode() and 'access_token' in session: try: from authlib.integrations.flask_client import OAuth oauth = OAuth(current_app) zitadel = oauth.create_client('zitadel') # Revoke token zitadel.revoke_token( token=session['access_token'], token_type_hint='access_token' ) except Exception as e: logger.warning(f'Failed to revoke token: {e}') # Log the logout if current_user: AuditLog.log_action( action='logout', resource_type='auth', user=current_user, details={'mode': auth_mode} ) return redirect(url_for('auth.login')) @auth_bp.route('/status') def status(): """ Check current authentication status. :return: JSON with current authentication status """ if 'user_id' in session and session.get('authenticated'): user_data = session.get('user', {}) return jsonify({ 'authenticated': True, 'user': user_data, 'dev_mode': AuthService.is_dev_mode(), 'mock_auth': AuthService.is_mock_auth_enabled() }) return jsonify({ 'authenticated': False, 'dev_mode': AuthService.is_dev_mode(), 'mock_auth': AuthService.is_mock_auth_enabled() }) @auth_bp.route('/userinfo') def userinfo(): """ Get detailed user information. Requires authentication. :return: JSON with user information """ if not ('user_id' in session and session.get('authenticated')): return jsonify({'error': 'Unauthorized'}), 401 user_id = session['user_id'] user = User.query.get(user_id) if not user: return jsonify({'error': 'User not found'}), 404 return jsonify({ 'user': user.to_dict(), 'tenants': [tenant.to_dict() for tenant in user.tenants], 'dev_mode': AuthService.is_dev_mode(), 'mock_auth': AuthService.is_mock_auth_enabled() })