chmln-app-dev/formeleon
5
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

gemini (kostenintensives) app-aufbauen

Browse Source
This commit is contained in:
br committed 2026-08-01 16:42:30 +02:00
1 parent c74bca1e13
commit 8232f4d8f7
11 files changed
+659 -519
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

README.md
+10
View File
@@ -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
```
app.py
+3 -253
View File
@@ -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()
docs/dev-guide.md
+13 -2
View File
@@ -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=<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=<ID> --period=daily` (Führt datenbank-agnostischen Export via SQLAlchemy aus und versendet die Mail)
### Tenant & User Verwaltung
* **Neuen Mandanten erstellen:** `flask create-tenant <Name> <Public-Key> [--allowed-origins=...]`
* **Neuen Benutzer erstellen:** `flask create-user <Email> <Name> [--role=admin|tenant_admin|user|viewer]`
routes/api.py
+19 -144
View File
@@ -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')
routes/auth.py
+18 -120
View File
@@ -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()
})
services/auth_service.py
+135
View File
@@ -0,0 +1,135 @@
# ==============================================================================
# Formeleon - Multi-tenant Form Generator
# Author: Björn Runden <br@runden.net>
# 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
services/submission_service.py
+121
View File
@@ -0,0 +1,121 @@
# ==============================================================================
# Formeleon - Multi-tenant Form Generator
# Author: Björn Runden <br@runden.net>
# 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
services/tenant_service.py
+59
View File
@@ -0,0 +1,59 @@
# ==============================================================================
# Formeleon - Multi-tenant Form Generator
# Author: Björn Runden <br@runden.net>
# 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
utils/cli.py
+150
View File
@@ -0,0 +1,150 @@
# ==============================================================================
# Formeleon - Multi-tenant Form Generator
# Author: Björn Runden <br@runden.net>
# 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}')
utils/helpers.py
+34
View File
@@ -0,0 +1,34 @@
# ==============================================================================
# Formeleon - Multi-tenant Form Generator
# Author: Björn Runden <br@runden.net>
# 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
}
utils/middleware.py
+97
View File
@@ -0,0 +1,97 @@
# ==============================================================================
# Formeleon - Multi-tenant Form Generator
# Author: Björn Runden <br@runden.net>
# 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