auth ist im kasten
This commit is contained in:
br committed
1 parent
8498781c85
commit
7f81932fa5
14 files changed
+843
-443
No files matched your search
+75
-196
@@ -8,99 +8,96 @@
|
||||
|
||||
"""
|
||||
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 os
|
||||
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 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 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
|
||||
"""Middleware to handle development mode and mock authentication."""
|
||||
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
|
||||
"""
|
||||
"""Initiates OIDC flow via Authlib or handles dev/mock mode."""
|
||||
next_url = request.args.get('next', url_for('admin.dashboard'))
|
||||
|
||||
# Mock authentication mode: create mock session and redirect
|
||||
# Mock authentication mode
|
||||
if AuthService.is_mock_auth_enabled():
|
||||
mock_user = AuthService.create_mock_user_session()
|
||||
session.clear()
|
||||
@@ -109,12 +106,9 @@ def login():
|
||||
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',
|
||||
@@ -122,10 +116,9 @@ def login():
|
||||
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
|
||||
# Development mode (without MOCK_AUTH)
|
||||
if AuthService.is_dev_mode():
|
||||
dev_user = AuthService.create_dev_user_session()
|
||||
session.clear()
|
||||
@@ -134,12 +127,9 @@ def login():
|
||||
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',
|
||||
@@ -147,135 +137,50 @@ def login():
|
||||
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']:
|
||||
# Production OIDC mode
|
||||
if not current_app.config.get('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
|
||||
_init_zitadel_client()
|
||||
|
||||
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
|
||||
# Nach dem Callback gewünschte Ziel-URL speichern
|
||||
session['oauth_next'] = next_url
|
||||
|
||||
# Build authorization URL
|
||||
redirect_uri = current_app.config.get('ZITADEL_REDIRECT_URI',
|
||||
url_for('auth.callback', _external=True))
|
||||
redirect_uri = current_app.config.get('ZITADEL_REDIRECT_URI') or 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)
|
||||
# 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.
|
||||
|
||||
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')
|
||||
"""OIDC callback handler."""
|
||||
if AuthService.is_mock_auth_enabled() or AuthService.is_dev_mode():
|
||||
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
|
||||
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:
|
||||
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)
|
||||
_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}')
|
||||
|
||||
# Create or update user
|
||||
user, error = AuthService.process_oidc_callback(user_info)
|
||||
|
||||
if error:
|
||||
return jsonify({'error': error}), 400
|
||||
|
||||
# Store in session
|
||||
# Login-State in Flask-Session ablegen
|
||||
session.clear()
|
||||
session['user_id'] = user.id
|
||||
session['user'] = {
|
||||
@@ -293,7 +198,6 @@ def callback():
|
||||
session['token_expires'] = token.get('expires_at')
|
||||
session.permanent = True
|
||||
|
||||
# Log the login
|
||||
AuditLog.log_action(
|
||||
action='login',
|
||||
resource_type='auth',
|
||||
@@ -301,50 +205,35 @@ def callback():
|
||||
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
|
||||
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.
|
||||
|
||||
In production: Also revokes tokens with Zitadel if configured
|
||||
In development/mock: Simply clears the session
|
||||
"""
|
||||
# Get current user for audit logging
|
||||
"""Logout route - clears session and redirects to login."""
|
||||
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
|
||||
token_to_revoke = session.get('access_token')
|
||||
session.clear()
|
||||
|
||||
# In production, revoke token with Zitadel
|
||||
if not AuthService.is_dev_mode() and 'access_token' in session:
|
||||
if not AuthService.is_dev_mode() and token_to_revoke:
|
||||
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'],
|
||||
_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}')
|
||||
|
||||
# Log the logout
|
||||
if current_user:
|
||||
AuditLog.log_action(
|
||||
action='logout',
|
||||
@@ -358,11 +247,7 @@ def logout():
|
||||
|
||||
@auth_bp.route('/status')
|
||||
def status():
|
||||
"""
|
||||
Check current authentication status.
|
||||
|
||||
:return: JSON with current authentication status
|
||||
"""
|
||||
"""Check current authentication status."""
|
||||
if 'user_id' in session and session.get('authenticated'):
|
||||
user_data = session.get('user', {})
|
||||
return jsonify({
|
||||
@@ -381,13 +266,7 @@ def status():
|
||||
|
||||
@auth_bp.route('/userinfo')
|
||||
def userinfo():
|
||||
"""
|
||||
Get detailed user information.
|
||||
|
||||
Requires authentication.
|
||||
|
||||
:return: JSON with user information
|
||||
"""
|
||||
"""Get detailed user information."""
|
||||
if not ('user_id' in session and session.get('authenticated')):
|
||||
return jsonify({'error': 'Unauthorized'}), 401
|
||||
|
||||
@@ -402,4 +281,4 @@ def userinfo():
|
||||
'tenants': [tenant.to_dict() for tenant in user.tenants],
|
||||
'dev_mode': AuthService.is_dev_mode(),
|
||||
'mock_auth': AuthService.is_mock_auth_enabled()
|
||||
})
|
||||
})
|
||||
Reference in new issue
Block a user