Add revision history for user account and permission changes

enferno/admin/models/UserHistory.py

@@ -0,0 +1,58 @@
+import json
+from typing import Any
+
+from sqlalchemy import JSON
+
+from enferno.extensions import db
+from enferno.utils.base import BaseMixin
+from enferno.utils.date_helper import DateHelper
+from enferno.utils.logging_utils import get_logger
+
+logger = get_logger()
+
+
+class UserHistory(db.Model, BaseMixin):
+    """
+    SQL Alchemy model for user account revisions.
+
+    Snapshots account and permission state so admins can answer
+    "when did this user get this permission, and who granted it".
+    Access is restricted to Admin at the endpoint level.
+    """
+
+    id = db.Column(db.Integer, primary_key=True)
+    # The user this revision describes. Revisions outlive the account: this is
+    # an evidence platform, so a hard delete must not erase the record of what
+    # an account was allowed to do. The snapshot in `data` carries the id,
+    # username, name and email, so a detached row still identifies its subject.
+    target_user_id = db.Column(
+        db.Integer,
+        db.ForeignKey("user.id", ondelete="SET NULL"),
+        index=True,
+    )
+    target_user = db.relationship(
+        "User",
+        backref=db.backref("history", order_by="UserHistory.updated_at"),
+        foreign_keys=[target_user_id],
+    )
+    data = db.Column(JSON)
+    # user tracking, who made the change. Also kept on their deletion, so the
+    # trail of what they did to other accounts survives.
+    user_id = db.Column(db.Integer, db.ForeignKey("user.id", ondelete="SET NULL"))
+    user = db.relationship("User", backref="user_revisions", foreign_keys=[user_id])
+
+    def to_dict(self) -> dict[str, Any]:
+        """Return a dictionary representation of the user revision."""
+        return {
+            "id": self.id,
+            "data": self.data,
+            "created_at": DateHelper.serialize_datetime(self.created_at),
+            "user": self.user.to_compact() if self.user else None,
+        }
+
+    def to_json(self) -> str:
+        """Return a JSON representation of the user revision."""
+        return json.dumps(self.to_dict(), sort_keys=True)
+
+    def __repr__(self):
+        return "<UserHistory {} -- Target {}>".format(self.id, self.target_user_id)

enferno/admin/models/__init__.py

@@ -43,5 +43,6 @@
 from .Query import Query
 from .Settings import Settings
 from .Source import Source
+from .UserHistory import UserHistory
 from .WorkflowStatus import WorkflowStatus
 from .Notification import Notification

enferno/admin/views/history.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from flask import Response
-from flask_security.decorators import current_user
+from flask_security.decorators import current_user, roles_required
 from sqlalchemy import desc
 
 from enferno.admin.models import (
@@ -13,8 +13,10 @@
     Incident,
     IncidentHistory,
     LocationHistory,
+    UserHistory,
 )
 from enferno.extensions import db
+from enferno.user.models import User
 from enferno.utils.http_response import HTTPResponse
 import enferno.utils.typing as t
 from . import admin, require_view_history
@@ -148,3 +150,34 @@ def api_locationhistory(locationid: t.id) -> Response:
     # For standardization
     response = {"items": [item.to_dict() for item in result]}
     return HTTPResponse.success(data=response)
+
+
+# User History Helpers
+
+
+@admin.route("/api/userhistory/<int:userid>")
+@roles_required("Admin")
+def api_userhistory(userid: t.id) -> Response:
+    """
+    Endpoint to get revision history of a user account.
+
+    Admin only: revisions carry account and permission state, and are not
+    gated by the view_history permissions used for content items.
+
+    Args:
+        - userid: id of the user.
+
+    Returns:
+        - json feed of the user's history / error.
+    """
+    if not db.session.get(User, userid):
+        return HTTPResponse.not_found("User not found")
+
+    result = (
+        UserHistory.query.filter_by(target_user_id=userid)
+        .order_by(desc(UserHistory.created_at))
+        .all()
+    )
+    # For standardization
+    response = {"items": [item.to_dict() for item in result]}
+    return HTTPResponse.success(data=response)

enferno/admin/views/users.py

@@ -311,6 +311,7 @@ def api_user_create(
     user.from_json(u)
     result = user.save()
     if result:
+        user.create_revision()
         # Record activity
         Activity.create(
             current_user, Activity.ACTION_CREATE, Activity.STATUS_SUCCESS, user.to_mini(), "user"
@@ -396,6 +397,7 @@ def api_user_update(
 
         user = user.from_json(u)
         if user.save():
+            user.create_revision()
             # Record activity
             Activity.create(
                 current_user,

enferno/user/models.py

@@ -1,5 +1,5 @@
 import json
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 from uuid import uuid4
 
 from flask import current_app, session, has_app_context, has_request_context
@@ -432,6 +432,54 @@ def from_json(self, item: dict) -> "User":
         self.active = item.get("active")
         return self
 
+    def to_history_dict(self) -> dict:
+        """
+        Snapshot serializer for the revision history.
+
+        Deliberately not to_dict(): that one masks names via the secure_*
+        properties based on the acting user, and carries the live password
+        reset key. A stored revision must be neither viewer-dependent nor
+        hold secrets.
+        """
+        return {
+            "id": self.id,
+            "name": self.name,
+            "username": self.username,
+            "email": self.email,
+            "active": self.active,
+            # sorted so two snapshots of the same roles compare equal: the
+            # relationship has no order_by, so its order is whatever the join
+            # returns, and an unstable order would read as a change on diff
+            "roles": [
+                {"id": role.id, "name": role.name}
+                for role in sorted(self.roles, key=lambda r: r.id)
+            ],
+            "view_usernames": self.view_usernames,
+            "view_simple_history": self.view_simple_history,
+            "view_full_history": self.view_full_history,
+            "can_self_assign": self.can_self_assign,
+            "can_edit_locations": self.can_edit_locations,
+            "can_export": self.can_export,
+            "can_import_web": self.can_import_web,
+            "can_access_media": self.can_access_media,
+        }
+
+    def create_revision(self, user_id: Optional[int] = None) -> None:
+        """
+        Store a snapshot of this user's account and permission state.
+
+        Args:
+            - user_id: id of the user making the change, defaults to the
+              current user. None when there is no acting user (CLI, install).
+        """
+        from enferno.admin.models import UserHistory
+
+        # current_user is unbound outside a request, so guard rather than
+        # attributing the change to an arbitrary user
+        if user_id is None and has_request_context():
+            user_id = getattr(current_user, "id", None)
+        UserHistory(target_user_id=self.id, data=self.to_history_dict(), user_id=user_id).save()
+
     @property
     def two_factor_devices(self) -> Dict[str, Any]:
         """

migrations/versions/a91c4b7e0d52_add_user_history.py

@@ -0,0 +1,80 @@
+"""add user history
+
+Revision ID: a91c4b7e0d52
+Revises: d4f7a2c9b310
+Create Date: 2026-08-11
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = "a91c4b7e0d52"
+down_revision = "d4f7a2c9b310"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "user_history",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("target_user_id", sa.Integer(), nullable=True),
+        sa.Column("data", sa.JSON(), nullable=True),
+        sa.Column("user_id", sa.Integer(), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=True),
+        sa.Column("updated_at", sa.DateTime(), nullable=True),
+        sa.Column("deleted", sa.Boolean(), server_default="false", nullable=False),
+        sa.ForeignKeyConstraint(["target_user_id"], ["user.id"], ondelete="SET NULL"),
+        sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_user_history_target_user_id"), "user_history", ["target_user_id"], unique=False
+    )
+
+    # Baseline snapshot per existing user. A revision is only meaningful against
+    # the one before it, and user accounts are edited rarely, so without a
+    # baseline the first edit of each existing account would be undiffable
+    # for as long as that account goes untouched. user_id is left null: no
+    # acting user made this change. Mirrors User.to_history_dict().
+    op.execute("""
+        INSERT INTO user_history (target_user_id, data, user_id, created_at, updated_at, deleted)
+        SELECT
+            u.id,
+            json_build_object(
+                'id', u.id,
+                'name', u.name,
+                'username', u.username,
+                'email', u.email,
+                'active', u.active,
+                'roles', COALESCE(
+                    (
+                        SELECT json_agg(json_build_object('id', r.id, 'name', r.name) ORDER BY r.id)
+                        FROM roles_users ru
+                        JOIN role r ON r.id = ru.role_id
+                        WHERE ru.user_id = u.id
+                    ),
+                    '[]'::json
+                ),
+                'view_usernames', u.view_usernames,
+                'view_simple_history', u.view_simple_history,
+                'view_full_history', u.view_full_history,
+                'can_self_assign', u.can_self_assign,
+                'can_edit_locations', u.can_edit_locations,
+                'can_export', u.can_export,
+                'can_import_web', u.can_import_web,
+                'can_access_media', u.can_access_media
+            ),
+            NULL,
+            timezone('utc', now()),
+            timezone('utc', now()),
+            false
+        FROM "user" u
+        """)
+
+
+def downgrade():
+    op.drop_index(op.f("ix_user_history_target_user_id"), table_name="user_history")
+    op.drop_table("user_history")

pyproject.toml

@@ -179,6 +179,7 @@ select = ["F"]  # pyflakes: unused imports, undefined names, syntax errors
 [tool.ruff.lint.per-file-ignores]
 "enferno/app.py" = ["F841"]       # security = Security() assigned for side effects
 "enferno/commands.py" = ["F811"]  # extract() name reused across CLI groups
+"enferno/admin/models/__init__.py" = ["F401"]  # re-export surface for the model package
 
 [dependency-groups]
 dev = [

tests/test_pentest_fixes.py

@@ -826,3 +826,60 @@ def has_role(self, r):
             assert mu.can_view_media() is True
         with patch.object(mu, "current_user", _Admin()):
             assert mu.can_view_media() is True
+
+
+# ---------------------------------------------------------------------------
+# BAY-01-001 (extension)  The user revision-history endpoint is new surface of
+# the same class the auditor flagged: a *history* route that can leak state the
+# caller cannot reach through the primary API. User revisions carry account and
+# permission state, so this one is Admin-only rather than gated on the
+# view_history permissions, and the auditor's exact profile (view_simple_history
+# with no Admin role) must be refused.
+# ---------------------------------------------------------------------------
+
+
+def test_bay_01_001_user_history_denied_to_history_viewer(
+    session, users, history_viewer_outside_group
+):
+    admin_user, _, _, _ = users
+    resp = history_viewer_outside_group.get(f"/admin/api/userhistory/{admin_user.id}")
+    assert resp.status_code == 403
+
+
+def test_bay_01_001_user_history_denied_to_non_admin_roles(request, session, users):
+    admin_user, _, _, _ = users
+    for fixture in ("da_client", "mod_client"):
+        client = request.getfixturevalue(fixture)
+        resp = client.get(f"/admin/api/userhistory/{admin_user.id}")
+        assert resp.status_code == 403, f"{fixture} reached user history"
+
+
+def test_bay_01_001_user_history_requires_auth(anonymous_client):
+    """JSON callers get 401; a browser caller is redirected to login. Either way
+    the payload is never served anonymously."""
+    resp = anonymous_client.get(
+        "/admin/api/userhistory/1", headers={"Content-Type": "application/json"}
+    )
+    assert resp.status_code == 401
+
+    resp = anonymous_client.get("/admin/api/userhistory/1")
+    assert resp.status_code == 302
+    assert "/login" in resp.headers.get("Location", "")
+
+
+def test_bay_01_001_user_history_snapshot_holds_no_credentials(session, users):
+    """The snapshot is persisted forever, so it must never carry a secret.
+    to_dict() would have: it includes force_reset, the live password reset key."""
+    admin_user, _, _, _ = users
+    admin_user.set_security_reset_key()
+    try:
+        reset_key = admin_user.security_reset_key
+        assert reset_key, "precondition: reset key is set"
+        snapshot = admin_user.to_history_dict()
+        for secret in ("password", "force_reset", "fs_uniquifier", "tf_totp_secret"):
+            assert secret not in snapshot
+        serialized = str(snapshot)
+        assert reset_key not in serialized
+        assert admin_user.password not in serialized
+    finally:
+        admin_user.unset_security_reset_key()