feat(usersessions): User session management

ChangeLog.rst

@@ -16,6 +16,9 @@ Note worthy changes
   available that can be used to prevent users from deactivating e.g. their TOTP
   authenticator.
 
+- Added a new app, user sessions, allowing users to view a list of all their
+  active sessions, as well as offering a means to end these sessions.
+
 
 Backwards incompatible changes
 ------------------------------

allauth/app_settings.py

@@ -22,6 +22,10 @@ def SOCIALACCOUNT_ENABLED(self):
     def MFA_ENABLED(self):
         return apps.is_installed("allauth.mfa")
 
+    @property
+    def USERSESSIONS_ENABLED(self):
+        return apps.is_installed("allauth.usersessions")
+
 
 _app_settings = AppSettings("ALLAUTH_")
 

allauth/templates/allauth/elements/table.html

@@ -0,0 +1,5 @@
+{% load allauth %}
+<table>
+    {% slot %}
+    {% endslot %}
+</table>

allauth/templates/usersessions/base_manage.html

@@ -0,0 +1 @@
+{% extends "allauth/layouts/manage.html" %}

allauth/templates/usersessions/messages/sessions_logged_out.txt

@@ -0,0 +1,2 @@
+{% load i18n %}
+{% blocktranslate %}Signed out of all other sessions.{% endblocktranslate %}

allauth/templates/usersessions/usersession_list.html

@@ -0,0 +1,69 @@
+{% extends "usersessions/base_manage.html" %}
+{% load allauth %}
+{% load i18n %}
+{% load humanize %}
+{% block head_title %}
+    {% trans "Sessions" %}
+{% endblock head_title %}
+{% block content %}
+    {% element h1 tags="usersessions,list" %}
+        {% trans "Sessions" %}
+    {% endelement %}
+    {% if session_count > 1 %}
+        {% url 'usersessions_list' as action_url %}
+    {% else %}
+        {% url 'account_logout' as action_url %}
+    {% endif %}
+    {% element form action=action_url method="post" tags="sessions" no_visible_fields=True %}
+        {% slot body %}
+            {% csrf_token %}
+            {% element table tags="sessions" %}
+                <thead>
+                    <tr>
+                        <th>{% translate "Started At" %}</th>
+                        <th>{% translate "IP Address" %}</th>
+                        <th>{% translate "Browser" %}</th>
+                        {% if show_last_seen_at %}
+                            <th>{% translate "Last seen at" %}</th>
+                        {% endif %}
+                    </tr>
+                </thead>
+                <tbody>
+                    {% for session in sessions %}
+                        <tr>
+                            <td>
+                                <span title="{{ session.created_at }}">{{ session.created_at|naturaltime }}</span>
+                            </td>
+                            <td>{{ session.ip }}</td>
+                            <td>{{ session.user_agent }}</td>
+                            {% if show_last_seen_at %}
+                                <td>
+                                    <span title="{{ session.last_seen_at }}">{{ session.last_seen_at|naturaltime }}</span>
+                                </td>
+                            {% endif %}
+                            <td>
+                                {% if session.is_current %}
+                                    {% element badge tags="session,current" %}
+                                        {% translate "Current" %}
+                                    {% endelement %}
+                                {% else %}
+                                {% endif %}
+                            </td>
+                        </tr>
+                    {% endfor %}
+                </tbody>
+            {% endelement %}
+        {% endslot %}
+        {% slot actions %}
+            {% if session_count > 1 %}
+                {% element button type="submit" %}
+                    {% translate "Sign Out Other Sessions" %}
+                {% endelement %}
+            {% else %}
+                {% element button type="submit" %}
+                    {% translate "Sign Out" %}
+                {% endelement %}
+            {% endif %}
+        {% endslot %}
+    {% endelement %}
+{% endblock content %}

allauth/urls.py

@@ -14,6 +14,9 @@
 if app_settings.SOCIALACCOUNT_ENABLED:
     urlpatterns += [path("social/", include("allauth.socialaccount.urls"))]
 
+if app_settings.USERSESSIONS_ENABLED:
+    urlpatterns += [path("sessions/", include("allauth.usersessions.urls"))]
+
 # Provider urlpatterns, as separate attribute (for reusability).
 provider_urlpatterns = []
 provider_classes = providers.registry.get_class_list()

allauth/usersessions/adapter.py

@@ -0,0 +1,18 @@
+from allauth.usersessions import app_settings
+from allauth.utils import import_attribute
+
+
+class DefaultUserSessionsAdapter:
+    """The adapter class allows you to override various functionality of the
+    ``allauth.usersessions`` app.  To do so, point ``settings.USERSESSIONS_ADAPTER`` to your own
+    class that derives from ``DefaultUserSessionsAdapter`` and override the behavior by
+    altering the implementation of the methods according to your own need.
+    """
+
+    def end_sessions(self, sessions):
+        for session in sessions:
+            session.end()
+
+
+def get_adapter():
+    return import_attribute(app_settings.ADAPTER)()

allauth/usersessions/admin.py

@@ -0,0 +1,9 @@
+from django.contrib import admin
+
+from allauth.usersessions.models import UserSession
+
+
+@admin.register(UserSession)
+class UserSessionAdmin(admin.ModelAdmin):
+    raw_id_fields = ("user",)
+    list_display = ("user", "created_at", "last_seen_at", "ip", "user_agent")

allauth/usersessions/app_settings.py

@@ -0,0 +1,30 @@
+class AppSettings(object):
+    def __init__(self, prefix):
+        self.prefix = prefix
+
+    def _setting(self, name, dflt):
+        from allauth.utils import get_setting
+
+        return get_setting(self.prefix + name, dflt)
+
+    @property
+    def ADAPTER(self):
+        return self._setting(
+            "ADAPTER", "allauth.usersessions.adapter.DefaultUserSessionsAdapter"
+        )
+
+    @property
+    def TRACK_ACTIVITY(self):
+        """Whether or not sessions are to be actively tracked. When tracking is
+        enabled, the last seen IP address and last seen timestamp will be kept
+        track of.
+        """
+        return self._setting("TRACK_ACTIVITY", False)
+
+
+_app_settings = AppSettings("USERSESSIONS_")
+
+
+def __getattr__(name):
+    # See https://peps.python.org/pep-0562/
+    return getattr(_app_settings, name)

allauth/usersessions/apps.py

@@ -0,0 +1,14 @@
+from django.apps import AppConfig
+from django.utils.translation import gettext_lazy as _
+
+
+class UserSessionsConfig(AppConfig):
+    name = "allauth.usersessions"
+    verbose_name = _("User Sessions")
+    default_auto_field = "django.db.models.BigAutoField"
+
+    def ready(self):
+        from allauth.account.signals import user_logged_in
+        from allauth.usersessions import signals
+
+        user_logged_in.connect(receiver=signals.on_user_logged_in)

allauth/usersessions/forms.py

@@ -0,0 +1,18 @@
+from django import forms
+
+from allauth.usersessions.adapter import get_adapter
+from allauth.usersessions.models import UserSession
+
+
+class ManageUserSessionsForm(forms.Form):
+    def __init__(self, *args, **kwargs):
+        self.request = kwargs.pop("request")
+        super().__init__(*args, **kwargs)
+
+    def save(self, request):
+        sessions_to_end = []
+        for session in UserSession.objects.filter(user=request.user):
+            if session.is_current():
+                continue
+            sessions_to_end.append(session)
+        get_adapter().end_sessions(sessions_to_end)

allauth/usersessions/middleware.py

@@ -0,0 +1,13 @@
+from allauth.usersessions import app_settings
+from allauth.usersessions.models import UserSession
+
+
+class UserSessionsMiddleware:
+    def __init__(self, get_response):
+        self.get_response = get_response
+
+    def __call__(self, request):
+        if app_settings.TRACK_ACTIVITY:
+            UserSession.objects.create_from_request(request)
+        response = self.get_response(request)
+        return response

allauth/usersessions/migrations/0001_initial.py

@@ -0,0 +1,55 @@
+# Generated by Django 4.2.6 on 2023-12-05 11:44
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+import django.utils.timezone
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="UserSession",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("created_at", models.DateTimeField(default=django.utils.timezone.now)),
+                ("ip", models.GenericIPAddressField()),
+                (
+                    "last_seen_at",
+                    models.DateTimeField(default=django.utils.timezone.now),
+                ),
+                (
+                    "session_key",
+                    models.CharField(
+                        editable=False,
+                        max_length=40,
+                        unique=True,
+                        verbose_name="session key",
+                    ),
+                ),
+                ("user_agent", models.CharField(max_length=200)),
+                ("data", models.JSONField(default=dict)),
+                (
+                    "user",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+        ),
+    ]

allauth/usersessions/models.py

@@ -0,0 +1,72 @@
+from importlib import import_module
+
+from django.conf import settings
+from django.db import models
+from django.utils import timezone
+from django.utils.translation import gettext_lazy as _
+
+from allauth.account.adapter import get_adapter
+from allauth.core import context
+
+
+class UserSessionManager(models.Manager):
+    def purge_and_list(self, user):
+        ret = []
+        sessions = UserSession.objects.filter(user=user)
+        for session in sessions.iterator():
+            if not session.purge():
+                ret.append(session)
+        return ret
+
+    def create_from_request(self, request):
+        if not request.user.is_authenticated or not request.session.session_key:
+            raise ValueError()
+        ua = request.META.get("HTTP_USER_AGENT", "")[
+            0 : UserSession._meta.get_field("user_agent").max_length
+        ]
+        UserSession.objects.update_or_create(
+            user=request.user,
+            session_key=request.session.session_key,
+            defaults=dict(
+                ip=get_adapter().get_client_ip(request),
+                user_agent=ua,
+                last_seen_at=timezone.now(),
+            ),
+        )
+
+
+class UserSession(models.Model):
+    objects = UserSessionManager()
+
+    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
+    created_at = models.DateTimeField(default=timezone.now)
+    ip = models.GenericIPAddressField()
+    last_seen_at = models.DateTimeField(default=timezone.now)
+    session_key = models.CharField(
+        _("session key"), max_length=40, unique=True, editable=False
+    )
+    user_agent = models.CharField(max_length=200)
+    data = models.JSONField(default=dict)
+
+    def __str__(self):
+        return f"{self.ip} ({self.user_agent})"
+
+    def exists(self):
+        engine = import_module(settings.SESSION_ENGINE)
+        store = engine.SessionStore()
+        return store.exists(self.session_key)
+
+    def purge(self):
+        if not self.exists():
+            self.delete()
+            return True
+        return False
+
+    def is_current(self):
+        return self.session_key == context.request.session.session_key
+
+    def end(self):
+        engine = import_module(settings.SESSION_ENGINE)
+        store = engine.SessionStore()
+        store.delete(self.session_key)
+        self.delete()

allauth/usersessions/signals.py

@@ -0,0 +1,6 @@
+from .models import UserSession
+
+
+def on_user_logged_in(sender, **kwargs):
+    request = kwargs["request"]
+    UserSession.objects.create_from_request(request)

allauth/usersessions/tests/test_views.py

@@ -0,0 +1,32 @@
+from django.test import Client
+from django.urls import reverse
+
+from allauth.usersessions.models import UserSession
+
+
+def test_overall_flow(user, user_password):
+    firefox = Client(HTTP_USER_AGENT="Mozilla Firefox")
+    nyxt = Client(HTTP_USER_AGENT="Nyxt")
+    for client in [firefox, nyxt]:
+        resp = client.post(
+            reverse("account_login"),
+            {"login": user.username, "password": user_password},
+        )
+        assert resp.status_code == 302
+    assert UserSession.objects.filter(user=user).count() == 2
+    sessions = list(UserSession.objects.filter(user=user).order_by("pk"))
+    assert sessions[0].user_agent == "Mozilla Firefox"
+    assert sessions[1].user_agent == "Nyxt"
+    for client in [firefox, nyxt]:
+        resp = firefox.get(reverse("usersessions_list"))
+        assert resp.status_code == 200
+    resp = firefox.post(reverse("usersessions_list"))
+    assert resp.status_code == 302
+    assert UserSession.objects.filter(user=user).count() == 1
+    assert UserSession.objects.filter(user=user, pk=sessions[0].pk).exists()
+    assert not UserSession.objects.filter(user=user, pk=sessions[1].pk).exists()
+    resp = nyxt.get(reverse("usersessions_list"))
+    assert resp.status_code == 302
+    assert resp["location"] == reverse("account_login") + "?next=" + reverse(
+        "usersessions_list"
+    )

allauth/usersessions/urls.py

@@ -0,0 +1,8 @@
+from django.urls import path
+
+from allauth.usersessions import views
+
+
+urlpatterns = [
+    path("", views.list_usersessions, name="usersessions_list"),
+]