j3d1/webvirtcloud
1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-09 19:56:48 +00:00
Code Issues Projects Releases Wiki Activity

feat(logs): changes to users, groups, owners, flavors and settings are logged

Browse source 
The Logs page now records the panel's own records too: users (add,
change with the changed fields, delete, block, unblock, an admin's
password set, a user's own password change including the first-login
one, own profile edits), groups with their members and permissions,
VM owners from the accounts pages and from the VM page, OTP QR code
mails, flavors from web and API, app settings and the theme,
removed-VM records and SSH keys. No row carries a password or a key.

- A user is created in one transaction: when a VM named in
  NEW_USER_DEFAULT_INSTANCES is missing, no half-made user stays.
- The OTP mail checks the email address before it creates a device;
  without one it writes a failed row and makes no device.
- The VM page's owner rows read "VM web-1 owner bob: add" and
  "...: remove" instead of "Add owner: bob" and "Delete owner: 17"; a
  second owner refused by the single-owner setting is a failed row.
  Owner rows keep the VM's host and name, so the VM's log tab shows
  them.
- A setting or theme posted with its current value is no change and
  writes no row; the old free-text setting rows are gone.

Host-side objects (networks, storage, interfaces, secrets, filters)
follow in the next layer.
This commit is contained in:
catborise 2026-10-23 10:55:55 +03:00
parent 8c0fbff8f7
commit e04adec87b
10 changed files with 468 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

121
admin/tests/test_audit_rows.py Normal file
View file

@ -0,0 +1,121 @@
"""Changes to users and groups write rows on the Logs page."""
from django.contrib.auth.models import Group, Permission, User
from django.test import TestCase, override_settings
from django.urls import reverse
from logs.models import Logs
ATTRIBUTES = {"max_instances": 2, "max_cpus": 2, "max_memory": 2048, "max_disk_size": 20}
def rows():
return list(Logs.objects.order_by("id").values_list("message", "success"))
class UserRowTests(TestCase):
def setUp(self):
self.admin = User.objects.create_superuser("audit-admin", "audit@example.com", "pw")
self.client.force_login(self.admin)
Logs.objects.all().delete() # the login's row
def create(self, username="carol", **extra):
return self.client.post(reverse("admin:user_create"),
{"username": username, "password": "Carol-pass-2026", "is_active": "on", **ATTRIBUTES, **extra})
def edit(self, user, **changes):
data = {"username": user.username, "email": user.email, "is_active": "on", **ATTRIBUTES,
"user_permissions": [p.pk for p in user.user_permissions.all()],
"groups": [g.pk for g in user.groups.all()], **changes}
return self.client.post(reverse("admin:user_update", args=[user.pk]), data)
def test_user_add_change_delete(self):
self.create()
carol = User.objects.get(username="carol")
self.edit(carol, email="carol@example.com")
self.edit(User.objects.get(pk=carol.pk)) # nothing changed
self.client.post(reverse("admin:user_delete", args=[carol.pk]))
self.assertEqual(rows(), [
("User carol: add", True),
("User carol: change email", True),
("User carol: delete", True),
])
def test_attributes_and_groups_are_named(self):
self.create()
carol = User.objects.get(username="carol")
permission = Permission.objects.get(content_type__app_label="accounts", codename="change_password")
carol.user_permissions.remove(permission)
Logs.objects.all().delete()
self.edit(carol, max_instances=5)
group = Group.objects.create(name="ops")
self.edit(carol, max_instances=5, groups=[group.pk], user_permissions=[permission.pk])
self.edit(carol, max_instances=5, groups=[group.pk], user_permissions=[permission.pk], username="carla")
self.edit(User.objects.get(pk=carol.pk), username="") # invalid: no row
self.assertEqual([m for m, _ in rows()], [
"User carol: change max_instances",
"User carol: change groups, user_permissions",
"User carol: change username",
])
self.assertTrue(User.objects.get(pk=carol.pk).user_permissions.filter(pk=permission.pk).exists())
def test_block_unblock_and_password(self):
self.create()
carol = User.objects.get(username="carol")
Logs.objects.all().delete()
self.client.post(reverse("admin:user_block", args=[carol.pk]))
self.client.post(reverse("admin:user_unblock", args=[carol.pk]))
self.client.post(reverse("admin:user_update_password", args=[carol.pk]),
{"password1": "New-carol-pass-1", "password2": "New-carol-pass-1"})
self.assertEqual(rows(), [
("User carol: block", True),
("User carol: unblock", True),
("User carol: set password", True),
])
self.assertFalse(Logs.objects.filter(message__contains="New-carol-pass-1").exists())
@override_settings(NEW_USER_DEFAULT_INSTANCES=["no-such-vm"])
def test_a_failing_default_vm_leaves_no_user(self):
from instances.models import Instance
with self.assertRaises(Instance.DoesNotExist):
self.create()
self.assertFalse(User.objects.filter(username="carol").exists())
self.assertEqual(rows(), [("User carol: add: DoesNotExist", False)])
def test_the_row_has_actor_and_ip(self):
self.client.post(
reverse("admin:user_create"), {"username": "carol", "password": "Carol-pass-2026", **ATTRIBUTES},
REMOTE_ADDR="192.0.2.9")
row = Logs.objects.get()
self.assertEqual((row.user, row.host, row.instance, row.ip), ("audit-admin", "-", "", "192.0.2.9"))
class GroupAndRemovedVMRowTests(TestCase):
def setUp(self):
self.client.force_login(User.objects.create_superuser("audit-admin", "audit@example.com", "pw"))
Logs.objects.all().delete() # the login's row
def test_group_add_change_delete(self):
member = User.objects.create_user("dave")
permission = Permission.objects.get(content_type__app_label="accounts", codename="change_password")
self.client.post(reverse("admin:group_create"), {"name": "ops"})
group = Group.objects.get(name="ops")
url = reverse("admin:group_update", args=[group.pk])
self.client.post(url, {"name": "ops", "users": [member.pk]})
self.client.post(url, {"name": "ops", "users": [member.pk], "permissions": [permission.pk]})
self.client.post(url, {"name": "ops", "users": [member.pk], "permissions": [permission.pk]}) # no change
self.client.post(reverse("admin:group_delete", args=[group.pk]))
self.assertEqual(rows(), [
("Group ops: add", True),
("Group ops: change users", True),
("Group ops: change permissions", True),
("Group ops: delete", True),
])
def test_a_removed_vm_record_delete(self):
from instances.models import InstanceTombstone
uuid = "55555555-5555-5555-5555-555555555555"
tombstone = InstanceTombstone.objects.create(uuid=uuid, name="vm", owners=[])
self.client.post(reverse("admin:tombstone_delete", args=[tombstone.pk]))
self.assertEqual(rows(), [(f"Removed VM vm ({uuid}): delete", True)])

61
admin/views.py
View file

@ -8,17 +8,24 @@ from django.contrib.auth import update_session_auth_hash
from django.contrib.auth.forms import AdminPasswordChangeForm
from django.contrib.auth.models import Group, User
from django.core.paginator import Paginator
from django.db import transaction
from django.shortcuts import get_object_or_404, redirect, render
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 instances.models import InstanceTombstone
from logs.audit import audit, changed_fields, snapshot
from logs.models import Logs
from . import forms
from .decorators import superuser_only
def group_snapshot(group):
"""snapshot() of a group with its members, which are a reverse relation snapshot() does not see."""
return snapshot(group) | {"users": frozenset(group.user_set.values_list("pk", flat=True))}
@superuser_only
def group_list(request):
groups = Group.objects.all()
@ -35,7 +42,8 @@ def group_list(request):
def group_create(request):
form = forms.GroupForm(request.POST or None)
if form.is_valid():
form.save()
with audit(request, None, f"Group {form.cleaned_data['name']}", "add"):
form.save()
return redirect("admin:group_list")
return render(
@ -51,9 +59,14 @@ def group_create(request):
@superuser_only
def group_update(request, pk):
group = get_object_or_404(Group, pk=pk)
subject, before = f"Group {group.name}", group_snapshot(group)
form = forms.GroupForm(request.POST or None, instance=group)
if form.is_valid():
form.save()
with audit(request, None, subject, "change") as entry:
form.save()
fields = changed_fields(before, group_snapshot(group))
entry.action = "change " + ", ".join(fields)
entry.skip = not fields
return redirect("admin:group_list")
return render(
@ -70,7 +83,8 @@ def group_update(request, pk):
def group_delete(request, pk):
group = get_object_or_404(Group, pk=pk)
if request.method == "POST":
group.delete()
with audit(request, None, f"Group {group.name}", "delete"):
group.delete()
return redirect("admin:group_list")
return render(
@ -98,11 +112,12 @@ def user_create(request):
user_form = forms.UserCreateForm(request.POST or None)
attributes_form = forms.UserAttributesForm(request.POST or None)
if user_form.is_valid() and attributes_form.is_valid():
user = user_form.save()
attributes = attributes_form.save(commit=False)
attributes.user = user
attributes.save()
add_default_instances(user)
with audit(request, None, f"User {user_form.cleaned_data['username']}", "add"), transaction.atomic():
user = user_form.save()
attributes = attributes_form.save(commit=False)
attributes.user = user
attributes.save()
add_default_instances(user)
return redirect("admin:user_list")
return render(
@ -120,13 +135,20 @@ def user_create(request):
def user_update(request, pk):
user = get_object_or_404(User, pk=pk)
attributes, attributes_created = UserAttributes.objects.get_or_create(user=user)
subject = f"User {user.username}"
before, before_attributes = snapshot(user), snapshot(attributes)
user_form = forms.UserForm(request.POST or None, instance=user)
attributes_form = forms.UserAttributesForm(
request.POST or None, instance=attributes
)
if user_form.is_valid() and attributes_form.is_valid():
user_form.save()
attributes_form.save()
with audit(request, None, subject, "change") as entry:
user_form.save()
attributes_form.save()
fields = set(changed_fields(before, snapshot(user))) | set(changed_fields(before_attributes, snapshot(attributes)))
fields = sorted(fields - {"password", "last_login"})
entry.action = "change " + ", ".join(fields)
entry.skip = not fields
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()
@ -151,7 +173,8 @@ def user_update_password(request, pk):
if request.method == "POST":
form = AdminPasswordChangeForm(user, request.POST)
if form.is_valid():
user = form.save()
with audit(request, None, f"User {user.username}", "set password"):
user = form.save()
update_session_auth_hash(request, user) # Important!
messages.success(
request, _("Password changed for %(user)s") % {"user": user.username}
@ -176,7 +199,8 @@ def user_update_password(request, pk):
def user_delete(request, pk):
user = get_object_or_404(User, pk=pk)
if request.method == "POST":
user.delete()
with audit(request, None, f"User {user.username}", "delete"):
user.delete()
return redirect("admin:user_list")
return render(
@ -191,8 +215,9 @@ def user_delete(request, pk):
@superuser_only
def user_block(request, pk):
user: User = get_object_or_404(User, pk=pk)
user.is_active = False
user.save()
with audit(request, None, f"User {user.username}", "block"):
user.is_active = False
user.save()
return redirect("admin:user_list")
@ -200,8 +225,9 @@ def user_block(request, pk):
@superuser_only
def user_unblock(request, pk):
user: User = get_object_or_404(User, pk=pk)
user.is_active = True
user.save()
with audit(request, None, f"User {user.username}", "unblock"):
user.is_active = True
user.save()
return redirect("admin:user_list")
@ -230,7 +256,8 @@ def tombstone_list(request):
def tombstone_delete(request, pk):
tombstone = get_object_or_404(InstanceTombstone, pk=pk)
if request.method == "POST":
tombstone.delete()
with audit(request, None, f"Removed VM {tombstone.name} ({tombstone.uuid})", "delete"):
tombstone.delete()
return redirect("admin:tombstone_list")
return render(