fix: Revoke DRF token after password change

api/app/settings/common.py

@@ -955,6 +955,7 @@
         "confirmation": "users.emails.ConfirmationEmail",
     },
     "SET_PASSWORD_RETYPE": True,
+    "LOGOUT_ON_PASSWORD_CHANGE": True,
     "PASSWORD_RESET_CONFIRM_RETYPE": True,
     "HIDE_USERS": True,
     "PERMISSIONS": {

api/custom_auth/jwt_cookie/services.py

@@ -1,5 +1,10 @@
 from django.conf import settings
+from django.utils import timezone
 from rest_framework.response import Response
+from rest_framework_simplejwt.token_blacklist.models import (
+    BlacklistedToken,
+    OutstandingToken,
+)
 from rest_framework_simplejwt.tokens import SlidingToken
 
 from custom_auth.jwt_cookie.constants import JWT_SLIDING_COOKIE_KEY
@@ -16,3 +21,15 @@ def authorise_response(user: FFAdminUser, response: Response) -> Response:
         samesite=settings.COOKIE_SAME_SITE,  # type: ignore[arg-type]
     )
     return response
+
+
+def blacklist_user_tokens(user: FFAdminUser) -> None:
+    outstanding_tokens = OutstandingToken.objects.filter(
+        user=user,
+        expires_at__gt=timezone.now(),
+        blacklistedtoken__isnull=True,
+    )
+    BlacklistedToken.objects.bulk_create(
+        [BlacklistedToken(token=token) for token in outstanding_tokens],
+        ignore_conflicts=True,
+    )

api/custom_auth/views.py

@@ -17,7 +17,7 @@
 from rest_framework.status import HTTP_204_NO_CONTENT
 from rest_framework.throttling import ScopedRateThrottle
 
-from custom_auth.jwt_cookie.services import authorise_response
+from custom_auth.jwt_cookie.services import authorise_response, blacklist_user_tokens
 from custom_auth.mfa.backends.application import CustomApplicationBackend
 from custom_auth.mfa.trench.command.authenticate_second_factor import (
     authenticate_second_step_command,
@@ -231,3 +231,23 @@ def reset_password(self, request: Request, *args: Any, **kwargs: Any) -> Respons
             UserPasswordResetRequest.objects.create(user=user)
 
         return Response(status=status.HTTP_204_NO_CONTENT)
+
+    @action(["post"], detail=False)
+    def set_password(self, request: Request, *args: Any, **kwargs: Any) -> Response:
+        user = request.user
+        assert isinstance(user, FFAdminUser)
+        response: Response = super().set_password(request, *args, **kwargs)
+        blacklist_user_tokens(user)
+        return response
+
+    @action(["post"], detail=False)
+    def reset_password_confirm(
+        self, request: Request, *args: Any, **kwargs: Any
+    ) -> Response:
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        user = serializer.user
+        response: Response = super().reset_password_confirm(request, *args, **kwargs)
+        Token.objects.filter(user=user).delete()
+        blacklist_user_tokens(user)
+        return response

api/tests/unit/custom_auth/jwt_cookie/test_unit_jwt_cookie_services.py

@@ -0,0 +1,42 @@
+from datetime import timedelta
+
+from django.conf import settings
+from django.utils import timezone
+from freezegun import freeze_time
+from rest_framework_simplejwt.token_blacklist.models import BlacklistedToken
+from rest_framework_simplejwt.tokens import SlidingToken
+
+from custom_auth.jwt_cookie.services import blacklist_user_tokens
+from users.models import FFAdminUser
+
+
+def test_blacklist_user_tokens__outstanding_tokens__blacklists_only_user_tokens(
+    staff_user: FFAdminUser,
+    admin_user: FFAdminUser,
+) -> None:
+    # Given
+    SlidingToken.for_user(staff_user)
+    SlidingToken.for_user(staff_user)
+    SlidingToken.for_user(admin_user)
+
+    # When
+    blacklist_user_tokens(staff_user)
+
+    # Then
+    assert BlacklistedToken.objects.filter(token__user=staff_user).count() == 2
+    assert not BlacklistedToken.objects.filter(token__user=admin_user).exists()
+
+
+def test_blacklist_user_tokens__expired_token__skips_token(
+    staff_user: FFAdminUser,
+) -> None:
+    # Given
+    SlidingToken.for_user(staff_user)
+    token_lifetime = settings.SIMPLE_JWT["SLIDING_TOKEN_LIFETIME"]
+
+    # When
+    with freeze_time(timezone.now() + token_lifetime + timedelta(seconds=1)):
+        blacklist_user_tokens(staff_user)
+
+    # Then
+    assert not BlacklistedToken.objects.exists()

api/tests/unit/custom_auth/test_unit_custom_auth_views.py

@@ -3,14 +3,21 @@
 from unittest.mock import MagicMock
 
 import pytest
+from django.contrib.auth.hashers import PBKDF2PasswordHasher, make_password
+from django.contrib.auth.tokens import default_token_generator
+from django.db import DatabaseError
 from django.urls import reverse
+from djoser import utils  # type: ignore[import-untyped]
 from freezegun import freeze_time
 from pytest_django.fixtures import SettingsWrapper
 from pytest_mock import MockerFixture
 from rest_framework import status
+from rest_framework.authtoken.models import Token
 from rest_framework.test import APIClient
 from rest_framework.throttling import ScopedRateThrottle
+from rest_framework_simplejwt.tokens import SlidingToken
 
+from custom_auth.jwt_cookie.constants import JWT_SLIDING_COOKIE_KEY
 from users.models import FFAdminUser, HubspotTracker
 
 
@@ -273,6 +280,206 @@ def test_create_user__no_hubspot_cookie__does_not_create_hubspot_tracker(
     mock_create_hubspot_contact_for_user.delay.assert_called_once_with(args=(user.id,))
 
 
+def test_set_password__token_auth__revokes_token_for_all_sessions(
+    staff_user: FFAdminUser,
+) -> None:
+    # Given
+    staff_user.password = make_password("old-password")
+    staff_user.save()
+    token = Token.objects.create(user=staff_user)
+    current_client = APIClient()
+    current_client.credentials(HTTP_AUTHORIZATION=f"Token {token.key}")
+    other_client = APIClient()
+    other_client.credentials(HTTP_AUTHORIZATION=f"Token {token.key}")
+    url = reverse("api-v1:custom_auth:ffadminuser-set-password")
+
+    # When
+    response = current_client.post(
+        url,
+        data={
+            "current_password": "old-password",
+            "new_password": "new-password-123!",
+            "re_new_password": "new-password-123!",
+        },
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+    assert not Token.objects.filter(user=staff_user).exists()
+    me_url = reverse("api-v1:custom_auth:ffadminuser-me")
+    assert other_client.get(me_url).status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_reset_password_confirm__token_auth__revokes_token(
+    staff_user: FFAdminUser,
+) -> None:
+    # Given
+    token = Token.objects.create(user=staff_user)
+    authenticated_client = APIClient()
+    authenticated_client.credentials(HTTP_AUTHORIZATION=f"Token {token.key}")
+    url = reverse("api-v1:custom_auth:ffadminuser-reset-password-confirm")
+
+    # When
+    response = APIClient().post(
+        url,
+        data={
+            "uid": utils.encode_uid(staff_user.pk),
+            "token": default_token_generator.make_token(staff_user),
+            "new_password": "new-password-123!",
+            "re_new_password": "new-password-123!",
+        },
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+    assert not Token.objects.filter(user=staff_user).exists()
+    me_url = reverse("api-v1:custom_auth:ffadminuser-me")
+    assert authenticated_client.get(me_url).status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_set_password__jwt_cookie_auth__revokes_all_sessions(
+    staff_user: FFAdminUser,
+) -> None:
+    # Given
+    staff_user.password = make_password("old-password")
+    staff_user.save()
+    current_client = APIClient()
+    current_client.cookies[JWT_SLIDING_COOKIE_KEY] = str(
+        SlidingToken.for_user(staff_user)
+    )
+    other_client = APIClient()
+    other_client.cookies[JWT_SLIDING_COOKIE_KEY] = str(
+        SlidingToken.for_user(staff_user)
+    )
+    url = reverse("api-v1:custom_auth:ffadminuser-set-password")
+
+    # When
+    response = current_client.post(
+        url,
+        data={
+            "current_password": "old-password",
+            "new_password": "new-password-123!",
+            "re_new_password": "new-password-123!",
+        },
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+    projects_url = reverse("api-v1:projects:project-list")
+    assert current_client.get(projects_url).status_code == status.HTTP_401_UNAUTHORIZED
+    assert other_client.get(projects_url).status_code == status.HTTP_401_UNAUTHORIZED
+
+
+def test_reset_password_confirm__jwt_cookie_auth__revokes_all_sessions(
+    staff_user: FFAdminUser,
+) -> None:
+    # Given
+    authenticated_client = APIClient()
+    authenticated_client.cookies[JWT_SLIDING_COOKIE_KEY] = str(
+        SlidingToken.for_user(staff_user)
+    )
+    url = reverse("api-v1:custom_auth:ffadminuser-reset-password-confirm")
+
+    # When
+    response = APIClient().post(
+        url,
+        data={
+            "uid": utils.encode_uid(staff_user.pk),
+            "token": default_token_generator.make_token(staff_user),
+            "new_password": "new-password-123!",
+            "re_new_password": "new-password-123!",
+        },
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+    projects_url = reverse("api-v1:projects:project-list")
+    assert (
+        authenticated_client.get(projects_url).status_code
+        == status.HTTP_401_UNAUTHORIZED
+    )
+
+
+def test_set_password__save_fails__keeps_token(
+    staff_user: FFAdminUser,
+    mocker: MagicMock,
+) -> None:
+    # Given
+    staff_user.password = make_password("old-password")
+    staff_user.save()
+    token = Token.objects.create(user=staff_user)
+    client = APIClient()
+    client.credentials(HTTP_AUTHORIZATION=f"Token {token.key}")
+    url = reverse("api-v1:custom_auth:ffadminuser-set-password")
+    mocker.patch.object(FFAdminUser, "save", side_effect=DatabaseError)
+
+    # When
+    with pytest.raises(DatabaseError):
+        client.post(
+            url,
+            data={
+                "current_password": "old-password",
+                "new_password": "new-password-123!",
+                "re_new_password": "new-password-123!",
+            },
+        )
+
+    # Then
+    assert Token.objects.filter(key=token.key).exists()
+
+
+def test_reset_password_confirm__save_fails__keeps_token(
+    staff_user: FFAdminUser,
+    mocker: MagicMock,
+) -> None:
+    # Given
+    token = Token.objects.create(user=staff_user)
+    url = reverse("api-v1:custom_auth:ffadminuser-reset-password-confirm")
+    mocker.patch.object(FFAdminUser, "save", side_effect=DatabaseError)
+
+    # When
+    with pytest.raises(DatabaseError):
+        APIClient().post(
+            url,
+            data={
+                "uid": utils.encode_uid(staff_user.pk),
+                "token": default_token_generator.make_token(staff_user),
+                "new_password": "new-password-123!",
+                "re_new_password": "new-password-123!",
+            },
+        )
+
+    # Then
+    assert Token.objects.filter(key=token.key).exists()
+
+
+def test_login__outdated_password_hash__keeps_token(
+    staff_user: FFAdminUser,
+    settings: SettingsWrapper,
+) -> None:
+    # Given
+    settings.PASSWORD_HASHERS = [
+        "django.contrib.auth.hashers.MD5PasswordHasher",
+        "django.contrib.auth.hashers.PBKDF2PasswordHasher",
+    ]
+    hasher = PBKDF2PasswordHasher()
+    staff_user.password = hasher.encode("password", hasher.salt(), iterations=1)
+    staff_user.save()
+    token = Token.objects.create(user=staff_user)
+    url = reverse("api-v1:custom_auth:custom-mfa-authtoken-login")
+
+    # When
+    response = APIClient().post(
+        url, data={"email": staff_user.email, "password": "password"}
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+    staff_user.refresh_from_db()
+    assert staff_user.password.startswith("md5$")
+    assert Token.objects.filter(key=token.key).exists()
+
+
 @pytest.mark.parametrize(
     "signup_data, expected_onboarding_data",
     [

frontend/common/constants.ts

@@ -91,6 +91,9 @@ const keywordsReactNative = {
   ...keywords,
   NPM_CLIENT: '@flagsmith/react-native',
 }
+// Set before logging out after a password change, so the login page can say why.
+export const PASSWORD_CHANGED_SESSION_KEY = 'password_changed'
+
 const Constants = {
   archivedTag: { color: '#8f8f8f', label: 'Archived' },
   codeHelp: {