feat(logs): admin changes and logins are recorded on the Logs page

The Logs page shows who added, changed or deleted a compute, from web
or API, and every login, failed login, lockout and logout, with the
client address and a Result column (Done, Failed, Unknown for older
rows). A failed change is recorded with its reason and the error still
reaches the user; a refused one is recorded too.

- logs.audit.write saves each row in its own savepoint and never
  raises, so a log write cannot break or undo the action it records.
  The audit() context manager writes one row when the change ends.
- Logs gains success and ip; user, host and instance are wider and
  date is indexed. The page orders by date then id.
- A compute edit names the changed fields; an edit that changes
  nothing writes no row, and an empty password field is no change.
- A compute's VMs leaving the panel get a system row each: owners
  kept on the Removed VMs page, moved to another compute, restored,
  left without owners, or the removed record expired.
- Job rows carry the result. A job's log entry that cannot be
  computed is logged and skipped instead of rolling back the finish
  (test_service.py changes this on purpose).
- Logins are recorded by signal receivers; CustomLoginView is removed
  and the plain LoginView is used. A wrong OTP code is a failed login,
  counted by django-axes toward the lockout.

Panel records and host-side objects follow in the next layers.

logs/tests/test_audit.py

@@ -0,0 +1,146 @@
+from unittest import mock
+
+import libvirt
+
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Group
+from django.db import IntegrityError, transaction
+from django.test import RequestFactory, TestCase
+
+from computes.models import Compute
+from logs.audit import audit, changed_fields, failed, request_ip, snapshot, write
+from logs.models import Logs
+from logs.views import addlogmsg
+from operations import claims
+from vrtManager.util import OperationError
+
+
+class WriteTests(TestCase):
+    def test_a_row_has_its_fields(self):
+        write("bob", "host1", "", "Compute c: delete", True, "192.0.2.7")
+        row = Logs.objects.get()
+        self.assertEqual((row.user, row.host, row.instance), ("bob", "host1", ""))
+        self.assertEqual(row.message, "Compute c: delete")
+        self.assertIs(row.success, True)
+        self.assertEqual(row.ip, "192.0.2.7")
+
+    def test_long_values_are_cut(self):
+        write("u" * 200, "h" * 300, "i" * 200, "m" * 5000)
+        row = Logs.objects.get()
+        self.assertEqual(len(row.user), 150)
+        self.assertEqual(len(row.host), 255)
+        self.assertEqual(len(row.instance), 120)
+        self.assertEqual(len(row.message), 5000)
+
+    def test_a_failed_write_inside_a_transaction_leaves_it_usable(self):
+        with self.assertLogs("logs.audit", "WARNING") as logs:
+            with transaction.atomic():
+                with mock.patch.object(Logs.objects, "create", side_effect=IntegrityError):
+                    write("bob", "-", "", "msg")
+                get_user_model().objects.create_user("after")
+        self.assertEqual(len(logs.output), 1)
+        self.assertTrue(get_user_model().objects.filter(username="after").exists())
+
+    def test_the_file_line_keeps_one_line(self):
+        with self.assertLogs("logs.audit", "WARNING") as logs:
+            with mock.patch.object(Logs.objects, "create", side_effect=IntegrityError):
+                write("bob", "-", "", "a\nb")
+        self.assertNotIn("\n", logs.records[0].getMessage())
+        self.assertIn("a\\nb", logs.records[0].getMessage())
+
+    def test_a_failing_file_logger_raises_nothing(self):
+        with mock.patch.object(Logs.objects, "create", side_effect=IntegrityError), mock.patch(
+            "logs.audit.logger.warning", side_effect=OSError
+        ):
+            self.assertIsNone(write("bob", "-", "", "msg"))
+
+    def test_ips(self):
+        rf = RequestFactory()
+        self.assertEqual(request_ip(rf.get("/", REMOTE_ADDR="192.0.2.7")), "192.0.2.7")
+        self.assertIsNone(request_ip(rf.get("/", REMOTE_ADDR="127.0.0.1", HTTP_X_REAL_IP="not-an-ip")))
+        self.assertEqual(request_ip(rf.get("/", REMOTE_ADDR="127.0.0.1", HTTP_X_REAL_IP="")), "127.0.0.1")
+        self.assertIsNone(request_ip(rf.get("/", REMOTE_ADDR="127.0.0.1", HTTP_X_REAL_IP="fe80::1%eth0")))
+        self.assertIsNone(request_ip(rf.get("/", REMOTE_ADDR="")))
+
+    def test_addlogmsg_still_writes(self):
+        addlogmsg("u", "h", "vm", "Destroy")
+        row = Logs.objects.get()
+        self.assertEqual((row.user, row.host, row.instance, row.message), ("u", "h", "vm", "Destroy"))
+        self.assertIsNone(row.success)
+        self.assertIsNone(row.ip)
+
+
+class AuditTests(TestCase):
+    def setUp(self):
+        self.user = get_user_model().objects.create_user("bob")
+        self.request = RequestFactory().get("/", REMOTE_ADDR="192.0.2.7")
+        self.request.user = self.user
+        self.compute = Compute(name="c", hostname="192.0.2.5", type=1)
+
+    def row(self):
+        return Logs.objects.get()
+
+    def test_a_block_that_ends_writes_done(self):
+        with audit(self.request, self.compute, "Compute c (192.0.2.5)", "delete"):
+            pass
+        row = self.row()
+        self.assertEqual(row.message, "Compute c (192.0.2.5): delete")
+        self.assertIs(row.success, True)
+        self.assertEqual((row.host, row.user, row.ip), ("c", "bob", "192.0.2.7"))
+
+    def test_a_block_that_raises_writes_failed_and_reraises(self):
+        error = OperationError("in use")
+        with self.assertRaises(OperationError) as caught:
+            with audit(self.request, self.compute, "Compute c", "delete"):
+                raise error
+        self.assertIs(caught.exception, error)
+        row = self.row()
+        self.assertEqual(row.message, "Compute c: delete: in use")
+        self.assertIs(row.success, False)
+
+    def reason_row(self, exc):
+        Logs.objects.all().delete()
+        with self.assertRaises(type(exc)):
+            with audit(self.request, None, "Compute c", "delete"):
+                raise exc
+        return self.row().message
+
+    def test_reasons(self):
+        self.assertEqual(
+            self.reason_row(claims.Busy(None)),
+            "Compute c: delete: A short operation is in progress on this resource; try again in a moment.",
+        )
+        self.assertNotIn("secret", self.reason_row(libvirt.libvirtError("password=secret")))
+        self.assertEqual(self.reason_row(ValueError("<secret/>")), "Compute c: delete: ValueError")
+
+    def test_the_block_sets_the_action_or_skips(self):
+        with audit(self.request, self.compute, "Compute c", "edit") as entry:
+            entry.action = "change hostname"
+        self.assertEqual(self.row().message, "Compute c: change hostname")
+        Logs.objects.all().delete()
+        with audit(self.request, self.compute, "Compute c", "edit") as entry:
+            entry.skip = True
+        self.assertFalse(Logs.objects.exists())
+        with self.assertRaises(ValueError):
+            with audit(self.request, self.compute, "Compute c", "edit") as entry:
+                entry.skip = True
+                raise ValueError
+        self.assertEqual(self.row().message, "Compute c: edit: ValueError")
+        self.assertIs(self.row().success, False)
+
+    def test_failed_writes_a_refusal(self):
+        failed(self.request, None, "Network lab", "create", "Name in use")
+        row = self.row()
+        self.assertEqual(row.message, "Network lab: create: Name in use")
+        self.assertEqual(row.host, "-")
+        self.assertIs(row.success, False)
+
+    def test_changed_fields(self):
+        compute = Compute.objects.create(name="c", hostname="192.0.2.5", type=1, password="pw")
+        before = snapshot(compute)
+        self.assertEqual(changed_fields(before, snapshot(compute)), [])
+        compute.hostname = "192.0.2.6"
+        self.assertEqual(changed_fields(before, snapshot(compute)), ["hostname"])
+        before = snapshot(self.user)
+        self.user.groups.add(Group.objects.create(name="g"))
+        self.assertEqual(changed_fields(before, snapshot(self.user)), ["groups"])

logs/tests/test_login_rows.py

@@ -0,0 +1,51 @@
+"""Logins, failed logins, lockouts and logouts write rows on the Logs page."""
+
+from appsettings.models import AppSettings
+from django.contrib.auth import get_user_model
+from django.test import TestCase, override_settings
+from django.urls import reverse
+
+from logs.models import Logs
+
+
+def rows():
+    return list(Logs.objects.order_by("id").values_list("user", "host", "message", "success", "ip"))
+
+
+@override_settings(OTP_ENABLED=False)
+class LoginRowsTests(TestCase):
+    def setUp(self):
+        get_user_model().objects.create_user("bob", password="right-pass-1")
+        AppSettings.objects.filter(key="LOGIN_FAILURE_LIMIT").update(value="3")
+
+    def login(self, password, username="bob"):
+        return self.client.post(reverse("accounts:login"), {"username": username, "password": password})
+
+    def test_a_login_and_a_logout_write_rows(self):
+        self.assertEqual(self.login("right-pass-1").status_code, 302)
+        self.client.post(reverse("accounts:logout"))
+        self.assertEqual(rows(), [
+            ("bob", "-", "Login", True, "127.0.0.1"),
+            ("bob", "-", "Logout", True, "127.0.0.1"),
+        ])
+
+    def test_a_failed_login_writes_the_typed_username(self):
+        self.assertEqual(self.login("wrong").status_code, 200)
+        self.assertEqual(rows(), [("bob", "-", "Login failed", False, "127.0.0.1")])
+
+    def test_an_empty_username_is_no_attempt(self):
+        # the form stops it before authentication: no row and no error
+        self.assertEqual(self.login("wrong", username="").status_code, 200)
+        self.assertEqual(rows(), [])
+
+    def test_the_attempt_at_the_limit_and_one_while_locked(self):
+        self.login("wrong")
+        self.login("wrong")
+        self.assertEqual(self.login("wrong").status_code, 429)
+        # django-axes sends the lockout from its own failure receiver, before ours
+        messages = [r[2] for r in rows()]
+        self.assertEqual(messages, ["Login failed", "Login failed", "Locked out", "Login failed"])
+
+        self.assertEqual(self.login("right-pass-1").status_code, 429)
+        self.assertEqual([r[2] for r in rows()][4:], ["Locked out", "Login failed"])
+        self.assertTrue(all(r[0] == "bob" and r[3] is False for r in rows()))