authentik.enterprise.providers.ws_federation.processors.sign_out
1from dataclasses import dataclass 2from urllib.parse import urlparse 3 4from django.http import HttpRequest 5from django.shortcuts import get_object_or_404 6 7from authentik.core.models import Application 8from authentik.enterprise.providers.ws_federation.models import WSFederationProvider 9from authentik.enterprise.providers.ws_federation.processors.constants import WS_FED_ACTION_SIGN_OUT 10 11 12@dataclass() 13class SignOutRequest: 14 wa: str 15 wtrealm: str 16 wreply: str 17 18 @staticmethod 19 def parse(request: HttpRequest) -> SignOutRequest: 20 action = request.GET.get("wa") 21 if action != WS_FED_ACTION_SIGN_OUT: 22 raise ValueError("Invalid action") 23 realm = request.GET.get("wtrealm") 24 if not realm: 25 raise ValueError("Missing Realm") 26 27 req = SignOutRequest( 28 wa=action, 29 wtrealm=realm, 30 wreply=request.GET.get("wreply"), 31 ) 32 33 _, provider = req.get_app_provider() 34 if not req.wreply: 35 req.wreply = provider.acs_url 36 reply = urlparse(req.wreply) 37 configured = urlparse(provider.acs_url) 38 if not (reply[:2] == configured[:2] and reply.path.startswith(configured.path)): 39 raise ValueError("Invalid wreply") 40 return req 41 42 def get_app_provider(self): 43 provider: WSFederationProvider = get_object_or_404( 44 WSFederationProvider, audience=self.wtrealm 45 ) 46 application = get_object_or_404(Application, provider=provider) 47 return application, provider
@dataclass()
class
SignOutRequest:
13@dataclass() 14class SignOutRequest: 15 wa: str 16 wtrealm: str 17 wreply: str 18 19 @staticmethod 20 def parse(request: HttpRequest) -> SignOutRequest: 21 action = request.GET.get("wa") 22 if action != WS_FED_ACTION_SIGN_OUT: 23 raise ValueError("Invalid action") 24 realm = request.GET.get("wtrealm") 25 if not realm: 26 raise ValueError("Missing Realm") 27 28 req = SignOutRequest( 29 wa=action, 30 wtrealm=realm, 31 wreply=request.GET.get("wreply"), 32 ) 33 34 _, provider = req.get_app_provider() 35 if not req.wreply: 36 req.wreply = provider.acs_url 37 reply = urlparse(req.wreply) 38 configured = urlparse(provider.acs_url) 39 if not (reply[:2] == configured[:2] and reply.path.startswith(configured.path)): 40 raise ValueError("Invalid wreply") 41 return req 42 43 def get_app_provider(self): 44 provider: WSFederationProvider = get_object_or_404( 45 WSFederationProvider, audience=self.wtrealm 46 ) 47 application = get_object_or_404(Application, provider=provider) 48 return application, provider
19 @staticmethod 20 def parse(request: HttpRequest) -> SignOutRequest: 21 action = request.GET.get("wa") 22 if action != WS_FED_ACTION_SIGN_OUT: 23 raise ValueError("Invalid action") 24 realm = request.GET.get("wtrealm") 25 if not realm: 26 raise ValueError("Missing Realm") 27 28 req = SignOutRequest( 29 wa=action, 30 wtrealm=realm, 31 wreply=request.GET.get("wreply"), 32 ) 33 34 _, provider = req.get_app_provider() 35 if not req.wreply: 36 req.wreply = provider.acs_url 37 reply = urlparse(req.wreply) 38 configured = urlparse(provider.acs_url) 39 if not (reply[:2] == configured[:2] and reply.path.startswith(configured.path)): 40 raise ValueError("Invalid wreply") 41 return req