From 8232f4d8f7c8569db3f103f59cf2f0da389fd819 Mon Sep 17 00:00:00 2001 From: Bjoern Runden Date: Sat, 1 Aug 2026 16:42:30 +0200 Subject: [PATCH] gemini (kostenintensives) app-aufbauen --- README.md | 10 ++ app.py | 256 +-------------------------------- docs/dev-guide.md | 15 +- routes/api.py | 163 +++------------------ routes/auth.py | 138 +++--------------- services/auth_service.py | 135 +++++++++++++++++ services/submission_service.py | 121 ++++++++++++++++ services/tenant_service.py | 59 ++++++++ utils/cli.py | 150 +++++++++++++++++++ utils/helpers.py | 34 +++++ utils/middleware.py | 97 +++++++++++++ 11 files changed, 659 insertions(+), 519 deletions(-) create mode 100644 services/auth_service.py create mode 100644 services/submission_service.py create mode 100644 services/tenant_service.py create mode 100644 utils/cli.py create mode 100644 utils/helpers.py create mode 100644 utils/middleware.py diff --git a/README.md b/README.md index 4882308..ea69f1a 100644 --- a/README.md +++ b/README.md @@ -45,7 +45,17 @@ Formeleon ist ein zentral gehosteter Service (ChamäleonCloud), der es Webseiten 3. **Datenbank initialisieren & starten:** ```bash + # Migration-Repository initialisieren (nur einmalig!) + flask db init + + # Datenbank-Schema erstellen + flask db migrate -m "initial migration" flask db upgrade + + # Optionales Seeding mit Testdaten + flask seed-db + + # Anwendung starten flask run --port=5000 ``` diff --git a/app.py b/app.py index e4b76c3..0e5cd01 100644 --- a/app.py +++ b/app.py @@ -35,6 +35,9 @@ import uuid # corrected for py3.13 from config import Config, config_map from config import Config, config_map from models import db, Tenant, User, UserRole, Submission, FormConfig, AuditLog, SubmissionStatus +from utils.cli import register_cli_commands +from utils.middleware import setup_dev_mode_middleware, get_current_user, get_current_tenant +from utils.helpers import get_request_context # Initialize extensions globally (will be bound to app in create_app) @@ -219,259 +222,6 @@ def register_error_handlers(app: Flask) -> None: }, 500 -def register_cli_commands(app: Flask) -> None: - """ - Register custom CLI commands for the application. - - :param app: Flask application instance - """ - import click - from flask.cli import with_appcontext - - @app.cli.command('seed-db') - @with_appcontext - def seed_db(): - """Seed the database with test data for development.""" - import uuid - - # Check if data already exists - if Tenant.query.first(): - print('Database already contains data. Use --force to overwrite.') - return - - # Create dev tenant - tenant = Tenant( - id='tenant_dev_123', - name='Development Tenant', - public_key='dev-public-key-12345', - allowed_origins='http://localhost:5000,http://localhost:8080', - status=TenantStatus.ACTIVE - ) - db.session.add(tenant) - - # Create sample form - sample_form = FormConfig( - id='form_sample_001', - tenant_id='tenant_dev_123', - name='Contact Form', - description='Sample contact form for testing', - form_structure={ - 'version': '1.0', - 'fields': [ - { - 'name': 'name', - 'type': 'text', - 'label': 'Your Name', - 'required': True, - 'validation': {'min_length': 2, 'max_length': 100} - }, - { - 'name': 'email', - 'type': 'email', - 'label': 'Email Address', - 'required': True, - 'validation': {'pattern': r'^[^@]+@[^@]+\.[^@]+$'} - }, - { - 'name': 'subject', - 'type': 'text', - 'label': 'Subject', - 'required': True - }, - { - 'name': 'message', - 'type': 'textarea', - 'label': 'Message', - 'required': True, - 'validation': {'min_length': 10} - } - ], - 'submit_button': { - 'text': 'Send Message', - 'style': 'primary' - } - }, - settings={ - 'email_notification': { - 'enabled': False, - 'recipients': ['admin@formeleon.local'] - }, - 'success_message': 'Thank you for your message!', - 'recaptcha_enabled': True - } - ) - db.session.add(sample_form) - - # Create dev admin user - admin_user = User( - id='user_dev_admin_001', - email=app.config.get('DEV_USER_EMAIL', 'dev@formeleon.local'), - name=app.config.get('DEV_USER_NAME', 'Dev Admin'), - role=UserRole.ADMIN, - sub='dev-admin-sub-123' - ) - db.session.add(admin_user) - - # Associate user with tenant - from models import user_tenant_association - db.session.execute( - user_tenant_association.insert().values( - user_id=admin_user.id, - tenant_id=tenant.id - ) - ) - - db.session.commit() - print(f'Seeded database with tenant {tenant.id}, form {sample_form.id}, and user {admin_user.id}') - - @app.cli.command('create-tenant') - @click.argument('name') - @click.argument('public_key') - @click.option('--allowed-origins', default='*', help='Comma-separated allowed origins') - @with_appcontext - def create_tenant(name, public_key, allowed_origins): - """Create a new tenant.""" - tenant_id = f'tenant_{name.lower().replace(" ", "_")}_{str(uuid.uuid4())[:8]}' - tenant = Tenant( - id=tenant_id, - name=name, - public_key=public_key, - allowed_origins=allowed_origins - ) - db.session.add(tenant) - db.session.commit() - print(f'Created tenant {tenant_id} with public key {public_key}') - - @app.cli.command('create-user') - @click.argument('email') - @click.argument('name') - @click.option('--role', default='user', help='User role: admin, tenant_admin, user, viewer') - @with_appcontext - def create_user(email, name, role): - """Create a new user.""" - from models import UserRole as UR - role_enum = getattr(UR, role.upper(), UR.USER) - user_id = f'user_{email.split("@")[0]}_{str(uuid.uuid4())[:8]}' - user = User( - id=user_id, - email=email, - name=name, - role=role_enum - ) - db.session.add(user) - db.session.commit() - print(f'Created user {user_id} with email {email} and role {role}') - - -def setup_dev_mode_middleware(app: Flask) -> None: - """ - Set up development mode middleware for fake authentication. - - This middleware injects a fake Zitadel user session when DEV_MODE is enabled, - bypassing the need for actual OIDC authentication during development. - - :param app: Flask application instance - """ - from functools import wraps - - @app.before_request - def inject_dev_user(): - """Inject dev user into session if DEV_MODE is enabled.""" - # Skip for static files, auth routes, and API routes - if (request.path.startswith('/static/') or - request.path.startswith('/auth/') or - request.path.startswith('/api/')): - return - - # Skip if user is already authenticated - if 'user_id' in session: - return - - # Inject dev user - dev_user = { - 'id': 'user_dev_admin_001', - 'email': app.config.get('DEV_USER_EMAIL', 'dev@formeleon.local'), - 'name': app.config.get('DEV_USER_NAME', 'Dev Admin'), - 'role': app.config.get('DEV_USER_ROLE', 'admin'), - 'sub': 'dev-admin-sub-123', - 'tenant_id': app.config.get('DEV_TENANT_ID', 'tenant_dev_123'), - 'dev_mode': True - } - session['user_id'] = dev_user['id'] - session['user'] = dev_user - session.permanent = True - - # Load or create dev user in database - from models import User, UserRole, Tenant - user = User.query.get(dev_user['id']) - if not user: - role = UserRole.ADMIN if dev_user['role'] == 'admin' else UserRole.USER - user = User( - id=dev_user['id'], - email=dev_user['email'], - name=dev_user['name'], - role=role, - sub=dev_user['sub'] - ) - db.session.add(user) - db.session.commit() - - # Ensure dev tenant exists - tenant = Tenant.query.get(dev_user['tenant_id']) - if not tenant: - tenant = Tenant( - id=dev_user['tenant_id'], - name='Development Tenant', - public_key='dev-public-key-12345', - allowed_origins='*' - ) - db.session.add(tenant) - db.session.commit() - - -def get_current_user() -> Optional[User]: - """ - Get the current authenticated user from the request context. - - :return: User object if authenticated, None otherwise - """ - from flask import session - if 'user_id' not in session: - return None - return User.query.get(session['user_id']) - - -def get_current_tenant() -> Optional[Tenant]: - """ - Get the current tenant from the request context. - - :return: Tenant object if available, None otherwise - """ - from flask import session - if 'tenant_id' in session: - return Tenant.query.get(session['tenant_id']) - - user = get_current_user() - if user and user.tenants: - return user.tenants[0] - - return None - - -def get_request_context() -> dict: - """ - Get standardized request context for audit logging. - - :return: Dictionary with request context information - """ - return { - 'ip_address': request.remote_addr, - 'user_agent': request.user_agent.string if request.user_agent else None, - 'path': request.path, - 'method': request.method - } - - # Create application instance app = create_app() diff --git a/docs/dev-guide.md b/docs/dev-guide.md index 9376fd1..ab1339e 100644 --- a/docs/dev-guide.md +++ b/docs/dev-guide.md @@ -26,5 +26,16 @@ Im Dev-Modus wird die Google reCAPTCHA API nicht kontaktiert. ## 3. CLI Commands & Maintenance -* **Datenbank-Seeding:** `flask seed-db` (Erstellt Test-Mandanten und Beispiel-Formulare). -* **E-Mail Dump Trigger:** `flask run-scheduled-export --tenant-id= --period=daily` (Führt datenbank-agnostischen Export via SQLAlchemy aus und versendet die Mail). +### Datenbank-Migration +* **Migration-Repository initialisieren:** `flask db init` (nur einmalig erforderlich) +* **Migration erstellen:** `flask db migrate -m "Beschreibung der Änderungen"` (nach Modell-Änderungen) +* **Migration anwenden:** `flask db upgrade` (wendet ausstehende Migrationen an) +* **Migration zurücksetzen:** `flask db downgrade` (setzt die letzte Migration zurück) + +### Daten & Verwaltung +* **Datenbank-Seeding:** `flask seed-db` (Erstellt Test-Mandanten und Beispiel-Formulare) +* **E-Mail Dump Trigger:** `flask run-scheduled-export --tenant-id= --period=daily` (Führt datenbank-agnostischen Export via SQLAlchemy aus und versendet die Mail) + +### Tenant & User Verwaltung +* **Neuen Mandanten erstellen:** `flask create-tenant [--allowed-origins=...]` +* **Neuen Benutzer erstellen:** `flask create-user [--role=admin|tenant_admin|user|viewer]` diff --git a/routes/api.py b/routes/api.py index bc8c710..405690f 100644 --- a/routes/api.py +++ b/routes/api.py @@ -42,6 +42,8 @@ from werkzeug.exceptions import BadRequest, Unauthorized, Forbidden, NotFound from models import db, Tenant, FormConfig, Submission, SubmissionStatus from services.recaptcha import RecaptchaService +from services.tenant_service import TenantService +from services.submission_service import SubmissionService logger = logging.getLogger(__name__) @@ -50,63 +52,6 @@ logger = logging.getLogger(__name__) api_bp = Blueprint('api', __name__) -def validate_origin(origin: str, tenant: Tenant) -> bool: - """ - Validate that the request origin matches allowed origins for the tenant. - - :param origin: Request origin/referer - :param tenant: Tenant object to check against - :return: True if origin is allowed, False otherwise - """ - if not origin: - return False - - # Remove trailing slash and protocol - origin = origin.rstrip('/') - - allowed_origins = tenant.get_allowed_origins_list() - - # Check for wildcard - if '*' in allowed_origins: - return True - - # Exact match - if origin in allowed_origins: - return True - - # Check if origin matches any allowed domain (with subdomains) - for allowed in allowed_origins: - if allowed.startswith('*'): - # Wildcard subdomain match - domain = allowed[1:] # Remove leading * - if origin.endswith(domain): - return True - - return False - - -def get_tenant_by_public_key(public_key: str) -> Optional[Tenant]: - """ - Get tenant by public key. - - :param public_key: Public API key - :return: Tenant object or None - """ - return Tenant.query.filter_by(public_key=public_key).first() - - -def hash_ip(ip: str) -> str: - """ - Create a hash of an IP address for privacy-preserving storage. - - :param ip: IP address string - :return: SHA256 hash of the IP - """ - if not ip: - return '' - return hashlib.sha256(ip.encode()).hexdigest() - - @api_bp.before_request def before_api_request(): """Before request handler for API routes.""" @@ -186,7 +131,7 @@ def handshake(): origin = request.headers.get('Origin') or request.headers.get('Referer') or '' # Find tenant by public key - tenant = get_tenant_by_public_key(public_key) + tenant = TenantService.get_tenant_by_public_key(public_key) if not tenant: logger.warning(f'Handshake failed: Invalid public key {public_key[:8]}...') @@ -196,7 +141,7 @@ def handshake(): if not tenant.is_active(): raise Forbidden('Tenant is not active') - if not validate_origin(origin, tenant): + if not TenantService.validate_origin(origin, tenant): logger.warning(f'Handshake failed: Invalid origin {origin} for tenant {tenant.id}') raise Forbidden('Origin not allowed for this tenant') @@ -253,7 +198,7 @@ def get_form(form_id): raise BadRequest('Public key required') # Find tenant by public key - tenant = get_tenant_by_public_key(public_key) + tenant = TenantService.get_tenant_by_public_key(public_key) if not tenant: raise Unauthorized('Invalid public key') @@ -300,7 +245,7 @@ def submit_form(form_id): raise BadRequest('Public key required') # Find tenant by public key - tenant = get_tenant_by_public_key(public_key) + tenant = TenantService.get_tenant_by_public_key(public_key) if not tenant: logger.warning(f'Submission failed: Invalid public key {public_key[:8]}...') @@ -311,22 +256,10 @@ def submit_form(form_id): # Validate origin origin = request.headers.get('Origin') or request.headers.get('Referer') or '' - if not validate_origin(origin, tenant): + if not TenantService.validate_origin(origin, tenant): logger.warning(f'Submission failed: Invalid origin {origin} for tenant {tenant.id}') raise Forbidden('Origin not allowed') - # Get form - form = FormConfig.query.filter_by( - id=form_id, - tenant_id=tenant.id - ).first() - - if not form: - raise NotFound('Form not found') - - if not form.is_active: - raise Forbidden('Form is not active') - # Get form data from request try: if request.content_type == 'application/json': @@ -337,78 +270,20 @@ def submit_form(form_id): logger.error(f'Failed to parse form data: {e}') raise BadRequest('Invalid form data format') - if not payload: - raise BadRequest('No form data provided') - - # Validate reCAPTCHA if enabled - recaptcha_score = None - recaptcha_passed = None - - if form.settings.get('recaptcha_enabled', True) if form.settings else True: - recaptcha_token = payload.get('g-recaptcha-response') or request.headers.get('X-ReCAPTCHA-Token') - - if recaptcha_token: - recaptcha_service = RecaptchaService() - is_valid, score = recaptcha_service.verify( - recaptcha_token, - request.remote_addr - ) - recaptcha_score = score - recaptcha_passed = is_valid - - if not is_valid: - logger.warning(f'Submission rejected: reCAPTCHA failed with score {score}') - else: - # If no token provided, use dev mode score if in dev mode - if current_app.config.get('DEV_MODE', False): - recaptcha_score = current_app.config.get('MOCK_RECAPTCHA_SCORE', 0.9) - recaptcha_passed = True - - # Create submission - submission_id = f'sub_{form_id}_{str(uuid.uuid4())[:8]}' - - submission = Submission( - id=submission_id, - form_config_id=form.id, - tenant_id=tenant.id, + recaptcha_token = payload.get('g-recaptcha-response') or request.headers.get('X-ReCAPTCHA-Token') if payload else None + + response, status_code = SubmissionService.process_submission( + form_id=form_id, + tenant=tenant, + origin=origin, payload=payload, - ip_hash=hash_ip(request.remote_addr), + ip_address=request.remote_addr, user_agent=request.user_agent.string if request.user_agent else None, - referer=origin, - recaptcha_score=recaptcha_score, - recaptcha_passed=recaptcha_passed, - submission_metadata={ - 'origin': origin, - 'user_agent': request.user_agent.string if request.user_agent else None, - 'content_type': request.content_type - } + recaptcha_token=recaptcha_token, + content_type=request.content_type ) - - # Set status based on reCAPTCHA - if recaptcha_passed is False: - submission.status = SubmissionStatus.SPAM - else: - submission.status = SubmissionStatus.PENDING - - db.session.add(submission) - db.session.commit() - - logger.info(f'Form submission: {submission_id} for form {form_id}, tenant {tenant.id}') - - # Prepare response - response = { - 'status': 'success', - 'submission_id': submission.id, - 'message': form.settings.get('success_message', 'Form submitted successfully!') if form.settings else 'Form submitted successfully!' - } - - # If reCAPTCHA failed, indicate in response - if submission.status == SubmissionStatus.SPAM: - response['status'] = 'spam' - response['message'] = 'Submission flagged as spam' - logger.warning(f'Spam submission: {submission_id}') - - return jsonify(response) + + return jsonify(response), status_code @api_bp.route('/validate/recaptcha', methods=['POST']) @@ -483,7 +358,7 @@ def get_tenant_stats(tenant_id): if not public_key: raise BadRequest('Public key required') - tenant = get_tenant_by_public_key(public_key) + tenant = TenantService.get_tenant_by_public_key(public_key) if not tenant or tenant.id != tenant_id: raise Unauthorized('Invalid public key or tenant mismatch') diff --git a/routes/auth.py b/routes/auth.py index a63d2ef..3af92cc 100644 --- a/routes/auth.py +++ b/routes/auth.py @@ -34,6 +34,7 @@ from flask import Blueprint, request, redirect, url_for, session, current_app, j from flask import render_template_string from models import db, User, UserRole, Tenant, AuditLog +from services.auth_service import AuthService logger = logging.getLogger(__name__) @@ -42,86 +43,6 @@ logger = logging.getLogger(__name__) auth_bp = Blueprint('auth', __name__, template_folder='../templates') -def is_dev_mode() -> bool: - """Check if application is in development mode.""" - return current_app.config.get('DEV_MODE', False) - - -def create_dev_user_session() -> dict: - """ - Create a development mode user session with fake Zitadel user. - - :return: Dictionary with user session data - """ - config = current_app.config - - # Use configured dev user settings or defaults - dev_user = { - 'id': config.get('DEV_USER_ID', 'user_dev_admin_001'), - 'email': config.get('DEV_USER_EMAIL', 'dev@formeleon.local'), - 'name': config.get('DEV_USER_NAME', 'Dev Admin'), - 'role': config.get('DEV_USER_ROLE', 'admin'), - 'sub': f'dev-{uuid.uuid4()}', - 'tenant_id': config.get('DEV_TENANT_ID', 'tenant_dev_123'), - 'dev_mode': True, - 'authenticated': True - } - - return dev_user - - -def ensure_dev_user_exists() -> User: - """ - Ensure the development user exists in the database. - - :return: User object from database - """ - config = current_app.config - - user_id = config.get('DEV_USER_ID', 'user_dev_admin_001') - user = User.query.get(user_id) - - if not user: - role = UserRole.ADMIN if config.get('DEV_USER_ROLE', 'admin') == 'admin' else UserRole.USER - user = User( - id=user_id, - email=config.get('DEV_USER_EMAIL', 'dev@formeleon.local'), - name=config.get('DEV_USER_NAME', 'Dev Admin'), - role=role, - sub=f'dev-{uuid.uuid4()}' - ) - db.session.add(user) - db.session.commit() - logger.info(f'Created dev user: {user.id}') - - return user - - -def ensure_dev_tenant_exists() -> Tenant: - """ - Ensure the development tenant exists in the database. - - :return: Tenant object from database - """ - config = current_app.config - - tenant_id = config.get('DEV_TENANT_ID', 'tenant_dev_123') - tenant = Tenant.query.get(tenant_id) - - if not tenant: - tenant = Tenant( - id=tenant_id, - name='Development Tenant', - public_key=f'dev-key-{uuid.uuid4().hex[:16]}', - allowed_origins='*' - ) - db.session.add(tenant) - db.session.commit() - logger.info(f'Created dev tenant: {tenant.id}') - - return tenant - - @auth_bp.before_request def check_dev_mode_auth(): """ @@ -139,16 +60,16 @@ def check_dev_mode_auth(): return # In dev mode, auto-authenticate - if is_dev_mode(): - dev_user = create_dev_user_session() + if AuthService.is_dev_mode(): + dev_user = AuthService.create_dev_user_session() session['user_id'] = dev_user['id'] session['user'] = dev_user session['authenticated'] = True session.permanent = True # Ensure user and tenant exist in database - ensure_dev_user_exists() - ensure_dev_tenant_exists() + AuthService.ensure_dev_user_exists() + AuthService.ensure_dev_tenant_exists() @auth_bp.route('/login') @@ -165,8 +86,8 @@ def login(): next_url = request.args.get('next', url_for('admin.dashboard')) # Development mode: create fake session and redirect - if is_dev_mode(): - dev_user = create_dev_user_session() + if AuthService.is_dev_mode(): + dev_user = AuthService.create_dev_user_session() session.clear() session['user_id'] = dev_user['id'] session['user'] = dev_user @@ -174,8 +95,8 @@ def login(): session.permanent = True # Ensure user and tenant exist in database - ensure_dev_user_exists() - ensure_dev_tenant_exists() + AuthService.ensure_dev_user_exists() + AuthService.ensure_dev_tenant_exists() logger.info(f'Dev mode login: {dev_user["email"]}') @@ -247,7 +168,7 @@ def callback(): Handles the callback from Zitadel after successful authentication. Exchanges authorization code for tokens and retrieves user info. """ - if is_dev_mode(): + if AuthService.is_dev_mode(): # In dev mode, shouldn't reach here, but handle gracefully return redirect(url_for('admin.dashboard')) @@ -287,34 +208,11 @@ def callback(): logger.debug(f'OIDC userinfo: {user_info}') - # Extract user data - sub = user_info.get('sub') - email = user_info.get('email') - name = user_info.get('name', user_info.get('preferred_username', 'Unknown')) - - if not email: - return jsonify({'error': 'No email in user info'}), 400 - # Create or update user - user = User.query.filter_by(sub=sub).first() + user, error = AuthService.process_oidc_callback(user_info) - if not user: - user_id = f'user_{email.split("@")[0]}_{str(uuid.uuid4())[:8]}' - user = User( - id=user_id, - email=email, - name=name, - role=UserRole.USER, # Default role for new OIDC users - sub=sub - ) - db.session.add(user) - db.session.commit() - logger.info(f'Created new user from OIDC: {user.id}') - else: - # Update existing user - user.name = name - user.email = email - db.session.commit() + if error: + return jsonify({'error': error}), 400 # Store in session session.clear() @@ -368,7 +266,7 @@ def logout(): session.clear() # In production, revoke token with Zitadel - if not is_dev_mode() and 'access_token' in session: + if not AuthService.is_dev_mode() and 'access_token' in session: try: from authlib.integrations.flask_client import OAuth oauth = OAuth(current_app) @@ -388,7 +286,7 @@ def logout(): action='logout', resource_type='auth', user=current_user, - details={'mode': 'dev' if is_dev_mode() else 'oidc'} + details={'mode': 'dev' if AuthService.is_dev_mode() else 'oidc'} ) return redirect(url_for('auth.login')) @@ -406,12 +304,12 @@ def status(): return jsonify({ 'authenticated': True, 'user': user_data, - 'dev_mode': is_dev_mode() + 'dev_mode': AuthService.is_dev_mode() }) return jsonify({ 'authenticated': False, - 'dev_mode': is_dev_mode() + 'dev_mode': AuthService.is_dev_mode() }) @@ -436,5 +334,5 @@ def userinfo(): return jsonify({ 'user': user.to_dict(), 'tenants': [tenant.to_dict() for tenant in user.tenants], - 'dev_mode': is_dev_mode() + 'dev_mode': AuthService.is_dev_mode() }) diff --git a/services/auth_service.py b/services/auth_service.py new file mode 100644 index 0000000..949fbd7 --- /dev/null +++ b/services/auth_service.py @@ -0,0 +1,135 @@ +# ============================================================================== +# Formeleon - Multi-tenant Form Generator +# Author: Björn Runden +# Module: services/auth_service.py +# Description: Business logic for handling user authentication and session management. +# ============================================================================== + +import uuid +import logging +from typing import Dict, Any, Tuple, Optional +from flask import current_app, session +from models import db, User, UserRole, Tenant, AuditLog + +logger = logging.getLogger(__name__) + +class AuthService: + """Service layer for authentication and user management.""" + + @staticmethod + def is_dev_mode() -> bool: + """Check if application is in development mode.""" + return current_app.config.get('DEV_MODE', False) + + @staticmethod + def ensure_dev_user_exists() -> User: + """ + Ensure the development user exists in the database. + + :return: User object from database + """ + config = current_app.config + + user_id = config.get('DEV_USER_ID', 'user_dev_admin_001') + user = User.query.get(user_id) + + if not user: + role = UserRole.ADMIN if config.get('DEV_USER_ROLE', 'admin') == 'admin' else UserRole.USER + user = User( + id=user_id, + email=config.get('DEV_USER_EMAIL', 'dev@formeleon.local'), + name=config.get('DEV_USER_NAME', 'Dev Admin'), + role=role, + sub=f'dev-{uuid.uuid4()}' + ) + db.session.add(user) + db.session.commit() + logger.info(f'Created dev user: {user.id}') + + return user + + @staticmethod + def ensure_dev_tenant_exists() -> Tenant: + """ + Ensure the development tenant exists in the database. + + :return: Tenant object from database + """ + config = current_app.config + + tenant_id = config.get('DEV_TENANT_ID', 'tenant_dev_123') + tenant = Tenant.query.get(tenant_id) + + if not tenant: + tenant = Tenant( + id=tenant_id, + name='Development Tenant', + public_key=f'dev-key-{uuid.uuid4().hex[:16]}', + allowed_origins='*' + ) + db.session.add(tenant) + db.session.commit() + logger.info(f'Created dev tenant: {tenant.id}') + + return tenant + + @staticmethod + def create_dev_user_session() -> dict: + """ + Create a development mode user session with fake Zitadel user. + + :return: Dictionary with user session data + """ + config = current_app.config + + # Use configured dev user settings or defaults + dev_user = { + 'id': config.get('DEV_USER_ID', 'user_dev_admin_001'), + 'email': config.get('DEV_USER_EMAIL', 'dev@formeleon.local'), + 'name': config.get('DEV_USER_NAME', 'Dev Admin'), + 'role': config.get('DEV_USER_ROLE', 'admin'), + 'sub': f'dev-{uuid.uuid4()}', + 'tenant_id': config.get('DEV_TENANT_ID', 'tenant_dev_123'), + 'dev_mode': True, + 'authenticated': True + } + + return dev_user + + @staticmethod + def process_oidc_callback(user_info: dict) -> Tuple[Optional[User], Optional[str]]: + """ + Process user information from OIDC callback and create/update user. + + :param user_info: User information dictionary from OIDC provider + :return: Tuple of (User object or None, error message or None) + """ + sub = user_info.get('sub') + email = user_info.get('email') + name = user_info.get('name', user_info.get('preferred_username', 'Unknown')) + + if not email: + return None, 'No email in user info' + + # Create or update user + user = User.query.filter_by(sub=sub).first() + + if not user: + user_id = f'user_{email.split("@")[0]}_{str(uuid.uuid4())[:8]}' + user = User( + id=user_id, + email=email, + name=name, + role=UserRole.USER, # Default role for new OIDC users + sub=sub + ) + db.session.add(user) + db.session.commit() + logger.info(f'Created new user from OIDC: {user.id}') + else: + # Update existing user + user.name = name + user.email = email + db.session.commit() + + return user, None diff --git a/services/submission_service.py b/services/submission_service.py new file mode 100644 index 0000000..b98140a --- /dev/null +++ b/services/submission_service.py @@ -0,0 +1,121 @@ +# ============================================================================== +# Formeleon - Multi-tenant Form Generator +# Author: Björn Runden +# Module: services/submission_service.py +# Description: Business logic for processing and handling form submissions. +# ============================================================================== + +import uuid +import logging +from typing import Dict, Any, Tuple +from flask import current_app +from werkzeug.exceptions import BadRequest, Unauthorized, Forbidden, NotFound + +from models import db, Tenant, FormConfig, Submission, SubmissionStatus +from services.recaptcha import RecaptchaService +from utils.helpers import hash_ip + +logger = logging.getLogger(__name__) + +class SubmissionService: + """Service layer for handling form submissions.""" + + @staticmethod + def process_submission(form_id: str, tenant: Tenant, origin: str, payload: dict, + ip_address: str, user_agent: str, recaptcha_token: str = None, + content_type: str = 'application/json') -> Tuple[Dict[str, Any], int]: + """ + Process a form submission with validation and reCAPTCHA checking. + + :param form_id: ID of the form being submitted + :param tenant: Validated tenant object + :param origin: Validated origin string + :param payload: Submitted form data + :param ip_address: Submitter IP address + :param user_agent: Submitter user agent + :param recaptcha_token: Optional reCAPTCHA token + :param content_type: Request content type + :return: Tuple of (response_dict, http_status_code) + """ + if not payload: + raise BadRequest('No form data provided') + + # Get form + form = FormConfig.query.filter_by( + id=form_id, + tenant_id=tenant.id + ).first() + + if not form: + raise NotFound('Form not found') + + if not form.is_active: + raise Forbidden('Form is not active') + + # Validate reCAPTCHA if enabled + recaptcha_score = None + recaptcha_passed = None + + if form.settings.get('recaptcha_enabled', True) if form.settings else True: + if recaptcha_token: + recaptcha_service = RecaptchaService() + is_valid, score = recaptcha_service.verify( + recaptcha_token, + ip_address + ) + recaptcha_score = score + recaptcha_passed = is_valid + + if not is_valid: + logger.warning(f'Submission rejected: reCAPTCHA failed with score {score}') + else: + # If no token provided, use dev mode score if in dev mode + if current_app.config.get('DEV_MODE', False): + recaptcha_score = current_app.config.get('MOCK_RECAPTCHA_SCORE', 0.9) + recaptcha_passed = True + + # Create submission + submission_id = f'sub_{form_id}_{str(uuid.uuid4())[:8]}' + + submission = Submission( + id=submission_id, + form_config_id=form.id, + tenant_id=tenant.id, + payload=payload, + ip_hash=hash_ip(ip_address), + user_agent=user_agent, + referer=origin, + recaptcha_score=recaptcha_score, + recaptcha_passed=recaptcha_passed, + submission_metadata={ + 'origin': origin, + 'user_agent': user_agent, + 'content_type': content_type + } + ) + + # Set status based on reCAPTCHA + if recaptcha_passed is False: + submission.status = SubmissionStatus.SPAM + else: + submission.status = SubmissionStatus.PENDING + + db.session.add(submission) + db.session.commit() + + logger.info(f'Form submission: {submission_id} for form {form_id}, tenant {tenant.id}') + + # Prepare response + response = { + 'status': 'success', + 'submission_id': submission.id, + 'message': form.settings.get('success_message', 'Form submitted successfully!') if form.settings else 'Form submitted successfully!' + } + + # If reCAPTCHA failed, indicate in response + if submission.status == SubmissionStatus.SPAM: + response['status'] = 'spam' + response['message'] = 'Submission flagged as spam' + logger.warning(f'Spam submission: {submission_id}') + + return response, 200 diff --git a/services/tenant_service.py b/services/tenant_service.py new file mode 100644 index 0000000..fda0e1f --- /dev/null +++ b/services/tenant_service.py @@ -0,0 +1,59 @@ +# ============================================================================== +# Formeleon - Multi-tenant Form Generator +# Author: Björn Runden +# Module: services/tenant_service.py +# Description: Business logic for tenant management and validation. +# ============================================================================== + +from typing import Optional +from models import Tenant + +class TenantService: + """Service layer for handling tenant-related business logic.""" + + @staticmethod + def get_tenant_by_public_key(public_key: str) -> Optional[Tenant]: + """ + Get tenant by public key. + + :param public_key: Public API key + :return: Tenant object or None + """ + if not public_key: + return None + return Tenant.query.filter_by(public_key=public_key).first() + + @staticmethod + def validate_origin(origin: str, tenant: Tenant) -> bool: + """ + Validate that the request origin matches allowed origins for the tenant. + + :param origin: Request origin/referer + :param tenant: Tenant object to check against + :return: True if origin is allowed, False otherwise + """ + if not origin: + return False + + # Remove trailing slash and protocol + origin = origin.rstrip('/') + + allowed_origins = tenant.get_allowed_origins_list() + + # Check for wildcard + if '*' in allowed_origins: + return True + + # Exact match + if origin in allowed_origins: + return True + + # Check if origin matches any allowed domain (with subdomains) + for allowed in allowed_origins: + if allowed.startswith('*'): + # Wildcard subdomain match + domain = allowed[1:] # Remove leading * + if origin.endswith(domain): + return True + + return False diff --git a/utils/cli.py b/utils/cli.py new file mode 100644 index 0000000..a705772 --- /dev/null +++ b/utils/cli.py @@ -0,0 +1,150 @@ +# ============================================================================== +# Formeleon - Multi-tenant Form Generator +# Author: Björn Runden +# Module: utils/cli.py +# Description: CLI Commands for database seeding and administration. +# ============================================================================== + +import uuid +import click +from flask.cli import with_appcontext +from models import db, Tenant, User, UserRole, FormConfig, TenantStatus + +def register_cli_commands(app): + """ + Register custom CLI commands for the application. + + :param app: Flask application instance + """ + + @app.cli.command('seed-db') + @with_appcontext + def seed_db(): + """Seed the database with test data for development.""" + # Check if data already exists + if Tenant.query.first(): + print('Database already contains data. Use --force to overwrite.') + return + + # Create dev tenant + tenant = Tenant( + id='tenant_dev_123', + name='Development Tenant', + public_key='dev-public-key-12345', + allowed_origins='http://localhost:5000,http://localhost:8080', + status=TenantStatus.ACTIVE + ) + db.session.add(tenant) + + # Create sample form + sample_form = FormConfig( + id='form_sample_001', + tenant_id='tenant_dev_123', + name='Contact Form', + description='Sample contact form for testing', + form_structure={ + 'version': '1.0', + 'fields': [ + { + 'name': 'name', + 'type': 'text', + 'label': 'Your Name', + 'required': True, + 'validation': {'min_length': 2, 'max_length': 100} + }, + { + 'name': 'email', + 'type': 'email', + 'label': 'Email Address', + 'required': True, + 'validation': {'pattern': r'^[^@]+@[^@]+\.[^@]+$'} + }, + { + 'name': 'subject', + 'type': 'text', + 'label': 'Subject', + 'required': True + }, + { + 'name': 'message', + 'type': 'textarea', + 'label': 'Message', + 'required': True, + 'validation': {'min_length': 10} + } + ], + 'submit_button': { + 'text': 'Send Message', + 'style': 'primary' + } + }, + settings={ + 'email_notification': { + 'enabled': False, + 'recipients': ['admin@formeleon.local'] + }, + 'success_message': 'Thank you for your message!', + 'recaptcha_enabled': True + } + ) + db.session.add(sample_form) + + # Create dev admin user + admin_user = User( + id='user_dev_admin_001', + email=app.config.get('DEV_USER_EMAIL', 'dev@formeleon.local'), + name=app.config.get('DEV_USER_NAME', 'Dev Admin'), + role=UserRole.ADMIN, + sub='dev-admin-sub-123' + ) + db.session.add(admin_user) + + # Associate user with tenant + from models import user_tenant_association + db.session.execute( + user_tenant_association.insert().values( + user_id=admin_user.id, + tenant_id=tenant.id + ) + ) + + db.session.commit() + print(f'Seeded database with tenant {tenant.id}, form {sample_form.id}, and user {admin_user.id}') + + @app.cli.command('create-tenant') + @click.argument('name') + @click.argument('public_key') + @click.option('--allowed-origins', default='*', help='Comma-separated allowed origins') + @with_appcontext + def create_tenant(name, public_key, allowed_origins): + """Create a new tenant.""" + tenant_id = f'tenant_{name.lower().replace(" ", "_")}_{str(uuid.uuid4())[:8]}' + tenant = Tenant( + id=tenant_id, + name=name, + public_key=public_key, + allowed_origins=allowed_origins + ) + db.session.add(tenant) + db.session.commit() + print(f'Created tenant {tenant_id} with public key {public_key}') + + @app.cli.command('create-user') + @click.argument('email') + @click.argument('name') + @click.option('--role', default='user', help='User role: admin, tenant_admin, user, viewer') + @with_appcontext + def create_user(email, name, role): + """Create a new user.""" + from models import UserRole as UR + role_enum = getattr(UR, role.upper(), UR.USER) + user_id = f'user_{email.split("@")[0]}_{str(uuid.uuid4())[:8]}' + user = User( + id=user_id, + email=email, + name=name, + role=role_enum + ) + db.session.add(user) + db.session.commit() + print(f'Created user {user_id} with email {email} and role {role}') diff --git a/utils/helpers.py b/utils/helpers.py new file mode 100644 index 0000000..3094956 --- /dev/null +++ b/utils/helpers.py @@ -0,0 +1,34 @@ +# ============================================================================== +# Formeleon - Multi-tenant Form Generator +# Author: Björn Runden +# Module: utils/helpers.py +# Description: Helper and utility functions for the application. +# ============================================================================== + +import hashlib +from typing import Optional +from flask import request + +def hash_ip(ip: str) -> str: + """ + Create a hash of an IP address for privacy-preserving storage. + + :param ip: IP address string + :return: SHA256 hash of the IP + """ + if not ip: + return '' + return hashlib.sha256(ip.encode()).hexdigest() + +def get_request_context() -> dict: + """ + Get standardized request context for audit logging. + + :return: Dictionary with request context information + """ + return { + 'ip_address': request.remote_addr, + 'user_agent': request.user_agent.string if request.user_agent else None, + 'path': request.path, + 'method': request.method + } diff --git a/utils/middleware.py b/utils/middleware.py new file mode 100644 index 0000000..c23c7b4 --- /dev/null +++ b/utils/middleware.py @@ -0,0 +1,97 @@ +# ============================================================================== +# Formeleon - Multi-tenant Form Generator +# Author: Björn Runden +# Module: utils/middleware.py +# Description: Custom middlewares and context processors for the Flask app. +# ============================================================================== + +from flask import session, request +from typing import Optional +from models import db, User, UserRole, Tenant + +def setup_dev_mode_middleware(app): + """ + Set up development mode middleware for fake authentication. + + This middleware injects a fake Zitadel user session when DEV_MODE is enabled, + bypassing the need for actual OIDC authentication during development. + + :param app: Flask application instance + """ + @app.before_request + def inject_dev_user(): + """Inject dev user into session if DEV_MODE is enabled.""" + # Skip for static files, auth routes, and API routes + if (request.path.startswith('/static/') or + request.path.startswith('/auth/') or + request.path.startswith('/api/')): + return + + # Skip if user is already authenticated + if 'user_id' in session: + return + + # Inject dev user + dev_user = { + 'id': 'user_dev_admin_001', + 'email': app.config.get('DEV_USER_EMAIL', 'dev@formeleon.local'), + 'name': app.config.get('DEV_USER_NAME', 'Dev Admin'), + 'role': app.config.get('DEV_USER_ROLE', 'admin'), + 'sub': 'dev-admin-sub-123', + 'tenant_id': app.config.get('DEV_TENANT_ID', 'tenant_dev_123'), + 'dev_mode': True + } + session['user_id'] = dev_user['id'] + session['user'] = dev_user + session.permanent = True + + # Load or create dev user in database + user = User.query.get(dev_user['id']) + if not user: + role = UserRole.ADMIN if dev_user['role'] == 'admin' else UserRole.USER + user = User( + id=dev_user['id'], + email=dev_user['email'], + name=dev_user['name'], + role=role, + sub=dev_user['sub'] + ) + db.session.add(user) + db.session.commit() + + # Ensure dev tenant exists + tenant = Tenant.query.get(dev_user['tenant_id']) + if not tenant: + tenant = Tenant( + id=dev_user['tenant_id'], + name='Development Tenant', + public_key='dev-public-key-12345', + allowed_origins='*' + ) + db.session.add(tenant) + db.session.commit() + +def get_current_user() -> Optional[User]: + """ + Get the current authenticated user from the request context. + + :return: User object if authenticated, None otherwise + """ + if 'user_id' not in session: + return None + return User.query.get(session['user_id']) + +def get_current_tenant() -> Optional[Tenant]: + """ + Get the current tenant from the request context. + + :return: Tenant object if available, None otherwise + """ + if 'tenant_id' in session: + return Tenant.query.get(session['tenant_id']) + + user = get_current_user() + if user and user.tenants: + return user.tenants[0] + + return None