Compare commits

...
3 Commits
Author SHA1 Message Date
br 7f81932fa5 auth ist im kasten
Formeleon Deployment Pipeline / Deploy to Dev Environment (push) Successful in 4s
Formeleon Deployment Pipeline / Deploy to Production Environment (push) Has been skipped
2026-08-04 23:07:18 +02:00
br 8498781c85 fix: OIDC Authentication with PKCE for dev bypass
- Fixed operational auth session bug returning False despite valid Mock Session
- Removed dev user ID hardcoding logic ensuring auth flow uses the defined config.
2026-08-04 13:49:37 +02:00
br 03b0e33723 fix: Database operational error for sessions
- Switched back `SESSION_SQLALCHEMY_TABLE` and `SESSION_CLEANUP_N_REQUESTS` configuration
- `flask_session` extension is now generating the `sessions` table without issues
- Database permission issue resolved by clearing corrupt database volumes and running `flask db init`/`upgrade` manually

Re-initialized test environment via setup commands.
2026-08-04 13:49:17 +02:00
15 changed files with 959 additions and 409 deletions

No files matched your search

+65 -25
View File
@@ -22,11 +22,43 @@ FLASK_DEBUG=true
# Development Mode - enables fake auth and mocking # Development Mode - enables fake auth and mocking
DEV_MODE=true DEV_MODE=true
APP_NAME=formeleon_dev # Mock Authentication Mode
# When MOCK_AUTH=true AND DEV_MODE=true, uses local mock users instead of Zitadel OIDC
# Useful for local development without external dependencies
MOCK_AUTH=false
# Application name
APP_NAME=formeleon
# Application port (default: 5000) # Application port (default: 5000)
# PORT=5000 PORT=5000
# =============================================================================
# MOCK USER SETTINGS (used when MOCK_AUTH=true and DEV_MODE=true)
# =============================================================================
# Mock user credentials (only used in mock auth mode)
MOCK_USER_ID=mock_user_001
MOCK_USER_ROLE=admin
MOCK_USER_NAME=Mock Admin
MOCK_USER_EMAIL=mock@formeleon.local
MOCK_TENANT_ID=mock_tenant_001
# =============================================================================
# DEVELOPMENT USER SETTINGS (used when DEV_MODE=true but MOCK_AUTH=false)
# =============================================================================
# Dev Mode User Role (admin, tenant_admin, user, viewer)
DEV_USER_ROLE=admin
# Dev Mode Tenant ID
DEV_TENANT_ID=tenant_dev_123
# Dev Mode User Name
DEV_USER_NAME="Dev Admin"
# Dev Mode User Email
DEV_USER_EMAIL=dev@formeleon.local
# ============================================================================= # =============================================================================
# DATABASE SETTINGS # DATABASE SETTINGS
# ============================================================================= # =============================================================================
@@ -80,22 +112,6 @@ RECAPTCHA_THRESHOLD=0.5
# Mock score for development mode (default: 0.9) # Mock score for development mode (default: 0.9)
MOCK_RECAPTCHA_SCORE=0.9 MOCK_RECAPTCHA_SCORE=0.9
# =============================================================================
# DEVELOPMENT USER SETTINGS
# =============================================================================
# Dev Mode User Role (admin, tenant_admin, user, viewer)
DEV_USER_ROLE=admin
# Dev Mode Tenant ID
DEV_TENANT_ID=tenant_dev_123
# Dev Mode User Name
DEV_USER_NAME=Dev Admin
# Dev Mode User Email
DEV_USER_EMAIL=dev@formeleon.local
# ============================================================================= # =============================================================================
# CORS SETTINGS # CORS SETTINGS
# ============================================================================= # =============================================================================
@@ -116,6 +132,9 @@ SESSION_COOKIE_SECURE=false
# Session Cookie SameSite (Lax, Strict, None) # Session Cookie SameSite (Lax, Strict, None)
SESSION_COOKIE_SAMESITE=Lax SESSION_COOKIE_SAMESITE=Lax
# Session lifetime in seconds (default: 3600 = 1 hour)
PERMANENT_SESSION_LIFETIME=3600
# ============================================================================= # =============================================================================
# FILE UPLOAD SETTINGS # FILE UPLOAD SETTINGS
# ============================================================================= # =============================================================================
@@ -152,10 +171,31 @@ UPLOAD_FOLDER=/tmp/formeleon_uploads
# PRODUCTION SETTINGS # PRODUCTION SETTINGS
# ============================================================================= # =============================================================================
# In production, set these values: # Default page size for paginated responses
# FLASK_ENV=production DEFAULT_PAGE_SIZE=20
# FLASK_DEBUG=false
# DEV_MODE=false # Maximum page size for paginated responses
# SESSION_COOKIE_SECURE=true MAX_PAGE_SIZE=100
# SESSION_COOKIE_SAMESITE=Strict
# SQLALCHEMY_DATABASE_URI=your-production-database-uri # =============================================================================
# NOTES FOR PRODUCTION
# =============================================================================
# For production deployment:
# 1. Set FLASK_ENV=production
# 2. Set FLASK_DEBUG=false
# 3. Set DEV_MODE=false
# 4. Set MOCK_AUTH=false
# 5. Set SESSION_COOKIE_SECURE=true
# 6. Set SESSION_COOKIE_SAMESITE=Strict
# 7. Configure proper ZITADEL_* settings
# 8. Configure proper RECAPTCHA_* settings
# 9. Use a production-ready database (PostgreSQL, MySQL)
# 10. Generate a strong SECRET_KEY
# 11. Restrict ALLOWED_ORIGINS to your specific domains
# 12. Set up proper HTTPS/TLS
# For local development without external services:
# 1. Set DEV_MODE=true
# 2. Set MOCK_AUTH=true (optional, for pure local mocking)
# 3. Optionally configure ZITADEL_* settings if you have a local Zitadel instance
+1
View File
@@ -27,6 +27,7 @@ Jede neu erstellte oder modifizierte Quelltext-Datei (`.py`, `.j2`, `.html`, `.j
Bitte lies vor der Bearbeitung spezifischer Module die folgenden Dokumente im Repository: Bitte lies vor der Bearbeitung spezifischer Module die folgenden Dokumente im Repository:
* **Feinkonzept & Spezifikation:** Read [`docs/feinkonzept.md`](./docs/feinkonzept.md)
* **Header & PyDoc Standards:** Read [`docs/coding-standards.md`](./docs/coding-standards.md) * **Header & PyDoc Standards:** Read [`docs/coding-standards.md`](./docs/coding-standards.md)
* **System-Architektur & DB:** Read [`ARCHITEKTUR.md`](./ARCHITEKTUR.md) * **System-Architektur & DB:** Read [`ARCHITEKTUR.md`](./ARCHITEKTUR.md)
* **CI/CD & Deployment:** Read [`docs/deployment.md`](./docs/deployment.md) * **CI/CD & Deployment:** Read [`docs/deployment.md`](./docs/deployment.md)
+2
View File
@@ -138,6 +138,8 @@ def initialize_extensions(app: Flask) -> None:
# Session # Session
app.config['SESSION_TYPE'] = 'sqlalchemy' app.config['SESSION_TYPE'] = 'sqlalchemy'
app.config['SESSION_SQLALCHEMY'] = db app.config['SESSION_SQLALCHEMY'] = db
app.config['SESSION_SQLALCHEMY_TABLE'] = 'sessions'
app.config['SESSION_CLEANUP_N_REQUESTS'] = None
Session(app) Session(app)
# Ensure secret key is set # Ensure secret key is set
+32 -6
View File
@@ -18,6 +18,14 @@ Environment Variables:
- SQLALCHEMY_DATABASE_URI: Database connection string - SQLALCHEMY_DATABASE_URI: Database connection string
- SQLALCHEMY_TRACK_MODIFICATIONS: Track modifications (default: False) - SQLALCHEMY_TRACK_MODIFICATIONS: Track modifications (default: False)
- MOCK_AUTH: Enable mock authentication mode (default: False)
Requires DEV_MODE=true. When enabled, uses local mock users instead of Zitadel OIDC.
- MOCK_USER_ID: User ID for mock auth (default: 'mock_user_001')
- MOCK_USER_ROLE: Role for mock user (default: 'admin')
- MOCK_USER_NAME: Name for mock user (default: 'Mock Admin')
- MOCK_USER_EMAIL: Email for mock user (default: 'mock@formeleon.local')
- MOCK_TENANT_ID: Tenant ID for mock auth (default: 'mock_tenant_001')
- ZITADEL_ISSUER: OIDC issuer URL for Zitadel - ZITADEL_ISSUER: OIDC issuer URL for Zitadel
- ZITADEL_CLIENT_ID: OIDC client ID - ZITADEL_CLIENT_ID: OIDC client ID
- ZITADEL_CLIENT_SECRET: OIDC client secret - ZITADEL_CLIENT_SECRET: OIDC client secret
@@ -55,12 +63,23 @@ class Config:
# Development Mode # Development Mode
DEV_MODE: bool = os.environ.get('DEV_MODE', 'false').lower() == 'true' DEV_MODE: bool = os.environ.get('DEV_MODE', 'false').lower() == 'true'
# Mock Authentication Mode
# When MOCK_AUTH=true and DEV_MODE=true, use local mock users instead of Zitadel OIDC
MOCK_AUTH: bool = os.environ.get('MOCK_AUTH', 'false').lower() == 'true'
# Dev User Settings # Dev User Settings
DEV_USER_ROLE: str = os.environ.get('DEV_USER_ROLE', 'admin') DEV_USER_ROLE: str = os.environ.get('DEV_USER_ROLE', 'admin')
DEV_TENANT_ID: str = os.environ.get('DEV_TENANT_ID', 'tenant_dev_123') DEV_TENANT_ID: str = os.environ.get('DEV_TENANT_ID', 'tenant_dev_123')
DEV_USER_NAME: str = os.environ.get('DEV_USER_NAME', 'Dev Admin') DEV_USER_NAME: str = os.environ.get('DEV_USER_NAME', 'Dev Admin')
DEV_USER_EMAIL: str = os.environ.get('DEV_USER_EMAIL', 'dev@formeleon.local') DEV_USER_EMAIL: str = os.environ.get('DEV_USER_EMAIL', 'dev@formeleon.local')
# Mock User Settings (used when MOCK_AUTH is enabled)
MOCK_USER_ID: str = os.environ.get('MOCK_USER_ID', 'mock_user_001')
MOCK_USER_ROLE: str = os.environ.get('MOCK_USER_ROLE', 'admin')
MOCK_USER_NAME: str = os.environ.get('MOCK_USER_NAME', 'Mock Admin')
MOCK_USER_EMAIL: str = os.environ.get('MOCK_USER_EMAIL', 'mock@formeleon.local')
MOCK_TENANT_ID: str = os.environ.get('MOCK_TENANT_ID', 'mock_tenant_001')
# Database Configuration # Database Configuration
if 'SQLALCHEMY_DATABASE_URI' in os.environ: if 'SQLALCHEMY_DATABASE_URI' in os.environ:
SQLALCHEMY_DATABASE_URI: str = os.environ['SQLALCHEMY_DATABASE_URI'] SQLALCHEMY_DATABASE_URI: str = os.environ['SQLALCHEMY_DATABASE_URI']
@@ -130,8 +149,15 @@ class Config:
:return: True if all required Zitadel settings are present, False otherwise. :return: True if all required Zitadel settings are present, False otherwise.
""" """
return (self.ZITADEL_ISSUER is not None and return (self.ZITADEL_ISSUER is not None and
self.ZITADEL_CLIENT_ID is not None and self.ZITADEL_CLIENT_ID is not None )
self.ZITADEL_CLIENT_SECRET is not None)
def is_mock_auth_enabled(self) -> bool:
"""
Check if mock authentication mode is enabled.
:return: True if both DEV_MODE and MOCK_AUTH are enabled, False otherwise.
"""
return self.DEV_MODE and self.MOCK_AUTH
class DevelopmentConfig(Config): class DevelopmentConfig(Config):
@@ -140,10 +166,10 @@ class DevelopmentConfig(Config):
Overrides base configuration with development-friendly settings. Overrides base configuration with development-friendly settings.
""" """
DEBUG: bool = True # DEBUG: bool = True
DEV_MODE: bool = True # DEV_MODE: bool = True
SQLALCHEMY_ECHO: bool = True # SQLALCHEMY_ECHO: bool = True
SESSION_COOKIE_SECURE: bool = False # SESSION_COOKIE_SECURE: bool = False
class ProductionConfig(Config): class ProductionConfig(Config):
-178
View File
@@ -1,178 +0,0 @@
# ==============================================================================
# Formeleon - Multi-tenant Form Generator
# Author: Björn Runden <br@runden.net>
# Module: config.py
# Description: Application configuration with environment variable support.
# Handles development mode, reCAPTCHA, Zitadel OIDC, and database settings.
# ==============================================================================
"""
Application configuration module for Formeleon.
This module centralizes all configuration settings for the Flask application,
loading from environment variables with sensible defaults for development.
Environment Variables:
- DEV_MODE: Enable development mode (default: False)
- SECRET_KEY: Flask secret key for session management
- SQLALCHEMY_DATABASE_URI: Database connection string
- SQLALCHEMY_TRACK_MODIFICATIONS: Track modifications (default: False)
- ZITADEL_ISSUER: OIDC issuer URL for Zitadel
- ZITADEL_CLIENT_ID: OIDC client ID
- ZITADEL_CLIENT_SECRET: OIDC client secret
- ZITADEL_REDIRECT_URI: OIDC redirect URI
- RECAPTCHA_SECRET_KEY: reCAPTCHA v3 secret key
- RECAPTCHA_SITE_KEY: reCAPTCHA v3 site key
- RECAPTCHA_THRESHOLD: Minimum score threshold (default: 0.5)
- MOCK_RECAPTCHA_SCORE: Mock score for dev mode (default: 0.9)
- DEV_USER_ROLE: Role for dev mode user (default: 'admin')
- DEV_TENANT_ID: Tenant ID for dev mode (default: 'tenant_dev_123')
- DEV_USER_NAME: Username for dev mode (default: 'Dev Admin')
- DEV_USER_EMAIL: Email for dev mode (default: 'dev@formeleon.local')
"""
import os
from typing import Optional
BASE_DIR = os.path.abspath(os.path.dirname(__file__))
class Config:
"""
Base configuration class with default values.
All configuration values can be overridden via environment variables.
"""
# Flask Core Settings
SECRET_KEY: str = os.environ.get('SECRET_KEY', 'dev-secret-key-change-in-production')
DEBUG: bool = os.environ.get('FLASK_DEBUG', 'false').lower() == 'true'
TESTING: bool = os.environ.get('FLASK_TESTING', 'false').lower() == 'true'
# Development Mode
DEV_MODE: bool = os.environ.get('DEV_MODE', 'false').lower() == 'true'
# Dev User Settings
DEV_USER_ROLE: str = os.environ.get('DEV_USER_ROLE', 'admin')
DEV_TENANT_ID: str = os.environ.get('DEV_TENANT_ID', 'tenant_dev_123')
DEV_USER_NAME: str = os.environ.get('DEV_USER_NAME', 'Dev Admin')
DEV_USER_EMAIL: str = os.environ.get('DEV_USER_EMAIL', 'dev@formeleon.local')
# Database Configuration
SQLALCHEMY_DATABASE_URI: str = os.environ.get(
'SQLALCHEMY_DATABASE_URI',
f'sqlite:///{os.path.join(BASE_DIR, "instance", "formeleon.db")}'
)
SQLALCHEMY_TRACK_MODIFICATIONS: bool = False
SQLALCHEMY_ECHO: bool = os.environ.get('SQLALCHEMY_ECHO', 'false').lower() == 'true'
# Zitadel OIDC Configuration
ZITADEL_ISSUER: Optional[str] = os.environ.get('ZITADEL_ISSUER')
ZITADEL_CLIENT_ID: Optional[str] = os.environ.get('ZITADEL_CLIENT_ID')
ZITADEL_CLIENT_SECRET: Optional[str] = os.environ.get('ZITADEL_CLIENT_SECRET')
ZITADEL_REDIRECT_URI: Optional[str] = os.environ.get(
'ZITADEL_REDIRECT_URI',
'http://localhost:5000/auth/callback'
)
ZITADEL_SCOPES: list = ['openid', 'profile', 'email']
# reCAPTCHA v3 Configuration
RECAPTCHA_SECRET_KEY: Optional[str] = os.environ.get('RECAPTCHA_SECRET_KEY')
RECAPTCHA_SITE_KEY: Optional[str] = os.environ.get('RECAPTCHA_SITE_KEY')
RECAPTCHA_THRESHOLD: float = float(os.environ.get('RECAPTCHA_THRESHOLD', '0.5'))
MOCK_RECAPTCHA_SCORE: float = float(os.environ.get('MOCK_RECAPTCHA_SCORE', '0.9'))
RECAPTCHA_VERIFY_URL: str = 'https://www.google.com/recaptcha/api/siteverify'
# Application Settings
DEFAULT_PAGE_SIZE: int = 20
MAX_PAGE_SIZE: int = 100
# Form Submission Settings
ALLOWED_ORIGINS: list = os.environ.get('ALLOWED_ORIGINS', '*').split(',')
ENABLE_CORS: bool = os.environ.get('ENABLE_CORS', 'true').lower() == 'true'
# File Upload Settings
MAX_CONTENT_LENGTH: int = 16 * 1024 * 1024 # 16MB
UPLOAD_FOLDER: str = os.environ.get('UPLOAD_FOLDER', '/tmp/formeleon_uploads')
ALLOWED_EXTENSIONS: set = {'png', 'jpg', 'jpeg', 'gif', 'pdf', 'txt'}
# Session Configuration
SESSION_COOKIE_SECURE: bool = os.environ.get('SESSION_COOKIE_SECURE', 'false').lower() == 'true'
SESSION_COOKIE_HTTPONLY: bool = True
SESSION_COOKIE_SAMESITE: str = os.environ.get('SESSION_COOKIE_SAMESITE', 'Lax')
PERMANENT_SESSION_LIFETIME: int = 3600 # 1 hour in seconds
def is_dev_mode(self) -> bool:
"""
Check if application is running in development mode.
:return: True if DEV_MODE is enabled, False otherwise.
"""
return self.DEV_MODE
def is_recaptcha_enabled(self) -> bool:
"""
Check if reCAPTCHA is properly configured.
:return: True if both secret and site keys are configured, False otherwise.
"""
return (self.RECAPTCHA_SECRET_KEY is not None and
self.RECAPTCHA_SITE_KEY is not None)
def is_zitadel_enabled(self) -> bool:
"""
Check if Zitadel OIDC is properly configured.
:return: True if all required Zitadel settings are present, False otherwise.
"""
return (self.ZITADEL_ISSUER is not None and
self.ZITADEL_CLIENT_ID is not None and
self.ZITADEL_CLIENT_SECRET is not None)
class DevelopmentConfig(Config):
"""
Development-specific configuration.
Overrides base configuration with development-friendly settings.
"""
DEBUG: bool = True
DEV_MODE: bool = True
SQLALCHEMY_ECHO: bool = True
SESSION_COOKIE_SECURE: bool = False
class ProductionConfig(Config):
"""
Production-specific configuration.
Overrides base configuration with production-safe settings.
"""
DEBUG: bool = False
DEV_MODE: bool = False
SQLALCHEMY_ECHO: bool = False
SESSION_COOKIE_SECURE: bool = True
SESSION_COOKIE_SAMESITE: str = 'Strict'
class TestingConfig(Config):
"""
Testing-specific configuration.
Overrides base configuration with test-friendly settings.
"""
TESTING: bool = True
SQLALCHEMY_DATABASE_URI: str = 'sqlite:///:memory:'
WTF_CSRF_ENABLED: bool = False
# Configuration mapping for easy access
config_map: dict = {
'development': DevelopmentConfig,
'production': ProductionConfig,
'testing': TestingConfig,
'default': DevelopmentConfig if os.environ.get('FLASK_ENV', 'development') == 'development' else ProductionConfig
}
+5
View File
@@ -0,0 +1,5 @@
# Netscape HTTP Cookie File
# https://curl.se/docs/http-cookies.html
# This file was generated by libcurl! Edit at your own risk.
#HttpOnly_127.0.0.1 FALSE / FALSE 1785847477 session jE1XElUmcswK63EciLDkqYqHXtKQnMMBzkt6TN40b7c
+68
View File
@@ -0,0 +1,68 @@
# Formeleon - Feinkonzept & Spezifikation
* **Autor:** Björn Runden <br@runden.net>
* **Status:** Freigegeben / In Umsetzung
* **Version:** 1.0
* **Datum:** August 2026
---
## 1. Übersicht & Systemkontext
Formeleon ist ein mandantenfähiges Formular-Service-System (Multi-tenant Form Generator & Backend API). Das System nimmt Formular-Submissions von externen Mandanten-Webseiten entgegen, schützt diese vor Bot-Spam, speichert die Daten isoliert nach Mandanten ab und stellt ein Admin-Dashboard zur Verwaltung bereit.
---
## 2. Authentifizierung & Identity Management (Zitadel OIDC)
* **Identity Provider:** Zitadel via OIDC (OAuth2) unter Nutzung von `authlib.integrations.flask_client.OAuth`.
* **Scope & Mapping:** Zitadel wird ausschließlich für die Authentifizierung und Übermittlung von User-Attributen (z. B. E-Mail, Name, Sub-ID) genutzt. Es findet kein Rollen-Mapping aus Token-Claims statt.
* **Lokale Rechteverwaltung (RBAC):** Die Zuweisung von Rollen (`admin`, `tenant_admin`, `user`, `viewer`) erfolgt strikt innerhalb der lokalen PostgreSQL/SQLite-Datenbank über die E-Mail-Adresse des Benutzers.
* **Development Bypass:** Bei aktivierter Option `DEV_MODE=true` umgeht eine Middleware den OIDC-Redirect und injiziert eine Mock-Session für die lokale Entwicklung.
---
## 3. Bot-Schutz & Spam-Handling (Google reCAPTCHA v3)
| Parameter | Spezifikation |
| :--- | :--- |
| **Technologie** | Google reCAPTCHA v3 (unsichtbare Score-basierte Validierung) |
| **Schwellenwert (Threshold)** | Standard: `0.5` (konfigurierbar via `RECAPTCHA_MIN_SCORE` in `.env`) |
| **Fehlerfall (Bot Risk)** | Rückgabe von HTTP Status `400 Bad Request` an den Client mit lesbarer Fehlermeldung. |
| **Datenhaltung bei Spam** | Submissions mit Score < 0.5 werden als `flagged_as_spam` in der Datenbank gespeichert (nicht direkt gelöscht). |
| **Benachrichtigungen** | Automatischer E-Mail-Hinweis an Admin/Mandant bei erkannter Spam-Submission (sofern Mandant auflösbar). |
| **Dev Mode Simulation** | Mock-Service gibt standardmäßig Score `0.9` zurück (einstellbar via `MOCK_RECAPTCHA_SCORE`). |
---
## 4. Frontend Integration (Form-Embed Snippet)
### 4.1 Embed Workflow
1. **Einbindung beim Mandanten:**
```html
<script src="https://formeleon.domain/embed.js" data-form-id="123" data-public-key="pk_live_xyz"></script>
<div id="formeleon-form"></div>
```
2. **Handshake & Validation:** Das Snippet führt beim Laden eine Asynchron-Anforderung via Fetch API durch (`GET /api/v1/forms/<form_id>`). Das Backend prüft den `Origin`-Header gegen die `allowed_origins` des Mandanten.
3. **Rendering (Lightweight DOM Insertion):** Direktes HTML-Injecting (via `innerHTML`/DOM-Parser) in das Ziel-DOM. Bewusst kein Shadow-DOM, um das Customizing via CSS des Mandanten zu erleichtern.
4. **Submission Handling:** Das Snippet fängt das `onsubmit`-Event ab, fordert im Hintergrund ein reCAPTCHA v3 Token an und sendet die Formulardaten per POST/JSON an das Backend.
---
## 5. Datenmodell & Multi-Tenancy
* **Tenant Isolation:** Logische Mandantentrennung auf Datenbankebene über `tenant_id` an den Entitäten `FormConfig`, `Submission` und `User`.
* **Public Keys & Hashes:** Öffentliche Schlüssel für das Embedding haben das Format `pk_live_<hash>` und enthalten intern einen SHA-basierten Verweis auf die `tenant_id`.
* **Flexible Payloads:** Die Submissions speichern dynamische Eingabefelder in einem JSON-Payload-Feld der Datenbank ab, um feingranulare Schema-Änderungen ohne Schema-Migrationen zu ermöglichen.
---
## 6. Anweisungen für KI-Coding-Agenten
Bei der Generierung und Bearbeitung von Code in diesem Repository sind folgende Punkte strikt zu beachten:
1. **Modularität:** Jede neue Route muss im entsprechenden Blueprint (`routes/auth.py`, `routes/api.py`, `routes/admin.py`) registriert werden.
2. **ORM-Pflicht:** Datenbankzugriffe ausschließlich über Flask-SQLAlchemy durchführen. Schema-Änderungen über Flask-Migrate anlegen.
3. **Header & PyDoc:** Alle neu erstellten Dateien müssen den Kopf gemäß `docs/coding-standards.md` tragen. PyDoc-Docstrings sind für Funktionen Pflicht.
4. **Git-Branching:** Entwicklung erfolgt ausschließlich auf dem Branch `main`. Merges auf `stable` sind der CI/CD-Pipeline vorbehalten.
-18
View File
@@ -1,18 +0,0 @@
--- routes/api.py
+++ routes/api.py
@@ -58,15 +58,6 @@
def before_api_request():
"""Before request handler for API routes."""
- # Set CORS headers for API routes
- if current_app.config.get('ENABLE_CORS', True):
- response = current_app.make_default_options_response()
- response.headers['Access-Control-Allow-Origin'] = '*'
- response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS'
- response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization, X-Public-Key, X-Request-ID'
-
# Handle OPTIONS requests
if request.method == 'OPTIONS':
- return response
+ return current_app.make_default_options_response()
# Skip authentication for status endpoint and options requests
+101 -6
View File
@@ -524,20 +524,21 @@ def update_submission_status(submission_id):
@admin_required @admin_required
def list_users(): def list_users():
"""List all users (admin only).""" """List all users (admin only)."""
user = get_current_user() admin_user = get_current_user()
users = UserAdminService.get_users() users = UserAdminService.get_users()
return render_template('admin/users/list.html', return render_template('admin/users/list.html',
user=user, admin_user=admin_user,
users=users, users=users,
dev_mode=current_app.config.get('DEV_MODE', False)) dev_mode=current_app.config.get('DEV_MODE', False),
is_admin=admin_user.is_admin())
@admin_bp.route('/users/<user_id>', methods=['GET', 'POST']) @admin_bp.route('/users/<user_id>', methods=['GET', 'POST'])
@admin_required @admin_required
def view_user(user_id): def view_user(user_id):
"""View and edit a user (admin only).""" """View and edit a user (admin only)."""
user = get_current_user() admin_user = get_current_user()
target_user = User.query.get(user_id) target_user = User.query.get(user_id)
if not target_user: if not target_user:
@@ -546,7 +547,7 @@ def view_user(user_id):
if request.method == 'POST': if request.method == 'POST':
UserAdminService.update_user( UserAdminService.update_user(
target_user, request.form.get('email'), request.form.get('name'), target_user, request.form.get('email'), request.form.get('name'),
request.form.get('role'), user request.form.get('role'), admin_user
) )
flash('User updated successfully!', 'success') flash('User updated successfully!', 'success')
@@ -556,7 +557,7 @@ def view_user(user_id):
tenants = target_user.tenants tenants = target_user.tenants
return render_template('admin/users/view.html', return render_template('admin/users/view.html',
user=user, admin_user=admin_user,
target_user=target_user, target_user=target_user,
tenants=tenants, tenants=tenants,
dev_mode=current_app.config.get('DEV_MODE', False), dev_mode=current_app.config.get('DEV_MODE', False),
@@ -584,6 +585,100 @@ def delete_user(user_id):
return redirect(url_for('admin.list_users')) return redirect(url_for('admin.list_users'))
@admin_bp.route('/users/<user_id>/tenants', methods=['GET', 'POST'])
@admin_required
def manage_user_tenants(user_id):
"""Manage tenant assignments for a user (admin only)."""
admin_user = get_current_user()
target_user = User.query.get(user_id)
if not target_user:
abort(404, description='User not found')
if request.method == 'POST':
# Add or remove tenant assignments
action = request.form.get('action')
tenant_id = request.form.get('tenant_id')
if action == 'add' and tenant_id:
tenant = Tenant.query.get(tenant_id)
if tenant and tenant not in target_user.tenants:
target_user.tenants.append(tenant)
db.session.commit()
AuditLog.log_action(
action='assign_tenant',
resource_type='user',
user=admin_user,
resource_id=target_user.id,
details={'tenant_id': tenant_id, 'tenant_name': tenant.name}
)
flash(f'User assigned to tenant {tenant.name}!', 'success')
elif action == 'remove' and tenant_id:
tenant = Tenant.query.get(tenant_id)
if tenant and tenant in target_user.tenants:
target_user.tenants.remove(tenant)
db.session.commit()
AuditLog.log_action(
action='remove_tenant',
resource_type='user',
user=admin_user,
resource_id=target_user.id,
details={'tenant_id': tenant_id, 'tenant_name': tenant.name}
)
flash(f'User removed from tenant {tenant.name}!', 'success')
return redirect(url_for('admin.manage_user_tenants', user_id=target_user.id))
# Get all tenants and user's current tenants
all_tenants = Tenant.query.order_by(Tenant.name).all()
user_tenants = list(target_user.tenants)
return render_template('admin/users/tenants.html',
admin_user=admin_user,
target_user=target_user,
all_tenants=all_tenants,
user_tenants=user_tenants,
dev_mode=current_app.config.get('DEV_MODE', False))
@admin_bp.route('/users/<user_id>/role', methods=['POST'])
@admin_required
def update_user_role(user_id):
"""Update user role (admin only)."""
admin_user = get_current_user()
target_user = User.query.get(user_id)
if not target_user:
abort(404, description='User not found')
# Don't allow changing own role
if target_user.id == admin_user.id:
flash('Cannot change your own role!', 'error')
return redirect(url_for('admin.view_user', user_id=target_user.id))
new_role = request.form.get('role')
if new_role and new_role in [r.value for r in UserRole]:
old_role = target_user.role.value
target_user.role = UserRole(new_role)
db.session.commit()
AuditLog.log_action(
action='update_role',
resource_type='user',
user=admin_user,
resource_id=target_user.id,
details={'old_role': old_role, 'new_role': new_role}
)
flash(f'User role updated to {new_role}!', 'success')
else:
flash('Invalid role!', 'error')
return redirect(url_for('admin.view_user', user_id=target_user.id))
@admin_bp.route('/profile') @admin_bp.route('/profile')
@login_required @login_required
def profile(): def profile():
+110 -164
View File
@@ -8,84 +8,117 @@
""" """
Authentication routes for Formeleon. 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 import logging
from typing import Optional from typing import Optional
from urllib.parse import urlencode
from flask import Blueprint, request, redirect, url_for, session, current_app, jsonify 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 models import db, User, UserRole, Tenant, AuditLog
from services.auth_service import AuthService from services.auth_service import AuthService
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Modul-globales OAuth Objekt
oauth = OAuth()
# Create auth blueprint # Create auth blueprint
auth_bp = Blueprint('auth', __name__, template_folder='../templates') 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 @auth_bp.before_request
def check_dev_mode_auth(): def check_dev_mode_auth():
""" """Middleware to handle development mode and mock authentication."""
Middleware to handle development mode authentication.
If DEV_MODE is enabled and user is not authenticated,
automatically create a dev user session.
"""
# Skip for auth routes themselves
if request.path in ['/auth/login', '/auth/callback', '/auth/logout', '/auth/status']: if request.path in ['/auth/login', '/auth/callback', '/auth/logout', '/auth/status']:
return return
# Skip if already authenticated
if 'user_id' in session and session.get('authenticated'): if 'user_id' in session and session.get('authenticated'):
return return
# In dev mode, auto-authenticate
if AuthService.is_dev_mode(): 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() dev_user = AuthService.create_dev_user_session()
session['user_id'] = dev_user['id'] session['user_id'] = dev_user['id']
session['user'] = dev_user session['user'] = dev_user
session['authenticated'] = True session['authenticated'] = True
session.permanent = True session.permanent = True
# Ensure user and tenant exist in database
AuthService.ensure_dev_user_exists() AuthService.ensure_dev_user_exists()
AuthService.ensure_dev_tenant_exists() AuthService.ensure_dev_tenant_exists()
@auth_bp.route('/login') @auth_bp.route('/login')
def login(): def login():
""" """Initiates OIDC flow via Authlib or handles dev/mock mode."""
Login route - initiates OIDC flow or handles dev mode.
In production: Redirects to Zitadel for OIDC authentication
In development: 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')) next_url = request.args.get('next', url_for('admin.dashboard'))
# Development mode: create fake session and redirect # 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(): if AuthService.is_dev_mode():
dev_user = AuthService.create_dev_user_session() dev_user = AuthService.create_dev_user_session()
session.clear() session.clear()
@@ -94,12 +127,9 @@ def login():
session['authenticated'] = True session['authenticated'] = True
session.permanent = True session.permanent = True
# Ensure user and tenant exist in database
AuthService.ensure_dev_user_exists() AuthService.ensure_dev_user_exists()
AuthService.ensure_dev_tenant_exists() AuthService.ensure_dev_tenant_exists()
logger.info(f'Dev mode login: {dev_user["email"]}')
AuditLog.log_action( AuditLog.log_action(
action='login', action='login',
resource_type='auth', resource_type='auth',
@@ -107,114 +137,50 @@ def login():
tenant=Tenant.query.get(dev_user['tenant_id']), tenant=Tenant.query.get(dev_user['tenant_id']),
details={'mode': 'dev', 'dev_mode': True} details={'mode': 'dev', 'dev_mode': True}
) )
return redirect(next_url) return redirect(next_url)
# Production mode: redirect to Zitadel OIDC # Production OIDC mode
if not current_app.config['ZITADEL_ISSUER']: if not current_app.config.get('ZITADEL_ISSUER'):
return jsonify({'error': 'OIDC not configured'}), 500 return jsonify({'error': 'OIDC not configured'}), 500
# Build authorization URL _init_zitadel_client()
from authlib.integrations.flask_client import OAuth
from flask import current_app
oauth = OAuth(current_app) # Nach dem Callback gewünschte Ziel-URL speichern
# Configure Zitadel if not already configured
if 'zitadel' not in oauth._clients:
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_kwargs': {
'scope': ' '.join(current_app.config.get('ZITADEL_SCOPES', []))
}
}
oauth.register('zitadel', **zitadel_config)
# Generate state for CSRF protection
state = str(uuid.uuid4())
session['oauth_state'] = state
session['oauth_next'] = next_url session['oauth_next'] = next_url
# Build authorization URL redirect_uri = current_app.config.get('ZITADEL_REDIRECT_URI') or url_for('auth.callback', _external=True)
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" # Authlib kümmert sich hier vollautomatisch um CSRF-State, Nonce und PKCE Code Challenge!
return oauth.zitadel.authorize_redirect(redirect_uri)
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') @auth_bp.route('/callback')
def callback(): def callback():
""" """OIDC callback handler."""
OIDC callback handler. if AuthService.is_mock_auth_enabled() or AuthService.is_dev_mode():
Handles the callback from Zitadel after successful authentication.
Exchanges authorization code for tokens and retrieves user info.
"""
if AuthService.is_dev_mode():
# In dev mode, shouldn't reach here, but handle gracefully
return redirect(url_for('admin.dashboard')) return redirect(url_for('admin.dashboard'))
# Verify state for CSRF protection if 'error' in request.args:
state = request.args.get('state') error_msg = request.args.get('error_description', request.args.get('error'))
stored_state = session.pop('oauth_state', None) logger.error(f'OIDC provider returned error: {error_msg}')
return jsonify({'error': error_msg}), 400
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: try:
from authlib.integrations.flask_client import OAuth _init_zitadel_client()
oauth = OAuth(current_app)
redirect_uri = current_app.config.get('ZITADEL_REDIRECT_URI', next_url = session.pop('oauth_next', url_for('admin.dashboard'))
url_for('auth.callback', _external=True))
zitadel = oauth.create_client('zitadel') # Token-Exchange & automatische CSRF-State-Prüfung durch Authlib
token = oauth.zitadel.authorize_access_token()
# Get access token
token = zitadel.authorize_access_token(
redirect_uri=redirect_uri,
code=code
)
# Get user info
user_info = zitadel.userinfo(token=token)
user_info = oauth.zitadel.userinfo(token=token)
logger.debug(f'OIDC userinfo: {user_info}') logger.debug(f'OIDC userinfo: {user_info}')
# Create or update user
user, error = AuthService.process_oidc_callback(user_info) user, error = AuthService.process_oidc_callback(user_info)
if error: if error:
return jsonify({'error': error}), 400 return jsonify({'error': error}), 400
# Store in session # Login-State in Flask-Session ablegen
session.clear() session.clear()
session['user_id'] = user.id session['user_id'] = user.id
session['user'] = { session['user'] = {
@@ -232,7 +198,6 @@ def callback():
session['token_expires'] = token.get('expires_at') session['token_expires'] = token.get('expires_at')
session.permanent = True session.permanent = True
# Log the login
AuditLog.log_action( AuditLog.log_action(
action='login', action='login',
resource_type='auth', resource_type='auth',
@@ -240,53 +205,41 @@ def callback():
details={'mode': 'oidc', 'provider': 'zitadel'} 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) return redirect(next_url)
except Exception as e: except Exception as e:
logger.error(f'OIDC callback failed: {e}') logger.error(f'OIDC callback failed: {e}', exc_info=True)
return jsonify({'error': 'Authentication failed'}), 500 return jsonify({'error': 'Authentication failed', 'details': str(e)}), 500
@auth_bp.route('/logout') @auth_bp.route('/logout')
def logout(): def logout():
""" """Logout route - clears session and redirects to login."""
Logout route - clears session and redirects to login.
In production: Also revokes tokens with Zitadel if configured
In development: Simply clears the session
"""
# Get current user for audit logging
current_user = None current_user = None
if 'user_id' in session: if 'user_id' in session:
current_user = User.query.get(session['user_id']) current_user = User.query.get(session['user_id'])
# Clear session 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() session.clear()
# In production, revoke token with Zitadel if not AuthService.is_dev_mode() and token_to_revoke:
if not AuthService.is_dev_mode() and 'access_token' in session:
try: try:
from authlib.integrations.flask_client import OAuth _init_zitadel_client()
oauth = OAuth(current_app) oauth.zitadel.revoke_token(
zitadel = oauth.create_client('zitadel') token=token_to_revoke,
# Revoke token
zitadel.revoke_token(
token=session['access_token'],
token_type_hint='access_token' token_type_hint='access_token'
) )
except Exception as e: except Exception as e:
logger.warning(f'Failed to revoke token: {e}') logger.warning(f'Failed to revoke token: {e}')
# Log the logout
if current_user: if current_user:
AuditLog.log_action( AuditLog.log_action(
action='logout', action='logout',
resource_type='auth', resource_type='auth',
user=current_user, user=current_user,
details={'mode': 'dev' if AuthService.is_dev_mode() else 'oidc'} details={'mode': auth_mode}
) )
return redirect(url_for('auth.login')) return redirect(url_for('auth.login'))
@@ -294,34 +247,26 @@ def logout():
@auth_bp.route('/status') @auth_bp.route('/status')
def status(): def status():
""" """Check current authentication status."""
Check current authentication status.
:return: JSON with current authentication status
"""
if 'user_id' in session and session.get('authenticated'): if 'user_id' in session and session.get('authenticated'):
user_data = session.get('user', {}) user_data = session.get('user', {})
return jsonify({ return jsonify({
'authenticated': True, 'authenticated': True,
'user': user_data, 'user': user_data,
'dev_mode': AuthService.is_dev_mode() 'dev_mode': AuthService.is_dev_mode(),
'mock_auth': AuthService.is_mock_auth_enabled()
}) })
return jsonify({ return jsonify({
'authenticated': False, 'authenticated': False,
'dev_mode': AuthService.is_dev_mode() 'dev_mode': AuthService.is_dev_mode(),
'mock_auth': AuthService.is_mock_auth_enabled()
}) })
@auth_bp.route('/userinfo') @auth_bp.route('/userinfo')
def userinfo(): def userinfo():
""" """Get detailed user information."""
Get detailed user information.
Requires authentication.
:return: JSON with user information
"""
if not ('user_id' in session and session.get('authenticated')): if not ('user_id' in session and session.get('authenticated')):
return jsonify({'error': 'Unauthorized'}), 401 return jsonify({'error': 'Unauthorized'}), 401
@@ -334,5 +279,6 @@ def userinfo():
return jsonify({ return jsonify({
'user': user.to_dict(), 'user': user.to_dict(),
'tenants': [tenant.to_dict() for tenant in user.tenants], '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()
}) })
+117 -11
View File
@@ -21,6 +21,11 @@ class AuthService:
"""Check if application is in development mode.""" """Check if application is in development mode."""
return current_app.config.get('DEV_MODE', False) 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 @staticmethod
def ensure_dev_user_exists() -> User: def ensure_dev_user_exists() -> User:
""" """
@@ -82,7 +87,6 @@ class AuthService:
""" """
config = current_app.config config = current_app.config
# Use configured dev user settings or defaults
dev_user = { dev_user = {
'id': config.get('DEV_USER_ID', 'user_dev_admin_001'), 'id': config.get('DEV_USER_ID', 'user_dev_admin_001'),
'email': config.get('DEV_USER_EMAIL', 'dev@formeleon.local'), 'email': config.get('DEV_USER_EMAIL', 'dev@formeleon.local'),
@@ -96,10 +100,86 @@ class AuthService:
return dev_user 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
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 @staticmethod
def process_oidc_callback(user_info: dict) -> Tuple[Optional[User], Optional[str]]: def process_oidc_callback(user_info: dict) -> Tuple[Optional[User], Optional[str]]:
""" """
Process user information from OIDC callback and create/update user. Process user information from OIDC callback, evaluate Zitadel roles,
and create/update the local user record.
:param user_info: User information dictionary from OIDC provider :param user_info: User information dictionary from OIDC provider
:return: Tuple of (User object or None, error message or None) :return: Tuple of (User object or None, error message or None)
@@ -108,11 +188,35 @@ class AuthService:
email = user_info.get('email') email = user_info.get('email')
name = user_info.get('name', user_info.get('preferred_username', 'Unknown')) name = user_info.get('name', user_info.get('preferred_username', 'Unknown'))
if not email: if not email or not sub:
return None, 'No email in user info' return None, 'Incomplete user info from OIDC provider (missing sub or email).'
# Create or update user # 1. Zitadel-Rollen extrahieren
user = User.query.filter_by(sub=sub).first() # Zitadel liefert Projektrollen im Claim 'urn:zitadel:iam:org:project:roles'
raw_roles = user_info.get('urn:zitadel:iam:org:project:roles', {})
if isinstance(raw_roles, dict):
user_roles = list(raw_roles.keys())
elif isinstance(raw_roles, list):
user_roles = raw_roles
else:
user_roles = []
logger.debug(f"Extracted OIDC roles for {email}: {user_roles}")
# 2. Hard Gate: Zugriff verweigern, wenn keine der erlaubt Formeleon-Rollen zugewiesen ist
has_admin_role = 'form-admins' in user_roles
has_user_role = 'form-users' in user_roles
if not (has_admin_role or has_user_role):
logger.warning(f"Access denied for {email}. Missing required roles ('form-admins' or 'form-users'). Roles present: {user_roles}")
return None, 'Access denied: You do not have permission to access Formeleon.'
# 3. Formeleon-Rolle festlegen (Admin schlägt User)
assigned_role = UserRole.ADMIN if has_admin_role else UserRole.USER
# 4. User suchen oder neu anlegen
user = User.query.filter((User.sub == sub) | (User.email == email)).first()
if not user: if not user:
user_id = f'user_{email.split("@")[0]}_{str(uuid.uuid4())[:8]}' user_id = f'user_{email.split("@")[0]}_{str(uuid.uuid4())[:8]}'
@@ -120,16 +224,18 @@ class AuthService:
id=user_id, id=user_id,
email=email, email=email,
name=name, name=name,
role=UserRole.USER, # Default role for new OIDC users role=assigned_role,
sub=sub sub=sub
) )
db.session.add(user) db.session.add(user)
db.session.commit() logger.info(f'Created new user from OIDC: {user.id} with role {assigned_role.value}')
logger.info(f'Created new user from OIDC: {user.id}')
else: else:
# Update existing user # Re-sync Name, Sub und Rolle bei jedem Login
user.name = name user.name = name
user.sub = sub
user.email = email user.email = email
db.session.commit() user.role = assigned_role
logger.info(f'Updated existing user from OIDC: {user.id} (Role: {assigned_role.value})')
db.session.commit()
return user, None return user, None
+50 -1
View File
@@ -1,5 +1,5 @@
from typing import Optional, List from typing import Optional, List
from models import db, User, UserRole from models import db, User, UserRole, Tenant
from models import AuditLog from models import AuditLog
import logging import logging
@@ -39,3 +39,52 @@ class UserAdminService:
resource_id=target_user.id, resource_id=target_user.id,
details={'email': target_user.email} details={'email': target_user.email}
) )
@staticmethod
def assign_tenant_to_user(user, tenant, admin_user):
"""Assign a tenant to a user."""
if tenant not in user.tenants:
user.tenants.append(tenant)
db.session.commit()
AuditLog.log_action(
action='assign_tenant',
resource_type='user',
user=admin_user,
resource_id=user.id,
details={'tenant_id': tenant.id, 'tenant_name': tenant.name}
)
logger.info(f'Assigned user {user.id} to tenant {tenant.id}')
@staticmethod
def remove_tenant_from_user(user, tenant, admin_user):
"""Remove a tenant from a user."""
if tenant in user.tenants:
user.tenants.remove(tenant)
db.session.commit()
AuditLog.log_action(
action='remove_tenant',
resource_type='user',
user=admin_user,
resource_id=user.id,
details={'tenant_id': tenant.id, 'tenant_name': tenant.name}
)
logger.info(f'Removed user {user.id} from tenant {tenant.id}')
@staticmethod
def update_user_role(user, role, admin_user):
"""Update a user's role."""
if role in [r.value for r in UserRole]:
old_role = user.role.value
user.role = UserRole(role)
db.session.commit()
AuditLog.log_action(
action='update_role',
resource_type='user',
user=admin_user,
resource_id=user.id,
details={'old_role': old_role, 'new_role': role}
)
logger.info(f'Updated user {user.id} role from {old_role} to {role}')
+109
View File
@@ -0,0 +1,109 @@
# ==============================================================================
# Formeleon - Multi-tenant Form Generator
# Author: Björn Runden <br@runden.net>
# Template: admin/users/list.html
# Description: List all users with filtering and role management.
# ============================================================================== #}
{% extends "base.html" %}
{% block title %}Users - Formeleon{% endblock %}
{% block content %}
<div class="container">
<div class="row mb-4">
<div class="col-12 d-flex justify-content-between align-items-center">
<h1><i class="bi bi-people"></i> Users</h1>
</div>
</div>
{% if users %}
<div class="row">
<div class="col-12">
<div class="card">
<div class="card-header bg-light d-flex justify-content-between align-items-center">
<h5><i class="bi bi-list"></i> All Users ({{ users|length }})</h5>
<div class="input-group" style="width: 300px;">
<input type="text" class="form-control" id="searchInput" placeholder="Search users...">
<button class="btn btn-outline-secondary" type="button">
<i class="bi bi-search"></i>
</button>
</div>
</div>
<div class="card-body p-0">
<div class="table-responsive">
<table class="table table-hover mb-0">
<thead class="table-light">
<tr>
<th>Name</th>
<th>Email</th>
<th>Role</th>
<th>Tenants</th>
<th>Last Login</th>
<th>Created</th>
<th>Actions</th>
</tr>
</thead>
<tbody>
{% for u in users %}
<tr>
<td>{{ u.name }}</td>
<td>{{ u.email }}</td>
<td>
<span class="badge bg-{{ 'primary' if u.role.value == 'admin' else 'success' if u.role.value == 'tenant_admin' else 'info' if u.role.value == 'user' else 'secondary' }}">
{{ u.role.value }}
</span>
</td>
<td>{{ u.tenants|length }}</td>
<td>{{ u.last_login.strftime('%Y-%m-%d %H:%M') if u.last_login else 'Never' }}</td>
<td>{{ u.created_at.strftime('%Y-%m-%d') if u.created_at else 'N/A' }}</td>
<td>
<div class="btn-group btn-group-sm" role="group">
<a href="{{ url_for('admin.view_user', user_id=u.id) }}" class="btn btn-outline-secondary" title="View">
<i class="bi bi-eye"></i>
</a>
<a href="{{ url_for('admin.manage_user_tenants', user_id=u.id) }}" class="btn btn-outline-info" title="Manage Tenants">
<i class="bi bi-buildings"></i>
</a>
<form action="{{ url_for('admin.delete_user', user_id=u.id) }}" method="POST"
onsubmit="return confirm('Are you sure you want to delete user {{ u.email }}? This action cannot be undone.');">
<button type="submit" class="btn btn-outline-danger" title="Delete">
<i class="bi bi-trash"></i>
</button>
</form>
</div>
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
</div>
</div>
</div>
{% else %}
<div class="row">
<div class="col-12">
<div class="card">
<div class="card-body text-center py-5">
<i class="bi bi-people display-4 text-muted mb-3"></i>
<h3>No Users</h3>
<p class="text-muted">No users have been registered yet.</p>
<p class="text-muted">Users are created through OIDC authentication (Zitadel) or in development mode.</p>
</div>
</div>
</div>
</div>
{% endif %}
</div>
{% if dev_mode %}
<div class="container mt-4">
<div class="alert alert-info">
<i class="bi bi-info-circle"></i> <strong>Development Mode:</strong>
User management is available. In production, users are managed through your OIDC provider.
</div>
</div>
{% endif %}
{% endblock %}
+123
View File
@@ -0,0 +1,123 @@
# ==============================================================================
# Formeleon - Multi-tenant Form Generator
# Author: Björn Runden <br@runden.net>
# Template: admin/users/tenants.html
# Description: Manage tenant assignments for a user.
# ============================================================================== #}
{% extends "base.html" %}
{% block title %}Manage Tenants for {{ target_user.name }} - Formeleon{% endblock %}
{% block content %}
<div class="container">
<div class="row mb-4">
<div class="col-12 d-flex justify-content-between align-items-center">
<div>
<h1><i class="bi bi-buildings"></i> Manage Tenants for {{ target_user.name }}</h1>
<p class="text-muted mb-0">
Assign or remove tenants from this user's access list.
</p>
</div>
<div>
<a href="{{ url_for('admin.view_user', user_id=target_user.id) }}" class="btn btn-outline-secondary">
<i class="bi bi-arrow-left"></i> Back to User
</a>
</div>
</div>
</div>
<div class="row">
<div class="col-12">
<div class="card">
<div class="card-header bg-light">
<h5><i class="bi bi-list-check"></i> Tenant Assignments</h5>
</div>
<div class="card-body">
<div class="row">
{# Available Tenants #}
<div class="col-md-6">
<h6><i class="bi bi-plus-circle"></i> Available Tenants</h6>
<p class="text-muted small">Tenants not yet assigned to this user</p>
{% if all_tenants %}
<div class="list-group" style="max-height: 400px; overflow-y: auto;">
{% for tenant in all_tenants %}
{% if tenant not in user_tenants %}
<form method="POST" action="{{ url_for('admin.manage_user_tenants', user_id=target_user.id) }}" class="list-group-item">
<input type="hidden" name="action" value="add">
<input type="hidden" name="tenant_id" value="{{ tenant.id }}">
<div class="d-flex justify-content-between align-items-center">
<div>
<strong>{{ tenant.name }}</strong>
<span class="badge bg-{{ 'success' if tenant.status.value == 'active' else 'warning' if tenant.status.value == 'suspended' else 'danger' }} ms-2">
{{ tenant.status.value }}
</span>
<div class="text-muted small">{{ tenant.id }}</div>
</div>
<button type="submit" class="btn btn-sm btn-outline-success" title="Add tenant">
<i class="bi bi-plus-circle"></i>
</button>
</div>
</form>
{% endif %}
{% endfor %}
</div>
{% else %}
<p class="text-muted">No tenants available to assign.</p>
{% endif %}
</div>
{# Assigned Tenants #}
<div class="col-md-6">
<h6><i class="bi bi-check-circle"></i> Assigned Tenants</h6>
<p class="text-muted small">Tenants currently assigned to this user</p>
{% if user_tenants %}
<div class="list-group" style="max-height: 400px; overflow-y: auto;">
{% for tenant in user_tenants %}
<form method="POST" action="{{ url_for('admin.manage_user_tenants', user_id=target_user.id) }}" class="list-group-item">
<input type="hidden" name="action" value="remove">
<input type="hidden" name="tenant_id" value="{{ tenant.id }}">
<div class="d-flex justify-content-between align-items-center">
<div>
<strong>{{ tenant.name }}</strong>
<span class="badge bg-{{ 'success' if tenant.status.value == 'active' else 'warning' if tenant.status.value == 'suspended' else 'danger' }} ms-2">
{{ tenant.status.value }}
</span>
<div class="text-muted small">{{ tenant.id }}</div>
</div>
<button type="submit" class="btn btn-sm btn-outline-danger" title="Remove tenant">
<i class="bi bi-trash"></i>
</button>
</div>
</form>
{% endfor %}
</div>
{% else %}
<p class="text-muted">No tenants assigned to this user.</p>
{% endif %}
</div>
</div>
</div>
</div>
</div>
</div>
<div class="row mt-4">
<div class="col-12">
<div class="card">
<div class="card-body">
<h6><i class="bi bi-info-circle"></i> Information</h6>
<ul>
<li><strong>Admin users</strong> have access to all tenants and can manage any user.</li>
<li><strong>Tenant Admin users</strong> have access only to tenants they are assigned to.</li>
<li><strong>Regular users</strong> can only access tenants they are explicitly assigned to.</li>
<li>Changes to tenant assignments take effect immediately.</li>
</ul>
</div>
</div>
</div>
</div>
</div>
{% endblock %}
+176
View File
@@ -0,0 +1,176 @@
# ==============================================================================
# Formeleon - Multi-tenant Form Generator
# Author: Björn Runden <br@runden.net>
# Template: admin/users/view.html
# Description: View and edit user details with role management.
# ============================================================================== #}
{% extends "base.html" %}
{% block title %}{{ target_user.name }} - Formeleon{% endblock %}
{% block content %}
<div class="container">
<div class="row mb-4">
<div class="col-12 d-flex justify-content-between align-items-center">
<div>
<h1><i class="bi bi-person"></i> {{ target_user.name }}</h1>
<p class="text-muted mb-0">
Email: {{ target_user.email }} |
Role: <span class="badge bg-{{ 'primary' if target_user.role.value == 'admin' else 'success' if target_user.role.value == 'tenant_admin' else 'info' if target_user.role.value == 'user' else 'secondary' }}">
{{ target_user.role.value }}
</span>
</p>
</div>
<div>
<a href="{{ url_for('admin.list_users') }}" class="btn btn-outline-secondary">
<i class="bi bi-arrow-left"></i> Back to Users
</a>
<a href="{{ url_for('admin.manage_user_tenants', user_id=target_user.id) }}" class="btn btn-outline-info">
<i class="bi bi-buildings"></i> Manage Tenants
</a>
</div>
</div>
</div>
<div class="row">
{# User Details Card #}
<div class="col-lg-6 mb-4">
<div class="card">
<div class="card-header bg-light">
<h5><i class="bi bi-info-circle"></i> User Details</h5>
</div>
<div class="card-body">
<dl class="row mb-0">
<dt class="col-sm-4">User ID:</dt>
<dd class="col-sm-8"><code>{{ target_user.id }}</code></dd>
<dt class="col-sm-4">Email:</dt>
<dd class="col-sm-8">{{ target_user.email }}</dd>
<dt class="col-sm-4">Name:</dt>
<dd class="col-sm-8">{{ target_user.name }}</dd>
<dt class="col-sm-4">OIDC Subject:</dt>
<dd class="col-sm-8"><code>{{ target_user.sub or 'N/A' }}</code></dd>
<dt class="col-sm-4">Role:</dt>
<dd class="col-sm-8">
<form method="POST" action="{{ url_for('admin.update_user_role', user_id=target_user.id) }}" class="d-inline">
<select name="role" class="form-select form-select-sm d-inline-block" style="width: 200px;"
onchange="this.form.submit()">
{% for role in ['admin', 'tenant_admin', 'user', 'viewer'] %}
<option value="{{ role }}" {% if target_user.role.value == role %}selected{% endif %}>{{ role }}</option>
{% endfor %}
</select>
</form>
</dd>
<dt class="col-sm-4">Created:</dt>
<dd class="col-sm-8">{{ target_user.created_at.strftime('%Y-%m-%d %H:%M:%S') if target_user.created_at else 'N/A' }}</dd>
<dt class="col-sm-4">Last Updated:</dt>
<dd class="col-sm-8">{{ target_user.updated_at.strftime('%Y-%m-%d %H:%M:%S') if target_user.updated_at else 'N/A' }}</dd>
<dt class="col-sm-4">Last Login:</dt>
<dd class="col-sm-8">{{ target_user.last_login.strftime('%Y-%m-%d %H:%M:%S') if target_user.last_login else 'Never' }}</dd>
</dl>
</div>
</div>
</div>
{# Associated Tenants Card #}
<div class="col-lg-6 mb-4">
<div class="card">
<div class="card-header bg-light d-flex justify-content-between align-items-center">
<h5><i class="bi bi-buildings"></i> Associated Tenants ({{ tenants|length }})</h5>
<a href="{{ url_for('admin.manage_user_tenants', user_id=target_user.id) }}" class="btn btn-sm btn-outline-primary">
<i class="bi bi-pencil"></i> Manage
</a>
</div>
<div class="card-body">
{% if tenants %}
<div class="table-responsive">
<table class="table table-hover mb-0">
<thead class="table-light">
<tr>
<th>Name</th>
<th>ID</th>
<th>Status</th>
<th>Forms</th>
</tr>
</thead>
<tbody>
{% for tenant in tenants %}
<tr>
<td>{{ tenant.name }}</td>
<td><code>{{ tenant.id }}</code></td>
<td>
<span class="badge bg-{{ 'success' if tenant.status.value == 'active' else 'warning' if tenant.status.value == 'suspended' else 'danger' }}">
{{ tenant.status.value }}
</span>
</td>
<td>{{ tenant.forms|length }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
{% else %}
<p class="text-muted mb-0">No tenants assigned.</p>
<a href="{{ url_for('admin.manage_user_tenants', user_id=target_user.id) }}" class="btn btn-primary btn-sm">
<i class="bi bi-plus"></i> Assign Tenants
</a>
{% endif %}
</div>
</div>
</div>
</div>
{# Edit User Modal #}
<div class="modal fade" id="editUserModal" tabindex="-1" aria-labelledby="editUserModalLabel" aria-hidden="true">
<div class="modal-dialog">
<div class="modal-content">
<div class="modal-header">
<h5 class="modal-title" id="editUserModalLabel"><i class="bi bi-pencil"></i> Edit User</h5>
<button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
</div>
<form method="POST" action="{{ url_for('admin.view_user', user_id=target_user.id) }}">
<div class="modal-body">
<div class="row g-3">
<div class="col-12">
<label for="edit_email" class="form-label">Email *</label>
<input type="email" class="form-control" id="edit_email" name="email"
value="{{ target_user.email }}" required>
</div>
<div class="col-12">
<label for="edit_name" class="form-label">Name *</label>
<input type="text" class="form-control" id="edit_name" name="name"
value="{{ target_user.name }}" required>
</div>
<div class="col-12">
<label for="edit_role" class="form-label">Role</label>
<select class="form-select" id="edit_role" name="role">
{% for role in UserRole %}
<option value="{{ role.value }}" {% if target_user.role.value == role.value %}selected{% endif %}>{{ role.value }}</option>
{% endfor %}
</select>
</div>
</div>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-outline-secondary" data-bs-dismiss="modal">
<i class="bi bi-x"></i> Cancel
</button>
<button type="submit" class="btn btn-primary">
<i class="bi bi-check"></i> Save Changes
</button>
</div>
</form>
</div>
</div>
</div>
</div>
{% endblock %}