# ============================================================================== # 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. """ import os import logging from typing import Optional from flask import Blueprint, request, redirect, url_for, session, current_app, jsonify from authlib.integrations.flask_client import OAuth from models import db, User, UserRole, Tenant, AuditLog from services.auth_service import AuthService logger = logging.getLogger(__name__) # Modul-globales OAuth Objekt oauth = OAuth() # Create auth blueprint auth_bp = Blueprint('auth', __name__, template_folder='../templates') os.environ['AUTHLIB_INSECURE_TRANSPORT'] = '1' def _init_zitadel_client(): """Hilfsfunktion: Registriert den Zitadel-Client sicher bei Authlib.""" # HTTP im Dev/Mock-Modus für Authlib erlauben (nur lokal!) if AuthService.is_dev_mode() or AuthService.is_mock_auth_enabled(): os.environ['AUTHLIB_INSECURE_TRANSPORT'] = '1' if 'zitadel' not in oauth._clients: issuer = current_app.config.get('ZITADEL_ISSUER', '').rstrip('/') client_secret = current_app.config.get('ZITADEL_CLIENT_SECRET') scopes = current_app.config.get('ZITADEL_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, 'server_metadata_url': f"{issuer}/.well-known/openid-configuration", 'client_kwargs': { 'scope': scope_string, } } # Pure PKCE (Public App) ohne Client Secret if not client_secret: zitadel_config['client_kwargs']['token_endpoint_auth_method'] = 'none' oauth.init_app(current_app) oauth.register('zitadel', **zitadel_config) @auth_bp.before_request def check_dev_mode_auth(): """Middleware to handle development mode and mock authentication.""" if request.path in ['/auth/login', '/auth/callback', '/auth/logout', '/auth/status']: return if 'user_id' in session and session.get('authenticated'): return if AuthService.is_dev_mode(): if AuthService.is_mock_auth_enabled(): mock_user = AuthService.create_mock_user_session() session['user_id'] = mock_user['id'] session['user'] = mock_user session['authenticated'] = True session.permanent = True AuthService.ensure_mock_user_exists() AuthService.ensure_mock_tenant_exists() else: dev_user = AuthService.create_dev_user_session() session['user_id'] = dev_user['id'] session['user'] = dev_user session['authenticated'] = True session.permanent = True AuthService.ensure_dev_user_exists() AuthService.ensure_dev_tenant_exists() @auth_bp.route('/login') def login(): """Initiates OIDC flow via Authlib or handles dev/mock mode.""" next_url = request.args.get('next', url_for('admin.dashboard')) # Mock authentication mode 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 AuthService.ensure_mock_user_exists() AuthService.ensure_mock_tenant_exists() 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) 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 AuthService.ensure_dev_user_exists() AuthService.ensure_dev_tenant_exists() 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 OIDC mode if not current_app.config.get('ZITADEL_ISSUER'): return jsonify({'error': 'OIDC not configured'}), 500 _init_zitadel_client() # Nach dem Callback gewünschte Ziel-URL speichern session['oauth_next'] = next_url redirect_uri = current_app.config.get('ZITADEL_REDIRECT_URI') or url_for('auth.callback', _external=True) # Authlib kümmert sich hier vollautomatisch um CSRF-State, Nonce und PKCE Code Challenge! return oauth.zitadel.authorize_redirect(redirect_uri) @auth_bp.route('/callback') def callback(): """OIDC callback handler.""" if AuthService.is_mock_auth_enabled() or AuthService.is_dev_mode(): return redirect(url_for('admin.dashboard')) if 'error' in request.args: error_msg = request.args.get('error_description', request.args.get('error')) logger.error(f'OIDC provider returned error: {error_msg}') return jsonify({'error': error_msg}), 400 try: _init_zitadel_client() next_url = session.pop('oauth_next', url_for('admin.dashboard')) # Token-Exchange & automatische CSRF-State-Prüfung durch Authlib token = oauth.zitadel.authorize_access_token() user_info = oauth.zitadel.userinfo(token=token) logger.debug(f'OIDC userinfo: {user_info}') user, error = AuthService.process_oidc_callback(user_info) if error: return jsonify({'error': error}), 400 # Login-State in Flask-Session ablegen 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 AuditLog.log_action( action='login', resource_type='auth', user=user, details={'mode': 'oidc', 'provider': 'zitadel'} ) return redirect(next_url) except Exception as e: logger.error(f'OIDC callback failed: {e}', exc_info=True) return jsonify({'error': 'Authentication failed', 'details': str(e)}), 500 @auth_bp.route('/logout') def logout(): """Logout route - clears session and redirects to login.""" current_user = None if 'user_id' in session: current_user = User.query.get(session['user_id']) auth_mode = 'mock' if AuthService.is_mock_auth_enabled() else ('dev' if AuthService.is_dev_mode() else 'oidc') token_to_revoke = session.get('access_token') session.clear() if not AuthService.is_dev_mode() and token_to_revoke: try: _init_zitadel_client() oauth.zitadel.revoke_token( token=token_to_revoke, token_type_hint='access_token' ) except Exception as e: logger.warning(f'Failed to revoke token: {e}') 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.""" 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.""" 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() })