import uuid from django.conf import settings from django.contrib import messages from django.contrib.auth import authenticate, get_user_model, login, logout, update_session_auth_hash from django.contrib.auth.decorators import login_required from django.contrib.auth.forms import PasswordChangeForm from django.contrib.auth.views import ( PasswordResetCompleteView, PasswordResetConfirmView, PasswordResetDoneView, PasswordResetView, ) from django.core.mail import send_mail from django.core.signing import BadSignature, SignatureExpired, TimestampSigner from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse, reverse_lazy from django.utils.http import url_has_allowed_host_and_scheme from django.views.decorators.http import require_http_methods, require_POST from apps.core.permissions.roles import UserRole from apps.core.authentication.forms import ( ExperienceEntryForm, IILMPPasswordResetForm, LoginForm, MFAEnrollForm, MFARemoveForm, MFAVerifyForm, ProfileForm, ProfileVisibilityForm, QualificationEntryForm, RegisterForm, ) from apps.core.authentication.models import MFADevice, ProfileVisibility, UserProfile from apps.core.authentication.profile_completion import profile_completion from apps.core.authentication.totp import new_secret, provisioning_uri, qr_code_data_uri, verify_totp from apps.core.audit.rep_helpers import action_create, action_update, audit_rep User = get_user_model() signer = TimestampSigner() # Used after MFA step — user loaded from DB has no .backend; multiple AUTHENTICATION_BACKENDS require explicit backend. MFA_LOGIN_BACKEND = "apps.core.authentication.backends.EmailBackend" SESSION_MFA_ENROLL_SECRET = "mfa_totp_enroll_secret" MFA_VERIFY_FAILS_KEY = "mfa_verify_failures" MFA_MAX_VERIFY_ATTEMPTS = 5 def _send_verification_email(user, request) -> None: token = signer.sign(str(user.pk)) verify_url = request.build_absolute_uri( reverse("accounts:verify_email", kwargs={"token": token}) ) try: from django.template.loader import render_to_string from apps.core.notifications.emailing import portal_base_url plain = ( "Welcome to RUFORUM IILMP. Open this link to verify your email:\n\n" f"{verify_url}\n\nIf you did not request this, ignore this message." ) html = render_to_string( "notifications/emails/generic.html", { "notification": None, "recipient_name": user.get_full_name() or user.email.split("@")[0], "headline": "Confirm your email", "preheader": "Verify your email to finish registration.", "body": ( "Welcome to IILMP. Use the button below to verify your email address.\n\n" "If you did not request this, you can ignore this message." ), "body_html": "", "action_url": verify_url, "action_label": "Verify email", "portal_base": portal_base_url(), "preferences_path": reverse("core:notification_inbox"), }, ) send_mail( subject="Verify your RUFORUM IILMP account", message=plain, from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[user.email], html_message=html, fail_silently=True, ) except Exception: pass def _role_based_redirect_url(user) -> str: """Return the landing URL for a user based on their role after login.""" role = getattr(user, "role", "") or "" if role == UserRole.PROGRAM_DIRECTOR: return reverse("rims_grants:director_dashboard") if role == UserRole.GRANTS_MANAGER: return reverse("rims_grants:manager_call_list") return settings.LOGIN_REDIRECT_URL def _safe_next(request, value=None) -> str: """Return a same-host ``next`` destination, or "" if missing/unsafe. Needed for OAuth SSO (Moodle): the provider sends users to ``/accounts/login/?next=/o/authorize/...`` and expects them bounced back to the authorize endpoint after signing in. Validated so it can't be abused as an open redirect. """ next_url = value if value is not None else ( request.POST.get("next") or request.GET.get("next") or "" ) if next_url and not url_has_allowed_host_and_scheme( next_url, allowed_hosts={request.get_host()}, require_https=request.is_secure() ): return "" return next_url def _post_login_redirect(request, user) -> str: """Destination after a completed login (incl. via MFA). Honours a ``next`` stashed in the session by ``login_view`` before the MFA hop — revalidated here so a stale/unsafe value can never redirect off-host — otherwise falls back to the role-based landing page. """ next_url = _safe_next(request, request.session.pop("post_login_next", "") or "") return next_url or _role_based_redirect_url(user) @require_http_methods(["GET", "POST"]) def login_view(request): next_url = _safe_next(request) if request.user.is_authenticated: return redirect(next_url or settings.LOGIN_REDIRECT_URL) form = LoginForm(request.POST or None) template = "authentication/login.html" if request.method == "POST" and form.is_valid(): user = authenticate( request, email=form.cleaned_data["email"], password=form.cleaned_data["password"], ) if user is not None: if not user.is_active: form.add_error(None, "This account is inactive.") elif not user.email_verified: # FRREP-UR005 — activate the account only on link confirmation. # Accounts predating this gate were backfilled verified; only # genuinely-unverified new registrations land here. form.add_error( None, "Please verify your email before signing in — check your inbox, " "or resend the verification link below.", ) else: pending_mfa = ( MFADevice.objects.filter( user=user, is_verified=True, device_type=MFADevice.DeviceType.TOTP, ) .exclude(secret="") .exists() ) if pending_mfa: request.session.pop(MFA_VERIFY_FAILS_KEY, None) request.session["mfa_user_id"] = user.pk request.session["post_login_next"] = next_url return redirect("accounts:mfa_verify") login(request, user) return redirect(next_url or _role_based_redirect_url(user)) else: form.add_error(None, "Invalid email or password.") if getattr(request, "htmx", False) and request.method == "POST": return render( request, template, {"form": form, "next": next_url}, status=400 if form.errors else 200, ) return render(request, template, {"form": form, "next": next_url}) def logout_view(request): logout(request) return redirect("accounts:login") @require_http_methods(["GET", "POST"]) def register_view(request): if request.user.is_authenticated: return redirect(settings.LOGIN_REDIRECT_URL) # Pre-select Applicant role when coming from a RIMS call/grant/scholarship page next_url = request.GET.get("next", "") initial = {} if any(kw in next_url for kw in ["/rims/", "/grants/", "/scholarships/", "/calls/"]): initial["role"] = "applicant" form = RegisterForm(request.POST or None, initial=initial) if request.method == "POST" and form.is_valid(): user = form.save() # WS2.4 / FRREP-UR013 — register events must be auditable. audit_rep( actor=user, action=action_create(), target_model="CustomUser", object_id=user.pk, object_repr=user.email, changes={ "email_verified": False, "institution_id": user.institution_id, "role": user.role or "", }, request=request, ) _send_verification_email(user, request) messages.success( request, "Registration successful. Check your email for a verification link.", ) return redirect("accounts:login") return render(request, "authentication/register.html", {"form": form}) @require_http_methods(["GET", "POST"]) def resend_verification_view(request): """FRREP-UR005 A3 — resend an expired/lost verification link. Always shows the same confirmation regardless of whether the email matches an account, to avoid leaking which emails are registered. """ sent = False if request.method == "POST": email = (request.POST.get("email") or "").strip().lower() user = User.objects.filter(email__iexact=email, is_active=True, email_verified=False).first() if user: _send_verification_email(user, request) sent = True return render(request, "authentication/resend_verification.html", {"sent": sent}) def verify_email_view(request, token): try: uid = signer.unsign(token, max_age=60 * 60 * 24 * 7) user = User.objects.get(pk=int(uid)) user.email_verified = True user.save(update_fields=["email_verified"]) # WS2.4 / FRREP-UR013 — capture the verification step. audit_rep( actor=user, action=action_update(), target_model="CustomUser", object_id=user.pk, object_repr=user.email, changes={"email_verified": True}, request=request, ) # FRREP-UR006 — nudge newly verified users toward finishing their # professional profile (qualifications, background, interests, # country, bio) once they're signed in. messages.success( request, "Email verified. You can sign in. Once you're in, take a moment to " "complete your professional profile — it helps us match you with " "relevant opportunities.", ) except (BadSignature, SignatureExpired, User.DoesNotExist, ValueError): messages.error(request, "Invalid or expired verification link.") return redirect("accounts:login") @require_http_methods(["GET", "POST"]) def mfa_verify_view(request): uid = request.session.get("mfa_user_id") if not uid: return redirect("accounts:login") user = User.objects.filter(pk=uid, is_active=True).first() if not user: request.session.pop("mfa_user_id", None) request.session.pop(MFA_VERIFY_FAILS_KEY, None) request.session.pop("post_login_next", None) return redirect("accounts:login") failures = int(request.session.get(MFA_VERIFY_FAILS_KEY, 0)) if failures >= MFA_MAX_VERIFY_ATTEMPTS: messages.error(request, "Too many incorrect codes. Sign in again.") request.session.pop("mfa_user_id", None) request.session.pop(MFA_VERIFY_FAILS_KEY, None) request.session.pop("post_login_next", None) return redirect("accounts:login") devices = list( MFADevice.objects.filter( user=user, is_verified=True, device_type=MFADevice.DeviceType.TOTP, ).exclude(secret="") ) if not devices: request.session.pop("mfa_user_id", None) request.session.pop(MFA_VERIFY_FAILS_KEY, None) login(request, user, backend=MFA_LOGIN_BACKEND) return redirect(_post_login_redirect(request, user)) form = MFAVerifyForm(request.POST or None) if request.method == "POST" and form.is_valid(): code = form.cleaned_data["code"] if any(verify_totp(d.secret, code) for d in devices): request.session.pop(MFA_VERIFY_FAILS_KEY, None) login(request, user, backend=MFA_LOGIN_BACKEND) del request.session["mfa_user_id"] return redirect(_post_login_redirect(request, user)) request.session[MFA_VERIFY_FAILS_KEY] = failures + 1 request.session.modified = True remaining = MFA_MAX_VERIFY_ATTEMPTS - failures - 1 err = "Invalid code." if remaining > 0: err += f" {remaining} attempt(s) left." form.add_error("code", err) return render(request, "authentication/mfa_verify.html", {"form": form}) @login_required @require_http_methods(["GET", "POST"]) def mfa_manage_view(request): devices = request.user.mfa_devices.filter(is_verified=True).order_by("-created_at") return render( request, "authentication/mfa_manage.html", {"devices": devices}, ) @login_required @require_http_methods(["GET", "POST"]) def password_change_view(request): """Self-service password change for a signed-in user. SRS General Functional Requirement — the platform must let users manage their own account, "including changing passwords". Previously the only path to a new password was the emailed reset flow; this closes that gap. Uses Django's ``PasswordChangeForm`` (verifies the current password and enforces the configured validators), keeps the session alive via ``update_session_auth_hash``, and writes an audit entry. """ form = PasswordChangeForm(user=request.user, data=request.POST or None) if request.method == "POST" and form.is_valid(): form.save() # Rotate the session hash so the user isn't logged out by the change. update_session_auth_hash(request, form.user) audit_rep( actor=request.user, action=action_update(), target_model="User", object_id=request.user.pk, object_repr=request.user.email, changes={"changed_fields": ["password"]}, request=request, ) messages.success(request, "Your password has been changed.") return redirect("accounts:profile") return render(request, "authentication/password_change.html", {"form": form}) @login_required @require_http_methods(["GET", "POST"]) def mfa_enroll_view(request): if request.method == "GET" and request.GET.get("restart"): request.session.pop(SESSION_MFA_ENROLL_SECRET, None) secret = request.session.get(SESSION_MFA_ENROLL_SECRET) if not secret: secret = new_secret() request.session[SESSION_MFA_ENROLL_SECRET] = secret request.session.modified = True otpauth_uri = provisioning_uri(secret, request.user.email) qr_data_uri = qr_code_data_uri(otpauth_uri) form = MFAEnrollForm(request.POST or None) if request.method == "POST" and form.is_valid(): if verify_totp(secret, form.cleaned_data["code"]): name = (form.cleaned_data.get("device_name") or "").strip() or "Authenticator app" MFADevice.objects.create( user=request.user, device_type=MFADevice.DeviceType.TOTP, name=name, secret=secret, is_verified=True, ) del request.session[SESSION_MFA_ENROLL_SECRET] messages.success(request, "Authenticator app registered. It will be required at next sign-in.") return redirect("accounts:mfa_manage") form.add_error("code", "Invalid code. Check the time on your phone and try again.") return render( request, "authentication/mfa_enroll.html", { "form": form, "qr_data_uri": qr_data_uri, "otpauth_uri": otpauth_uri, "secret_key": secret, }, ) @login_required @require_POST def mfa_remove_view(request, pk): device = get_object_or_404( MFADevice, pk=pk, user=request.user, is_verified=True, ) form = MFARemoveForm(request.POST) if form.is_valid() and request.user.check_password(form.cleaned_data["password"]): device.delete() messages.success(request, "Authenticator removed from your account.") elif form.is_valid(): messages.error(request, "Incorrect password.") else: messages.error(request, "Could not remove device. Check the form and try again.") return redirect("accounts:mfa_manage") class IILMPPasswordResetView(PasswordResetView): form_class = IILMPPasswordResetForm template_name = "authentication/password_reset.html" email_template_name = "authentication/emails/password_reset.txt" html_email_template_name = "authentication/emails/password_reset.html" success_url = reverse_lazy("accounts:password_reset_done") class IILMPPasswordResetDoneView(PasswordResetDoneView): template_name = "authentication/password_reset_done.html" class IILMPPasswordResetConfirmView(PasswordResetConfirmView): template_name = "authentication/password_reset_confirm.html" success_url = reverse_lazy("accounts:password_reset_complete") class IILMPPasswordResetCompleteView(PasswordResetCompleteView): template_name = "authentication/password_reset_complete.html" @require_http_methods(["GET", "POST"]) def profile_view(request): if not request.user.is_authenticated: return redirect("accounts:login") profile, _ = UserProfile.objects.get_or_create(user=request.user) visibility, _ = ProfileVisibility.objects.get_or_create(user=request.user) # WS5.2 / FRREP-UR012 — visibility section POSTs with a `section` # discriminator. If the discriminator is missing, treat it as a profile # POST so older clients (and existing tests) keep working. if request.method == "POST": section = request.POST.get("section") or "profile" else: section = None form = ProfileForm( request.POST if section == "profile" else None, request.FILES if section == "profile" else None, instance=profile, user=request.user, ) visibility_form = ProfileVisibilityForm( request.POST if section == "visibility" else None, instance=visibility, ) # Optional resume-destination — e.g. an applicant sent here to complete # their profile before applying gets returned to the apply flow instead # of stranded on the profile page. Validated so it can't be abused as an # open redirect. next_url = request.POST.get("next") or request.GET.get("next") or "" if next_url and not url_has_allowed_host_and_scheme( next_url, allowed_hosts={request.get_host()}, require_https=request.is_secure() ): next_url = "" if request.method == "POST": if section == "profile" and form.is_valid(): form.save() # WS2.4 / FRREP-UR013 — profile updates must be auditable. audit_rep( actor=request.user, action=action_update(), target_model="UserProfile", object_id=profile.pk, object_repr=request.user.email, changes={"changed_fields": sorted(form.changed_data)}, request=request, ) messages.success(request, "Profile updated.") return redirect(next_url or "accounts:profile") if section == "visibility" and visibility_form.is_valid(): visibility_form.save() audit_rep( actor=request.user, action=action_update(), target_model="ProfileVisibility", object_id=visibility.pk, object_repr=request.user.email, changes={"changed_fields": sorted(visibility_form.changed_data)}, request=request, ) messages.success(request, "Privacy settings updated.") return redirect("accounts:profile") return render( request, "authentication/profile.html", { "form": form, "visibility_form": visibility_form, "next": next_url, "completion": profile_completion(request.user), }, ) # ── Repeatable profile entries (qualifications / experience) ── # # UserProfile.qualifications / UserProfile.experience are free-form JSON # lists of dicts (see models.py), not separate models — there's no natural # "add another" via a ModelForm/formset here, so each entry gets a stable # uuid4 `id` on creation and these views operate on that list in place. Since # every view below always reads/writes request.user's own UserProfile (never # an id passed in the URL for *which user*), there's no cross-user access to # guard against — `entry_id` only ever selects a dict within the current # user's own list. @login_required @require_POST def qualification_save_view(request): profile, _ = UserProfile.objects.get_or_create(user=request.user) form = QualificationEntryForm(request.POST) if not form.is_valid(): messages.error(request, "Please fix the errors in the qualification form: " + " ".join( f"{f}: {', '.join(errs)}" for f, errs in form.errors.items() )) return redirect("accounts:profile") entry_id = request.POST.get("entry_id") or uuid.uuid4().hex is_edit = bool(request.POST.get("entry_id")) entry = { "id": entry_id, "institution": form.cleaned_data["institution"], "qualification": form.cleaned_data["qualification"], "field_of_study": form.cleaned_data["field_of_study"], "year": form.cleaned_data["year"], } entries = [e for e in profile.qualifications if e.get("id") != entry_id] entries.append(entry) entries.sort(key=lambda e: e.get("year") or 0, reverse=True) profile.qualifications = entries profile.save(update_fields=["qualifications"]) audit_rep( actor=request.user, action=action_update() if is_edit else action_create(), target_model="UserProfile.qualifications", object_id=profile.pk, object_repr=f"{request.user.email} — {entry['qualification']}", changes={"entry": entry}, request=request, ) messages.success(request, "Qualification saved.") return redirect("accounts:profile") @login_required @require_POST def qualification_delete_view(request, entry_id): profile, _ = UserProfile.objects.get_or_create(user=request.user) profile.qualifications = [e for e in profile.qualifications if e.get("id") != entry_id] profile.save(update_fields=["qualifications"]) audit_rep( actor=request.user, action=action_update(), target_model="UserProfile.qualifications", object_id=profile.pk, object_repr=request.user.email, changes={"removed_entry_id": entry_id}, request=request, ) messages.success(request, "Qualification removed.") return redirect("accounts:profile") @login_required @require_POST def experience_save_view(request): profile, _ = UserProfile.objects.get_or_create(user=request.user) form = ExperienceEntryForm(request.POST) if not form.is_valid(): messages.error(request, "Please fix the errors in the experience form: " + " ".join( f"{f}: {', '.join(errs)}" for f, errs in form.errors.items() )) return redirect("accounts:profile") entry_id = request.POST.get("entry_id") or uuid.uuid4().hex is_edit = bool(request.POST.get("entry_id")) entry = { "id": entry_id, "organization": form.cleaned_data["organization"], "title": form.cleaned_data["title"], "start_year": form.cleaned_data["start_year"], "end_year": form.cleaned_data["end_year"], "is_current": form.cleaned_data["is_current"], "description": form.cleaned_data["description"], } entries = [e for e in profile.experience if e.get("id") != entry_id] entries.append(entry) entries.sort(key=lambda e: (e.get("is_current") or False, e.get("start_year") or 0), reverse=True) profile.experience = entries profile.save(update_fields=["experience"]) audit_rep( actor=request.user, action=action_update() if is_edit else action_create(), target_model="UserProfile.experience", object_id=profile.pk, object_repr=f"{request.user.email} — {entry['title']}", changes={"entry": entry}, request=request, ) messages.success(request, "Experience saved.") return redirect("accounts:profile") @login_required @require_POST def experience_delete_view(request, entry_id): profile, _ = UserProfile.objects.get_or_create(user=request.user) profile.experience = [e for e in profile.experience if e.get("id") != entry_id] profile.save(update_fields=["experience"]) audit_rep( actor=request.user, action=action_update(), target_model="UserProfile.experience", object_id=profile.pk, object_repr=request.user.email, changes={"removed_entry_id": entry_id}, request=request, ) messages.success(request, "Experience removed.") return redirect("accounts:profile")