Log security relevant events to an audit log

Successful logins, user creation and deletion, password changes and the
enabled and admin toggles are logged with actor, target and IP. So are
failed logins, a wrong current password and denied access. Values are
JSON encoded, so client input cannot break the line or add fields.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

max/audit.py

@@ -0,0 +1,20 @@
+from flask import current_app, request
+
+from max.log_format import format_log_fields
+
+
+def audit(event, actor=None, target=None, **fields):
+    """
+    Logs a security relevant event as one line, starting with "audit". actor is the
+    acting user, None for anonymous requests. target is the affected user, if any.
+    """
+    values = dict(
+        event=event,
+        actor_user_id=actor.id if actor is not None else None,
+        actor_email=actor.email if actor is not None else None,
+        ip=getattr(request, "real_remote_addr", None),
+    )
+    if target is not None:
+        values.update(target_user_id=target.id, target_email=target.email)
+    values.update(fields)
+    current_app.logger.info("audit %s", format_log_fields(**values))

max/auth/login.py

@@ -16,28 +16,29 @@ LOGIN_FAILED_MESSAGE = "Email or password is wrong, or the account is deactivate
 
 
 def login(email, password):
+    """Returns (error, None) if the login failed, else (None, user)"""
     if not isinstance(email, str) or not isinstance(password, str):
-        return t(LOGIN_FAILED_MESSAGE)
+        return t(LOGIN_FAILED_MESSAGE), None
     if not email or not password:
-        return t(LOGIN_FAILED_MESSAGE)
+        return t(LOGIN_FAILED_MESSAGE), None
 
     user = get_user_by_email(email)
     if user is None:
         check_password(password, DUMMY_HASH)
-        return t(LOGIN_FAILED_MESSAGE)
+        return t(LOGIN_FAILED_MESSAGE), None
 
     if not check_password(password, user.passwordhash):
-        return t(LOGIN_FAILED_MESSAGE)
+        return t(LOGIN_FAILED_MESSAGE), None
 
     if not user.enabled:
-        return t(LOGIN_FAILED_MESSAGE)
+        return t(LOGIN_FAILED_MESSAGE), None
 
     rehash_password_if_needed(user, password)
 
     # Start a new session (new id, new CSRF token) against session fixation.
     session.clear()
     session["user_id"] = user.id
-    return None
+    return None, user
 
 
 def logout():

max/auth/routes.py

@@ -1,5 +1,6 @@
 from flask import flash, session, url_for
 
+from max.audit import audit
 from max.csrf import no_csrf
 from max.permissions import AllowAnyAccess
 from max.translations import t
@@ -46,15 +47,14 @@ class Login(AllowAnyAccess, BaseTemplateGetView):
         email = self.request.form.get("email")
         password = self.request.form.get("password")
 
-        error = login(email, password)
+        error, user = login(email, password)
         if error:
-            self.app.logger.info(
-                "Failed login attempt from %s", self.request.real_remote_addr
-            )
+            audit("login.failed", attempted_email=email)
             self.context["message"] = error
             self.context["message_category"] = "error"
             self.context["email"] = email
         else:
+            audit("login", actor=user)
             flash(t("Login was successful!"), category="success")
             return self.redirect("/")
 

max/db/users.py

@@ -66,12 +66,13 @@ def update_user(cur, id, **fields):
 def create_user(
     cur, email, passwordhash, enabled, is_admin, can_use_different_alias_domain, note
 ):
+    """Returns the id of the new user"""
     email_id = create_email(cur, email)
     cur.execute(
         """\
         INSERT INTO users
         (email_id, passwordhash, enabled, is_admin, can_use_different_alias_domain, note)
-        VALUES (%s, %s, %s, %s, %s, %s)""",
+        VALUES (%s, %s, %s, %s, %s, %s) RETURNING id""",
         [
             email_id,
             passwordhash,
@@ -81,6 +82,7 @@ def create_user(
             note,
         ],
     )
+    return cur.fetchone()[0]
 
 
 @with_cursor

max/log_format.py

@@ -0,0 +1,12 @@
+import json
+
+
+def format_log_fields(**fields):
+    """
+    Formats the fields as "key=value key=value". Values are JSON encoded, so values from
+    the client cannot break the line or add fields.
+    """
+    return " ".join(
+        f"{key}={json.dumps(value, ensure_ascii=False)}"
+        for key, value in fields.items()
+    )

max/routes/user/change_password.py

@@ -1,5 +1,6 @@
 from flask import flash
 
+from max.audit import audit
 from max.auth import (
     PASSWORD_MESSAGE_ARGUMENTS,
     change_password,
@@ -43,6 +44,11 @@ class UserChangePassword(AllowAdminOrSelf, BackToUsersMixin, FetchUserTemplateGe
                 flash(t("The current password must be given"), category="error")
                 return
             if not check_password(current_password, self.user.passwordhash):
+                audit(
+                    "user.change_password.wrong_current",
+                    actor=self.auth_user,
+                    target=self.user,
+                )
                 flash(t("The current password is wrong"), category="error")
                 return
 
@@ -56,5 +62,6 @@ class UserChangePassword(AllowAdminOrSelf, BackToUsersMixin, FetchUserTemplateGe
             return
 
         change_password(self.user, pw1)
+        audit("user.change_password", actor=self.auth_user, target=self.user)
         flash(t("Password changed successfully!"), category="success")
         return self.redirect()

max/routes/user/create.py

@@ -1,5 +1,6 @@
 from flask import flash, url_for
 
+from max.audit import audit
 from max.auth import PASSWORD_MESSAGE_ARGUMENTS, hash_password, validate_password
 from max.db import EmailAlreadyExists, create_user, get_domains
 from max.permissions import AllowAdmin
@@ -96,7 +97,7 @@ class UserCreate(AllowAdmin, CheckEmailMixin, BaseTemplateGetView):
 
         passwordhash = hash_password(self.password)
         try:
-            create_user(
+            user_id = create_user(
                 full_email,
                 passwordhash,
                 self.enabled,
@@ -114,6 +115,15 @@ class UserCreate(AllowAdmin, CheckEmailMixin, BaseTemplateGetView):
                 flash(t("Email already exists."))
             return
 
+        audit(
+            "user.create",
+            actor=self.auth_user,
+            target_user_id=user_id,
+            target_email=full_email,
+            enabled=self.enabled,
+            is_admin=self.is_admin,
+            can_use_different_alias_domain=self.can_use_different_alias_domain,
+        )
         flash(
             t("Creation of {email} was successful!", email=full_email),
             category="success",

max/routes/user/delete.py

@@ -1,5 +1,6 @@
 from flask import flash, session, url_for
 
+from max.audit import audit
 from max.db import delete_user, is_last_enabled_admin
 from max.permissions import AllowAdmin
 from max.sessions import session_storage
@@ -30,6 +31,7 @@ class UserDelete(AllowAdmin, BackToUsersMixin, FetchUserTemplateGetView):
 
         delete_user(self.user)
         session_storage.remove_user_sessions(self.user.id)
+        audit("user.delete", actor=self.auth_user, target=self.user)
         flash(
             t("{email} was deleted successfully", email=self.user.email),
             category="success",

max/routes/user/toggle_admin.py

@@ -1,5 +1,6 @@
 from flask import flash
 
+from max.audit import audit
 from max.db import is_last_enabled_admin, update_user
 from max.permissions import AllowAdmin
 from max.translations import t
@@ -18,6 +19,12 @@ class UserToggleAdmin(AllowAdmin, FetchUserMixin, BackToUsersMixin, BaseView):
             flash(message, category="error")
         else:
             update_user(self.user.id, is_admin=(not self.user.is_admin))
+            audit(
+                "user.toggle_admin",
+                actor=self.auth_user,
+                target=self.user,
+                is_admin=not self.user.is_admin,
+            )
 
         _, back_url = self.get_back_text_and_url()
         return self.redirect(back_url)

max/routes/user/toggle_enabled.py

@@ -1,5 +1,6 @@
 from flask import flash
 
+from max.audit import audit
 from max.db import is_last_enabled_admin, update_user
 from max.permissions import AllowAdmin
 from max.sessions import session_storage
@@ -21,6 +22,12 @@ class UserToggleEnabled(AllowAdmin, FetchUserMixin, BackToUsersMixin, BaseView):
             )
         else:
             update_user(self.user.id, enabled=(not self.user.enabled))
+            audit(
+                "user.toggle_enabled",
+                actor=self.auth_user,
+                target=self.user,
+                enabled=not self.user.enabled,
+            )
             if self.user.enabled:  # the user was deactivated
                 session_storage.remove_user_sessions(self.user.id)
 

max/views.py

@@ -5,10 +5,10 @@ from flask import (
     redirect as flask_redirect,
     render_template,
     request,
-    session,
 )
 from werkzeug.exceptions import MethodNotAllowed
 
+from max.audit import audit
 from max.exceptions import ImproperlyConfigured, PermissionDenied
 
 
@@ -78,11 +78,12 @@ class BaseView:
 
     def check_access(self):
         if not self.can_access():
-            self.app.logger.debug(
-                "Access denied for user %s to %s (%s)",
-                session.get("user_id"),
-                self.request.path,
-                self.__class__.__name__,
+            audit(
+                "access_denied",
+                actor=getattr(self, "auth_user", None),
+                method=self.request.method,
+                path=self.request.path,
+                view=self.__class__.__name__,
             )
             raise PermissionDenied()