84 lines
3.0 KiB
Python
84 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
from urllib.parse import urlsplit
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth import login as django_login
|
|
from django.contrib.auth import logout as django_logout
|
|
from django.urls import reverse
|
|
from drf_spectacular.utils import OpenApiResponse, extend_schema
|
|
from rest_framework import serializers, status
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
|
|
from ..authentication import SessionAwareJWTAuthentication
|
|
from ..serializers import (
|
|
BrowserSessionRequestSerializer,
|
|
BrowserSessionResponseSerializer,
|
|
)
|
|
|
|
SESSION_BACKEND = "django.contrib.auth.backends.ModelBackend"
|
|
|
|
|
|
def validate_oidc_return_url(value: str) -> str:
|
|
"""Allow redirects only back to this portal's authorization endpoint."""
|
|
|
|
candidate = (value or "").strip()
|
|
if not candidate:
|
|
return "/"
|
|
|
|
parsed = urlsplit(candidate)
|
|
expected_path = reverse("oauth2_provider:authorize")
|
|
if parsed.path.rstrip("/") != expected_path.rstrip("/") or parsed.fragment:
|
|
raise serializers.ValidationError("invalid_oidc_return_url")
|
|
|
|
if not parsed.scheme and not parsed.netloc:
|
|
return candidate
|
|
|
|
issuer = urlsplit(settings.OIDC_ISSUER_URL)
|
|
if (parsed.scheme, parsed.netloc) != (issuer.scheme, issuer.netloc):
|
|
raise serializers.ValidationError("invalid_oidc_return_url")
|
|
return candidate
|
|
|
|
|
|
class BrowserSessionAPIView(APIView):
|
|
"""Turn an existing portal JWT into a short-lived server browser session."""
|
|
|
|
authentication_classes = [SessionAwareJWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
@extend_schema(
|
|
tags=["Auth"],
|
|
summary="Создать браузерную сессию для корпоративного SSO",
|
|
request=BrowserSessionRequestSerializer,
|
|
responses={200: BrowserSessionResponseSerializer},
|
|
)
|
|
def post(self, request):
|
|
if not settings.OIDC_PROVIDER_ENABLED:
|
|
return Response(status=status.HTTP_404_NOT_FOUND)
|
|
|
|
serializer = BrowserSessionRequestSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
redirect_url = validate_oidc_return_url(
|
|
serializer.validated_data.get("next", "")
|
|
)
|
|
|
|
django_login(request._request, request.user, backend=SESSION_BACKEND)
|
|
request._request.session.set_expiry(settings.OIDC_BROWSER_SESSION_SECONDS)
|
|
return Response({"redirect_url": redirect_url}, status=status.HTTP_200_OK)
|
|
|
|
|
|
class BrowserSessionLogoutAPIView(APIView):
|
|
authentication_classes = [SessionAwareJWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
@extend_schema(
|
|
tags=["Auth"],
|
|
summary="Завершить браузерную SSO-сессию",
|
|
responses={204: OpenApiResponse(description="Сессия завершена.")},
|
|
)
|
|
def post(self, request):
|
|
django_logout(request._request)
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|