diff --git a/.env.example b/.env.example index 6dd3510..75071ac 100644 --- a/.env.example +++ b/.env.example @@ -24,3 +24,11 @@ CORS_ALLOWED_ORIGINS=http://localhost:3000,http://127.0.0.1:3000 LOG_LEVEL=INFO BACKUP_ENCRYPTION_KEY= BACKUP_KEY_ID=default + +# IAS ECP SSO (keep disabled without a mounted P-256 public key) +SSO_ENABLED=false +SSO_PROVIDER_URL=https://divopk.vniicentr.ru/sso-redirect/ +SSO_SERVICE_CODE=fkc +SSO_ISSUER=dvr +SSO_PUBLIC_KEY_PATH= +SSO_SESSION_TTL_SECONDS=7200 diff --git a/.env.prod.example b/.env.prod.example index 1566a64..679a68c 100644 --- a/.env.prod.example +++ b/.env.prod.example @@ -25,5 +25,13 @@ COLLECTSTATIC_ON_MIGRATE=1 BACKUP_ENCRYPTION_KEY=CHANGE_ME_BASE64URL_32B_KEY BACKUP_KEY_ID=default +# IAS ECP SSO (mount the PEM outside the repository and webroot) +SSO_ENABLED=false +SSO_PROVIDER_URL=https://divopk.vniicentr.ru/sso-redirect/ +SSO_SERVICE_CODE=fkc +SSO_ISSUER=dvr +SSO_PUBLIC_KEY_PATH=/run/secrets/sso-public-key.pem +SSO_SESSION_TTL_SECONDS=7200 + WEB_IMAGE=registry.example.com/state_corp-web:latest CELERY_IMAGE=registry.example.com/state_corp-celery:latest diff --git a/README.md b/README.md index 9bf0c12..b2dea39 100644 --- a/README.md +++ b/README.md @@ -249,3 +249,8 @@ make services-up make migrate make pre-commit ``` +## Вход через ИАС ЕЦП + +Дополнительный SSO-вход включается переменной `SSO_ENABLED=true`. Публичный P-256 PEM-ключ провайдера должен быть смонтирован вне репозитория и webroot, а путь к нему передан в `SSO_PUBLIC_KEY_PATH`. + +Используются настройки `SSO_PROVIDER_URL`, `SSO_SERVICE_CODE` (по умолчанию `fkc`), `SSO_ISSUER` (по умолчанию `dvr`) и `SSO_SESSION_TTL_SECONDS` (по умолчанию `7200`). Перед включением зарегистрируйте callback `https:///api/v1/auth/sso/` в ИАС ЕЦП и сначала включите backend, затем frontend-кнопку. Для отката достаточно установить `SSO_ENABLED=false`; JWT-вход продолжит работать. diff --git a/pyproject.toml b/pyproject.toml index babb58b..fe32ab4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,7 @@ dependencies = [ "pytz==2024.1", # Security "cryptography==42.0.5", + "PyJWT==2.10.1", "djangorestframework-simplejwt>=5.3.1", "drf-yasg>=1.21.10", "pillow>=12.1.0", diff --git a/src/apps/core/openapi.py b/src/apps/core/openapi.py index f566821..5cc3362 100644 --- a/src/apps/core/openapi.py +++ b/src/apps/core/openapi.py @@ -38,6 +38,14 @@ API предназначен в основном для чтения и мони форм и импорт реестров выполняются через специализированные маршруты и фоновые задачи. """.strip() +OPENAPI_INFO = openapi.Info( + title=OPENAPI_PROJECT_TITLE, + default_version="v1", + description=OPENAPI_PROJECT_DESCRIPTION, + contact=openapi.Contact(email="contact@state-corp.local"), + license=openapi.License(name="BSD License"), +) + OPENAPI_TAG_DESCRIPTIONS = OrderedDict( [ ( diff --git a/src/apps/user/migrations/0009_ssoidentity.py b/src/apps/user/migrations/0009_ssoidentity.py new file mode 100644 index 0000000..1243af2 --- /dev/null +++ b/src/apps/user/migrations/0009_ssoidentity.py @@ -0,0 +1,64 @@ +from django.conf import settings +from django.db import migrations, models +import django.db.models.deletion + + +class Migration(migrations.Migration): + dependencies = [ + ("user", "0008_auto_20260328_1630"), + ] + + operations = [ + migrations.CreateModel( + name="SsoIdentity", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("issuer", models.CharField(max_length=64, verbose_name="issuer")), + ( + "subject", + models.CharField(max_length=255, verbose_name="external user id"), + ), + ( + "last_authenticated_at", + models.DateTimeField(verbose_name="last authenticated at"), + ), + ( + "created_at", + models.DateTimeField(auto_now_add=True, verbose_name="created at"), + ), + ( + "updated_at", + models.DateTimeField(auto_now=True, verbose_name="updated at"), + ), + ( + "user", + models.OneToOneField( + on_delete=django.db.models.deletion.CASCADE, + related_name="sso_identity", + to=settings.AUTH_USER_MODEL, + verbose_name="user", + ), + ), + ], + options={ + "verbose_name": "SSO identity", + "verbose_name_plural": "SSO identities", + "db_table": "user_sso_identities", + }, + ), + migrations.AddConstraint( + model_name="ssoidentity", + constraint=models.UniqueConstraint( + fields=("issuer", "subject"), + name="user_sso_identity_issuer_subject_unique", + ), + ), + ] diff --git a/src/apps/user/models.py b/src/apps/user/models.py index ea2d000..b1aead9 100644 --- a/src/apps/user/models.py +++ b/src/apps/user/models.py @@ -109,3 +109,33 @@ class Profile(models.Model): parts = [self.first_name, self.mid_name, self.last_name] full_name = " ".join(part for part in parts if part) return full_name or self.user.username + + +class SsoIdentity(models.Model): + """Неизменяемая связь локального пользователя с учётной записью ИАС ЕЦП.""" + + user = models.OneToOneField( + User, + on_delete=models.CASCADE, + related_name="sso_identity", + verbose_name=_("user"), + ) + issuer = models.CharField(_("issuer"), max_length=64) + subject = models.CharField(_("external user id"), max_length=255) + last_authenticated_at = models.DateTimeField(_("last authenticated at")) + created_at = models.DateTimeField(_("created at"), auto_now_add=True) + updated_at = models.DateTimeField(_("updated at"), auto_now=True) + + class Meta: + db_table = "user_sso_identities" + verbose_name = _("SSO identity") + verbose_name_plural = _("SSO identities") + constraints = [ + models.UniqueConstraint( + fields=("issuer", "subject"), + name="user_sso_identity_issuer_subject_unique", + ) + ] + + def __str__(self): + return f"{self.issuer}:{self.subject}" diff --git a/src/apps/user/sso_services.py b/src/apps/user/sso_services.py new file mode 100644 index 0000000..5bd8632 --- /dev/null +++ b/src/apps/user/sso_services.py @@ -0,0 +1,221 @@ +import hashlib +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path + +import jwt +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec +from django.conf import settings +from django.contrib.auth import get_user_model +from django.core.cache import cache +from django.core.exceptions import ValidationError as DjangoValidationError +from django.core.validators import validate_email +from django.db import transaction +from django.db.models import Q +from django.utils import timezone +from jwt import InvalidTokenError + +from .models import SsoIdentity + +User = get_user_model() + + +class SsoAuthenticationError(Exception): + """Безопасная ошибка проверки или связывания SSO-пользователя.""" + + +@dataclass(frozen=True) +class SsoClaims: + issuer: str + subject: str + username: str + email: str + first_name: str + middle_name: str + last_name: str + jti: str + expires_at: int + + +def _load_public_key(): + key_path = Path(settings.SSO_PUBLIC_KEY_PATH) + try: + key_data = key_path.read_bytes() + public_key = serialization.load_pem_public_key(key_data) + except (OSError, ValueError, TypeError) as exc: + raise SsoAuthenticationError("Публичный ключ SSO недоступен.") from exc + + if not isinstance(public_key, ec.EllipticCurvePublicKey) or not isinstance( + public_key.curve, ec.SECP256R1 + ): + raise SsoAuthenticationError("Публичный ключ SSO должен использовать P-256.") + return public_key + + +def _decode_sso_payload(encoded_token: str) -> dict: + try: + header = jwt.get_unverified_header(encoded_token) + if header.get("alg") != "ES256": + raise SsoAuthenticationError("Недопустимый алгоритм SSO-токена.") + + return jwt.decode( + encoded_token, + _load_public_key(), + algorithms=["ES256"], + issuer=settings.SSO_ISSUER, + leeway=settings.SSO_CLOCK_SKEW_SECONDS, + options={ + "require": [ + "iss", + "exp", + "iat", + "jti", + "token_type", + "sso_target", + "user_id", + "username", + "email", + ] + }, + ) + except SsoAuthenticationError: + raise + except (InvalidTokenError, ValueError, TypeError) as exc: + raise SsoAuthenticationError("SSO-токен не прошёл проверку.") from exc + + +def _validate_token_lifetime(payload: dict) -> tuple[int, int]: + issued_at = payload.get("iat") + expires_at = payload.get("exp") + if ( + not isinstance(issued_at, int) + or isinstance(issued_at, bool) + or not isinstance(expires_at, int) + or isinstance(expires_at, bool) + ): + raise SsoAuthenticationError("Некорректное время жизни SSO-токена.") + if expires_at <= issued_at or ( + expires_at - issued_at > settings.SSO_TOKEN_MAX_AGE_SECONDS + ): + raise SsoAuthenticationError("Недопустимое время жизни SSO-токена.") + return issued_at, expires_at + + +def verify_sso_token(encoded_token: str) -> SsoClaims: + """Проверяет ES256 SSO JWT и возвращает только нужные локальному сервису claims.""" + payload = _decode_sso_payload(encoded_token) + + if payload.get("token_type") != "sso": + raise SsoAuthenticationError("Недопустимый тип SSO-токена.") + if payload.get("sso_target") != settings.SSO_SERVICE_CODE: + raise SsoAuthenticationError("SSO-токен предназначен другому сервису.") + + _, expires_at = _validate_token_lifetime(payload) + + username = payload.get("username") + email = payload.get("email") + jti = payload.get("jti") + subject = payload.get("user_id") + if not isinstance(username, str) or not username.strip(): + raise SsoAuthenticationError("В SSO-токене отсутствует username.") + if not isinstance(email, str) or not email.strip(): + raise SsoAuthenticationError("В SSO-токене отсутствует email.") + if not isinstance(jti, str) or not jti.strip(): + raise SsoAuthenticationError("В SSO-токене отсутствует jti.") + if isinstance(subject, bool) or not isinstance(subject, int | str): + raise SsoAuthenticationError("В SSO-токене отсутствует user_id.") + if not str(subject).strip(): + raise SsoAuthenticationError("В SSO-токене отсутствует user_id.") + + try: + validate_email(email) + User._meta.get_field("username").run_validators(username) + except DjangoValidationError as exc: + raise SsoAuthenticationError("Некорректные данные пользователя в SSO-токене.") from exc + + return SsoClaims( + issuer=str(payload["iss"]), + subject=str(subject), + username=username, + email=email, + first_name=str(payload.get("first_name") or ""), + middle_name=str(payload.get("surname") or ""), + last_name=str(payload.get("last_name") or ""), + jti=jti, + expires_at=expires_at, + ) + + +def reserve_sso_token(claims: SsoClaims) -> None: + """Атомарно помечает одноразовый SSO-токен использованным.""" + digest = hashlib.sha256(f"{claims.issuer}\0{claims.jti}".encode()).hexdigest() + cache_key = f"sso:jti:{digest}" + now = int(datetime.now(tz=UTC).timestamp()) + timeout = max(1, claims.expires_at - now + settings.SSO_CLOCK_SKEW_SECONDS) + if not cache.add(cache_key, "used", timeout=timeout): + raise SsoAuthenticationError("SSO-токен уже использован.") + + +def _ensure_user_claims_available(claims: SsoClaims, *, exclude_user_id=None) -> None: + conflicts = User.objects.filter( + Q(username=claims.username) | Q(email__iexact=claims.email) + ) + if exclude_user_id is not None: + conflicts = conflicts.exclude(id=exclude_user_id) + if conflicts.exists(): + raise SsoAuthenticationError("Данные SSO-пользователя уже заняты.") + + +@transaction.atomic +def provision_sso_user(claims: SsoClaims): + """Создаёт или обновляет локального пользователя по неизменяемой SSO identity.""" + identity = ( + SsoIdentity.objects.select_for_update() + .select_related("user", "user__profile") + .filter(issuer=claims.issuer, subject=claims.subject) + .first() + ) + + if identity is not None: + user = identity.user + if not user.is_active: + raise SsoAuthenticationError("Локальная учётная запись отключена.") + _ensure_user_claims_available(claims, exclude_user_id=user.id) + else: + username_user = User.objects.filter(username=claims.username).first() + email_users = list(User.objects.filter(email__iexact=claims.email)[:2]) + if username_user is not None or email_users: + if ( + username_user is None + or len(email_users) != 1 + or email_users[0].id != username_user.id + ): + raise SsoAuthenticationError("Неоднозначное совпадение локальной учётной записи.") + user = username_user + if not user.is_active: + raise SsoAuthenticationError("Локальная учётная запись отключена.") + else: + user = User(username=claims.username, email=claims.email, is_verified=True) + user.set_unusable_password() + user.save() + + identity = SsoIdentity.objects.create( + user=user, + issuer=claims.issuer, + subject=claims.subject, + last_authenticated_at=timezone.now(), + ) + + user.username = claims.username + user.email = claims.email + user.is_verified = True + user.save(update_fields=["username", "email", "is_verified", "updated_at"]) + profile = user.profile + profile.first_name = claims.first_name + profile.mid_name = claims.middle_name + profile.last_name = claims.last_name + profile.save(update_fields=["first_name", "mid_name", "last_name", "updated_at"]) + identity.last_authenticated_at = timezone.now() + identity.save(update_fields=["last_authenticated_at", "updated_at"]) + return user diff --git a/src/apps/user/sso_views.py b/src/apps/user/sso_views.py new file mode 100644 index 0000000..5db8e7f --- /dev/null +++ b/src/apps/user/sso_views.py @@ -0,0 +1,181 @@ +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +from django.conf import settings +from django.contrib.auth import login +from django.db import IntegrityError +from django.http import HttpResponse +from django.middleware.csrf import get_token +from drf_yasg import openapi +from drf_yasg.utils import swagger_auto_schema +from rest_framework import status +from rest_framework.parsers import FormParser +from rest_framework.permissions import AllowAny +from rest_framework.response import Response +from rest_framework.views import APIView + +from .sso_services import ( + SsoAuthenticationError, + provision_sso_user, + reserve_sso_token, + verify_sso_token, +) + + +def validate_local_redirect_path(value: str | None) -> str: + """Возвращает безопасный локальный путь для завершения SSO-перехода.""" + redirect_path = value or "/main" + parsed = urlsplit(redirect_path) + has_control_character = any( + ord(character) < 32 or ord(character) == 127 for character in redirect_path + ) + + if ( + not redirect_path.startswith("/") + or redirect_path.startswith("//") + or "\\" in redirect_path + or has_control_character + or parsed.scheme + or parsed.netloc + ): + raise ValueError("redirect_path должен быть относительным локальным путём") + + return redirect_path + + +def build_sso_provider_url(redirect_path: str) -> str: + """Добавляет код сервиса и обратный путь к URL провайдера SSO.""" + provider_url = urlsplit(settings.SSO_PROVIDER_URL) + query = dict(parse_qsl(provider_url.query, keep_blank_values=True)) + query.update({"target": settings.SSO_SERVICE_CODE, "path": redirect_path}) + return urlunsplit( + ( + provider_url.scheme, + provider_url.netloc, + provider_url.path, + urlencode(query), + provider_url.fragment, + ) + ) + + +class SsoStartView(APIView): + """Перенаправляет браузер на вход ИАС ЕЦП.""" + + authentication_classes = [] + permission_classes = [AllowAny] + + @swagger_auto_schema( + tags=["Аутентификация"], + operation_summary="Начать вход через ИАС ЕЦП", + manual_parameters=[ + openapi.Parameter( + "redirect_path", + openapi.IN_QUERY, + description="Относительный локальный путь возврата", + type=openapi.TYPE_STRING, + required=False, + ) + ], + responses={307: "Переход к провайдеру SSO", 400: "Некорректный путь", 503: "SSO отключён"}, + ) + def get(self, request): + if not settings.SSO_ENABLED: + return Response( + {"detail": "SSO отключён."}, + status=status.HTTP_503_SERVICE_UNAVAILABLE, + ) + + try: + redirect_path = validate_local_redirect_path( + request.query_params.get("redirect_path") + ) + except ValueError as exc: + return Response( + {"detail": str(exc)}, + status=status.HTTP_400_BAD_REQUEST, + ) + + response = HttpResponse(status=307) + response["Location"] = build_sso_provider_url(redirect_path) + return response + + +class SsoCallbackView(APIView): + """Принимает одноразовый SSO JWT и создаёт локальную Django-сессию.""" + + authentication_classes = [] + parser_classes = [FormParser] + permission_classes = [AllowAny] + + @swagger_auto_schema( + tags=["Аутентификация"], + operation_summary="Завершить вход через ИАС ЕЦП", + manual_parameters=[ + openapi.Parameter( + "sso_token", + openapi.IN_FORM, + description="Одноразовый ES256 SSO JWT", + type=openapi.TYPE_STRING, + required=True, + ), + openapi.Parameter( + "redirect_path", + openapi.IN_FORM, + description="Относительный локальный путь возврата", + type=openapi.TYPE_STRING, + required=False, + ), + ], + responses={ + 303: "SSO-сессия создана", + 400: "Некорректная форма или путь", + 403: "SSO-аутентификация отклонена", + 503: "SSO отключён", + }, + ) + def post(self, request): + if not settings.SSO_ENABLED: + return Response( + {"detail": "SSO отключён."}, + status=status.HTTP_503_SERVICE_UNAVAILABLE, + ) + + if request.content_type != "application/x-www-form-urlencoded": + return Response( + {"detail": "Ожидается application/x-www-form-urlencoded."}, + status=status.HTTP_400_BAD_REQUEST, + ) + + encoded_token = request.data.get("sso_token") + if not isinstance(encoded_token, str) or not encoded_token: + return Response( + {"detail": "sso_token обязателен."}, + status=status.HTTP_400_BAD_REQUEST, + ) + + try: + redirect_path = validate_local_redirect_path( + request.data.get("redirect_path") + ) + except ValueError as exc: + return Response( + {"detail": str(exc)}, + status=status.HTTP_400_BAD_REQUEST, + ) + + try: + claims = verify_sso_token(encoded_token) + reserve_sso_token(claims) + user = provision_sso_user(claims) + except (IntegrityError, SsoAuthenticationError): + return Response( + {"detail": "SSO-аутентификация отклонена."}, + status=status.HTTP_403_FORBIDDEN, + ) + + login(request, user, backend="django.contrib.auth.backends.ModelBackend") + request.session.set_expiry(settings.SSO_SESSION_TTL_SECONDS) + get_token(request) + response = HttpResponse(status=303) + response["Location"] = redirect_path + return response diff --git a/src/apps/user/views.py b/src/apps/user/views.py index a9c0650..a840df0 100644 --- a/src/apps/user/views.py +++ b/src/apps/user/views.py @@ -2,7 +2,7 @@ from urllib.parse import urlencode from apps.core.models import BackgroundJob from apps.core.services import BackgroundJobService -from django.contrib.auth import authenticate, get_user_model +from django.contrib.auth import authenticate, get_user_model, logout from django.contrib.auth.hashers import check_password from django.core.paginator import Paginator from django.db.models import F @@ -169,7 +169,7 @@ class LogoutView(APIView): """ Выход пользователя. - Логаут на JWT означает удаление токенов на клиенте. + Завершает Django-сессию. JWT по-прежнему удаляются на клиенте. """ permission_classes = [IsAuthenticated] @@ -177,12 +177,15 @@ class LogoutView(APIView): @swagger_auto_schema( tags=[AUTH_TAG], operation_summary="Выход", - operation_description="Выход из системы (удаление токенов на клиенте).", + operation_description=( + "Завершение Django-сессии; JWT удаляются на клиенте." + ), responses={200: "Успешный выход"}, ) def post(self, request): - # Для JWT логаут означает удаление токенов на клиенте. - # Сервер не хранит сессию и ничего не инвалидирует. + # Для JWT это сохраняет прежнее поведение: токены удаляются клиентом. + # Для SSO серверная сессия должна быть явно завершена. + logout(request) return Response({"message": "Успешный выход"}, status=status.HTTP_200_OK) diff --git a/src/core/api_v1_urls.py b/src/core/api_v1_urls.py index 705b263..f79ba7a 100644 --- a/src/core/api_v1_urls.py +++ b/src/core/api_v1_urls.py @@ -11,6 +11,7 @@ from apps.core.views import ( BackgroundJobStatusView, BackgroundJobStreamView, ) +from apps.user.sso_views import SsoCallbackView, SsoStartView from django.urls import include, path app_name = "api_v1" @@ -22,6 +23,8 @@ jobs_urlpatterns = [ ] urlpatterns = [ + path("auth/sso/", SsoCallbackView.as_view(), name="sso-callback"), + path("auth/sso/start/", SsoStartView.as_view(), name="sso-start"), path("analytics/", include("apps.organization.analytics_root_urls")), path("exchange/", include("apps.exchange.urls")), path("dictionaries/", include("apps.organization.dictionary_urls")), diff --git a/src/core/urls.py b/src/core/urls.py index 9dfcd68..9bd7e22 100644 --- a/src/core/urls.py +++ b/src/core/urls.py @@ -5,27 +5,19 @@ The `urlpatterns` list routes URLs to views. """ from apps.core.openapi import ( - OPENAPI_PROJECT_DESCRIPTION, - OPENAPI_PROJECT_TITLE, + OPENAPI_INFO, RussianTagSchemaGenerator, ) from django.conf import settings from django.conf.urls.static import static from django.contrib import admin from django.urls import include, path -from drf_yasg import openapi from drf_yasg.views import get_schema_view from rest_framework import permissions # Swagger schema view schema_view = get_schema_view( - openapi.Info( - title=OPENAPI_PROJECT_TITLE, - default_version="v1", - description=OPENAPI_PROJECT_DESCRIPTION, - contact=openapi.Contact(email="contact@state-corp.local"), - license=openapi.License(name="BSD License"), - ), + OPENAPI_INFO, public=True, generator_class=RussianTagSchemaGenerator, permission_classes=(permissions.AllowAny,), diff --git a/src/settings/base.py b/src/settings/base.py index 74c0b3c..7f524fb 100644 --- a/src/settings/base.py +++ b/src/settings/base.py @@ -22,6 +22,17 @@ EXCHANGE_KEY_ID = os.getenv("EXCHANGE_KEY_ID", "dev-shared-token") # Read old encrypted archives after rotation; never accepted as HTTP credentials. EXCHANGE_PREVIOUS_SHARED_TOKEN = os.getenv("EXCHANGE_PREVIOUS_SHARED_TOKEN", "") EXCHANGE_PREVIOUS_KEY_ID = os.getenv("EXCHANGE_PREVIOUS_KEY_ID", "") +SSO_ENABLED = os.getenv("SSO_ENABLED", "false").strip().lower() == "true" +SSO_PROVIDER_URL = os.getenv( + "SSO_PROVIDER_URL", + "https://divopk.vniicentr.ru/sso-redirect/", +) +SSO_SERVICE_CODE = os.getenv("SSO_SERVICE_CODE", "fkc") +SSO_ISSUER = os.getenv("SSO_ISSUER", "dvr") +SSO_PUBLIC_KEY_PATH = os.getenv("SSO_PUBLIC_KEY_PATH", "") +SSO_SESSION_TTL_SECONDS = int(os.getenv("SSO_SESSION_TTL_SECONDS", "7200")) +SSO_TOKEN_MAX_AGE_SECONDS = 60 +SSO_CLOCK_SKEW_SECONDS = 5 warnings.filterwarnings( "ignore", @@ -258,10 +269,14 @@ SOURCE_RECORD_EXPORT_DOWNLOAD_TICKET_TTL_SECONDS = int( DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField" AUTH_USER_MODEL = "user.User" LOGIN_URL = "/auth/login/" +SESSION_COOKIE_HTTPONLY = True +SESSION_COOKIE_SAMESITE = "Lax" +CSRF_COOKIE_SAMESITE = "Lax" REST_FRAMEWORK = { "DEFAULT_AUTHENTICATION_CLASSES": [ "rest_framework_simplejwt.authentication.JWTAuthentication", + "rest_framework.authentication.SessionAuthentication", ], "DEFAULT_PERMISSION_CLASSES": [ "rest_framework.permissions.IsAuthenticatedOrReadOnly", @@ -320,6 +335,7 @@ CORS_ALLOWED_ORIGINS = ["http://localhost:3000", "http://127.0.0.1:3000"] CORS_ALLOW_CREDENTIALS = True SWAGGER_SETTINGS = { + "DEFAULT_INFO": "apps.core.openapi.OPENAPI_INFO", "SECURITY_DEFINITIONS": { "Bearer": { "type": "apiKey", diff --git a/src/settings/test.py b/src/settings/test.py index 7d6ef01..05e2dc4 100644 --- a/src/settings/test.py +++ b/src/settings/test.py @@ -93,6 +93,7 @@ REST_FRAMEWORK = { **globals().get("REST_FRAMEWORK", {}), "DEFAULT_AUTHENTICATION_CLASSES": [ "rest_framework_simplejwt.authentication.JWTAuthentication", + "rest_framework.authentication.SessionAuthentication", ], "TEST_REQUEST_DEFAULT_FORMAT": "json", # Disable throttling for tests diff --git a/tests/apps/user/test_sso.py b/tests/apps/user/test_sso.py new file mode 100644 index 0000000..1c560d8 --- /dev/null +++ b/tests/apps/user/test_sso.py @@ -0,0 +1,330 @@ +import os +import tempfile +from datetime import UTC, datetime, timedelta +from urllib.parse import parse_qs, urlencode, urlsplit + +import jwt +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec +from django.apps import apps +from django.contrib.auth import get_user_model +from django.test import SimpleTestCase, override_settings +from rest_framework.test import APIClient, APITestCase + +from .factories import UserFactory + +User = get_user_model() + + +@override_settings( + SSO_ENABLED=True, + SSO_PROVIDER_URL="https://divopk.vniicentr.ru/sso-redirect/", + SSO_SERVICE_CODE="fkc", +) +class SsoStartViewTest(SimpleTestCase): + def test_redirects_to_provider_with_target_and_local_path(self): + response = self.client.get( + "/api/v1/auth/sso/start/", + {"redirect_path": "/open-data?scope=rosatom"}, + ) + + self.assertEqual(response.status_code, 307) + redirect = urlsplit(response["Location"]) + self.assertEqual( + f"{redirect.scheme}://{redirect.netloc}{redirect.path}", + "https://divopk.vniicentr.ru/sso-redirect/", + ) + self.assertEqual( + parse_qs(redirect.query), + {"target": ["fkc"], "path": ["/open-data?scope=rosatom"]}, + ) + + def test_rejects_absolute_redirect_path(self): + response = self.client.get( + "/api/v1/auth/sso/start/", + {"redirect_path": "https://attacker.example/steal"}, + ) + + self.assertEqual(response.status_code, 400) + + +class SsoCallbackViewTest(APITestCase): + callback_url = "/api/v1/auth/sso/" + + def setUp(self): + self.private_key = ec.generate_private_key(ec.SECP256R1()) + public_key = self.private_key.public_key().public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + key_file = tempfile.NamedTemporaryFile(delete=False) + key_file.write(public_key) + key_file.close() + self.public_key_path = key_file.name + self.settings_override = override_settings( + SSO_ENABLED=True, + SSO_PROVIDER_URL="https://divopk.vniicentr.ru/sso-redirect/", + SSO_SERVICE_CODE="fkc", + SSO_ISSUER="dvr", + SSO_PUBLIC_KEY_PATH=self.public_key_path, + SSO_SESSION_TTL_SECONDS=7200, + SSO_TOKEN_MAX_AGE_SECONDS=60, + SSO_CLOCK_SKEW_SECONDS=5, + SESSION_COOKIE_SECURE=True, + SESSION_COOKIE_HTTPONLY=True, + SESSION_COOKIE_SAMESITE="Lax", + CSRF_COOKIE_SECURE=True, + CSRF_COOKIE_SAMESITE="Lax", + ) + self.settings_override.enable() + self.addCleanup(self.settings_override.disable) + self.addCleanup(os.unlink, self.public_key_path) + + def _claims(self, **overrides): + issued_at = datetime.now(UTC).replace(microsecond=0) + claims = { + "token_type": "sso", + "iss": "dvr", + "iat": int(issued_at.timestamp()), + "exp": int((issued_at + timedelta(seconds=60)).timestamp()), + "jti": f"{self._testMethodName}-token", + "user_id": 2564, + "username": "test30", + "email": "test30@example.com", + "first_name": "Иван", + "last_name": "Иванов", + "surname": "Иванович", + "sso_target": "fkc", + "groups": ["Аналитика"], + "okpo": "07508753", + } + claims.update(overrides) + return claims + + def _token(self, **overrides): + return self._encode_claims(self._claims(**overrides)) + + def _encode_claims(self, claims, *, key=None, algorithm="ES256"): + return jwt.encode( + claims, + key or self.private_key, + algorithm=algorithm, + ) + + def _post_callback(self, token: str, redirect_path: str = "/open-data"): + return self.client.post( + self.callback_url, + urlencode({"sso_token": token, "redirect_path": redirect_path}), + content_type="application/x-www-form-urlencoded", + ) + + def test_valid_token_creates_regular_user_identity_and_session(self): + response = self._post_callback(self._token()) + + self.assertEqual(response.status_code, 303) + self.assertEqual(response["Location"], "/open-data") + user = User.objects.get(username="test30") + self.assertFalse(user.has_usable_password()) + self.assertTrue(user.is_verified) + self.assertFalse(user.is_staff) + self.assertEqual(user.profile.first_name, "Иван") + self.assertEqual(user.profile.mid_name, "Иванович") + self.assertEqual(user.profile.last_name, "Иванов") + + identity_model = apps.get_model("user", "SsoIdentity") + identity = identity_model.objects.get(user=user) + self.assertEqual(identity.issuer, "dvr") + self.assertEqual(identity.subject, "2564") + self.assertIn("sessionid", response.cookies) + self.assertIn("csrftoken", response.cookies) + self.assertTrue(response.cookies["sessionid"]["secure"]) + self.assertTrue(response.cookies["sessionid"]["httponly"]) + self.assertEqual(response.cookies["sessionid"]["samesite"], "Lax") + self.assertTrue(response.cookies["csrftoken"]["secure"]) + self.assertEqual(response.cookies["csrftoken"]["samesite"], "Lax") + self.assertAlmostEqual(self.client.session.get_expiry_age(), 7200, delta=1) + self.assertEqual( + self.client.session.get("_auth_user_id"), + str(user.id), + ) + + current_user_response = self.client.get("/api/v1/users/me/") + self.assertEqual(current_user_response.status_code, 200) + self.assertEqual(current_user_response.data["username"], "test30") + + def test_rejects_replayed_jti(self): + token = self._token() + + first_response = self._post_callback(token) + replay_response = self._post_callback(token) + + self.assertEqual(first_response.status_code, 303) + self.assertEqual(replay_response.status_code, 403) + + def test_links_existing_user_only_when_username_and_email_both_match(self): + existing_user = UserFactory.create_user( + username="test30", + email="TEST30@example.com", + is_staff=True, + ) + + response = self._post_callback(self._token()) + + self.assertEqual(response.status_code, 303) + self.assertEqual(User.objects.count(), 1) + existing_user.refresh_from_db() + self.assertTrue(existing_user.is_staff) + identity_model = apps.get_model("user", "SsoIdentity") + self.assertTrue(identity_model.objects.filter(user=existing_user).exists()) + + def test_rejects_partial_local_account_collision(self): + UserFactory.create_user( + username="test30", + email="another@example.com", + ) + + response = self._post_callback(self._token()) + + self.assertEqual(response.status_code, 403) + self.assertEqual(User.objects.count(), 1) + + def test_rejects_invalid_signature_and_algorithm(self): + another_private_key = ec.generate_private_key(ec.SECP256R1()) + invalid_signature = self._encode_claims( + self._claims(jti="invalid-signature"), + key=another_private_key, + ) + invalid_algorithm = self._encode_claims( + self._claims(jti="invalid-algorithm"), + key="test-only-hmac-secret", + algorithm="HS256", + ) + + signature_response = self._post_callback(invalid_signature) + algorithm_response = self._post_callback(invalid_algorithm) + + self.assertEqual(signature_response.status_code, 403) + self.assertEqual(algorithm_response.status_code, 403) + self.assertFalse(User.objects.exists()) + + def test_rejects_wrong_issuer_type_and_target(self): + invalid_claims = ( + {"iss": "other", "jti": "wrong-issuer"}, + {"token_type": "access", "jti": "wrong-type"}, + {"sso_target": "other", "jti": "wrong-target"}, + ) + + for overrides in invalid_claims: + with self.subTest(overrides=overrides): + response = self._post_callback(self._token(**overrides)) + self.assertEqual(response.status_code, 403) + + self.assertFalse(User.objects.exists()) + + def test_rejects_expired_future_and_overlong_tokens(self): + now = datetime.now(UTC).replace(microsecond=0) + invalid_lifetimes = ( + { + "iat": int((now - timedelta(seconds=120)).timestamp()), + "exp": int((now - timedelta(seconds=60)).timestamp()), + "jti": "expired", + }, + { + "iat": int((now + timedelta(seconds=30)).timestamp()), + "exp": int((now + timedelta(seconds=60)).timestamp()), + "jti": "future", + }, + { + "iat": int(now.timestamp()), + "exp": int((now + timedelta(seconds=61)).timestamp()), + "jti": "overlong", + }, + ) + + for overrides in invalid_lifetimes: + with self.subTest(overrides=overrides): + response = self._post_callback(self._token(**overrides)) + self.assertEqual(response.status_code, 403) + + self.assertFalse(User.objects.exists()) + + def test_rejects_missing_required_claim(self): + claims = self._claims() + claims.pop("email") + + response = self._post_callback(self._encode_claims(claims)) + + self.assertEqual(response.status_code, 403) + self.assertFalse(User.objects.exists()) + + def test_rejects_callback_open_redirect_before_creating_user(self): + unsafe_paths = ( + "//attacker.example/steal", + "/\\attacker.example/steal", + "/main\nSet-Cookie: stolen=true", + ) + + for index, redirect_path in enumerate(unsafe_paths): + with self.subTest(redirect_path=redirect_path): + response = self._post_callback( + self._token(jti=f"unsafe-redirect-{index}"), + redirect_path=redirect_path, + ) + self.assertEqual(response.status_code, 400) + self.assertFalse(User.objects.exists()) + + def test_rejects_non_form_callback_body(self): + response = self.client.post( + self.callback_url, + {"sso_token": self._token(), "redirect_path": "/main"}, + format="json", + ) + + self.assertEqual(response.status_code, 400) + self.assertFalse(User.objects.exists()) + + def test_rejects_inactive_linked_user(self): + first_response = self._post_callback(self._token(jti="initial-login")) + user = User.objects.get(username="test30") + user.is_active = False + user.save(update_fields=["is_active"]) + + second_response = self._post_callback(self._token(jti="inactive-login")) + + self.assertEqual(first_response.status_code, 303) + self.assertEqual(second_response.status_code, 403) + user.refresh_from_db() + self.assertFalse(user.is_active) + + def test_session_logout_requires_csrf_and_clears_server_session(self): + csrf_client = APIClient(enforce_csrf_checks=True) + callback_response = csrf_client.post( + self.callback_url, + urlencode( + { + "sso_token": self._token(jti="csrf-login"), + "redirect_path": "/main", + } + ), + content_type="application/x-www-form-urlencoded", + ) + self.assertEqual(callback_response.status_code, 303) + + missing_csrf_response = csrf_client.post("/api/v1/users/logout/") + csrf_token = csrf_client.cookies["csrftoken"].value + logout_response = csrf_client.post( + "/api/v1/users/logout/", + HTTP_X_CSRFTOKEN=csrf_token, + ) + + self.assertEqual(missing_csrf_response.status_code, 403) + self.assertEqual(logout_response.status_code, 200) + self.assertNotIn("_auth_user_id", csrf_client.session) + + @override_settings(SSO_ENABLED=False) + def test_disabled_sso_returns_service_unavailable(self): + start_response = self.client.get("/api/v1/auth/sso/start/") + callback_response = self._post_callback(self._token()) + + self.assertEqual(start_response.status_code, 503) + self.assertEqual(callback_response.status_code, 503) diff --git a/uv.lock b/uv.lock index 7d3cbdb..f8966c5 100644 --- a/uv.lock +++ b/uv.lock @@ -2430,6 +2430,7 @@ dependencies = [ { name = "pillow" }, { name = "playwright" }, { name = "psycopg2-binary" }, + { name = "pyjwt" }, { name = "pylint" }, { name = "python-dateutil" }, { name = "python-decouple" }, @@ -2560,6 +2561,7 @@ requires-dist = [ { name = "playwright", specifier = ">=1.57.0" }, { name = "pre-commit", marker = "extra == 'dev'", specifier = "==3.6.0" }, { name = "psycopg2-binary", specifier = "==2.9.9" }, + { name = "pyjwt", specifier = "==2.10.1" }, { name = "pylint", specifier = ">=3.0" }, { name = "pytest", marker = "extra == 'dev'", specifier = "==7.4.4" }, { name = "pytest-cov", marker = "extra == 'dev'", specifier = "==4.1.0" },