如何改造Django ModelViewSet实现SAML ACS回调以保存name_id与session_index

1. 更新URL配置，添加ACS回调路由 #

由于ModelViewSet的路由规则无法直接适配ACS的POST回调需求，需要单独添加ACS的URL条目：

from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import Saml2AccountView, saml_acs_callback

router = DefaultRouter()
router.register("saml2/account", Saml2AccountView, basename="account")

urlpatterns = [
    path("", include(router.urls)),
    # 匹配指定的ACS URL规则
    path("saml2/sso/acs/", saml_acs_callback, name="saml_acs"),
]

2. 完善视图逻辑，实现ACS回调与会话信息存储 #

利用django-saml2-auth的核心方法，改造登录、登出逻辑，并新增ACS回调视图处理SAML响应，同时保存name_id和session_index到会话：

from django.http import HttpResponseRedirect
from django.contrib.auth import login, logout as auth_logout
from django_saml2_auth.views import _create_saml_request, _process_saml_response
from django_saml2_auth.utils import get_realm, get_saml_client
from rest_framework import viewsets, status
from rest_framework.response import Response
from rest_framework.decorators import api_view

class Saml2AccountView(viewsets.ModelViewSet):
    @action(detail=False, methods=['get'])
    def login(self, request, *args, **kwargs):
        # 创建SAML登录请求，跳转至新IAM认证页面
        saml_client = get_saml_client(get_realm())
        request_id, saml_request = _create_saml_request(saml_client, "login")
        
        # 将request_id存入会话，用于ACS回调验证
        request.session['saml_request_id'] = request_id
        
        # 构造跳转至新IAM登录端点的URL
        login_url = f"https://example.com/saml2/account/login?SAMLRequest={saml_request}"
        return HttpResponseRedirect(login_url)

    @action(detail=False, methods=['get'])
    def logout(self, request, *args, **kwargs):
        # 从会话中取出登录时保存的SAML会话信息
        name_id = request.session.get('saml_name_id')
        session_index = request.session.get('saml_session_index')
        
        if not name_id or not session_index:
            return Response({"error": "缺少登出所需的SAML会话参数"}, status=status.HTTP_400_BAD_REQUEST)
        
        # 创建携带name_id和session_index的SAML登出请求
        saml_client = get_saml_client(get_realm())
        request_id, saml_request = _create_saml_request(
            saml_client, 
            "logout",
            name_id=name_id,
            session_index=session_index
        )
        
        # 清除本地会话
        auth_logout(request)
        
        # 跳转至新IAM登出端点
        logout_url = f"https://example.com/saml2/account/logout?SAMLRequest={saml_request}"
        return HttpResponseRedirect(logout_url)

# ACS回调视图，处理POST提交的SAML响应
@api_view(['POST'])
def saml_acs_callback(request):
    try:
        saml_response = request.POST.get('SAMLResponse')
        request_id = request.session.get('saml_request_id')
        
        if not saml_response or not request_id:
            return Response({"error": "无效的SAML响应或请求ID"}, status=status.HTTP_400_BAD_REQUEST)
        
        # 处理SAML响应，获取本地用户和SAML属性
        user, saml_attributes = _process_saml_response(saml_response, request_id)
        
        # 提取并保存name_id和session_index到会话
        name_id = saml_attributes.get('name_id')
        session_index = saml_attributes.get('session_index')
        
        if user and name_id and session_index:
            login(request, user)
            request.session['saml_name_id'] = name_id
            request.session['saml_session_index'] = session_index
            
            return Response({"message": "登录成功"}, status=status.HTTP_200_OK)
        else:
            return Response({"error": "SAML响应缺少必要的会话信息"}, status=status.HTTP_400_BAD_REQUEST)
    except Exception as e:
        return Response({"error": f"SAML响应处理失败: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)

3. 核心要点说明 #

• ACS回调必须使用POST方法，因此单独定义视图而非复用ModelViewSet的action。
• 登录流程中需保存request_id到会话，用于ACS回调时验证响应的合法性。
• ACS回调完成后，将name_id和session_index存入Django会话，确保登出时能构造符合新IAM要求的LogoutRequest。
• 登出时必须携带这两个参数，否则新IAM会拒绝处理登出请求。

内容的提问来源于stack exchange，提问作者chesschi