🔒 fix(identity): код TOTP принимается один раз

Отметка `totp_last_used_at` писалась, но ни с чем не сравнивалась. Код живёт
30 секунд и принимается с окном ±1 интервал, значит подсмотренный код работал
второй раз ещё полторы минуты — RFC 6238 §5.2 требует запоминать интервал.
Появилось поле `totp_last_counter`: `verify_totp` возвращает номер принятого
интервала, и коды до него включительно больше не проходят. Тем же путём идёт
подтверждение при настройке 2FA — код, которым её включили, на входе не
сработает.

Заодно два соседних. Шаг с кодом доставал пользователя без `is_active`: пароль
приняли раньше, и сотрудник, отключённый между шагами, всё равно входил.
И у начатого входа не было срока — брошенная вкладка ждала код сколько угодно
долго; теперь пять минут.

Три теста.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

apps/backend/chatballs/identity/auth/sessions.py

@@ -1,3 +1,5 @@
+import time
+
 from django.contrib.auth import authenticate, login, logout
 from django.utils.decorators import method_decorator
 from django.views.decorators.csrf import csrf_protect, ensure_csrf_cookie
@@ -9,7 +11,7 @@ from rest_framework.views import APIView
 
 from chatballs.identity.audit import record_audit_event
 from chatballs.identity.auth.common import _challenge_payload, _user_payload
-from chatballs.identity.auth.totp_utils import TOTP_SESSION_KEY
+from chatballs.identity.auth.totp_utils import TOTP_SESSION_KEY, TOTP_STARTED_KEY
 from chatballs.identity.models import AuditResult
 from chatballs.identity.sessions import remember_device
 
@@ -36,12 +38,15 @@ class LoginView(APIView):
         email = str(body.get("email", ""))
         password = str(body.get("password", ""))
         request.session.pop(TOTP_SESSION_KEY, None)
+        request.session.pop(TOTP_STARTED_KEY, None)
         user = authenticate(request, username=email, password=password)
         if user is None:
             record_audit_event(action="identity.login_failed", result=AuditResult.DENIED, request=request)
             return Response({"detail": "Invalid credentials"}, status=401)
         if user.totp_enabled:
             request.session[TOTP_SESSION_KEY] = user.id
+            # Шаг с кодом ждёт не вечно: см. TOTP_CHALLENGE_TTL_SECONDS.
+            request.session[TOTP_STARTED_KEY] = time.time()
             record_audit_event(
                 action="identity.login_totp_required",
                 actor=user,

apps/backend/chatballs/identity/auth/totp.py

@@ -1,7 +1,7 @@
+import time
 from urllib.parse import quote
 
 from django.contrib.auth import login
-from django.utils import timezone
 from django.utils.decorators import method_decorator
 from django.views.decorators.csrf import csrf_protect
 from rest_framework.permissions import AllowAny, IsAuthenticated
@@ -13,16 +13,31 @@ from rest_framework.views import APIView
 from chatballs.identity.audit import record_audit_event
 from chatballs.identity.auth.common import _user_payload
 from chatballs.identity.auth.totp_utils import (
+    TOTP_CHALLENGE_TTL_SECONDS,
     TOTP_ISSUER,
     TOTP_PERIOD_SECONDS,
     TOTP_SESSION_KEY,
+    TOTP_STARTED_KEY,
     _ensure_totp_secret,
-    _verify_totp,
+    accept_totp_code,
 )
 from chatballs.identity.models import AuditResult, HumanUser
 from chatballs.identity.sessions import remember_device
 
 
+def _drop_challenge(request: Request) -> None:
+    request.session.pop(TOTP_SESSION_KEY, None)
+    request.session.pop(TOTP_STARTED_KEY, None)
+
+
+def _challenge_expired(request: Request) -> bool:
+    """Начатый вход, к которому не вернулись, перестаёт ждать код."""
+    started = request.session.get(TOTP_STARTED_KEY)
+    if not isinstance(started, int | float):
+        return True
+    return time.time() - started > TOTP_CHALLENGE_TTL_SECONDS
+
+
 class TotpSetupView(APIView):
     permission_classes = [IsAuthenticated]
 
@@ -51,8 +66,10 @@ class TotpConfirmView(APIView):
     permission_classes = [IsAuthenticated]
 
     def post(self, request: Request) -> Response:
-        secret = _ensure_totp_secret(request.user)
-        if not _verify_totp(secret, str(request.data.get("code", ""))):
+        _ensure_totp_secret(request.user)
+        # Тот же приём кода, что и при входе: интервал запоминается, поэтому
+        # код, которым включили 2FA, не сработает ещё раз на входе.
+        if not accept_totp_code(request.user, str(request.data.get("code", ""))):
             record_audit_event(
                 action="identity.totp_setup_failed",
                 actor=request.user,
@@ -62,8 +79,7 @@ class TotpConfirmView(APIView):
             return Response({"detail": "Invalid TOTP code"}, status=400)
 
         request.user.totp_enabled = True
-        request.user.totp_last_used_at = timezone.now()
-        request.user.save(update_fields=["totp_enabled", "totp_last_used_at"])
+        request.user.save(update_fields=["totp_enabled"])
         record_audit_event(
             action="identity.totp_enabled",
             actor=request.user,
@@ -81,21 +97,22 @@ class TotpVerifyView(APIView):
 
     def post(self, request: Request) -> Response:
         pending_user_id = request.session.get(TOTP_SESSION_KEY)
-        if not pending_user_id:
+        if not pending_user_id or _challenge_expired(request):
+            _drop_challenge(request)
             return Response({"detail": "TOTP challenge is not active"}, status=401)
 
         try:
+            # is_active обязателен: пароль приняли раньше, и без этой проверки
+            # отключённый между шагами сотрудник всё равно вошёл бы.
             user = HumanUser.objects.prefetch_related("memberships__organization").get(
-                id=pending_user_id
+                id=pending_user_id, is_active=True
             )
         except HumanUser.DoesNotExist:
-            request.session.pop(TOTP_SESSION_KEY, None)
+            _drop_challenge(request)
             return Response({"detail": "TOTP challenge is not active"}, status=401)
 
-        if (
-            not user.totp_enabled
-            or not user.totp_secret
-            or not _verify_totp(user.totp_secret, str(request.data.get("code", "")))
+        if not user.totp_enabled or not accept_totp_code(
+            user, str(request.data.get("code", ""))
         ):
             record_audit_event(
                 action="identity.totp_verify_failed",
@@ -105,10 +122,9 @@ class TotpVerifyView(APIView):
             )
             return Response({"detail": "Invalid TOTP code"}, status=400)
 
-        request.session.pop(TOTP_SESSION_KEY, None)
-        # Отметка «последний код принят …» в карточке 2FA (кадр P1).
-        user.totp_last_used_at = timezone.now()
-        user.save(update_fields=["totp_last_used_at"])
+        # Отметку «последний код принят …» для карточки 2FA (кадр P1) уже
+        # проставил accept_totp_code вместе с номером интервала.
+        _drop_challenge(request)
         login(request, user)
         remember_device(request)
         record_audit_event(

apps/backend/chatballs/identity/auth/totp_utils.py

@@ -8,8 +8,15 @@ import time
 from chatballs.identity.models import HumanUser
 
 TOTP_SESSION_KEY = "identity_pending_totp_user_id"
+# Когда пароль приняли и остался только код. Без отметки времени начатая и
+# брошенная попытка входа висела бы в сессии сколько угодно долго: чужая
+# вкладка через неделю дописывала бы код и получала вход.
+TOTP_STARTED_KEY = "identity_pending_totp_started_at"
+TOTP_CHALLENGE_TTL_SECONDS = 5 * 60
 TOTP_ISSUER = "Chatballs"
 TOTP_PERIOD_SECONDS = 30
+# Допуск на расхождение часов с телефоном: интервал до и после текущего.
+TOTP_WINDOW = (-1, 0, 1)
 
 
 def _generate_totp_secret() -> str:
@@ -22,24 +29,55 @@ def _decode_totp_secret(secret: str) -> bytes:
     return base64.b32decode(normalized + padding)
 
 
+def current_counter(now: int | None = None) -> int:
+    return int(time.time() if now is None else now) // TOTP_PERIOD_SECONDS
+
+
 def _totp_code(secret: str, for_time: int | None = None) -> str:
-    timestamp = int(time.time() if for_time is None else for_time)
-    counter = timestamp // TOTP_PERIOD_SECONDS
-    digest = hmac.new(_decode_totp_secret(secret), struct.pack(">Q", counter), hashlib.sha1).digest()
+    return _code_for_counter(secret, current_counter(for_time))
+
+
+def _code_for_counter(secret: str, counter: int) -> str:
+    digest = hmac.new(
+        _decode_totp_secret(secret), struct.pack(">Q", counter), hashlib.sha1
+    ).digest()
     offset = digest[-1] & 0x0F
     code = struct.unpack(">I", digest[offset : offset + 4])[0] & 0x7FFFFFFF
     return f"{code % 1_000_000:06d}"
 
 
-def _verify_totp(secret: str, code: str) -> bool:
+def verify_totp(secret: str, code: str, *, after_counter: int = 0) -> int | None:
+    """Номер интервала принятого кода или None.
+
+    ``after_counter`` — последний уже использованный интервал этого
+    пользователя: коды до него включительно не принимаются повторно
+    (RFC 6238 §5.2). Возвращённый номер вызывающий обязан сохранить, иначе
+    защиты от повтора нет.
+    """
     normalized = "".join(character for character in code if character.isdigit())
-    if len(normalized) != 6:
+    if len(normalized) != 6 or not secret:
+        return None
+    now = current_counter()
+    for offset in TOTP_WINDOW:
+        counter = now + offset
+        if counter <= after_counter:
+            continue
+        if hmac.compare_digest(_code_for_counter(secret, counter), normalized):
+            return counter
+    return None
+
+
+def accept_totp_code(user: HumanUser, code: str) -> bool:
+    """Принять код и запомнить его интервал, чтобы он не сработал второй раз."""
+    from django.utils import timezone
+
+    counter = verify_totp(user.totp_secret, code, after_counter=user.totp_last_counter)
+    if counter is None:
         return False
-    now = int(time.time())
-    return any(
-        hmac.compare_digest(_totp_code(secret, now + (offset * TOTP_PERIOD_SECONDS)), normalized)
-        for offset in (-1, 0, 1)
-    )
+    user.totp_last_counter = counter
+    user.totp_last_used_at = timezone.now()
+    user.save(update_fields=["totp_last_counter", "totp_last_used_at"])
+    return True
 
 
 def _ensure_totp_secret(user: HumanUser) -> str:

apps/backend/chatballs/identity/migrations/0031_totp_last_counter.py

@@ -0,0 +1,16 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('identity', '0030_instance_turn'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='humanuser',
+            name='totp_last_counter',
+            field=models.BigIntegerField(default=0),
+        ),
+    ]

apps/backend/chatballs/identity/models.py

@@ -70,6 +70,10 @@ class HumanUser(AbstractUser):
     # Когда последний раз принимался код аутентификатора — подпись в карточке
     # «Двухфакторная аутентификация» (дизайн-базлайн v2, кадр P1).
     totp_last_used_at = models.DateTimeField(null=True, blank=True, db_default=None)
+    # Номер последнего принятого интервала TOTP (RFC 6238 §5.2). Код живёт
+    # 30 секунд и принимается с окном ±1 интервал, то есть подсмотренный код
+    # без этой отметки принимался бы второй раз ещё полторы минуты.
+    totp_last_counter = models.BigIntegerField(default=0)
     # Внешний вид — глобальная настройка пользователя (не membership):
     # тема и акцентный HEX-цвет; пустой акцент — дефолтный синий #1677ff.
     ui_theme = models.CharField(max_length=8, choices=UiTheme.choices, default=UiTheme.SYSTEM)

apps/backend/chatballs/identity/tests.py

@@ -1,4 +1,5 @@
 import json
+import time
 from unittest import mock
 
 from django.conf import settings
@@ -12,7 +13,11 @@ from chatballs.testing import TenantAPIClient as APIClient
 from rest_framework.throttling import ScopedRateThrottle
 
 from chatballs.identity.bootstrap import bootstrap_owner
-from chatballs.identity.auth.totp_utils import _totp_code
+from chatballs.identity.auth.totp_utils import (
+    TOTP_CHALLENGE_TTL_SECONDS,
+    TOTP_STARTED_KEY,
+    _totp_code,
+)
 from chatballs.identity.models import (
     AuditEvent,
     EmployeeGroup,
@@ -430,6 +435,82 @@ class AuthEndpointTests(TestCase):
         self.assertEqual(verify_response.status_code, 200)
         self.assertTrue(verify_response.json()["authenticated"])
 
+    def test_totp_code_is_not_accepted_twice(self) -> None:
+        # Код живёт 30 секунд и принимается с окном ±1 интервал: без отметки
+        # использованного интервала подсмотренный код работал бы ещё полторы
+        # минуты (RFC 6238 §5.2).
+        owner = HumanUser.objects.get(email="owner@example.com")
+        owner.totp_secret = "JBSWY3DPEHPK3PXP"
+        owner.totp_enabled = True
+        owner.save(update_fields=["totp_secret", "totp_enabled"])
+        code = _totp_code(owner.totp_secret)
+
+        def attempt():
+            self.client.post(
+                "/api/v1/auth/login/",
+                data=json.dumps({"email": "owner@example.com", "password": "temporary-password"}),
+                content_type="application/json",
+            )
+            return self.client.post(
+                "/api/v1/auth/totp/verify/",
+                data=json.dumps({"code": code}),
+                content_type="application/json",
+            )
+
+        self.assertEqual(attempt().status_code, 200)
+        self.client.post("/api/v1/auth/logout/")
+
+        replayed = attempt()
+
+        self.assertEqual(replayed.status_code, 400)
+        self.assertFalse(replayed.json().get("authenticated", False))
+
+    def test_totp_step_does_not_let_in_a_deactivated_user(self) -> None:
+        # Пароль приняли раньше; если между шагами сотрудника отключили,
+        # второй шаг обязан это заметить.
+        owner = HumanUser.objects.get(email="owner@example.com")
+        owner.totp_secret = "JBSWY3DPEHPK3PXP"
+        owner.totp_enabled = True
+        owner.save(update_fields=["totp_secret", "totp_enabled"])
+        self.client.post(
+            "/api/v1/auth/login/",
+            data=json.dumps({"email": "owner@example.com", "password": "temporary-password"}),
+            content_type="application/json",
+        )
+
+        HumanUser.objects.filter(pk=owner.pk).update(is_active=False)
+
+        verify_response = self.client.post(
+            "/api/v1/auth/totp/verify/",
+            data=json.dumps({"code": _totp_code(owner.totp_secret)}),
+            content_type="application/json",
+        )
+
+        self.assertEqual(verify_response.status_code, 401)
+
+    def test_abandoned_totp_challenge_expires(self) -> None:
+        owner = HumanUser.objects.get(email="owner@example.com")
+        owner.totp_secret = "JBSWY3DPEHPK3PXP"
+        owner.totp_enabled = True
+        owner.save(update_fields=["totp_secret", "totp_enabled"])
+        self.client.post(
+            "/api/v1/auth/login/",
+            data=json.dumps({"email": "owner@example.com", "password": "temporary-password"}),
+            content_type="application/json",
+        )
+
+        session = self.client.session
+        session[TOTP_STARTED_KEY] = time.time() - TOTP_CHALLENGE_TTL_SECONDS - 1
+        session.save()
+
+        verify_response = self.client.post(
+            "/api/v1/auth/totp/verify/",
+            data=json.dumps({"code": _totp_code(owner.totp_secret)}),
+            content_type="application/json",
+        )
+
+        self.assertEqual(verify_response.status_code, 401)
+
 
 @override_settings(ROOT_URLCONF="chatballs_backend.urls_admin")
 class DjangoAdminTests(TestCase):