from datetime import timedelta from django.conf import settings from django.contrib import messages from django.contrib.auth import authenticate, login, logout from django.contrib.auth.decorators import login_required from django.contrib.auth.models import Group from django.core.mail import send_mail from django.core.paginator import Paginator from django.db.models import Q from django.shortcuts import get_object_or_404, redirect, render from django.utils import timezone from django.views.decorators.http import require_POST from core.audit import create_audit_log from core.decorators import capability_required from core.permissions import Capability, ROLE_CAPABILITIES, Role, capability_label from .forms import ( ChangePasswordForm, EmployeeForm, GroupForm, LoginForm, UserCreateForm, UserEditForm, UserPasswordResetForm, ) from .models import Employee, GroupCapability, User from .rbac import ROLE_GROUP_PREFIX, SYSTEM_ROLES, ensure_role_groups, role_group_name, sync_user_role_group PROVISIONED_CREDENTIALS_SESSION_KEY = "provisioned_credentials" PROVISIONED_CREDENTIALS_TTL_HOURS = 24 def _provisioned_credentials_bucket(request): bucket = request.session.get(PROVISIONED_CREDENTIALS_SESSION_KEY, {}) if not isinstance(bucket, dict): return {} return bucket def _cleanup_provisioned_credentials(request): bucket = _provisioned_credentials_bucket(request) if not bucket: return {} cutoff_ts = (timezone.now() - timedelta(hours=PROVISIONED_CREDENTIALS_TTL_HOURS)).timestamp() cleaned = {} changed = False for key, payload in bucket.items(): created_at_ts = payload.get("created_at_ts") if not created_at_ts: changed = True continue if created_at_ts < cutoff_ts: changed = True continue cleaned[key] = payload if changed: request.session[PROVISIONED_CREDENTIALS_SESSION_KEY] = cleaned request.session.modified = True return cleaned def _store_provisioned_credentials(request, user, temporary_password): bucket = _cleanup_provisioned_credentials(request) bucket[str(user.pk)] = { "username": user.username, "temporary_password": temporary_password, "created_at_ts": timezone.now().timestamp(), } request.session[PROVISIONED_CREDENTIALS_SESSION_KEY] = bucket request.session.modified = True def _get_provisioned_credentials(request, user_id, *, pop=False): bucket = _cleanup_provisioned_credentials(request) key = str(user_id) payload = bucket.get(key) if pop and key in bucket: bucket.pop(key, None) request.session[PROVISIONED_CREDENTIALS_SESSION_KEY] = bucket request.session.modified = True return payload def login_view(request): if request.user.is_authenticated: if request.user.must_change_password: return redirect('accounts:profile') return redirect('finance:dashboard') if request.method == 'POST': form = LoginForm(request.POST) if form.is_valid(): username = form.cleaned_data['username'] password = form.cleaned_data['password'] user_obj = User.objects.filter(username=username).first() if user_obj and user_obj.is_locked: messages.error( request, "Account is temporarily locked due to failed attempts. " f"Try again after {user_obj.locked_until:%Y-%m-%d %H:%M UTC}.", ) return render(request, 'accounts/login.html', {'form': form}) user = authenticate(request, username=username, password=password) if user: if not user.is_active: messages.error(request, 'Your account is inactive. Contact administrator.') elif user.is_locked: messages.error( request, "Account is temporarily locked. " f"Try again after {user.locked_until:%Y-%m-%d %H:%M UTC}.", ) else: user.reset_failed_login_state() login(request, user) create_audit_log('login', user, metadata={'source': 'web'}) if user.must_change_password: messages.warning( request, 'You must change your password before continuing.', ) return redirect('accounts:profile') return redirect(request.GET.get('next', 'finance:dashboard')) else: if user_obj and user_obj.is_active: user_obj.register_failed_login() user_obj.refresh_from_db(fields=["locked_until"]) if user_obj.is_locked: messages.error( request, "Too many failed attempts. " f"Account locked until {user_obj.locked_until:%Y-%m-%d %H:%M UTC}.", ) else: messages.error(request, 'Invalid username or password.') else: messages.error(request, 'Invalid username or password.') else: form = LoginForm() return render(request, 'accounts/login.html', {'form': form}) def logout_view(request): if request.user.is_authenticated: create_audit_log('logout', request.user, metadata={'source': 'web'}) logout(request) return redirect('accounts:login') @capability_required(Capability.USERS_MANAGE) def user_list(request): users = User.objects.prefetch_related('groups').order_by('username') query = (request.GET.get("q") or "").strip() role = (request.GET.get("role") or "").strip() status = (request.GET.get("status") or "").strip() if query: users = users.filter( Q(username__icontains=query) | Q(first_name__icontains=query) | Q(last_name__icontains=query) | Q(email__icontains=query) | Q(contact__icontains=query) ) if role: users = users.filter(role=role) if status == "active": users = users.filter(is_active=True) elif status == "inactive": users = users.filter(is_active=False) paginator = Paginator(users, 20) page = request.GET.get("page") users = paginator.get_page(page) return render( request, 'accounts/user_list.html', { 'users': users, 'query': query, 'current_role': role, 'current_status': status, 'role_choices': User.assignable_role_choices(), }, ) @capability_required(Capability.USERS_MANAGE) def user_create(request): if request.method == 'POST': form = UserCreateForm(request.POST) if form.is_valid(): user = form.save() sync_user_role_group(user) _store_provisioned_credentials(request, user, form.created_password) create_audit_log( 'create', user, new_values={ 'username': user.username, 'role': user.role, 'is_active': user.is_active, }, ) messages.success(request, f"User {user.username} created successfully.") return redirect("accounts:user_credentials", pk=user.pk) else: initial = {} employee_id = (request.GET.get("employee") or "").strip() if employee_id.isdigit(): employee = Employee.objects.filter( pk=int(employee_id), is_active=True, user__isnull=True ).first() if employee: initial = { "employee": employee.pk, "first_name": employee.first_name, "last_name": employee.last_name, "email": employee.email, "contact": employee.phone, "role": employee.role, } form = UserCreateForm(initial=initial) return render( request, 'accounts/user_form.html', { 'form': form, 'title': 'Create User Account', 'auto_username': True, }, ) @capability_required(Capability.USERS_MANAGE) def user_edit(request, pk): user = get_object_or_404(User, pk=pk) if request.method == 'POST': form = UserEditForm(request.POST, instance=user) if form.is_valid(): updated_user = form.save() sync_user_role_group(updated_user) create_audit_log( 'update', updated_user, metadata={'changed_by': request.user.username}, ) messages.success(request, 'User updated.') return redirect('accounts:user_list') else: form = UserEditForm(instance=user) return render( request, 'accounts/user_form.html', { 'form': form, 'title': f'Edit User: {user.username}', 'user_obj': user, 'auto_username': False, }, ) @capability_required(Capability.USERS_MANAGE) @require_POST def user_toggle(request, pk): user = get_object_or_404(User, pk=pk) if user == request.user: messages.warning(request, 'You cannot deactivate your own account.') return redirect('accounts:user_list') user.is_active = not user.is_active user.save(update_fields=["is_active"]) status = 'activated' if user.is_active else 'deactivated' create_audit_log( 'update', user, metadata={'account_status': status, 'changed_by': request.user.username}, ) messages.success(request, f'{user.username} has been {status}.') return redirect('accounts:user_list') @capability_required(Capability.USERS_MANAGE) def user_reset_password(request, pk): user = get_object_or_404(User, pk=pk) if request.method == "POST": form = UserPasswordResetForm(request.POST) if form.is_valid(): new_password = form.cleaned_data["new_password"] user.set_password(new_password) user.must_change_password = form.cleaned_data["must_change_password"] user.reset_failed_login_state() user.save(update_fields=["password", "must_change_password"]) _store_provisioned_credentials(request, user, new_password) create_audit_log( 'update', user, metadata={'password_reset_by': request.user.username}, ) messages.success(request, f'Password reset for {user.username}.') return redirect("accounts:user_credentials", pk=user.pk) else: form = UserPasswordResetForm() return render( request, "accounts/user_reset_password.html", {"form": form, "user_obj": user}, ) @capability_required(Capability.USERS_MANAGE) def user_credentials(request, pk): user = get_object_or_404(User, pk=pk) credential = _get_provisioned_credentials(request, user.pk) if request.method == "POST": action = request.POST.get("action") if action == "send_email": if not credential: messages.error(request, "No temporary credential is available for this user.") elif not user.email: messages.error(request, "This user does not have an email address.") else: subject = "Your BizTrack account credentials" message = ( "Hello,\n\n" "An account has been created for you in BizTrack.\n\n" f"Username: {credential['username']}\n" f"Temporary Password: {credential['temporary_password']}\n\n" "For security, you must change this password immediately after your first login.\n" ) from_email = getattr(settings, "DEFAULT_FROM_EMAIL", "no-reply@biztrack.local") try: send_mail( subject, message, from_email, [user.email], fail_silently=False, ) except Exception as exc: messages.error(request, f"Unable to send email: {exc}") else: messages.success(request, f"Credentials sent to {user.email}.") elif action == "clear": _get_provisioned_credentials(request, user.pk, pop=True) messages.success(request, "Credential slip cleared from this session.") return redirect("accounts:user_list") return redirect("accounts:user_credentials", pk=user.pk) return render( request, "accounts/user_credentials.html", {"user_obj": user, "credential": credential}, ) @capability_required(Capability.USERS_MANAGE) def role_list(request): ensure_role_groups() role_rows = [] for role in SYSTEM_ROLES: caps = sorted(ROLE_CAPABILITIES.get(role, set())) group_name = role_group_name(role) group = Group.objects.filter(name=group_name).first() role_rows.append( { "code": role, "label": role.replace("_", " ").title(), "group_name": group_name, "group_members": group.user_set.count() if group else 0, "capabilities": [{"code": cap, "label": capability_label(cap)} for cap in caps], } ) return render(request, "accounts/role_list.html", {"roles": role_rows}) @capability_required(Capability.USERS_MANAGE) @require_POST def role_groups_sync(request): groups = ensure_role_groups() for user in User.objects.all(): sync_user_role_group(user) messages.success( request, f"Role groups synchronized successfully ({len(groups)} groups).", ) return redirect("accounts:role_list") @capability_required(Capability.USERS_MANAGE) def group_list(request): query = (request.GET.get("q") or "").strip() groups = Group.objects.exclude(name__startswith=ROLE_GROUP_PREFIX).order_by("name") if query: groups = groups.filter(name__icontains=query) rows = [] for group in groups.prefetch_related("user_set"): rows.append( { "id": group.id, "name": group.name, "users_count": group.user_set.count(), "capabilities_count": GroupCapability.objects.filter(group=group).count(), } ) return render( request, "accounts/group_list.html", {"groups": rows, "query": query}, ) @capability_required(Capability.USERS_MANAGE) def group_create(request): if request.method == "POST": form = GroupForm(request.POST) if form.is_valid(): group = form.save() create_audit_log('create', request.user, metadata={'group': group.name}) messages.success(request, f"Group {group.name} created.") return redirect("accounts:group_list") else: form = GroupForm() return render( request, "accounts/group_form.html", {"form": form, "title": "Create Group"}, ) @capability_required(Capability.USERS_MANAGE) def group_edit(request, pk): group = get_object_or_404(Group, pk=pk) if group.name.startswith(ROLE_GROUP_PREFIX): messages.error(request, "System role groups are managed automatically.") return redirect("accounts:group_list") if request.method == "POST": form = GroupForm(request.POST, instance=group) if form.is_valid(): form.save() create_audit_log('update', request.user, metadata={'group': group.name}) messages.success(request, f"Group {group.name} updated.") return redirect("accounts:group_list") else: form = GroupForm(instance=group) current_capabilities = GroupCapability.objects.filter(group=group).order_by("capability") return render( request, "accounts/group_form.html", { "form": form, "title": f"Edit Group: {group.name}", "group_obj": group, "current_capabilities": current_capabilities, }, ) @capability_required(Capability.USERS_MANAGE) def group_delete(request, pk): group = get_object_or_404(Group, pk=pk) if group.name.startswith(ROLE_GROUP_PREFIX): messages.error(request, "System role groups cannot be deleted.") return redirect("accounts:group_list") if request.method == "POST": group_name = group.name group.delete() create_audit_log('delete', request.user, metadata={'group': group_name}) messages.success(request, f"Group {group_name} deleted.") return redirect("accounts:group_list") @capability_required(Capability.USERS_MANAGE) def employee_list(request): query = (request.GET.get("q") or "").strip() employees = Employee.objects.select_related("user").order_by("first_name", "last_name") if query: employees = employees.filter( Q(code__icontains=query) | Q(first_name__icontains=query) | Q(last_name__icontains=query) | Q(email__icontains=query) | Q(phone__icontains=query) ) return render( request, "accounts/employee_list.html", {"employees": employees, "query": query}, ) @capability_required(Capability.USERS_MANAGE) def employee_create(request): if request.method == "POST": form = EmployeeForm(request.POST) if form.is_valid(): employee = form.save() create_audit_log( "create", request.user, metadata={"employee": employee.code}, ) messages.success(request, f"Employee {employee.full_name} created.") return redirect("accounts:employee_list") else: form = EmployeeForm() return render( request, "accounts/employee_form.html", {"form": form, "title": "Create Employee"}, ) @capability_required(Capability.USERS_MANAGE) def employee_edit(request, pk): employee = get_object_or_404(Employee, pk=pk) if request.method == "POST": form = EmployeeForm(request.POST, instance=employee) if form.is_valid(): form.save() create_audit_log( "update", request.user, metadata={"employee": employee.code}, ) messages.success(request, f"Employee {employee.full_name} updated.") return redirect("accounts:employee_list") else: form = EmployeeForm(instance=employee) return render( request, "accounts/employee_form.html", {"form": form, "title": f"Edit Employee: {employee.full_name}"}, ) @login_required def profile(request): if request.method == 'POST': form = ChangePasswordForm(request.POST, user=request.user) if form.is_valid(): request.user.set_password(form.cleaned_data['new_password']) request.user.must_change_password = False request.user.reset_failed_login_state() request.user.save(update_fields=['password', 'must_change_password']) create_audit_log( 'update', request.user, metadata={'profile_password_change': True}, ) messages.success(request, 'Password changed. Please log in again.') return redirect('accounts:login') else: form = ChangePasswordForm(user=request.user) if request.user.must_change_password: messages.warning( request, "Password change is required before continuing.", ) return render(request, 'accounts/profile.html', {'form': form})