登陆优化

core/backend/apps/accounts/tests.py

@@ -4,7 +4,7 @@ from django.conf import settings
 from django.test import TestCase
 from rest_framework.test import APIClient
 
-from apps.accounts.models import Invitation, Team, TeamMember, User
+from apps.accounts.models import Invitation, LoginSession, Team, TeamMember, User
 from apps.billing.models import CreditAccount, CreditLedger
 
 
@@ -65,6 +65,45 @@ class AuthApiTests(TestCase):
             self.assertEqual(genesis.count(), 0)
 
 
+class SingleDeviceLoginTests(TestCase):
+    def setUp(self):
+        self.user = User.objects.create_user(username="single-device", password="strong-password")
+
+    def test_new_login_invalidates_previous_device_and_session(self):
+        first = APIClient()
+        first_login = first.post(
+            "/api/auth/login/",
+            {"username": self.user.username, "password": "strong-password"},
+            format="json",
+            HTTP_USER_AGENT="Device A",
+            REMOTE_ADDR="10.0.0.1",
+        )
+        self.assertEqual(first_login.status_code, 200)
+        first_token = first_login.data["token"]
+
+        second = APIClient()
+        second_login = second.post(
+            "/api/auth/login/",
+            {"username": self.user.username, "password": "strong-password"},
+            format="json",
+            HTTP_USER_AGENT="Device B",
+            REMOTE_ADDR="10.0.0.2",
+        )
+        self.assertEqual(second_login.status_code, 200)
+        second_token = second_login.data["token"]
+        self.assertNotEqual(second_token, first_token)
+
+        first.credentials(HTTP_AUTHORIZATION=f"Token {first_token}")
+        self.assertEqual(first.get("/api/auth/me/").status_code, 401)
+
+        second.credentials(HTTP_AUTHORIZATION=f"Token {second_token}")
+        self.assertEqual(second.get("/api/auth/me/").status_code, 200)
+
+        active = LoginSession.objects.filter(user=self.user, revoked_at__isnull=True)
+        self.assertEqual(active.count(), 1)
+        self.assertEqual(active.get().user_agent, "Device B")
+
+
 class InvitationFlowTests(TestCase):
     def _register(self, client, username, **extra):
         return client.post(

core/backend/apps/accounts/views.py

@@ -4,6 +4,7 @@ from pathlib import Path
 
 from django.contrib.auth import authenticate
 from django.db import transaction
+from django.utils import timezone
 from rest_framework import status
 from rest_framework.authtoken.models import Token
 from rest_framework.decorators import api_view, parser_classes, permission_classes
@@ -67,8 +68,7 @@ def _client_ip(request):
 
 
 def record_login_session(request, user):
-    """登录成功后记录设备会话(UA / IP)。去重:同一台电脑(UA)+ 同一 IP 视为同一台设备,
-    已存在未下线的同设备会话则只刷新 last_seen_at,不再新增一行(避免「在用设备」列表里同设备重复堆叠)。"""
+    """记录当前登录设备(UA / IP)。同一设备重复写入时只刷新最近活跃时间。"""
     try:
         user_agent = (request.META.get("HTTP_USER_AGENT") or "")[:400]
         ip_address = _client_ip(request)
@@ -87,14 +87,29 @@ def record_login_session(request, user):
         pass
 
 
+def issue_single_device_token(request, user):
+    """签发账号唯一有效 Token。
+
+    锁住用户行后再旋转 Token，确保两个设备同时登录时严格由最后完成的登录获胜；
+    旧设备持有的 Token 会立即在下一次请求时得到 401。LoginSession 仅保留当前
+    设备为 active，供设置页展示和审计。
+    """
+    with transaction.atomic():
+        locked_user = User.objects.select_for_update().get(pk=user.pk)
+        LoginSession.objects.filter(user=locked_user, revoked_at__isnull=True).update(revoked_at=timezone.now())
+        Token.objects.filter(user=locked_user).delete()
+        token = Token.objects.create(user=locked_user)
+        record_login_session(request, locked_user)
+    return token
+
+
 @api_view(["POST"])
 @permission_classes([])
 def register(request):
     serializer = RegisterSerializer(data=request.data)
     serializer.is_valid(raise_exception=True)
     data = serializer.save()
-    token, _ = Token.objects.get_or_create(user=data["user"])
-    record_login_session(request, data["user"])
+    token = issue_single_device_token(request, data["user"])
     return Response(auth_payload(data["user"], data["team"], token), status=status.HTTP_201_CREATED)
 
 
@@ -142,15 +157,17 @@ def login(request):
     # 团队被平台超管停用:其成员一律不准登录(平台超管无团队,不受影响)
     if team is not None and team.status == Team.Status.DISABLED:
         return Response({"detail": "团队已停用,请联系平台管理员"}, status=status.HTTP_400_BAD_REQUEST)
-    token, _ = Token.objects.get_or_create(user=user)
-    record_login_session(request, user)
+    token = issue_single_device_token(request, user)
     return Response(auth_payload(user, team, token))
 
 
 @api_view(["POST"])
 @permission_classes([IsAuthenticated])
 def logout(request):
-    Token.objects.filter(user=request.user).delete()
+    with transaction.atomic():
+        locked_user = User.objects.select_for_update().get(pk=request.user.pk)
+        LoginSession.objects.filter(user=locked_user, revoked_at__isnull=True).update(revoked_at=timezone.now())
+        Token.objects.filter(user=locked_user).delete()
     return Response(status=status.HTTP_204_NO_CONTENT)
 
 
@@ -189,8 +206,7 @@ def change_password(request):
         return Response({"new_password": ["新密码至少 8 位"]}, status=status.HTTP_400_BAD_REQUEST)
     user.set_password(new_password)
     user.save(update_fields=["password"])
-    Token.objects.filter(user=user).delete()
-    token, _ = Token.objects.get_or_create(user=user)
+    token = issue_single_device_token(request, user)
     return Response({"token": token.key})
 
 
@@ -508,8 +524,6 @@ def revoke_login_session(request, session_id):
     只吊销某一台,唯一能真正下线的做法 = 旋转 token(删旧 + 发新),令所有端旧 token
     立即失效,再把新 token 发回当前设备(同 revoke_other_sessions 的思路)。
     被下线的目标设备下次请求即 401,真正离线。"""
-    from django.utils import timezone
-
     target = LoginSession.objects.filter(user=request.user, id=session_id).first()
     if not target:
         return Response({"revoked": 0})
@@ -520,10 +534,7 @@ def revoke_login_session(request, session_id):
         revoked_at__isnull=True,
     ).update(revoked_at=timezone.now())
     # 真正吊销:旋转用户 token(令被下线设备旧 token 立即失效),再发新 token 给当前设备。
-    Token.objects.filter(user=request.user).delete()
-    token, _ = Token.objects.get_or_create(user=request.user)
-    # 刷新当前设备这条会话(旋转后当前设备等价于「重新登录」),避免它被自己下线后又冒出来
-    record_login_session(request, request.user)
+    token = issue_single_device_token(request, request.user)
     return Response({"revoked": updated, "token": token.key})
 
 
@@ -531,10 +542,5 @@ def revoke_login_session(request, session_id):
 @permission_classes([IsAuthenticated])
 def revoke_other_sessions(request):
     """下线除当前外的所有其他设备:旋转 token(令其他端 token 失效)+ 标记会话已下线。"""
-    from django.utils import timezone
-
-    LoginSession.objects.filter(user=request.user, revoked_at__isnull=True).update(revoked_at=timezone.now())
-    Token.objects.filter(user=request.user).delete()
-    token, _ = Token.objects.get_or_create(user=request.user)
-    record_login_session(request, request.user)
+    token = issue_single_device_token(request, request.user)
     return Response({"token": token.key})