241 lines
8.5 KiB
Python
241 lines
8.5 KiB
Python
# ==============================================================================
|
|
# 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 is_mock_auth_enabled() -> bool:
|
|
"""Check if mock authentication mode is enabled."""
|
|
return current_app.config.get('MOCK_AUTH', False) and current_app.config.get('DEV_MODE', False)
|
|
|
|
@staticmethod
|
|
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
|
|
|
|
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 create_mock_user_session() -> dict:
|
|
"""
|
|
Create a mock authentication user session with local mock user.
|
|
|
|
:return: Dictionary with mock user session data
|
|
"""
|
|
config = current_app.config
|
|
|
|
mock_user = {
|
|
'id': config.get('MOCK_USER_ID', 'mock_user_001'),
|
|
'email': config.get('MOCK_USER_EMAIL', 'mock@formeleon.local'),
|
|
'name': config.get('MOCK_USER_NAME', 'Mock Admin'),
|
|
'role': config.get('MOCK_USER_ROLE', 'admin'),
|
|
'sub': f'mock-{uuid.uuid4()}',
|
|
'tenant_id': config.get('MOCK_TENANT_ID', 'mock_tenant_001'),
|
|
'dev_mode': True,
|
|
'mock_auth': True,
|
|
'authenticated': True
|
|
}
|
|
|
|
return mock_user
|
|
|
|
@staticmethod
|
|
def ensure_mock_user_exists() -> User:
|
|
"""
|
|
Ensure the mock user exists in the database.
|
|
|
|
:return: User object from database
|
|
"""
|
|
config = current_app.config
|
|
|
|
user_id = config.get('MOCK_USER_ID', 'mock_user_001')
|
|
user = User.query.get(user_id)
|
|
|
|
if not user:
|
|
role = UserRole.ADMIN if config.get('MOCK_USER_ROLE', 'admin') == 'admin' else UserRole.USER
|
|
user = User(
|
|
id=user_id,
|
|
email=config.get('MOCK_USER_EMAIL', 'mock@formeleon.local'),
|
|
name=config.get('MOCK_USER_NAME', 'Mock Admin'),
|
|
role=role,
|
|
sub=f'mock-{uuid.uuid4()}'
|
|
)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
logger.info(f'Created mock user: {user.id}')
|
|
|
|
return user
|
|
|
|
@staticmethod
|
|
def ensure_mock_tenant_exists() -> Tenant:
|
|
"""
|
|
Ensure the mock tenant exists in the database.
|
|
|
|
:return: Tenant object from database
|
|
"""
|
|
config = current_app.config
|
|
|
|
tenant_id = config.get('MOCK_TENANT_ID', 'mock_tenant_001')
|
|
tenant = Tenant.query.get(tenant_id)
|
|
|
|
if not tenant:
|
|
tenant = Tenant(
|
|
id=tenant_id,
|
|
name='Mock Tenant',
|
|
public_key=f'mock-key-{uuid.uuid4().hex[:16]}',
|
|
allowed_origins='*'
|
|
)
|
|
db.session.add(tenant)
|
|
db.session.commit()
|
|
logger.info(f'Created mock tenant: {tenant.id}')
|
|
|
|
return tenant
|
|
|
|
@staticmethod
|
|
def process_oidc_callback(user_info: dict) -> Tuple[Optional[User], Optional[str]]:
|
|
"""
|
|
Process user information from OIDC callback, evaluate Zitadel roles,
|
|
and create/update the local user record.
|
|
|
|
:param user_info: User information dictionary from OIDC provider
|
|
: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 or not sub:
|
|
return None, 'Incomplete user info from OIDC provider (missing sub or email).'
|
|
|
|
# 1. Zitadel-Rollen extrahieren
|
|
# Zitadel liefert Projektrollen im Claim 'urn:zitadel:iam:org:project:roles'
|
|
raw_roles = user_info.get('urn:zitadel:iam:org:project:roles', {})
|
|
|
|
if isinstance(raw_roles, dict):
|
|
user_roles = list(raw_roles.keys())
|
|
elif isinstance(raw_roles, list):
|
|
user_roles = raw_roles
|
|
else:
|
|
user_roles = []
|
|
|
|
logger.debug(f"Extracted OIDC roles for {email}: {user_roles}")
|
|
|
|
# 2. Hard Gate: Zugriff verweigern, wenn keine der erlaubt Formeleon-Rollen zugewiesen ist
|
|
has_admin_role = 'form-admins' in user_roles
|
|
has_user_role = 'form-users' in user_roles
|
|
|
|
if not (has_admin_role or has_user_role):
|
|
logger.warning(f"Access denied for {email}. Missing required roles ('form-admins' or 'form-users'). Roles present: {user_roles}")
|
|
return None, 'Access denied: You do not have permission to access Formeleon.'
|
|
|
|
# 3. Formeleon-Rolle festlegen (Admin schlägt User)
|
|
assigned_role = UserRole.ADMIN if has_admin_role else UserRole.USER
|
|
|
|
# 4. User suchen oder neu anlegen
|
|
user = User.query.filter((User.sub == sub) | (User.email == email)).first()
|
|
|
|
if not user:
|
|
user_id = f'user_{email.split("@")[0]}_{str(uuid.uuid4())[:8]}'
|
|
user = User(
|
|
id=user_id,
|
|
email=email,
|
|
name=name,
|
|
role=assigned_role,
|
|
sub=sub
|
|
)
|
|
db.session.add(user)
|
|
logger.info(f'Created new user from OIDC: {user.id} with role {assigned_role.value}')
|
|
else:
|
|
# Re-sync Name, Sub und Rolle bei jedem Login
|
|
user.name = name
|
|
user.sub = sub
|
|
user.email = email
|
|
user.role = assigned_role
|
|
logger.info(f'Updated existing user from OIDC: {user.id} (Role: {assigned_role.value})')
|
|
|
|
db.session.commit()
|
|
return user, None |