Files
formeleon/routes/admin.py
T

759 lines
24 KiB
Python

# ==============================================================================
# Formeleon - Multi-tenant Form Generator
# Author: Björn Runden <br@runden.net>
# Module: routes/admin.py
# Description: Admin Dashboard routes for form management.
# Provides CRUD operations for tenants, forms, and submissions.
# ==============================================================================
"""
Admin Dashboard routes for Formeleon.
This module provides the management interface for:
- Tenant management (CRUD)
- Form configuration (CRUD)
- Submission viewing and export
- User management
- Dashboard analytics
Routes:
GET /admin/ - Dashboard overview
GET /admin/tenants/ - List all tenants
GET /admin/tenants/<id> - View/edit tenant
GET /admin/forms/ - List all forms
GET /admin/forms/<id> - View/edit form
GET /admin/forms/create - Create new form
GET /admin/submissions/ - List submissions
GET /admin/submissions/<id> - View submission details
GET /admin/users/ - List users
GET /admin/users/<id> - View/edit user
Authentication is handled via the auth middleware or dev mode bypass.
"""
import uuid
import logging
from typing import Optional, List, Dict, Any
from datetime import datetime, timezone
from flask import Blueprint, request, redirect, url_for, session, current_app, jsonify
from flask import render_template, flash, abort
from models import db, Tenant, FormConfig, Submission, User, UserRole, TenantStatus, SubmissionStatus
from models import AuditLog
from services.recaptcha import RecaptchaService
logger = logging.getLogger(__name__)
# Create admin blueprint
admin_bp = Blueprint('admin', __name__, template_folder='../templates', static_folder='../static')
def login_required(func):
"""Decorator to require login for admin routes."""
def wrapper(*args, **kwargs):
if not (session.get('user_id') and session.get('authenticated')):
return redirect(url_for('auth.login', next=request.url))
return func(*args, **kwargs)
wrapper.__name__ = func.__name__
return wrapper
def admin_required(func):
"""Decorator to require admin privileges."""
def wrapper(*args, **kwargs):
if not (session.get('user_id') and session.get('authenticated')):
return redirect(url_for('auth.login', next=request.url))
user = User.query.get(session['user_id'])
if not user or not user.is_admin():
abort(403, description='Admin privileges required')
return func(*args, **kwargs)
wrapper.__name__ = func.__name__
return wrapper
def get_current_user() -> Optional[User]:
"""Get current authenticated user."""
if 'user_id' in session:
return User.query.get(session['user_id'])
return None
def get_current_tenant() -> Optional[Tenant]:
"""Get current tenant from session or user context."""
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
@admin_bp.before_request
def before_request():
"""Before request handler for admin routes."""
# Ensure user exists in database if authenticated in session
if session.get('user_id'):
user = User.query.get(session['user_id'])
if not user:
# Clean up invalid session
session.clear()
return redirect(url_for('auth.login'))
@admin_bp.route('/')
@login_required
def dashboard():
"""
Admin dashboard overview.
Shows:
- Number of tenants
- Number of forms
- Recent submissions
- Quick access links
"""
user = get_current_user()
# Get statistics
tenant_count = Tenant.query.count()
form_count = FormConfig.query.count()
submission_count = Submission.query.count()
# Recent submissions
recent_submissions = Submission.query.order_by(
Submission.created_at.desc()
).limit(10).all()
# Recent forms
recent_forms = FormConfig.query.order_by(
FormConfig.created_at.desc()
).limit(5).all()
# Tenant list
tenants = Tenant.query.order_by(Tenant.created_at.desc()).limit(5).all()
return render_template('admin/dashboard.html',
user=user,
tenant_count=tenant_count,
form_count=form_count,
submission_count=submission_count,
recent_submissions=recent_submissions,
recent_forms=recent_forms,
tenants=tenants,
dev_mode=current_app.config.get('DEV_MODE', False))
@admin_bp.route('/tenants/')
@login_required
def list_tenants():
"""List all tenants."""
user = get_current_user()
# Filter tenants based on user role
if user.role == UserRole.ADMIN:
tenants = Tenant.query.order_by(Tenant.created_at.desc()).all()
else:
# Non-admin users can only see their assigned tenants
tenants = user.tenants if user else []
return render_template('admin/tenants/list.html',
user=user,
tenants=tenants,
dev_mode=current_app.config.get('DEV_MODE', False))
@admin_bp.route('/tenants/create', methods=['GET', 'POST'])
@admin_required
def create_tenant():
"""Create a new tenant."""
user = get_current_user()
if request.method == 'POST':
# Validate form data
name = request.form.get('name', '').strip()
public_key = request.form.get('public_key', '').strip()
allowed_origins = request.form.get('allowed_origins', '*').strip()
status = request.form.get('status', 'active')
# Generate tenant ID
tenant_id = f'tenant_{name.lower().replace(" ", "_")}_{str(uuid.uuid4())[:8]}'
# Create tenant
tenant = Tenant(
id=tenant_id,
name=name,
public_key=public_key or str(uuid.uuid4()).hex[:32],
allowed_origins=allowed_origins,
status=TenantStatus(status) if status in [s.value for s in TenantStatus] else TenantStatus.ACTIVE
)
db.session.add(tenant)
db.session.commit()
# Audit log
AuditLog.log_action(
action='create',
resource_type='tenant',
user=user,
resource_id=tenant.id,
details={'name': name, 'public_key': public_key}
)
flash('Tenant created successfully!', 'success')
return redirect(url_for('admin.list_tenants'))
return render_template('admin/tenants/create.html',
user=user,
dev_mode=current_app.config.get('DEV_MODE', False))
@admin_bp.route('/tenants/<tenant_id>', methods=['GET', 'POST'])
@login_required
def view_tenant(tenant_id):
"""View and edit a tenant."""
user = get_current_user()
tenant = Tenant.query.get(tenant_id)
if not tenant:
abort(404, description='Tenant not found')
# Check permissions
if user.role != UserRole.ADMIN and tenant not in user.tenants:
abort(403, description='Access denied')
if request.method == 'POST':
if not user.is_admin():
abort(403, description='Admin privileges required to edit tenant')
# Update tenant
tenant.name = request.form.get('name', tenant.name)
tenant.public_key = request.form.get('public_key', tenant.public_key)
tenant.allowed_origins = request.form.get('allowed_origins', tenant.allowed_origins)
status = request.form.get('status')
if status in [s.value for s in TenantStatus]:
tenant.status = TenantStatus(status)
db.session.commit()
# Audit log
AuditLog.log_action(
action='update',
resource_type='tenant',
user=user,
resource_id=tenant.id,
details={'name': tenant.name, 'status': tenant.status.value}
)
flash('Tenant updated successfully!', 'success')
return redirect(url_for('admin.view_tenant', tenant_id=tenant.id))
# Get tenant forms
forms = FormConfig.query.filter_by(tenant_id=tenant.id).order_by(
FormConfig.created_at.desc()
).all()
return render_template('admin/tenants/view.html',
user=user,
tenant=tenant,
forms=forms,
dev_mode=current_app.config.get('DEV_MODE', False),
is_admin=user.is_admin())
@admin_bp.route('/tenants/<tenant_id>/delete', methods=['POST'])
@admin_required
def delete_tenant(tenant_id):
"""Delete a tenant."""
user = get_current_user()
tenant = Tenant.query.get(tenant_id)
if not tenant:
abort(404, description='Tenant not found')
# Delete tenant (cascade will handle related forms and submissions)
db.session.delete(tenant)
db.session.commit()
# Audit log
AuditLog.log_action(
action='delete',
resource_type='tenant',
user=user,
resource_id=tenant.id,
details={'name': tenant.name}
)
flash('Tenant deleted successfully!', 'success')
return redirect(url_for('admin.list_tenants'))
@admin_bp.route('/forms/')
@login_required
def list_forms():
"""List all forms for the current tenant/user."""
user = get_current_user()
tenant = get_current_tenant()
# Get forms based on user access
if user.role == UserRole.ADMIN:
# Admin can see all forms
forms = FormConfig.query.join(Tenant).order_by(
FormConfig.created_at.desc()
).all()
elif tenant:
# Tenant user sees only their tenant's forms
forms = FormConfig.query.filter_by(tenant_id=tenant.id).order_by(
FormConfig.created_at.desc()
).all()
else:
forms = []
return render_template('admin/forms/list.html',
user=user,
forms=forms,
tenant=tenant,
dev_mode=current_app.config.get('DEV_MODE', False),
is_admin=user.is_admin())
@admin_bp.route('/forms/create', methods=['GET', 'POST'])
@login_required
def create_form():
"""Create a new form."""
user = get_current_user()
tenant = get_current_tenant()
if not tenant:
flash('No tenant selected. Please select a tenant.', 'error')
return redirect(url_for('admin.dashboard'))
if request.method == 'POST':
# Get form data
name = request.form.get('name', '').strip()
description = request.form.get('description', '').strip()
# Parse form structure from JSON
form_structure_str = request.form.get('form_structure', '{}')
try:
import json
form_structure = json.loads(form_structure_str)
except json.JSONDecodeError:
form_structure = {'version': '1.0', 'fields': []}
# Parse settings
settings_str = request.form.get('settings', '{}')
try:
settings = json.loads(settings_str)
except json.JSONDecodeError:
settings = {}
# Generate form ID
form_id = f'form_{name.lower().replace(" ", "_")}_{str(uuid.uuid4())[:8]}'
# Create form
form = FormConfig(
id=form_id,
tenant_id=tenant.id,
name=name,
description=description,
form_structure=form_structure,
settings=settings,
is_active=request.form.get('is_active', 'true').lower() == 'true'
)
db.session.add(form)
db.session.commit()
# Audit log
AuditLog.log_action(
action='create',
resource_type='form',
user=user,
tenant=tenant,
resource_id=form.id,
details={'name': name, 'tenant_id': tenant.id}
)
flash('Form created successfully!', 'success')
return redirect(url_for('admin.view_form', form_id=form.id))
# Provide default form structure for new forms
default_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
},
{
'name': 'message',
'type': 'textarea',
'label': 'Message',
'required': True
}
],
'submit_button': {
'text': 'Submit',
'style': 'primary'
}
}
return render_template('admin/forms/create.html',
user=user,
tenant=tenant,
default_form_structure=default_form_structure,
dev_mode=current_app.config.get('DEV_MODE', False))
@admin_bp.route('/forms/<form_id>', methods=['GET', 'POST'])
@login_required
def view_form(form_id):
"""View and edit a form."""
user = get_current_user()
form = FormConfig.query.get(form_id)
if not form:
abort(404, description='Form not found')
# Check permissions
tenant = Tenant.query.get(form.tenant_id)
if not tenant:
abort(404, description='Tenant not found')
if user.role != UserRole.ADMIN and tenant not in user.tenants:
abort(403, description='Access denied')
if request.method == 'POST':
# Update form
form.name = request.form.get('name', form.name)
form.description = request.form.get('description', form.description)
form.is_active = request.form.get('is_active', 'true').lower() == 'true'
# Update form structure
form_structure_str = request.form.get('form_structure', '{}')
try:
import json
form.form_structure = json.loads(form_structure_str)
except json.JSONDecodeError:
pass # Keep existing structure
# Update settings
settings_str = request.form.get('settings', '{}')
try:
import json
form.settings = json.loads(settings_str)
except json.JSONDecodeError:
pass # Keep existing settings
db.session.commit()
# Audit log
AuditLog.log_action(
action='update',
resource_type='form',
user=user,
tenant=tenant,
resource_id=form.id,
details={'name': form.name, 'is_active': form.is_active}
)
flash('Form updated successfully!', 'success')
return redirect(url_for('admin.view_form', form_id=form.id))
# Get form submissions
submissions = Submission.query.filter_by(form_config_id=form.id).order_by(
Submission.created_at.desc()
).limit(10).all()
return render_template('admin/forms/view.html',
user=user,
form=form,
tenant=tenant,
submissions=submissions,
dev_mode=current_app.config.get('DEV_MODE', False),
is_admin=user.is_admin())
@admin_bp.route('/forms/<form_id>/delete', methods=['POST'])
@login_required
def delete_form(form_id):
"""Delete a form."""
user = get_current_user()
form = FormConfig.query.get(form_id)
if not form:
abort(404, description='Form not found')
tenant = Tenant.query.get(form.tenant_id)
if not tenant:
abort(404, description='Tenant not found')
# Check permissions
if user.role != UserRole.ADMIN and tenant not in user.tenants:
abort(403, description='Access denied')
# Delete form (cascade will handle submissions)
db.session.delete(form)
db.session.commit()
# Audit log
AuditLog.log_action(
action='delete',
resource_type='form',
user=user,
tenant=tenant,
resource_id=form.id,
details={'name': form.name}
)
flash('Form deleted successfully!', 'success')
return redirect(url_for('admin.list_forms'))
@admin_bp.route('/submissions/')
@login_required
def list_submissions():
"""List form submissions."""
user = get_current_user()
tenant = get_current_tenant()
# Get submissions based on user access
if user.role == UserRole.ADMIN:
submissions = Submission.query.join(FormConfig).order_by(
Submission.created_at.desc()
).all()
elif tenant:
submissions = Submission.query.filter_by(tenant_id=tenant.id).order_by(
Submission.created_at.desc()
).all()
else:
submissions = []
return render_template('admin/submissions/list.html',
user=user,
submissions=submissions,
tenant=tenant,
dev_mode=current_app.config.get('DEV_MODE', False),
is_admin=user.is_admin())
@admin_bp.route('/submissions/<submission_id>')
@login_required
def view_submission(submission_id):
"""View submission details."""
user = get_current_user()
submission = Submission.query.get(submission_id)
if not submission:
abort(404, description='Submission not found')
# Check permissions
form = FormConfig.query.get(submission.form_config_id)
tenant = Tenant.query.get(submission.tenant_id)
if not form or not tenant:
abort(404, description='Related form or tenant not found')
if user.role != UserRole.ADMIN and tenant not in user.tenants:
abort(403, description='Access denied')
return render_template('admin/submissions/view.html',
user=user,
submission=submission,
form=form,
tenant=tenant,
dev_mode=current_app.config.get('DEV_MODE', False),
is_admin=user.is_admin())
@admin_bp.route('/submissions/<submission_id>/status', methods=['POST'])
@login_required
def update_submission_status(submission_id):
"""Update submission status."""
user = get_current_user()
submission = Submission.query.get(submission_id)
if not submission:
abort(404, description='Submission not found')
tenant = Tenant.query.get(submission.tenant_id)
if not tenant:
abort(404, description='Tenant not found')
if user.role != UserRole.ADMIN and tenant not in user.tenants:
abort(403, description='Access denied')
status = request.form.get('status')
if status in [s.value for s in SubmissionStatus]:
submission.status = SubmissionStatus(status)
db.session.commit()
# Audit log
AuditLog.log_action(
action='update_status',
resource_type='submission',
user=user,
tenant=tenant,
resource_id=submission.id,
details={'status': status}
)
flash('Submission status updated!', 'success')
return redirect(url_for('admin.view_submission', submission_id=submission.id))
@admin_bp.route('/users/')
@admin_required
def list_users():
"""List all users (admin only)."""
user = get_current_user()
users = User.query.order_by(User.created_at.desc()).all()
return render_template('admin/users/list.html',
user=user,
users=users,
dev_mode=current_app.config.get('DEV_MODE', False))
@admin_bp.route('/users/<user_id>', methods=['GET', 'POST'])
@admin_required
def view_user(user_id):
"""View and edit a user (admin only)."""
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':
# Update user
target_user.email = request.form.get('email', target_user.email)
target_user.name = request.form.get('name', target_user.name)
role = request.form.get('role')
if role in [r.value for r in UserRole]:
target_user.role = UserRole(role)
db.session.commit()
# Audit log
AuditLog.log_action(
action='update',
resource_type='user',
user=user,
resource_id=target_user.id,
details={'email': target_user.email, 'role': target_user.role.value}
)
flash('User updated successfully!', 'success')
return redirect(url_for('admin.view_user', user_id=target_user.id))
# Get user tenants
tenants = target_user.tenants
return render_template('admin/users/view.html',
user=user,
target_user=target_user,
tenants=tenants,
dev_mode=current_app.config.get('DEV_MODE', False),
UserRole=UserRole)
@admin_bp.route('/users/<user_id>/delete', methods=['POST'])
@admin_required
def delete_user(user_id):
"""Delete a user (admin only)."""
user = get_current_user()
target_user = User.query.get(user_id)
if not target_user:
abort(404, description='User not found')
# Don't allow deleting self
if target_user.id == user.id:
flash('Cannot delete your own account!', 'error')
return redirect(url_for('admin.view_user', user_id=target_user.id))
db.session.delete(target_user)
db.session.commit()
# Audit log
AuditLog.log_action(
action='delete',
resource_type='user',
user=user,
resource_id=target_user.id,
details={'email': target_user.email}
)
flash('User deleted successfully!', 'success')
return redirect(url_for('admin.list_users'))
@admin_bp.route('/profile')
@login_required
def profile():
"""User profile page."""
user = get_current_user()
if not user:
return redirect(url_for('auth.login'))
return render_template('admin/profile.html',
user=user,
dev_mode=current_app.config.get('DEV_MODE', False))
@admin_bp.route('/api/recaptcha-config')
@login_required
def get_recaptcha_config():
"""Get reCAPTCHA configuration for client-side rendering."""
recaptcha_service = RecaptchaService()
return jsonify({
'enabled': recaptcha_service.is_enabled(),
'site_key': recaptcha_service.get_site_key(),
'dev_mode': recaptcha_service.is_dev_mode(),
'mock_score': recaptcha_service.get_dev_mode_score() if recaptcha_service.is_dev_mode() else None
})
@admin_bp.route('/settings')
@admin_required
def settings():
"""Application settings (admin only)."""
user = get_current_user()
# Get system statistics
tenant_count = Tenant.query.count()
user_count = User.query.count()
form_count = FormConfig.query.count()
submission_count = Submission.query.count()
# Get audit logs
audit_logs = AuditLog.query.order_by(AuditLog.created_at.desc()).limit(50).all()
return render_template('admin/settings.html',
user=user,
tenant_count=tenant_count,
user_count=user_count,
form_count=form_count,
submission_count=submission_count,
audit_logs=audit_logs,
dev_mode=current_app.config.get('DEV_MODE', False))