diff --git a/open_api_framework/admin.py b/open_api_framework/admin.py new file mode 100644 index 0000000..5bca0ad --- /dev/null +++ b/open_api_framework/admin.py @@ -0,0 +1,40 @@ +from django.contrib import admin + +from sessionprofile.models import SessionProfile + +from open_api_framework.utils import get_session_store + + +@admin.register(SessionProfile) +class SessionProfileAdmin(admin.ModelAdmin): + list_display = ["session_key", "user", "exists"] + + @property + def SessionStore(self): + + return get_session_store() + + def get_queryset(self, request): + qs = super().get_queryset(request) + return qs.filter(user=request.user) + + def has_add_permission(self, request, obj=None): + return False + + def has_change_permission(self, request, obj=None): + return False + + @admin.display(boolean=True) + def exists(self, obj): + return self.SessionStore().exists(obj.session_key) + + def delete_model(self, request, obj): + self.SessionStore(obj.session_key).flush() + super().delete_model(request, obj) + + def delete_queryset(self, request, queryset): + + for session_profile in queryset.iterator(): + self.SessionStore(session_profile.session_key).flush() + + super().delete_queryset(request, queryset) diff --git a/open_api_framework/conf/base.py b/open_api_framework/conf/base.py index 8a507a1..e39a95a 100644 --- a/open_api_framework/conf/base.py +++ b/open_api_framework/conf/base.py @@ -231,19 +231,20 @@ "mozilla_django_oidc_db", "log_outgoing_requests", "django_setup_configuration", + "sessionprofile", "open_api_framework", PROJECT_DIRNAME, ] MIDDLEWARE = [ "django.middleware.security.SecurityMiddleware", + "sessionprofile.middleware.SessionProfileMiddleware", "django.contrib.sessions.middleware.SessionMiddleware", "corsheaders.middleware.CorsMiddleware", "django.middleware.common.CommonMiddleware", "django.middleware.csrf.CsrfViewMiddleware", "django.contrib.auth.middleware.AuthenticationMiddleware", "maykin_2fa.middleware.OTPMiddleware", - "mozilla_django_oidc_db.middleware.SessionRefresh", "django.contrib.messages.middleware.MessageMiddleware", "django.middleware.clickjacking.XFrameOptionsMiddleware", "axes.middleware.AxesMiddleware", @@ -537,6 +538,12 @@ SESSION_COOKIE_NAME = f"{PROJECT_DIRNAME}_sessionid" SESSION_ENGINE = "django.contrib.sessions.backends.cache" +SESSION_COOKIE_AGE = config( + "SESSION_COOKIE_AGE", + default=1209600, + help_text="For how long, in seconds, the session cookie will be valid.", +) + LOGIN_URL = reverse_lazy("admin:login") LOGIN_REDIRECT_URL = reverse_lazy("admin:index") diff --git a/open_api_framework/utils.py b/open_api_framework/utils.py new file mode 100644 index 0000000..8d95d62 --- /dev/null +++ b/open_api_framework/utils.py @@ -0,0 +1,8 @@ +from importlib import import_module + +from django.conf import settings +from django.contrib.sessions.backends.base import SessionBase + + +def get_session_store() -> SessionBase: + return import_module(settings.SESSION_ENGINE).SessionStore diff --git a/pyproject.toml b/pyproject.toml index d21574b..ce8603e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,6 +56,7 @@ dependencies = [ "flower>=2.0.1", "maykin-2fa>=1.0.1", "django-setup-configuration>=0.1.0", + "django-sessionprofile>=3.0.0", ] [project.urls] @@ -72,6 +73,7 @@ tests = [ "isort", "black", "flake8", + "factory-boy", ] coverage = [ "pytest-cov", diff --git a/testapp/settings.py b/testapp/settings.py index 9aecf64..9349b5a 100644 --- a/testapp/settings.py +++ b/testapp/settings.py @@ -44,11 +44,13 @@ "django.contrib.messages", "django.contrib.admin", "open_api_framework", + "sessionprofile", "testapp", ] MIDDLEWARE = [ "django.middleware.security.SecurityMiddleware", + "sessionprofile.middleware.SessionProfileMiddleware", "django.contrib.sessions.middleware.SessionMiddleware", "django.middleware.common.CommonMiddleware", "django.middleware.csrf.CsrfViewMiddleware", @@ -78,3 +80,5 @@ # These are excluded from generate_envvar_docs test by their group VARIABLE_TO_BE_EXCLUDED = config("VARIABLE_TO_BE_EXCLUDED1", "foo", group="Excluded") VARIABLE_TO_BE_EXCLUDED = config("VARIABLE_TO_BE_EXCLUDED2", "bar", group="Excluded") + +SESSION_ENGINE = "django.contrib.sessions.backends.db" diff --git a/tests/factories.py b/tests/factories.py new file mode 100644 index 0000000..170ea43 --- /dev/null +++ b/tests/factories.py @@ -0,0 +1,17 @@ +import factory.fuzzy +from sessionprofile.models import SessionProfile + +from open_api_framework.utils import get_session_store + + +class SessionProfileFactory(factory.django.DjangoModelFactory): + + session_key = factory.fuzzy.FuzzyText(length=40) + + class Meta: + model = SessionProfile + + @factory.post_generation + def session(self, create, extracted, **kwargs): + SessionStore = get_session_store() + SessionStore(self.session_key).save(True) diff --git a/tests/test_admin.py b/tests/test_admin.py new file mode 100644 index 0000000..e31d519 --- /dev/null +++ b/tests/test_admin.py @@ -0,0 +1,197 @@ +from django.contrib.sessions.backends.cache import SessionStore as CachedSessionStore +from django.contrib.sessions.backends.db import SessionStore as DBSessionStore +from django.contrib.sessions.models import Session +from django.test import override_settings +from django.urls import reverse + +import pytest +from sessionprofile.models import SessionProfile + +from open_api_framework.utils import get_session_store + +from .factories import SessionProfileFactory + + +@pytest.fixture +def session_changelist_url(): + return reverse("admin:sessionprofile_sessionprofile_changelist") + + +def test_session_profile_sanity(client, admin_user, session_changelist_url): + + client.force_login(admin_user) + response = client.get(session_changelist_url) + assert response.status_code == 200 + + assert SessionProfile.objects.count() == 1 + + session = SessionProfile.objects.get() + assert client.session.session_key == session.session_key + + +def test_only_session_profile_of_user_shown( + client, admin_user, django_user_model, session_changelist_url +): + + other_admin = django_user_model.objects.create_superuser("garry") + + client.force_login(other_admin) + response = client.get(session_changelist_url) + assert response.status_code == 200 + + client.force_login(admin_user) + response = client.get(session_changelist_url) + assert response.status_code == 200 + + # two sessions, one for each user + assert SessionProfile.objects.count() == 2 + + # Session created after response, needs to be called again + response = client.get(session_changelist_url) + + admin_user_session = SessionProfile.objects.get(user=admin_user) + assert admin_user_session.session_key in response.content.decode() + + other_user_session = SessionProfile.objects.get(user=other_admin) + assert other_user_session.session_key not in response.content.decode() + + # should only be able to access own page + change_url = reverse( + "admin:sessionprofile_sessionprofile_change", + args=[admin_user_session.session_key], + ) + response = client.get(change_url) + assert response.status_code == 200 + + change_url = reverse( + "admin:sessionprofile_sessionprofile_change", + args=[other_user_session.session_key], + ) + response = client.get(change_url) + assert response.status_code == 302 + assert response.url == reverse("admin:index") + + +def test_cant_delete_other_users_session(client, admin_user, django_user_model): + client.force_login(admin_user) + + other_admin = django_user_model.objects.create_superuser("garry") + + other_user_session = SessionProfileFactory(user=other_admin) + + delete_url = reverse( + "admin:sessionprofile_sessionprofile_delete", + args=[other_user_session.session_key], + ) + + response = client.post(delete_url, {"post": "yes"}) + assert response.status_code == 302 + + SessionStore = get_session_store() + + assert SessionStore().exists(other_user_session.session_key) + + +def test_delete_with_session_db_backend(client, admin_user, session_changelist_url): + client.force_login(admin_user) + + session = SessionProfileFactory(user=admin_user) + + assert SessionProfile.objects.count() == 1 + # sesison created by login + assert Session.objects.count() == 2 + assert DBSessionStore().exists(session.session_key) + + url = reverse("admin:sessionprofile_sessionprofile_delete", args=[session.pk]) + + response = client.post(url, {"post": "yes"}) + assert response.status_code == 302 + + # new session saved upon request + assert SessionProfile.objects.count() == 1 + assert SessionProfile.objects.count() != session + assert Session.objects.count() == 1 + assert not DBSessionStore().exists(session.session_key) + + +@override_settings(SESSION_ENGINE="django.contrib.sessions.backends.cache") +def test_delete_with_session_cache_backend(client, admin_user, session_changelist_url): + client.force_login(admin_user) + + session = SessionProfileFactory(user=admin_user) + + assert SessionProfile.objects.count() == 1 + assert Session.objects.count() == 0 + assert CachedSessionStore().exists(session.session_key) + + url = reverse("admin:sessionprofile_sessionprofile_delete", args=[session.pk]) + + response = client.post(url, {"post": "yes"}) + assert response.status_code == 302 + + # new session saved upon request + assert SessionProfile.objects.count() == 1 + assert SessionProfile.objects.count() != session + assert Session.objects.count() == 0 + assert not CachedSessionStore().exists(session.session_key) + + +def test_delete_action_with_session_db_backend( + client, admin_user, session_changelist_url +): + client.force_login(admin_user) + sessions = SessionProfileFactory.create_batch(5, user=admin_user) + + # one created from user login + assert Session.objects.count() == 6 + assert SessionProfile.objects.count() == 5 + + session_keys = [session.session_key for session in sessions] + for session_key in session_keys: + assert DBSessionStore().exists(session_key) + + response = client.post( + session_changelist_url, + {"action": "delete_selected", "_selected_action": session_keys, "post": "yes"}, + ) + assert response.status_code == 302 + + # one is created as the post request is sent + assert SessionProfile.objects.count() == 1 + assert Session.objects.count() == 1 + + for session_key in session_keys: + assert not DBSessionStore().exists(session_key) + + +@override_settings(SESSION_ENGINE="django.contrib.sessions.backends.cache") +def test_delete_action_with_session_cache_backend( + client, admin_user, session_changelist_url +): + + client.force_login(admin_user) + sessions = SessionProfileFactory.create_batch(5, user=admin_user) + + # no db sessions are created + assert Session.objects.count() == 0 + assert SessionProfile.objects.count() == 5 + + session_keys = [session.session_key for session in sessions] + + # sessions are created + for session_key in session_keys: + assert CachedSessionStore().exists(session_key) + + response = client.post( + session_changelist_url, + {"action": "delete_selected", "_selected_action": session_keys, "post": "yes"}, + ) + assert response.status_code == 302 + + # one is created as the post request is sent + assert SessionProfile.objects.count() == 1 + assert Session.objects.count() == 0 + + # sessions should be deleted + for session_key in session_keys: + assert not CachedSessionStore().exists(session_key)