diff --git a/apps/backend/chatballs/identity/auth/sessions.py b/apps/backend/chatballs/identity/auth/sessions.py index 3f6db4a..5f28eab 100644 --- a/apps/backend/chatballs/identity/auth/sessions.py +++ b/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, diff --git a/apps/backend/chatballs/identity/auth/totp.py b/apps/backend/chatballs/identity/auth/totp.py index 757f32c..578e148 100644 --- a/apps/backend/chatballs/identity/auth/totp.py +++ b/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( diff --git a/apps/backend/chatballs/identity/auth/totp_utils.py b/apps/backend/chatballs/identity/auth/totp_utils.py index ae6f825..963e3be 100644 --- a/apps/backend/chatballs/identity/auth/totp_utils.py +++ b/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: diff --git a/apps/backend/chatballs/identity/migrations/0031_totp_last_counter.py b/apps/backend/chatballs/identity/migrations/0031_totp_last_counter.py new file mode 100644 index 0000000..e386b15 --- /dev/null +++ b/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), + ), + ] diff --git a/apps/backend/chatballs/identity/models.py b/apps/backend/chatballs/identity/models.py index 9872a59..fbc174e 100644 --- a/apps/backend/chatballs/identity/models.py +++ b/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) diff --git a/apps/backend/chatballs/identity/tests.py b/apps/backend/chatballs/identity/tests.py index 5d8376a..27e37ac 100644 --- a/apps/backend/chatballs/identity/tests.py +++ b/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):