mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 11:16:47 +00:00
2cbcb8ab2c
No view used sensitive_post_parameters, so once ADMINS is set the error mail of a failing POST would carry the password or key in plain text. The user create, edit and password views, the password change, the compute forms, the host secrets, the VM's root password and its console settings now mark their POST data sensitive.
381 lines
14 KiB
Python
381 lines
14 KiB
Python
import logging
|
|
import os
|
|
from datetime import timedelta
|
|
|
|
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
|
|
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.core.signing import BadSignature
|
|
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
|
|
from django.utils import timezone
|
|
from django.utils.http import url_has_allowed_host_and_scheme
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django.views.decorators.cache import never_cache
|
|
from django.views.decorators.http import require_POST
|
|
from django.views.decorators.debug import sensitive_post_parameters
|
|
|
|
from accounts.apps import admin_password_path
|
|
from computes.utils import lock_ownership
|
|
from logs.audit import audit, audit_vm, changed_fields, failed, snapshot
|
|
|
|
from accounts.forms import ApiTokenForm, EmailOTPForm, ProfileForm, UserSSHKeyForm
|
|
from accounts.models import ApiToken, User, UserAttributes, UserInstance, UserSSHKey
|
|
|
|
from . import forms
|
|
from .utils import OTP_LINK, OTP_LINK_MAX_AGE, get_existing_totp_device, get_user_totp_device, otp_qr_svg, send_email_with_otp
|
|
|
|
|
|
def profile(request):
|
|
publickeys = UserSSHKey.objects.filter(user_id=request.user.id)
|
|
before = snapshot(request.user)
|
|
profile_form = ProfileForm(request.POST or None, instance=request.user)
|
|
ssh_key_form = UserSSHKeyForm()
|
|
|
|
if profile_form.is_valid():
|
|
with audit(request, None, f"User {request.user.username}", "change") as entry:
|
|
profile_form.save()
|
|
fields = changed_fields(before, snapshot(request.user))
|
|
entry.action = "change " + ", ".join(fields)
|
|
entry.skip = not fields
|
|
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,
|
|
"api_tokens": ApiToken.objects.filter(user=request.user),
|
|
"api_token_form": ApiTokenForm(),
|
|
},
|
|
)
|
|
|
|
|
|
@require_POST
|
|
@never_cache
|
|
def api_token_create(request):
|
|
"""A REST API token, shown once: only its digest is kept. It is made in
|
|
a session that passed the login, and its OTP when OTP is on."""
|
|
form = ApiTokenForm(request.POST)
|
|
if not form.is_valid():
|
|
return render(request, "common/form.html", {"form": form, "title": _("Create an API token")})
|
|
ApiToken.objects.filter(user=request.user, expiry__lt=timezone.now()).delete()
|
|
if ApiToken.objects.filter(user=request.user).count() >= ApiToken.LIMIT:
|
|
form.add_error(None, _("A user has at most %(limit)s API tokens: delete one first.") % {"limit": ApiToken.LIMIT})
|
|
return render(request, "common/form.html", {"form": form, "title": _("Create an API token")})
|
|
name, days = form.cleaned_data["name"], form.cleaned_data["days"]
|
|
with audit(request, None, f"API token {name}", "add"):
|
|
_token, key = ApiToken.objects.create(request.user, expiry=timedelta(days=days) if days else None, name=name)
|
|
# shown by a GET, so reloading the page does not create another token
|
|
request.session["api_token_created"] = {"name": name, "key": key}
|
|
return redirect("accounts:api_token_created")
|
|
|
|
|
|
@never_cache
|
|
def api_token_created(request):
|
|
"""The token just created, shown once."""
|
|
created = request.session.pop("api_token_created", None)
|
|
if created is None:
|
|
return redirect(reverse("accounts:profile") + "#apitokens")
|
|
return render(request, "api_token_created.html", created)
|
|
|
|
|
|
def api_token_delete(request, token_key, user_id=None):
|
|
"""Delete one of the user's own tokens or, given user_id, a superuser
|
|
deletes one of that user's."""
|
|
if user_id is None:
|
|
token = get_object_or_404(ApiToken, token_key=token_key, user=request.user)
|
|
subject = f"API token {token.name}"
|
|
back = reverse("accounts:profile")
|
|
elif request.user.is_superuser:
|
|
token = get_object_or_404(ApiToken, token_key=token_key, user_id=user_id)
|
|
subject = f"API token {token.name} of {token.user.username}"
|
|
back = reverse("accounts:account", args=[user_id])
|
|
else:
|
|
raise PermissionDenied
|
|
if request.method == "POST":
|
|
with audit(request, None, subject, "delete"):
|
|
token.delete()
|
|
messages.success(request, _("API token deleted"))
|
|
return redirect(back + "#apitokens")
|
|
|
|
return render(
|
|
request,
|
|
"common/confirm_delete.html",
|
|
{
|
|
"object": subject,
|
|
"title": _("Delete API token"),
|
|
"consequence": _("Scripts and tools that use it can no longer reach the API."),
|
|
},
|
|
)
|
|
|
|
|
|
def ssh_key_create(request):
|
|
key_form = UserSSHKeyForm(request.POST or None, user=request.user)
|
|
if key_form.is_valid():
|
|
with audit(request, None, f"SSH key {key_form.cleaned_data['keyname']}", "add"):
|
|
key_form.save()
|
|
messages.success(request, _("SSH key added"))
|
|
return redirect(reverse("accounts:profile") + "#sshkeys")
|
|
|
|
return render(
|
|
request,
|
|
"common/form.html",
|
|
{
|
|
"form": key_form,
|
|
"title": _("Add SSH key"),
|
|
},
|
|
)
|
|
|
|
|
|
def ssh_key_delete(request, pk, user_id=None):
|
|
"""Delete one of the user's own keys or, given user_id, a superuser
|
|
deletes one of that user's."""
|
|
if user_id is None:
|
|
ssh_key = get_object_or_404(UserSSHKey, pk=pk, user=request.user)
|
|
subject = f"SSH key {ssh_key.keyname}"
|
|
back = reverse("accounts:profile") + "#sshkeys"
|
|
elif request.user.is_superuser:
|
|
ssh_key = get_object_or_404(UserSSHKey, pk=pk, user_id=user_id)
|
|
subject = f"SSH key {ssh_key.keyname} of {ssh_key.user.username}"
|
|
back = reverse("accounts:account", args=[user_id]) + "#public-keys"
|
|
else:
|
|
raise PermissionDenied
|
|
if request.method == "POST":
|
|
with audit(request, None, subject, "delete"):
|
|
ssh_key.delete()
|
|
messages.success(request, _("SSH key deleted"))
|
|
return redirect(back)
|
|
|
|
return render(
|
|
request,
|
|
"common/confirm_delete.html",
|
|
{
|
|
"object": subject,
|
|
"title": _("Delete SSH key"),
|
|
"consequence": _("VMs that already have it keep it; VMs created from now on do not get it."),
|
|
},
|
|
)
|
|
|
|
|
|
@superuser_only
|
|
def account(request, user_id):
|
|
user = get_object_or_404(User, id=user_id)
|
|
user_insts = UserInstance.objects.filter(user_id=user_id).select_related("instance__compute")
|
|
publickeys = UserSSHKey.objects.filter(user_id=user_id)
|
|
|
|
return render(
|
|
request,
|
|
"account.html",
|
|
{
|
|
"user": user,
|
|
"user_insts": user_insts,
|
|
"publickeys": publickeys,
|
|
"api_tokens": ApiToken.objects.filter(user_id=user_id),
|
|
"otp_enabled": settings.OTP_ENABLED,
|
|
},
|
|
)
|
|
|
|
|
|
@sensitive_post_parameters()
|
|
@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
|
|
if not request.user.has_usable_password(): # an LDAP user: the password is changed in LDAP
|
|
raise PermissionDenied
|
|
form = PasswordChangeForm(request.user, request.POST or None)
|
|
|
|
if form.is_valid():
|
|
with audit(request, None, f"User {request.user.username}", "change password"), 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, "must_change": must_change})
|
|
|
|
|
|
@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():
|
|
instance = form.cleaned_data["instance"]
|
|
# the form's one-owner check again, with the insert in one step, as on the VM page
|
|
with audit_vm(request, instance, f"add owner {form.cleaned_data['user'].username}") as entry, \
|
|
transaction.atomic():
|
|
lock_ownership()
|
|
form.full_clean()
|
|
entry.skip = not form.is_valid()
|
|
if form.is_valid():
|
|
form.save()
|
|
if form.is_valid():
|
|
return redirect(reverse("accounts:account", args=[user.id]))
|
|
|
|
return render(
|
|
request,
|
|
"common/form.html",
|
|
{
|
|
"form": form,
|
|
"title": _("Create User Instance"),
|
|
},
|
|
)
|
|
|
|
|
|
def _next_or(request, default):
|
|
"""To the ?next page, the VM page an owner was changed from, when it is
|
|
on this site; else to default."""
|
|
next_url = request.GET.get("next")
|
|
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(default)
|
|
|
|
|
|
@superuser_only
|
|
def user_instance_update(request, pk):
|
|
user_instance = get_object_or_404(UserInstance, pk=pk)
|
|
instance, before = user_instance.instance, snapshot(user_instance)
|
|
form = forms.UserInstanceForm(request.POST or None, instance=user_instance)
|
|
if form.is_valid():
|
|
with audit_vm(request, instance, "change owner") as entry:
|
|
form.save()
|
|
fields = changed_fields(before, snapshot(user_instance))
|
|
entry.action = f"change owner {user_instance.user.username}: " + ", ".join(fields)
|
|
entry.skip = not fields
|
|
return _next_or(request, 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, instance = user_instance.user, user_instance.instance
|
|
with audit_vm(request, instance, f"remove owner {user.username}"):
|
|
user_instance.delete()
|
|
return _next_or(request, reverse("accounts:account", args=[user.id]))
|
|
|
|
return render(
|
|
request,
|
|
"common/confirm_delete.html",
|
|
{"object": user_instance,
|
|
"consequence": _("The user loses access to this VM; the VM itself is not changed.")},
|
|
)
|
|
|
|
|
|
# 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 a link to 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(request, user)
|
|
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"),
|
|
},
|
|
)
|
|
|
|
|
|
@login_not_required
|
|
def otp_qr(request, token):
|
|
"""The OTP QR code of the user a mailed link (send_email_with_otp) names."""
|
|
if not settings.OTP_ENABLED:
|
|
raise Http404
|
|
try:
|
|
user_id = OTP_LINK.unsign(token, max_age=OTP_LINK_MAX_AGE)
|
|
except BadSignature:
|
|
raise Http404
|
|
user = get_object_or_404(get_user_model(), pk=user_id, is_active=True)
|
|
device = get_existing_totp_device(user)
|
|
if device is None:
|
|
raise Http404
|
|
response = render(request, "accounts/otp_qr.html", {"user": user, "qr_svg": otp_qr_svg(device)})
|
|
response["Cache-Control"] = "no-store"
|
|
return response
|
|
|
|
|
|
@require_POST
|
|
@superuser_only
|
|
def admin_email_otp(request, user_id):
|
|
user = get_object_or_404(get_user_model(), pk=user_id)
|
|
subject = f"User {user.username}"
|
|
if user.email != "":
|
|
try:
|
|
with audit(request, None, subject, "mail OTP QR code"):
|
|
get_user_totp_device(user)
|
|
send_email_with_otp(request, user)
|
|
except OSError as err: # the SMTP server refused it, or did not answer within EMAIL_TIMEOUT
|
|
messages.error(request, _("The OTP QR code could not be mailed to %(user)s: %(error)s") % {
|
|
"user": user, "error": err})
|
|
else:
|
|
messages.success(
|
|
request, _("OTP QR code was emailed to user %(user)s") % {"user": user}
|
|
)
|
|
else:
|
|
failed(request, None, subject, "mail OTP QR code", "no email address")
|
|
messages.error(request, _("User email not set, failed to send QR code"))
|
|
return redirect("accounts:account", user.id)
|