bostame/workdock-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

snapshot: preserve role management and user lifecycle controls

Browse Source
This commit is contained in:
Md Bayazid Bostame
2026-03-26 10:07:49 +01:00
parent 438334bd92
commit b585287004
17 changed files with 1137 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

344
backend/workflows/views.py
View File

@@ -2,6 +2,7 @@ from pathlib import Path
from datetime import timedelta
from tempfile import NamedTemporaryFile
import json
from functools import wraps
from celery import current_app
from django.conf import settings
@@ -10,16 +11,21 @@ from django.db import IntegrityError
from django.db.models import Q
from django.shortcuts import get_object_or_404, redirect, render
from django.contrib import messages
from django.contrib.auth.decorators import login_required, user_passes_test
from django.contrib.auth import get_user_model
from django.contrib.auth.decorators import login_required
from django.contrib.auth.tokens import default_token_generator
from django.http import JsonResponse
from django.views.decorators.http import require_POST
from django.views.decorators.csrf import ensure_csrf_cookie
from django.utils import timezone
from django.utils.encoding import force_bytes
from django.utils.http import urlsafe_base64_encode
from django.utils.translation import gettext as _, gettext_lazy
from django.utils.translation import get_language, override
from django.urls import reverse
from .backup_ops import create_backup_bundle, delete_backup_bundle, list_backup_bundles, verify_backup_bundle
from .forms import OffboardingRequestForm, OnboardingRequestForm
from .forms import OffboardingRequestForm, OnboardingRequestForm, UserManagementCreateForm
from .form_builder import (
DEFAULT_FIELD_ORDER,
LOCKED_FIELD_RULES,
@@ -30,6 +36,7 @@ from .form_builder import (
)
from .models import AdminAuditLog, EmployeeProfile, FormFieldConfig, FormOption, IntroChecklistItem, NotificationRule, NotificationTemplate, OffboardingRequest, OnboardingIntroductionSession, OnboardingRequest, ScheduledWelcomeEmail, SystemEmailConfig, WorkflowConfig
from .emailing import send_system_email
from .roles import ROLE_GROUP_NAMES, ROLE_LABELS, 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 .tasks import (
DEFAULT_NOTIFICATION_TEMPLATES,
@@ -112,8 +119,19 @@ def healthz(request):
)
def _is_staff(user) -> bool:
return user.is_authenticated and user.is_staff
def _require_capability(capability: str):
def decorator(view_func):
@wraps(view_func)
@login_required
def wrapped(request, *args, **kwargs):
if not user_has_capability(request.user, capability):
messages.error(request, _('Sie haben keine Berechtigung für diese Aktion.'))
return redirect('home')
return view_func(request, *args, **kwargs)
return wrapped
return decorator
def _display_user_name(user) -> str:
@@ -218,6 +236,10 @@ def _audit_action_label(action: str) -> str:
'mail_settings_saved': _('Mail-Einstellungen gespeichert'),
'email_routing_saved': _('E-Mail-Routing gespeichert'),
'notification_rules_saved': _('Benachrichtigungsregeln gespeichert'),
'user_created': _('Benutzer erstellt'),
'user_updated': _('Benutzer aktualisiert'),
'user_password_reset_sent': _('Passwort-Reset-Link versendet'),
'user_deleted': _('Benutzer gelöscht'),
'backup_created': _('Backup erstellt'),
'backup_verified': _('Backup verifiziert'),
'backup_deleted': _('Backup gelöscht'),
@@ -311,36 +333,220 @@ def home(request):
'nextcloud_enabled': is_nextcloud_enabled(),
'email_test_mode': is_email_test_mode(),
'workflow_config': config,
'role_label': get_user_role_label(request.user),
},
)
@login_required
@user_passes_test(_is_staff)
def _user_management_rows():
user_model = get_user_model()
role_order = {
ROLE_SUPER_ADMIN: 0,
'admin': 1,
'it_staff': 2,
'staff': 3,
}
rows = []
for user in user_model.objects.all().order_by('-is_active', 'username'):
role_key = get_user_role_key(user)
rows.append(
{
'user': user,
'role_key': role_key,
'role_label': str(ROLE_LABELS[role_key]),
'role_sort': role_order.get(role_key, 99),
'display_name': _display_user_name(user),
}
)
rows.sort(key=lambda item: (not item['user'].is_active, item['role_sort'], item['user'].username.lower()))
return rows
def _render_user_management(request, create_form=None, status_code: int = 200):
return render(
request,
'workflows/user_management.html',
{
'create_form': create_form or UserManagementCreateForm(),
'rows': _user_management_rows(),
'role_choices': [(key, str(ROLE_LABELS[key])) for key in ROLE_GROUP_NAMES],
},
status=status_code,
)
def _super_admin_user_count() -> int:
user_model = get_user_model()
return sum(1 for user in user_model.objects.all() if get_user_role_key(user) == ROLE_SUPER_ADMIN and user.is_active)
def _would_remove_last_super_admin(user, new_role_key: str | None = None, new_is_active: bool | None = None, deleting: bool = False) -> bool:
if get_user_role_key(user) != ROLE_SUPER_ADMIN or not user.is_active:
return False
if _super_admin_user_count() > 1:
return False
if deleting:
return True
if new_role_key is not None and new_role_key != ROLE_SUPER_ADMIN:
return True
if new_is_active is not None and not new_is_active:
return True
return False
@_require_capability('manage_users')
def user_management_page(request):
return _render_user_management(request)
@_require_capability('manage_users')
@require_POST
def create_user_from_admin(request):
form = UserManagementCreateForm(request.POST)
if not form.is_valid():
messages.error(request, _('Benutzer konnte nicht erstellt werden. Bitte prüfen Sie die Eingaben.'))
return _render_user_management(request, create_form=form, status_code=400)
user = form.save()
_audit(
request,
'user_created',
target_type='user',
target_id=user.id,
target_label=_display_user_name(user),
details={'username': user.username, 'role': get_user_role_key(user)},
)
messages.success(request, _('Benutzer wurde erstellt: %(username)s') % {'username': user.username})
return redirect('user_management_page')
@_require_capability('manage_users')
@require_POST
def update_user_from_admin(request, user_id: int):
user_model = get_user_model()
target_user = get_object_or_404(user_model, id=user_id)
role_key = (request.POST.get('role_key') or '').strip()
is_active = request.POST.get('is_active') == 'on'
new_password = (request.POST.get('new_password') or '').strip()
if role_key not in ROLE_GROUP_NAMES:
messages.error(request, _('Ungültige Rolle.'))
return redirect('user_management_page')
if target_user == request.user and (role_key != ROLE_SUPER_ADMIN or not is_active):
messages.error(request, _('Der aktuell angemeldete Super Admin kann sich hier nicht selbst sperren oder herabstufen.'))
return redirect('user_management_page')
if _would_remove_last_super_admin(target_user, new_role_key=role_key, new_is_active=is_active):
messages.error(request, _('Der letzte aktive Super Admin kann nicht deaktiviert oder herabgestuft werden.'))
return redirect('user_management_page')
assign_user_role(target_user, role_key)
target_user.is_active = is_active
if new_password:
target_user.set_password(new_password)
target_user.save()
_audit(
request,
'user_updated',
target_type='user',
target_id=target_user.id,
target_label=_display_user_name(target_user),
details={'username': target_user.username, 'role': role_key, 'is_active': is_active, 'password_changed': bool(new_password)},
)
messages.success(request, _('Benutzer wurde aktualisiert: %(username)s') % {'username': target_user.username})
return redirect('user_management_page')
@_require_capability('manage_users')
@require_POST
def send_password_reset_from_admin(request, user_id: int):
user_model = get_user_model()
target_user = get_object_or_404(user_model, id=user_id)
email = (target_user.email or '').strip()
if not email:
messages.error(request, _('Für diesen Benutzer ist keine E-Mail-Adresse hinterlegt.'))
return redirect('user_management_page')
uid = urlsafe_base64_encode(force_bytes(target_user.pk))
token = default_token_generator.make_token(target_user)
reset_path = reverse('password_reset_confirm', kwargs={'uidb64': uid, 'token': token})
reset_url = request.build_absolute_uri(reset_path)
send_system_email(
subject=_('Passwort zurücksetzen für %(username)s') % {'username': target_user.username},
body=_(
'Hallo %(name)s,\n\n'
'für Ihr Konto wurde ein Link zum Zurücksetzen des Passworts erstellt.\n'
'Bitte öffnen Sie den folgenden Link:\n'
'%(url)s\n\n'
'Wenn Sie diese Anfrage nicht erwartet haben, können Sie diese E-Mail ignorieren.'
) % {
'name': _display_user_name(target_user),
'url': reset_url,
},
to=[email],
)
_audit(
request,
'user_password_reset_sent',
target_type='user',
target_id=target_user.id,
target_label=_display_user_name(target_user),
details={'username': target_user.username, 'email': email},
)
messages.success(request, _('Passwort-Reset-Link wurde versendet: %(username)s') % {'username': target_user.username})
return redirect('user_management_page')
@_require_capability('manage_users')
@require_POST
def delete_user_from_admin(request, user_id: int):
user_model = get_user_model()
target_user = get_object_or_404(user_model, id=user_id)
if target_user == request.user:
messages.error(request, _('Der aktuell angemeldete Super Admin kann sich hier nicht selbst löschen.'))
return redirect('user_management_page')
if _would_remove_last_super_admin(target_user, deleting=True):
messages.error(request, _('Der letzte aktive Super Admin kann nicht gelöscht werden.'))
return redirect('user_management_page')
target_label = _display_user_name(target_user)
username = target_user.username
target_user.delete()
_audit(
request,
'user_deleted',
target_type='user',
target_label=target_label,
details={'username': username},
)
messages.success(request, _('Benutzer wurde gelöscht: %(username)s') % {'username': username})
return redirect('user_management_page')
@_require_capability('view_docs')
def handbook_page(request):
return render(request, 'workflows/handbook.html')
@login_required
@user_passes_test(_is_staff)
@_require_capability('view_docs')
def project_wiki_page(request):
return render(request, 'workflows/project_wiki.html')
@login_required
@user_passes_test(_is_staff)
@_require_capability('view_docs')
def developer_handbook_page(request):
return render(request, 'workflows/developer_handbook.html')
@login_required
@user_passes_test(_is_staff)
@_require_capability('view_docs')
def release_checklist_page(request):
return render(request, 'workflows/release_checklist.html')
@login_required
@user_passes_test(_is_staff)
@_require_capability('view_audit_log')
def audit_log_page(request):
action = (request.GET.get('action') or '').strip()
user_query = (request.GET.get('user') or '').strip()
@@ -380,8 +586,7 @@ def audit_log_page(request):
)
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_backups')
def backup_recovery_page(request):
return render(
request,
@@ -392,8 +597,7 @@ def backup_recovery_page(request):
)
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_backups')
@require_POST
def create_backup_from_admin(request):
try:
@@ -411,8 +615,7 @@ def create_backup_from_admin(request):
return redirect('backup_recovery_page')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_backups')
@require_POST
def verify_backup_from_admin(request, backup_name: str):
try:
@@ -430,8 +633,7 @@ def verify_backup_from_admin(request, backup_name: str):
return redirect('backup_recovery_page')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_backups')
@require_POST
def delete_backup_from_admin(request, backup_name: str):
try:
@@ -449,8 +651,7 @@ def delete_backup_from_admin(request, backup_name: str):
return redirect('backup_recovery_page')
@login_required
@user_passes_test(_is_staff)
@_require_capability('access_requests_dashboard')
def request_timeline_page(request, kind: str, request_id: int):
if kind == 'onboarding':
obj = get_object_or_404(OnboardingRequest, id=request_id)
@@ -569,8 +770,12 @@ def request_timeline_page(request, kind: str, request_id: int):
@login_required
def requests_dashboard(request):
if not user_has_capability(request.user, 'access_requests_dashboard'):
messages.error(request, _('Sie haben keine Berechtigung für diese Aktion.'))
return redirect('home')
if request.method == 'POST':
if not request.user.is_staff:
if not user_has_capability(request.user, 'delete_requests'):
messages.error(request, _('Sie haben keine Berechtigung für diese Aktion.'))
return redirect('requests_dashboard')
@@ -765,6 +970,13 @@ def requests_dashboard(request):
{'value': 'failed', 'label': _request_status_label('failed', language_code)},
]
has_filters = any([search_query, type_filter, status_filter, department_filter, date_from, date_to])
column_count = 4
if user_has_capability(request.user, 'delete_requests'):
column_count += 1
if user_has_capability(request.user, 'run_intro_session') or user_has_capability(request.user, 'generate_intro_pdfs'):
column_count += 1
if user_has_capability(request.user, 'access_requests_dashboard'):
column_count += 1
return render(
request,
'workflows/requests_dashboard.html',
@@ -779,6 +991,7 @@ def requests_dashboard(request):
'departments': departments,
'status_choices': status_choices,
'has_filters': has_filters,
'column_count': column_count,
'onboarding_total': onboarding_total,
'offboarding_total': offboarding_total,
'combined_total': onboarding_total + offboarding_total,
@@ -838,8 +1051,7 @@ def onboarding_success(request, request_id: int):
return render(request, 'workflows/onboarding_success.html', {'obj': obj, 'pdf_url': pdf_url})
@login_required
@user_passes_test(_is_staff)
@_require_capability('generate_intro_pdfs')
@require_POST
def generate_onboarding_intro_pdf(request, request_id: int):
obj = get_object_or_404(OnboardingRequest, id=request_id)
@@ -851,8 +1063,7 @@ def generate_onboarding_intro_pdf(request, request_id: int):
return redirect('requests_dashboard')
@login_required
@user_passes_test(_is_staff)
@_require_capability('generate_intro_pdfs')
@require_POST
def generate_onboarding_intro_session_pdf(request, request_id: int):
onboarding = get_object_or_404(OnboardingRequest, id=request_id)
@@ -869,8 +1080,7 @@ def generate_onboarding_intro_session_pdf(request, request_id: int):
return redirect('onboarding_intro_session_page', request_id=request_id)
@login_required
@user_passes_test(_is_staff)
@_require_capability('run_intro_session')
def onboarding_intro_session_page(request, request_id: int):
onboarding = get_object_or_404(OnboardingRequest, id=request_id)
session, _created = OnboardingIntroductionSession.objects.get_or_create(onboarding_request=onboarding)
@@ -1024,8 +1234,7 @@ def offboarding_success(request, request_id: int):
return render(request, 'workflows/offboarding_success.html', {'obj': obj, 'pdf_url': pdf_url})
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_builders')
def form_builder_page(request):
language_code = get_language()
form_type = request.GET.get('form_type', 'onboarding')
@@ -1200,8 +1409,7 @@ def form_builder_page(request):
)
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_builders')
def intro_builder_page(request):
if request.method == 'POST':
delete_id = (request.POST.get('delete_item_id') or '').strip()
@@ -1311,8 +1519,7 @@ def intro_builder_page(request):
)
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_integrations')
def integrations_setup_page(request):
config, _ = WorkflowConfig.objects.get_or_create(name='Default')
kind = (request.GET.get('kind') or 'nextcloud').strip().lower()
@@ -1342,8 +1549,7 @@ def integrations_setup_page(request):
)
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_welcome_emails')
def welcome_emails_page(request):
rows = ScheduledWelcomeEmail.objects.select_related('onboarding_request').order_by('-send_at', '-id')[:200]
config, _ = WorkflowConfig.objects.get_or_create(name='Default')
@@ -1373,8 +1579,7 @@ def welcome_emails_page(request):
)
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_welcome_emails')
@require_POST
def trigger_welcome_email_now(request, schedule_id: int):
scheduled = ScheduledWelcomeEmail.objects.filter(id=schedule_id).first()
@@ -1395,8 +1600,7 @@ def trigger_welcome_email_now(request, schedule_id: int):
return redirect('welcome_emails_page')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_welcome_emails')
@require_POST
def save_welcome_email_settings(request):
config, _ = WorkflowConfig.objects.get_or_create(name='Default')
@@ -1498,8 +1702,7 @@ def _parse_selected_schedule_ids(raw: str) -> list[int]:
return parsed
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_welcome_emails')
@require_POST
def bulk_welcome_email_action(request):
action = (request.POST.get('bulk_action') or '').strip().lower()
@@ -1569,8 +1772,7 @@ def bulk_welcome_email_action(request):
return redirect('welcome_emails_page')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_welcome_emails')
@require_POST
def pause_welcome_email(request, schedule_id: int):
scheduled = ScheduledWelcomeEmail.objects.filter(id=schedule_id).first()
@@ -1589,8 +1791,7 @@ def pause_welcome_email(request, schedule_id: int):
return redirect('welcome_emails_page')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_welcome_emails')
@require_POST
def resume_welcome_email(request, schedule_id: int):
scheduled = ScheduledWelcomeEmail.objects.filter(id=schedule_id).first()
@@ -1612,8 +1813,7 @@ def resume_welcome_email(request, schedule_id: int):
return redirect('welcome_emails_page')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_welcome_emails')
@require_POST
def cancel_welcome_email(request, schedule_id: int):
scheduled = ScheduledWelcomeEmail.objects.filter(id=schedule_id).first()
@@ -1633,8 +1833,7 @@ def cancel_welcome_email(request, schedule_id: int):
return redirect('welcome_emails_page')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_builders')
@require_POST
def form_builder_save_order(request):
try:
@@ -1698,8 +1897,7 @@ def form_builder_save_order(request):
return JsonResponse({'ok': True, 'saved_count': len(configs)})
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_integrations')
@require_POST
def send_test_email(request):
mode = 'TEST_MODE_ON' if is_email_test_mode() else 'TEST_MODE_OFF'
@@ -1718,8 +1916,7 @@ def send_test_email(request):
return _redirect_back(request, 'home')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_integrations')
@require_POST
def nextcloud_test_upload(request):
filename = f"nextcloud_test_{timezone.now().strftime('%Y%m%d_%H%M%S')}.txt"
@@ -1751,8 +1948,7 @@ def nextcloud_test_upload(request):
return _redirect_back(request, 'home')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_integrations')
@require_POST
def toggle_nextcloud_enabled(request):
config, _ = WorkflowConfig.objects.get_or_create(name='Default')
@@ -1766,8 +1962,7 @@ def toggle_nextcloud_enabled(request):
return _redirect_back(request, 'home')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_integrations')
@require_POST
def toggle_email_mode(request):
config, _ = WorkflowConfig.objects.get_or_create(name='Default')
@@ -1781,8 +1976,7 @@ def toggle_email_mode(request):
return _redirect_back(request, 'home')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_integrations')
@require_POST
def save_integrations_settings(request):
config, _ = WorkflowConfig.objects.get_or_create(name='Default')
@@ -1820,8 +2014,7 @@ def save_integrations_settings(request):
return redirect('home')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_integrations')
@require_POST
def save_nextcloud_settings(request):
config, _ = WorkflowConfig.objects.get_or_create(name='Default')
@@ -1846,8 +2039,7 @@ def save_nextcloud_settings(request):
return redirect('/admin-tools/integrations/?kind=nextcloud')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_integrations')
@require_POST
def save_workflow_rules(request):
config, _ = WorkflowConfig.objects.get_or_create(name='Default')
@@ -1877,8 +2069,7 @@ def save_workflow_rules(request):
return redirect('/admin-tools/integrations/?kind=rules')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_integrations')
@require_POST
def save_backup_settings(request):
config, _ = WorkflowConfig.objects.get_or_create(name='Default')
@@ -1930,8 +2121,7 @@ def save_backup_settings(request):
return redirect('/admin-tools/integrations/?kind=backup')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_integrations')
@require_POST
def save_mail_settings(request):
config, _ = WorkflowConfig.objects.get_or_create(name='Default')
@@ -1971,8 +2161,7 @@ def save_mail_settings(request):
return redirect('/admin-tools/integrations/?kind=mail')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_integrations')
@require_POST
def save_email_routing_settings(request):
config, _ = WorkflowConfig.objects.get_or_create(name='Default')
@@ -2039,8 +2228,7 @@ def save_email_routing_settings(request):
return redirect('/admin-tools/integrations/?kind=emails')
@login_required
@user_passes_test(_is_staff)
@_require_capability('manage_integrations')
@require_POST
def save_notification_rules(request):
rule_ids = request.POST.getlist('rule_ids')
@@ -2103,8 +2291,7 @@ def save_notification_rules(request):
return redirect('/admin-tools/integrations/?kind=emails')
@login_required
@user_passes_test(_is_staff)
@_require_capability('delete_requests')
@require_POST
def delete_request_from_dashboard(request, kind: str, request_id: int):
if kind == 'onboarding':
@@ -2122,8 +2309,7 @@ def delete_request_from_dashboard(request, kind: str, request_id: int):
return redirect('requests_dashboard')
@login_required
@user_passes_test(_is_staff)
@_require_capability('retry_requests')
@require_POST
def retry_request_from_dashboard(request, kind: str, request_id: int):
if kind == 'onboarding':