mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 19:26:48 +00:00
961b43a42e
The login page showed its sign-in card to a user who was already signed in, with the navbar squeezed beside it. Both login views, with and without OTP, now send that user to the page in next, or to the default page when next is missing or points elsewhere. Every permission check answers 403 instead of redirecting to the login, so this cannot loop. The page test renders the login page signed out. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
262 lines
8.4 KiB
Python
262 lines
8.4 KiB
Python
import logging
|
|
import os
|
|
|
|
from admin.decorators import superuser_only
|
|
from django.conf import settings
|
|
from django.contrib import messages
|
|
from django.contrib.auth import get_user_model, update_session_auth_hash, login as auth_login
|
|
from django.contrib.auth.decorators import login_not_required, login_required
|
|
from django.contrib.auth.forms import PasswordChangeForm
|
|
from django.core.exceptions import PermissionDenied
|
|
from django.db import transaction
|
|
from django.shortcuts import get_object_or_404, redirect, render
|
|
from django.urls import reverse
|
|
from django.core.cache import cache
|
|
from django.http import Http404, HttpResponseRedirect
|
|
from django.utils.http import url_has_allowed_host_and_scheme
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django.views.decorators.http import require_POST
|
|
|
|
from accounts.apps import admin_password_path
|
|
from instances.models import Instance
|
|
|
|
from accounts.forms import EmailOTPForm, ProfileForm, UserSSHKeyForm
|
|
from accounts.models import User, UserAttributes, UserInstance, UserSSHKey
|
|
|
|
from . import forms
|
|
from .utils import get_existing_totp_device, get_user_totp_device, send_email_with_otp
|
|
from django.contrib.auth.views import LoginView
|
|
from logs.views import addlogmsg
|
|
|
|
class CustomLoginView(LoginView):
|
|
redirect_authenticated_user = True
|
|
|
|
def form_valid(self, form):
|
|
username = form.cleaned_data['username']
|
|
addlogmsg(username, "-", "-", "Logged In")
|
|
auth_login(self.request, form.get_user())
|
|
return HttpResponseRedirect(self.get_success_url())
|
|
|
|
def form_invalid(self, form):
|
|
username = form.cleaned_data['username']
|
|
addlogmsg(username, "-", "-", "Failed Login Attempt")
|
|
return self.render_to_response(self.get_context_data(form=form))
|
|
|
|
def profile(request):
|
|
publickeys = UserSSHKey.objects.filter(user_id=request.user.id)
|
|
profile_form = ProfileForm(request.POST or None, instance=request.user)
|
|
ssh_key_form = UserSSHKeyForm()
|
|
|
|
if profile_form.is_valid():
|
|
profile_form.save()
|
|
messages.success(request, _("Profile updated"))
|
|
return redirect("accounts:profile")
|
|
|
|
return render(
|
|
request,
|
|
"profile.html",
|
|
{
|
|
"publickeys": publickeys,
|
|
"profile_form": profile_form,
|
|
"ssh_key_form": ssh_key_form,
|
|
},
|
|
)
|
|
|
|
|
|
def ssh_key_create(request):
|
|
key_form = UserSSHKeyForm(request.POST or None, user=request.user)
|
|
if key_form.is_valid():
|
|
key_form.save()
|
|
messages.success(request, _("SSH key added"))
|
|
return redirect("accounts:profile")
|
|
|
|
return render(
|
|
request,
|
|
"common/form.html",
|
|
{
|
|
"form": key_form,
|
|
"title": _("Add SSH key"),
|
|
},
|
|
)
|
|
|
|
|
|
def ssh_key_delete(request, pk):
|
|
ssh_key = get_object_or_404(UserSSHKey, pk=pk, user=request.user)
|
|
if request.method == "POST":
|
|
ssh_key.delete()
|
|
messages.success(request, _("SSH key deleted"))
|
|
return redirect("accounts:profile")
|
|
|
|
return render(
|
|
request,
|
|
"common/confirm_delete.html",
|
|
{
|
|
"object": ssh_key,
|
|
"title": _("Delete SSH key"),
|
|
},
|
|
)
|
|
|
|
|
|
@superuser_only
|
|
def account(request, user_id):
|
|
user = User.objects.get(id=user_id)
|
|
user_insts = UserInstance.objects.filter(user_id=user_id)
|
|
instances = Instance.objects.all().order_by("name")
|
|
publickeys = UserSSHKey.objects.filter(user_id=user_id)
|
|
|
|
return render(
|
|
request,
|
|
"account.html",
|
|
{
|
|
"user": user,
|
|
"user_insts": user_insts,
|
|
"instances": instances,
|
|
"publickeys": publickeys,
|
|
"otp_enabled": settings.OTP_ENABLED,
|
|
},
|
|
)
|
|
|
|
|
|
@login_required
|
|
def change_password(request):
|
|
must_change = UserAttributes.objects.filter(
|
|
user=request.user, must_change_password=True
|
|
).exists()
|
|
if not must_change and not request.user.has_perm("accounts.change_password"):
|
|
raise PermissionDenied
|
|
form = PasswordChangeForm(request.user, request.POST or None)
|
|
|
|
if form.is_valid():
|
|
with transaction.atomic():
|
|
user = form.save()
|
|
UserAttributes.objects.filter(user=user).update(must_change_password=False)
|
|
update_session_auth_hash(request, user)
|
|
request.session.pop("must_change_password", None)
|
|
if must_change and not UserAttributes.objects.filter(must_change_password=True).exists():
|
|
try:
|
|
os.remove(admin_password_path())
|
|
except FileNotFoundError:
|
|
pass
|
|
except OSError:
|
|
logging.getLogger(__name__).warning("Could not remove the generated admin password file")
|
|
messages.success(request, _("Password Changed"))
|
|
return redirect("accounts:profile")
|
|
|
|
return render(request, "accounts/change_password_form.html", {"form": form})
|
|
|
|
|
|
@superuser_only
|
|
def user_instance_create(request, user_id):
|
|
user = get_object_or_404(User, pk=user_id)
|
|
|
|
form = forms.UserInstanceForm(request.POST or None, initial={"user": user})
|
|
if form.is_valid():
|
|
form.save()
|
|
return redirect(reverse("accounts:account", args=[user.id]))
|
|
|
|
return render(
|
|
request,
|
|
"common/form.html",
|
|
{
|
|
"form": form,
|
|
"title": _("Create User Instance"),
|
|
},
|
|
)
|
|
|
|
|
|
@superuser_only
|
|
def user_instance_update(request, pk):
|
|
user_instance = get_object_or_404(UserInstance, pk=pk)
|
|
form = forms.UserInstanceForm(request.POST or None, instance=user_instance)
|
|
if form.is_valid():
|
|
form.save()
|
|
return redirect(reverse("accounts:account", args=[user_instance.user.id]))
|
|
|
|
return render(
|
|
request,
|
|
"common/form.html",
|
|
{
|
|
"form": form,
|
|
"title": _("Update User Instance"),
|
|
},
|
|
)
|
|
|
|
|
|
@superuser_only
|
|
def user_instance_delete(request, pk):
|
|
user_instance = get_object_or_404(UserInstance, pk=pk)
|
|
if request.method == "POST":
|
|
user = user_instance.user
|
|
user_instance.delete()
|
|
next_url = request.GET.get("next", None)
|
|
if next_url and url_has_allowed_host_and_scheme(
|
|
url=next_url,
|
|
allowed_hosts={request.get_host()},
|
|
require_https=request.is_secure(),
|
|
):
|
|
return redirect(next_url)
|
|
return redirect(reverse("accounts:account", args=[user.id]))
|
|
|
|
return render(
|
|
request,
|
|
"common/confirm_delete.html",
|
|
{"object": user_instance},
|
|
)
|
|
|
|
|
|
# a user gets at most one OTP mail per this many seconds
|
|
EMAIL_OTP_INTERVAL = 300
|
|
|
|
|
|
@login_not_required
|
|
def email_otp(request):
|
|
"""
|
|
"Lost OTP?": mail the OTP QR code to the user with this address. Open to
|
|
anyone, so it mails only one exact, active match, at most once per
|
|
EMAIL_OTP_INTERVAL, and answers the same in every case.
|
|
"""
|
|
if not settings.OTP_ENABLED:
|
|
raise Http404
|
|
form = EmailOTPForm(request.POST or None)
|
|
if form.is_valid():
|
|
users = list(get_user_model().objects.filter(email__iexact=form.cleaned_data["email"], is_active=True)[:2])
|
|
# the cache is per process: with several workers the bound is per worker
|
|
if len(users) == 1 and cache.add(f"email_otp:{users[0].pk}", True, EMAIL_OTP_INTERVAL):
|
|
user = users[0]
|
|
# resend only to a user who already enrolled: creating a device here
|
|
# would let anyone who knows an email force OTP on that account
|
|
device = get_existing_totp_device(user)
|
|
if device is not None:
|
|
try:
|
|
send_email_with_otp(user, device)
|
|
except Exception:
|
|
logging.getLogger(__name__).exception("Could not mail the OTP QR code to user %s", user.pk)
|
|
|
|
messages.success(
|
|
request, _("OTP Sent to %(email)s") % {"email": form.cleaned_data["email"]}
|
|
)
|
|
return redirect("accounts:login")
|
|
|
|
return render(
|
|
request,
|
|
"accounts/email_otp_form.html",
|
|
{
|
|
"form": form,
|
|
"title": _("Email OTP"),
|
|
},
|
|
)
|
|
|
|
|
|
@require_POST
|
|
@superuser_only
|
|
def admin_email_otp(request, user_id):
|
|
user = get_object_or_404(get_user_model(), pk=user_id)
|
|
device = get_user_totp_device(user)
|
|
if user.email != "":
|
|
send_email_with_otp(user, device)
|
|
messages.success(
|
|
request, _("OTP QR code was emailed to user %(user)s") % {"user": user}
|
|
)
|
|
else:
|
|
messages.error(request, _("User email not set, failed to send QR code"))
|
|
return redirect("accounts:account", user.id)
|