diff --git a/routes/auth.py b/routes/auth.py index 3af92cc..0274db1 100644 --- a/routes/auth.py +++ b/routes/auth.py @@ -46,10 +46,11 @@ auth_bp = Blueprint('auth', __name__, template_folder='../templates') @auth_bp.before_request def check_dev_mode_auth(): """ - Middleware to handle development mode authentication. + Middleware to handle development mode and mock authentication. - If DEV_MODE is enabled and user is not authenticated, - automatically create a dev user session. + 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']: @@ -61,31 +62,70 @@ def check_dev_mode_auth(): # In dev mode, auto-authenticate if AuthService.is_dev_mode(): - 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() + 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 mode. + Login route - initiates OIDC flow or handles dev/mock mode. In production: Redirects to Zitadel for OIDC authentication - In development: Creates fake session and redirects to admin + 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')) - # Development mode: create fake session and redirect + # 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() @@ -122,18 +162,33 @@ def login(): # 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': current_app.config.get('ZITADEL_CLIENT_SECRET'), - 'authorize_url': f"{current_app.config.get('ZITADEL_ISSUER')}/v1/oauth/authorize", - 'access_token_url': f"{current_app.config.get('ZITADEL_ISSUER')}/v1/oauth/token", - 'userinfo_endpoint': f"{current_app.config.get('ZITADEL_ISSUER')}/v1/userinfo", + 'client_secret': client_secret, + # Automatische Konfiguration über Zitadels Discovery Endpoint + 'server_metadata_url': f"{issuer}/.well-known/openid-configuration", 'client_kwargs': { - 'scope': ' '.join(current_app.config.get('ZITADEL_SCOPES', [])) + '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' } } - oauth.register('zitadel', **zitadel_config) - + # 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 @@ -143,7 +198,7 @@ def login(): redirect_uri = current_app.config.get('ZITADEL_REDIRECT_URI', url_for('auth.callback', _external=True)) - authorization_url = f"{current_app.config.get('ZITADEL_ISSUER')}/v1/oauth/authorize" + authorization_url = f"{current_app.config.get('ZITADEL_ISSUER')}/v2/oauth/authorize" params = { 'client_id': current_app.config.get('ZITADEL_CLIENT_ID'), @@ -167,7 +222,13 @@ def callback(): 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')) @@ -255,13 +316,16 @@ def logout(): Logout route - clears session and redirects to login. In production: Also revokes tokens with Zitadel if configured - In development: Simply clears the session + 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() @@ -286,7 +350,7 @@ def logout(): action='logout', resource_type='auth', user=current_user, - details={'mode': 'dev' if AuthService.is_dev_mode() else 'oidc'} + details={'mode': auth_mode} ) return redirect(url_for('auth.login')) @@ -304,12 +368,14 @@ def status(): return jsonify({ 'authenticated': True, 'user': user_data, - 'dev_mode': AuthService.is_dev_mode() + 'dev_mode': AuthService.is_dev_mode(), + 'mock_auth': AuthService.is_mock_auth_enabled() }) return jsonify({ 'authenticated': False, - 'dev_mode': AuthService.is_dev_mode() + 'dev_mode': AuthService.is_dev_mode(), + 'mock_auth': AuthService.is_mock_auth_enabled() }) @@ -334,5 +400,6 @@ def userinfo(): return jsonify({ 'user': user.to_dict(), 'tenants': [tenant.to_dict() for tenant in user.tenants], - 'dev_mode': AuthService.is_dev_mode() + 'dev_mode': AuthService.is_dev_mode(), + 'mock_auth': AuthService.is_mock_auth_enabled() }) diff --git a/services/auth_service.py b/services/auth_service.py index 949fbd7..ec35abc 100644 --- a/services/auth_service.py +++ b/services/auth_service.py @@ -20,6 +20,11 @@ class AuthService: def is_dev_mode() -> bool: """Check if application is in development mode.""" return current_app.config.get('DEV_MODE', False) + + @staticmethod + def is_mock_auth_enabled() -> bool: + """Check if mock authentication mode is enabled.""" + return current_app.config.get('MOCK_AUTH', False) and current_app.config.get('DEV_MODE', False) @staticmethod def ensure_dev_user_exists() -> User: @@ -95,6 +100,82 @@ class AuthService: } return dev_user + + @staticmethod + def create_mock_user_session() -> dict: + """ + Create a mock authentication user session with local mock user. + + :return: Dictionary with mock user session data + """ + config = current_app.config + + # Use configured mock user settings or defaults + mock_user = { + 'id': config.get('MOCK_USER_ID', 'mock_user_001'), + 'email': config.get('MOCK_USER_EMAIL', 'mock@formeleon.local'), + 'name': config.get('MOCK_USER_NAME', 'Mock Admin'), + 'role': config.get('MOCK_USER_ROLE', 'admin'), + 'sub': f'mock-{uuid.uuid4()}', + 'tenant_id': config.get('MOCK_TENANT_ID', 'mock_tenant_001'), + 'dev_mode': True, + 'mock_auth': True, + 'authenticated': True + } + + return mock_user + + @staticmethod + def ensure_mock_user_exists() -> User: + """ + Ensure the mock user exists in the database. + + :return: User object from database + """ + config = current_app.config + + user_id = config.get('MOCK_USER_ID', 'mock_user_001') + user = User.query.get(user_id) + + if not user: + role = UserRole.ADMIN if config.get('MOCK_USER_ROLE', 'admin') == 'admin' else UserRole.USER + user = User( + id=user_id, + email=config.get('MOCK_USER_EMAIL', 'mock@formeleon.local'), + name=config.get('MOCK_USER_NAME', 'Mock Admin'), + role=role, + sub=f'mock-{uuid.uuid4()}' + ) + db.session.add(user) + db.session.commit() + logger.info(f'Created mock user: {user.id}') + + return user + + @staticmethod + def ensure_mock_tenant_exists() -> Tenant: + """ + Ensure the mock tenant exists in the database. + + :return: Tenant object from database + """ + config = current_app.config + + tenant_id = config.get('MOCK_TENANT_ID', 'mock_tenant_001') + tenant = Tenant.query.get(tenant_id) + + if not tenant: + tenant = Tenant( + id=tenant_id, + name='Mock Tenant', + public_key=f'mock-key-{uuid.uuid4().hex[:16]}', + allowed_origins='*' + ) + db.session.add(tenant) + db.session.commit() + logger.info(f'Created mock tenant: {tenant.id}') + + return tenant @staticmethod def process_oidc_callback(user_info: dict) -> Tuple[Optional[User], Optional[str]]: