diff --git a/backend/workflows/tests/test_account_ui.py b/backend/workflows/tests/test_account_ui.py
index 9716e2e..86b19fa 100644
--- a/backend/workflows/tests/test_account_ui.py
+++ b/backend/workflows/tests/test_account_ui.py
@@ -1,7 +1,9 @@
from django.contrib.auth import get_user_model
from django.test import Client, TestCase
+from django.utils import timezone
from workflows.models import UserProfile
+from workflows.totp import generate_totp_token
class AccountUISmokeTests(TestCase):
@@ -55,3 +57,59 @@ class AccountUISmokeTests(TestCase):
self.assertEqual(self.user.email, 'updated@example.com')
self.assertEqual(profile.phone_number, '030 123456')
self.assertEqual(profile.job_title, 'IT Manager')
+
+ def test_totp_can_be_enabled_from_account(self):
+ response = self.client.post(
+ '/account/',
+ {
+ 'account_form': 'totp_enable',
+ 'current_password': 'secret-12345',
+ 'verification_code': '000000',
+ },
+ HTTP_HOST='localhost',
+ follow=True,
+ )
+ self.assertEqual(response.status_code, 200)
+ self.user.refresh_from_db()
+ profile = self.user.profile
+ pending_secret = self.client.session.get('account_totp_pending_secret')
+ self.assertTrue(pending_secret)
+ valid_code = generate_totp_token(pending_secret, int(timezone.now().timestamp()))
+
+ response = self.client.post(
+ '/account/',
+ {
+ 'account_form': 'totp_enable',
+ 'current_password': 'secret-12345',
+ 'verification_code': valid_code,
+ },
+ HTTP_HOST='localhost',
+ follow=True,
+ )
+ self.assertEqual(response.status_code, 200)
+ profile.refresh_from_db()
+ self.assertTrue(profile.totp_enabled)
+ self.assertTrue(profile.totp_secret)
+
+ def test_login_requires_totp_when_enabled(self):
+ profile = self.user.profile
+ profile.totp_secret = 'JBSWY3DPEHPK3PXP'
+ profile.totp_enabled = True
+ profile.save(update_fields=['totp_secret', 'totp_enabled', 'updated_at'])
+
+ client = Client()
+ response = client.post(
+ '/accounts/login/',
+ {'username': 'profile-user', 'password': 'secret-12345'},
+ HTTP_HOST='localhost',
+ )
+ self.assertEqual(response.status_code, 200)
+ self.assertContains(response, 'TOTP-Code')
+
+ token = generate_totp_token(profile.totp_secret, int(timezone.now().timestamp()))
+ response = client.post(
+ '/accounts/login/',
+ {'username': 'profile-user', 'password': 'secret-12345', 'otp_code': token},
+ HTTP_HOST='localhost',
+ )
+ self.assertEqual(response.status_code, 302)
diff --git a/backend/workflows/totp.py b/backend/workflows/totp.py
new file mode 100644
index 0000000..a15063b
--- /dev/null
+++ b/backend/workflows/totp.py
@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import secrets
+import struct
+from urllib.parse import quote
+
+
+def generate_totp_secret(length: int = 20) -> str:
+ return base64.b32encode(secrets.token_bytes(length)).decode('ascii').rstrip('=')
+
+
+def normalize_totp_token(value: str | None) -> str:
+ return ''.join(ch for ch in (value or '').strip() if ch.isdigit())
+
+
+def _secret_bytes(secret: str) -> bytes:
+ padded = secret.strip().replace(' ', '').upper()
+ padding = '=' * ((8 - len(padded) % 8) % 8)
+ return base64.b32decode(padded + padding, casefold=True)
+
+
+def generate_totp_token(secret: str, for_time: int, *, digits: int = 6, period: int = 30) -> str:
+ counter = int(for_time // period)
+ key = _secret_bytes(secret)
+ msg = struct.pack('>Q', counter)
+ digest = hmac.new(key, msg, hashlib.sha1).digest()
+ offset = digest[-1] & 0x0F
+ code_int = struct.unpack('>I', digest[offset:offset + 4])[0] & 0x7FFFFFFF
+ return str(code_int % (10**digits)).zfill(digits)
+
+
+def verify_totp_token(secret: str, token: str, *, for_time: int, digits: int = 6, period: int = 30, window: int = 1) -> bool:
+ normalized = normalize_totp_token(token)
+ if len(normalized) != digits:
+ return False
+ for offset in range(-window, window + 1):
+ candidate_time = for_time + (offset * period)
+ if generate_totp_token(secret, candidate_time, digits=digits, period=period) == normalized:
+ return True
+ return False
+
+
+def build_otpauth_uri(secret: str, *, account_name: str, issuer: str) -> str:
+ label = quote(f'{issuer}:{account_name}')
+ issuer_q = quote(issuer)
+ return f'otpauth://totp/{label}?secret={secret}&issuer={issuer_q}&algorithm=SHA1&digits=6&period=30'
diff --git a/backend/workflows/views.py b/backend/workflows/views.py
index 6b9611a..695f488 100644
--- a/backend/workflows/views.py
+++ b/backend/workflows/views.py
@@ -24,7 +24,7 @@ from django.utils.translation import gettext as _, gettext_lazy
from django.utils.translation import get_language, override
from django.urls import reverse
-from .app_registry import build_portal_app_sections, get_portal_app_registry_rows
+from .app_registry import build_portal_app_sections, get_portal_app_registry_rows, normalize_portal_app_sort_orders
from .backup_ops import (
create_backup_bundle,
delete_backup_bundle,
@@ -33,7 +33,7 @@ from .backup_ops import (
verify_backup_bundle,
)
from .branding import get_branding_email_copy, get_company_email_domain, get_default_notification_templates, get_portal_trial_config, is_trial_expired
-from .forms import AccountAvatarForm, AccountDetailsForm, OffboardingRequestForm, OnboardingRequestForm, PortalBrandingForm, PortalCompanyConfigForm, PortalTrialConfigForm, UserManagementCreateForm
+from .forms import AccountAvatarForm, AccountDetailsForm, AccountTOTPDisableForm, AccountTOTPEnableForm, OffboardingRequestForm, OnboardingRequestForm, PortalBrandingForm, PortalCompanyConfigForm, PortalTrialConfigForm, UserManagementCreateForm
from .form_builder import (
DEFAULT_FIELD_ORDER,
LOCKED_FIELD_RULES,
@@ -46,6 +46,7 @@ from .models import AdminAuditLog, AsyncTaskLog, EmployeeProfile, FormFieldConfi
from .emailing import send_system_email
from .roles import ROLE_GROUP_NAMES, ROLE_LABELS, ROLE_PLATFORM_OWNER, ROLE_SUPER_ADMIN, assign_user_role, get_user_role_key, get_user_role_label, user_has_capability
from .services import get_email_test_redirect, is_email_test_mode, is_nextcloud_enabled, upload_to_nextcloud
+from .totp import build_otpauth_uri, generate_totp_secret
from .tasks import (
_generate_onboarding_intro_pdf,
_generate_onboarding_intro_session_pdf,
@@ -128,10 +129,22 @@ def healthz(request):
@login_required
def account_profile_page(request):
+ session_secret_key = 'account_totp_pending_secret'
profile, created = UserProfile.objects.get_or_create(user=request.user)
+ pending_totp_secret = request.session.get(session_secret_key) or ''
+ if profile.totp_enabled:
+ pending_totp_secret = ''
+ request.session.pop(session_secret_key, None)
+ elif not pending_totp_secret:
+ pending_totp_secret = generate_totp_secret()
+ request.session[session_secret_key] = pending_totp_secret
+
avatar_form = AccountAvatarForm(instance=profile)
details_form = AccountDetailsForm(user=request.user, profile=profile)
+ totp_enable_form = AccountTOTPEnableForm(user=request.user, secret=pending_totp_secret)
+ totp_disable_form = AccountTOTPDisableForm(user=request.user, profile=profile)
account_edit_open = False
+ totp_edit_open = False
if request.method == 'POST':
form_kind = (request.POST.get('account_form') or '').strip()
if form_kind == 'avatar':
@@ -149,6 +162,28 @@ def account_profile_page(request):
messages.success(request, _('Profildaten gespeichert.'))
return redirect('account_profile_page')
messages.error(request, _('Profildaten konnten nicht gespeichert werden.'))
+ elif form_kind == 'totp_enable':
+ totp_edit_open = True
+ totp_enable_form = AccountTOTPEnableForm(request.POST, user=request.user, secret=pending_totp_secret)
+ if totp_enable_form.is_valid():
+ profile.enable_totp(pending_totp_secret)
+ request.session.pop(session_secret_key, None)
+ messages.success(request, _('TOTP wurde aktiviert.'))
+ return redirect('account_profile_page')
+ messages.error(request, _('TOTP konnte nicht aktiviert werden.'))
+ elif form_kind == 'totp_disable':
+ totp_edit_open = True
+ totp_disable_form = AccountTOTPDisableForm(request.POST, user=request.user, profile=profile)
+ if totp_disable_form.is_valid():
+ profile.disable_totp()
+ request.session.pop(session_secret_key, None)
+ messages.success(request, _('TOTP wurde deaktiviert.'))
+ return redirect('account_profile_page')
+ messages.error(request, _('TOTP konnte nicht deaktiviert werden.'))
+
+ branding_context = get_branding_email_copy()
+ totp_account_name = (request.user.email or request.user.username or '').strip()
+ totp_issuer = (branding_context.get('portal_title') or branding_context.get('company_name') or 'Workdock').strip()
return render(
request,
'workflows/account_profile.html',
@@ -157,8 +192,13 @@ def account_profile_page(request):
'account_user_profile': profile,
'avatar_form': avatar_form,
'details_form': details_form,
+ 'totp_enable_form': totp_enable_form,
+ 'totp_disable_form': totp_disable_form,
'account_edit_open': account_edit_open,
+ 'totp_edit_open': totp_edit_open,
'role_label': get_user_role_label(request.user),
+ 'totp_pending_secret': pending_totp_secret,
+ 'totp_otpauth_uri': '' if profile.totp_enabled else build_otpauth_uri(pending_totp_secret, account_name=totp_account_name, issuer=totp_issuer),
},
)
@@ -426,6 +466,7 @@ def job_monitor_page(request):
@require_POST
def save_portal_app_registry(request):
rows = get_portal_app_registry_rows()
+ updated_configs = []
for row in rows:
config = row['config']
key = config.key
@@ -448,6 +489,9 @@ def save_portal_app_registry(request):
config.action_label_override = (request.POST.get(f'action_label_override__{key}') or '').strip()
config.action_label_override_en = (request.POST.get(f'action_label_override_en__{key}') or '').strip()
config.save()
+ updated_configs.append(config)
+
+ normalize_portal_app_sort_orders()
_audit(
request,
@@ -607,6 +651,8 @@ def portal_branding_page(request):
{
'form': form,
'branding': branding,
+ 'branding_sections': _build_branding_sections(form, branding),
+ 'editing_branding_section': '',
},
)
@@ -615,7 +661,18 @@ def portal_branding_page(request):
@require_POST
def save_portal_branding(request):
branding, created = PortalBranding.objects.get_or_create(name='Default')
- form = PortalBrandingForm(request.POST, request.FILES, instance=branding)
+ section_key = (request.POST.get('section_key') or '').strip()
+ data = request.POST.copy()
+ for field_name in PortalBrandingForm.Meta.fields:
+ if field_name not in data:
+ field = PortalBranding._meta.get_field(field_name)
+ if getattr(field, 'many_to_many', False):
+ continue
+ if getattr(field, 'null', False) and getattr(branding, field_name, None) is None:
+ data[field_name] = ''
+ else:
+ data[field_name] = getattr(branding, field_name, '') or ''
+ form = PortalBrandingForm(data, request.FILES, instance=branding)
if not form.is_valid():
messages.error(request, _('Branding konnte nicht gespeichert werden. Bitte prüfen Sie die Eingaben.'))
return render(
@@ -624,6 +681,8 @@ def save_portal_branding(request):
{
'form': form,
'branding': branding,
+ 'branding_sections': _build_branding_sections(form, branding),
+ 'editing_branding_section': section_key,
},
status=400,
)
@@ -650,10 +709,76 @@ def save_portal_branding(request):
{
'form': PortalBrandingForm(instance=branding),
'branding': branding,
+ 'branding_sections': _build_branding_sections(PortalBrandingForm(instance=branding), branding),
+ 'editing_branding_section': '',
},
)
+def _build_branding_sections(form, branding):
+ sections = [
+ {
+ 'key': 'identity',
+ 'title': _('Identität'),
+ 'subtitle': _('Titel, Firmenname und zentrale Spracheinstellungen.'),
+ 'fields': ['portal_title', 'company_name', 'company_domain', 'default_language', 'login_subtitle'],
+ 'field_full': {'login_subtitle'},
+ 'hint_map': {
+ 'company_domain': _('Wird für E-Mail-Vorschläge und Domain-bezogene Standardtexte verwendet, z. B. tub.co.'),
+ },
+ },
+ {
+ 'key': 'appearance',
+ 'title': _('Farben & Erscheinungsbild'),
+ 'subtitle': _('Zentrale visuelle Markenwerte und Browser-Icon.'),
+ 'fields': ['primary_color', 'secondary_color', 'logo_image', 'favicon_image'],
+ 'field_full': set(),
+ 'hint_map': {
+ 'logo_image': _('Erlaubte Formate: SVG, PNG, JPG, JPEG, WEBP. Maximal 5 MB.'),
+ 'favicon_image': _('Erlaubte Formate: ICO, PNG, SVG, WEBP. Maximal 2 MB.'),
+ },
+ },
+ {
+ 'key': 'communication',
+ 'title': _('Kommunikation'),
+ 'subtitle': _('Absender, Support und PDF-Branding für ausgehende Kommunikation.'),
+ 'fields': ['support_email', 'sender_display_name', 'pdf_letterhead'],
+ 'field_full': {'pdf_letterhead'},
+ 'hint_map': {
+ 'sender_display_name': _('Wird für ausgehende System-E-Mails als Anzeigename verwendet.'),
+ 'pdf_letterhead': _('Erlaubtes Format: PDF. Maximal 10 MB.'),
+ },
+ },
+ {
+ 'key': 'legal',
+ 'title': _('Footer & Rechtliches'),
+ 'subtitle': _('Gemeinsame Footer-Texte und rechtliche Hinweise für die Shell.'),
+ 'fields': ['footer_text', 'legal_notice', 'footer_text_en', 'legal_notice_en'],
+ 'field_full': {'legal_notice', 'legal_notice_en'},
+ 'hint_map': {},
+ },
+ ]
+ for section in sections:
+ rows = []
+ for field_name in section['fields']:
+ field = form[field_name]
+ value = getattr(branding, field_name, '') or ''
+ is_file = bool(getattr(field.field.widget, 'input_type', '') == 'file')
+ rows.append(
+ {
+ 'name': field_name,
+ 'bound_field': field,
+ 'label': field.label,
+ 'value': value,
+ 'is_file': is_file,
+ 'is_full': field_name in section.get('field_full', set()),
+ 'hint': section.get('hint_map', {}).get(field_name, ''),
+ }
+ )
+ section['rows'] = rows
+ return sections
+
+
@_require_capability('manage_company_config')
def portal_company_config_page(request):
company_config, created = PortalCompanyConfig.objects.get_or_create(name='Default')
@@ -664,6 +789,8 @@ def portal_company_config_page(request):
{
'form': form,
'company_config': company_config,
+ 'company_config_sections': _build_company_config_sections(form, company_config),
+ 'editing_company_section': '',
},
)
@@ -672,7 +799,12 @@ def portal_company_config_page(request):
@require_POST
def save_portal_company_config(request):
company_config, created = PortalCompanyConfig.objects.get_or_create(name='Default')
- form = PortalCompanyConfigForm(request.POST, instance=company_config)
+ section_key = (request.POST.get('section_key') or '').strip()
+ data = request.POST.copy()
+ for field_name in PortalCompanyConfigForm.Meta.fields:
+ if field_name not in data:
+ data[field_name] = getattr(company_config, field_name, '') or ''
+ form = PortalCompanyConfigForm(data, instance=company_config)
if not form.is_valid():
messages.error(request, _('Firmenkonfiguration konnte nicht gespeichert werden. Bitte prüfen Sie die Eingaben.'))
return render(
@@ -681,6 +813,8 @@ def save_portal_company_config(request):
{
'form': form,
'company_config': company_config,
+ 'company_config_sections': _build_company_config_sections(form, company_config),
+ 'editing_company_section': section_key,
},
status=400,
)
@@ -708,10 +842,56 @@ def save_portal_company_config(request):
{
'form': PortalCompanyConfigForm(instance=company_config),
'company_config': company_config,
+ 'company_config_sections': _build_company_config_sections(PortalCompanyConfigForm(instance=company_config), company_config),
+ 'editing_company_section': '',
},
)
+def _build_company_config_sections(form, company_config):
+ sections = [
+ {
+ 'key': 'profile',
+ 'title': _('Firmenprofil'),
+ 'subtitle': _('Rechtlicher Name und zentrale Stammdaten der Firma.'),
+ 'fields': ['legal_company_name', 'phone_number', 'website_url', 'country'],
+ },
+ {
+ 'key': 'address',
+ 'title': _('Adresse & Register'),
+ 'subtitle': _('Anschrift sowie optionale Register- und Steuerangaben.'),
+ 'fields': ['street_address', 'postal_code', 'city', 'registration_number', 'vat_id'],
+ },
+ {
+ 'key': 'contacts',
+ 'title': _('Kontaktpunkte'),
+ 'subtitle': _('Zentrale Ansprechpartner für HR, IT und Operations.'),
+ 'fields': ['hr_contact_email', 'it_contact_email', 'operations_contact_email'],
+ },
+ {
+ 'key': 'public',
+ 'title': _('Recht & Öffentlichkeit'),
+ 'subtitle': _('Öffentliche Links für Website, Impressum und Datenschutz.'),
+ 'fields': ['imprint_url', 'privacy_url'],
+ 'hint': _('Diese Links können später im Portal-Footer oder in öffentlichen Seiten verwendet werden.'),
+ },
+ ]
+ for section in sections:
+ rows = []
+ for field_name in section['fields']:
+ field = form[field_name]
+ rows.append(
+ {
+ 'name': field_name,
+ 'bound_field': field,
+ 'label': field.label,
+ 'value': getattr(company_config, field_name, '') or '',
+ }
+ )
+ section['rows'] = rows
+ return sections
+
+
@_require_capability('manage_trial_lifecycle')
def portal_trial_config_page(request):
trial_config = get_portal_trial_config()