state-corp-backend/src/apps/user/sso_views.py

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