#!/usr/bin/env python3 """ OpenID Connect (OIDC) Identity Provider (IdP) in Flask Implementiert den Authorization Code Flow mit SQLite/PostgreSQL Datenbank """ from flask import Flask, request, redirect, jsonify, render_template, session import jwt from jwt.algorithms import RSAAlgorithm from datetime import datetime, timedelta from functools import wraps from flask_limiter import Limiter from flask_limiter.util import get_remote_address from flask_migrate import Migrate import json import os # Import from app.core from app.core.database import db from app.core.security import generate_secure_token, create_id_token from models import User, AuthorizationCode, AccessToken, AuditLog, cleanup_expired_tokens, seed_db, Client from config import get_config # Flask App initialisieren app = Flask(__name__) # Config laden basierend auf FLASK_ENV env = os.environ.get('FLASK_ENV', 'development') app.config.from_object(get_config(env)) # Für Kompatibilität: Config-Werte als Modul-Level Variablen ISSUER = app.config['OIDC_ISSUER'] # Datenbank initialisieren db.init_app(app) migrate = Migrate(app, db) # Rate Limiter initialisieren limiter = Limiter( app=app, key_func=get_remote_address, default_limits=["200 per day", "50 per hour"], storage_uri="memory://" ) @app.cli.command("seed") def seed(): """Seeds the database with initial users.""" seed_db() # Admin Authentication Decorator def admin_required(f): """Decorator für Admin-geschützte Routes""" @wraps(f) def decorated_function(*args, **kwargs): # Check if user is logged in user_id = session.get('user_id') if not user_id: return redirect('/login?redirect=admin') # Check if user is admin user = User.query.get(user_id) if not user or not user.is_admin or not user.is_active: session.clear() return redirect('/login?error=access_denied') return f(*args, **kwargs) return decorated_function # Admin Routes @app.route('/admin/logout') def admin_logout(): """Logout - clears session and redirects to login""" session.clear() return redirect('/login') @app.route('/admin/users') @admin_required def admin_users(): """Admin Dashboard - User List - Thin endpoint using UserService""" from app.services import UserService user_service = UserService() # Get admin and user data via service admin_id = session.get('admin_user_id') or session.get('user_id') admin_user = user_service.get_user_by_id(admin_id) users_data = user_service.get_all_users(page=1, per_page=1000) stats = user_service.get_user_statistics() return render_template( 'admin/dashboard.html', admin_user=admin_user, users=users_data['users'], total_users=stats['total_users'], active_users=stats['active_users'], admin_users=stats['admin_users'], inactive_users=stats['inactive_users'], admin_count=stats['admin_users'], message=request.args.get('message') ) @app.route('/admin/user/create', methods=['GET', 'POST']) @admin_required def admin_create_user(): """Create New User - Thin endpoint using UserService""" if request.method == 'GET': return render_template('admin/create_user.html', error=None) # POST: Create user via service from app.services import UserService user_service = UserService() result = user_service.create_user( username=request.form.get('username'), email=request.form.get('email'), name=request.form.get('name'), password=request.form.get('password'), role=request.form.get('role', 'user'), permissions_str=request.form.get('permissions', ''), is_admin=request.form.get('is_admin') == 'on', is_active=request.form.get('is_active') == 'on', admin_id=session.get('admin_user_id') or session.get('user_id'), ip_address=request.remote_addr, user_agent=request.headers.get('User-Agent') ) if not result['success']: return render_template('admin/create_user.html', error=result['error']) return redirect('/admin/users?message=User%20created%20successfully') @app.route('/admin/user//edit', methods=['GET', 'POST']) @admin_required def admin_edit_user(user_id): """Edit User - Thin endpoint using UserService""" from app.services import UserService user_service = UserService() # Get user for display user = user_service.get_user_by_id(user_id) if not user: return "User not found", 404 if request.method == 'GET': return render_template('admin/edit_user.html', user=user, error=None) # POST: Update user via service result = user_service.update_user( user_id=user_id, username=request.form.get('username'), email=request.form.get('email'), name=request.form.get('name'), role=request.form.get('role', 'user'), permissions_str=request.form.get('permissions', ''), is_admin=request.form.get('is_admin') == 'on', is_active=request.form.get('is_active') == 'on', new_password=request.form.get('new_password') ) if not result['success']: return render_template('admin/edit_user.html', user=user, error=result['error']) return redirect('/admin/users?message=User%20updated%20successfully') @app.route('/admin/user//deactivate', methods=['POST']) @admin_required def admin_deactivate_user(user_id): """Deactivate User - Thin endpoint using UserService""" from app.services import UserService user_service = UserService() result = user_service.deactivate_user(user_id) if not result['success']: return redirect(f'/admin/users?message={result["error"]}') return redirect('/admin/users?message=User%20deactivated') @app.route('/admin/user//activate', methods=['POST']) @admin_required def admin_activate_user(user_id): """Activate User - Thin endpoint using UserService""" from app.services import UserService user_service = UserService() result = user_service.activate_user(user_id) if not result['success']: return redirect(f'/admin/users?message={result["error"]}') return redirect('/admin/users?message=User%20activated') @app.route('/admin/user//delete', methods=['POST']) @admin_required def admin_delete_user(user_id): """Delete User - Thin endpoint using UserService""" from app.services import UserService user_service = UserService() result = user_service.delete_user( user_id=user_id, admin_id=session.get('admin_user_id') or session.get('user_id'), ip_address=request.remote_addr, user_agent=request.headers.get('User-Agent') ) if not result['success']: return redirect(f'/admin/users?message={result["error"]}') return redirect('/admin/users?message=User%20deleted') @app.route('/admin/clients') @admin_required def admin_clients(): """Admin Dashboard - Client List - Thin endpoint using ClientService""" from app.services import ClientService, UserService client_service = ClientService() user_service = UserService() admin_id = session.get('admin_user_id') or session.get('user_id') admin_user = user_service.get_user_by_id(admin_id) clients = client_service.get_all_clients() return render_template( 'admin/clients.html', admin_user=admin_user, clients=clients, message=request.args.get('message') ) @app.route('/admin/analytics') @admin_required def admin_analytics(): """Admin Analytics Dashboard - Shows active sessions and client usage""" from app.services import AnalyticsService analytics_service = AnalyticsService() # Get active sessions data analytics_data = analytics_service.get_active_sessions() return render_template( 'admin/analytics.html', summary=analytics_data['summary'], by_client=analytics_data['by_client'], detailed_sessions=analytics_data['detailed_sessions'] ) @app.route('/admin/client/create', methods=['GET', 'POST']) @admin_required def admin_create_client(): """Create New OIDC Client - Thin endpoint using ClientService""" if request.method == 'GET': return render_template('admin/create_client.html', error=None) # POST: Create client via service from app.services import ClientService client_service = ClientService() result = client_service.create_client( client_name=request.form.get('client_name'), redirect_uris_str=request.form.get('redirect_uris', ''), allowed_scopes_str=request.form.get('allowed_scopes', 'openid, profile, email'), client_id=request.form.get('client_id'), client_secret=request.form.get('client_secret') ) if not result['success']: return render_template('admin/create_client.html', error=result['error']) return redirect('/admin/clients?message=Client%20created%20successfully') @app.route('/admin/client//edit', methods=['GET', 'POST']) @admin_required def admin_edit_client(client_id): """Edit OIDC Client - Thin endpoint using ClientService""" from app.services import ClientService client_service = ClientService() # Get client for display client = client_service.get_client_by_id(client_id) if not client: return "Client not found", 404 if request.method == 'GET': return render_template('admin/edit_client.html', client=client, error=None) # POST: Update client via service result = client_service.update_client( client_id_pk=client_id, client_name=request.form.get('client_name'), redirect_uris_str=request.form.get('redirect_uris', ''), allowed_scopes_str=request.form.get('allowed_scopes', ''), new_client_secret=request.form.get('new_client_secret') ) if not result['success']: return render_template('admin/edit_client.html', client=client, error=result['error']) return redirect('/admin/clients?message=Client%20updated%20successfully') @app.route('/admin/client//delete', methods=['POST']) @admin_required def admin_delete_client(client_id): """Delete OIDC Client - Thin endpoint using ClientService""" from app.services import ClientService client_service = ClientService() result = client_service.delete_client(client_id) if not result['success']: return redirect(f'/admin/clients?message={result["error"]}') return redirect('/admin/clients?message=Client%20deleted') # Health Check Endpoint @app.route('/health') def health_check(): """ Health Check Endpoint für Load Balancer und Monitoring Prüft: - App läuft - Datenbankverbindung funktioniert """ try: # Test DB Connection mit einfacher Query db.session.execute(db.text('SELECT 1')) db_status = "healthy" except Exception as e: db_status = f"unhealthy: {str(e)}" return jsonify({ "status": "unhealthy", "database": db_status, "timestamp": datetime.utcnow().isoformat() }), 503 return jsonify({ "status": "healthy", "database": db_status, "timestamp": datetime.utcnow().isoformat(), "version": "1.0.0" }), 200 # OIDC Routes @app.route('/.well-known/openid-configuration') def openid_configuration(): """OIDC Discovery Endpoint""" return jsonify({ "issuer": ISSUER, "authorization_endpoint": f"{ISSUER}/authorize", "token_endpoint": f"{ISSUER}/token", "userinfo_endpoint": f"{ISSUER}/userinfo", "registration_endpoint": f"{ISSUER}/register", "jwks_uri": f"{ISSUER}/jwks", "response_types_supported": ["code"], "subject_types_supported": ["public"], "id_token_signing_alg_values_supported": ["RS256"], "scopes_supported": ["openid", "profile", "email"], "token_endpoint_auth_methods_supported": ["client_secret_post", "client_secret_basic"], "claims_supported": ["sub", "email", "name", "preferred_username"] }) @app.route('/authorize', methods=['GET', 'POST']) def authorize(): """ Authorization Endpoint - Thin endpoint using OIDCService GET: Validates request and shows login form POST: Authenticates user and creates authorization code """ from app.services import OIDCService oidc_service = OIDCService() if request.method == 'GET': # Validate authorization request result = oidc_service.validate_authorization_request( client_id=request.args.get('client_id'), redirect_uri=request.args.get('redirect_uri'), response_type=request.args.get('response_type'), scope=request.args.get('scope', ''), state=request.args.get('state', '') ) if not result['success']: return result['error'], 400 # Store auth request in session for POST session['auth_request'] = result['auth_request'] return render_template('login.html', error=None, success=None) elif request.method == 'POST': # Get stored auth request auth_request = session.get('auth_request') if not auth_request: return "Session expired. Please restart authorization", 400 # Authorize with credentials result = oidc_service.authorize_with_credentials( username=request.form.get('username'), password=request.form.get('password'), client_id=auth_request['client_id'], redirect_uri=auth_request['redirect_uri'], scope=auth_request['scope'], state=auth_request.get('state', '') ) if not result['success']: return render_template('login.html', error=result['error'], success=None) return redirect(result['redirect_url']) @app.route('/token', methods=['POST']) @limiter.limit("20 per minute") def token(): """Token Endpoint - Thin endpoint using OIDCService""" from app.services import OIDCService oidc_service = OIDCService() result = oidc_service.exchange_code_for_token( grant_type=request.form.get('grant_type'), code=request.form.get('code'), redirect_uri=request.form.get('redirect_uri'), client_id=request.form.get('client_id'), client_secret=request.form.get('client_secret') ) if 'error' in result: status_code = 401 if result['error'] == 'invalid_client' else 400 return jsonify(result), status_code return jsonify(result) @app.route('/userinfo', methods=['GET']) def userinfo(): """UserInfo Endpoint - Thin endpoint using OIDCService""" from app.services import OIDCService oidc_service = OIDCService() # Get Authorization header auth_header = request.headers.get('Authorization', '') result = oidc_service.get_userinfo(auth_header) if 'error' in result: return jsonify(result), 401 return jsonify(result) @app.route('/register', methods=['GET', 'POST']) def register(): """User Registration Endpoint - Thin endpoint using AuthService""" if request.method == 'GET': return render_template('register.html', error=None) # POST: Register user via service from app.services import AuthService auth_service = AuthService() result = auth_service.register_user( username=request.form.get('username'), email=request.form.get('email'), name=request.form.get('name'), password=request.form.get('password'), password_confirm=request.form.get('password_confirm') ) if not result['success']: return render_template('register.html', error=result['error']) # Success: Show login page with success message return render_template('login.html', error=None, success=result['message']) @app.route('/login', methods=['GET', 'POST']) @limiter.limit("10 per minute") def login(): """User Login Endpoint - Thin endpoint using AuthService""" if request.method == 'GET': return render_template('login.html', error=None, success=None) # POST: Authenticate user via service from app.services import AuthService auth_service = AuthService() result = auth_service.authenticate_user( username=request.form.get('username'), password=request.form.get('password'), ip_address=request.remote_addr, user_agent=request.headers.get('User-Agent') ) if not result['success']: return render_template('login.html', error=result['error'], success=None) # Success: Save user in session session['user_id'] = result['user'].id # Redirect based on user type or redirect parameter redirect_param = request.args.get('redirect') if redirect_param == 'admin' and result['user'].is_admin: return redirect('/admin/users') elif result['user'].is_admin: return redirect('/admin/users') else: return redirect('/dashboard') @app.route('/dashboard') def dashboard(): """User Dashboard - Zeigt eingeloggte User-Informationen""" user_id = session.get('user_id') if not user_id: return redirect('/login') user = User.query.get(user_id) if not user: session.clear() return redirect('/login') # Regular user dashboard return render_template('dashboard.html', user=user) @app.route('/my-sessions') def my_sessions(): """User Analytics - Shows user's own active sessions""" user_id = session.get('user_id') if not user_id: return redirect('/login') user = User.query.get(user_id) if not user: session.clear() return redirect('/login') # Get user-specific analytics from app.services import AnalyticsService analytics_service = AnalyticsService() analytics_data = analytics_service.get_user_analytics(user_id) return render_template( 'user_analytics.html', user=user, summary=analytics_data['summary'], active_sessions=analytics_data['active_sessions'] ) @app.route('/logout') def logout(): """User Logout""" session.clear() return redirect('/') @app.route('/change-password', methods=['GET', 'POST']) def change_password(): """Password Change Endpoint - Thin endpoint using AuthService""" if request.method == 'GET': return render_template('change_password.html', error=None, success=None) # POST: Change password via service from app.services import AuthService auth_service = AuthService() result = auth_service.change_password( username=request.form.get('username'), current_password=request.form.get('current_password'), new_password=request.form.get('new_password'), new_password_confirm=request.form.get('new_password_confirm') ) if not result['success']: return render_template('change_password.html', error=result['error'], success=None) return render_template('change_password.html', error=None, success=result['message']) @app.route('/jwks') def jwks(): """JWKS Endpoint - Liefert Public Keys für Token-Validierung""" from cryptography.hazmat.primitives import serialization from cryptography.hazmat.backends import default_backend # Load PEM string as cryptography key object public_key_pem = app.config['OIDC_JWT_PUBLIC_KEY'] public_key = serialization.load_pem_public_key( public_key_pem.encode('utf-8'), backend=default_backend() ) # Convert to JWK format jwk = RSAAlgorithm.to_jwk(public_key, as_dict=True) return jsonify({ "keys": [jwk] }) @app.route('/admin/cleanup') def admin_cleanup(): """Admin Endpoint - Löscht abgelaufene Tokens und Codes""" codes, tokens = cleanup_expired_tokens() return jsonify({ "deleted_authorization_codes": codes, "deleted_access_tokens": tokens }) @app.route('/') def index(): """Landing Page - Öffentliche Startseite ohne sensitive Daten""" return render_template('index.html') if __name__ == '__main__': print("=" * 60) print("🔐 OIDC Identity Provider gestartet") print("=" * 60) print(f"Issuer: {ISSUER}") print(f"Discovery: {ISSUER}/.well-known/openid-configuration") print(f"\nDatenbank: {app.config['SQLALCHEMY_DATABASE_URI']}") print("\nUser Management:") print(f" - Registrierung: {ISSUER}/register") print(f" - Passwort ändern: {ISSUER}/change-password") print("=" * 60) app.run(host='0.0.0.0', port=5000, debug=True)