authentik.providers.saml.processors.logout_request

  1"""SAML LogoutRequest Processor"""
  2
  3import base64
  4from urllib.parse import quote, urlencode
  5
  6import xmlsec
  7from lxml import etree  # nosec
  8from lxml.etree import Element, _Element
  9
 10from authentik.common.saml.constants import (
 11    DEFAULT_ISSUER,
 12    DIGEST_ALGORITHM_TRANSLATION_MAP,
 13    NS_MAP,
 14    NS_SAML_ASSERTION,
 15    NS_SAML_PROTOCOL,
 16    SAML_NAME_ID_FORMAT_EMAIL,
 17    SIGN_ALGORITHM_TRANSFORM_MAP,
 18)
 19from authentik.core.models import User
 20from authentik.lib.xml import remove_xml_newlines
 21from authentik.providers.saml.models import SAMLProvider
 22from authentik.providers.saml.utils import get_random_id
 23from authentik.providers.saml.utils.encoding import deflate_and_base64_encode
 24from authentik.providers.saml.utils.time import get_time_string
 25
 26
 27class LogoutRequestProcessor:
 28    """Generate SAML LogoutRequest messages"""
 29
 30    provider: SAMLProvider
 31    user: User | None
 32    destination: str
 33    name_id: str | None
 34    name_id_format: str
 35    session_index: str | None
 36    relay_state: str | None
 37    issuer: str | None
 38
 39    _issue_instant: str
 40    _request_id: str
 41
 42    def __init__(  # noqa: PLR0913
 43        self,
 44        provider: SAMLProvider,
 45        user: User | None,
 46        destination: str,
 47        name_id: str | None = None,
 48        name_id_format: str = SAML_NAME_ID_FORMAT_EMAIL,
 49        session_index: str | None = None,
 50        relay_state: str | None = None,
 51        issuer: str | None = None,
 52    ):
 53        self.provider = provider
 54        self.user = user
 55        self.destination = destination
 56        self.name_id = name_id or (user.email if user else None)
 57        self.name_id_format = name_id_format
 58        self.session_index = session_index
 59        self.relay_state = relay_state
 60        self.issuer = issuer
 61
 62        self._issue_instant = get_time_string()
 63        self._request_id = get_random_id()
 64
 65    def _get_issuer_value(self) -> str:
 66        """Get issuer value from session, with fallback to provider"""
 67        if self.issuer:
 68            return self.issuer
 69        if self.provider.issuer_override:
 70            return self.provider.issuer_override
 71        return DEFAULT_ISSUER
 72
 73    def get_issuer(self) -> Element:
 74        """Get Issuer element"""
 75        issuer = Element(f"{{{NS_SAML_ASSERTION}}}Issuer")
 76        issuer.text = self._get_issuer_value()
 77        return issuer
 78
 79    def get_name_id(self) -> Element:
 80        """Get NameID element"""
 81        name_id = Element(f"{{{NS_SAML_ASSERTION}}}NameID")
 82        name_id.attrib["Format"] = self.name_id_format
 83        name_id.text = self.name_id
 84        return name_id
 85
 86    def build(self) -> Element:
 87        """Build a SAML LogoutRequest as etree Element"""
 88        logout_request = Element(f"{{{NS_SAML_PROTOCOL}}}LogoutRequest", nsmap=NS_MAP)
 89        logout_request.attrib["ID"] = self._request_id
 90        logout_request.attrib["Version"] = "2.0"
 91        logout_request.attrib["IssueInstant"] = self._issue_instant
 92        logout_request.attrib["Destination"] = self.destination
 93
 94        logout_request.append(self.get_issuer())
 95        logout_request.append(self.get_name_id())
 96
 97        if self.session_index:
 98            session_index_element = Element(f"{{{NS_SAML_PROTOCOL}}}SessionIndex")
 99            session_index_element.text = self.session_index
100            logout_request.append(session_index_element)
101
102        return logout_request
103
104    def encode_post(self) -> str:
105        """Encode LogoutRequest for POST binding"""
106        logout_request = self.build()
107        if self.provider.signing_kp and self.provider.sign_logout_request:
108            self._sign_logout_request(logout_request)
109        return base64.b64encode(etree.tostring(logout_request)).decode()
110
111    def encode_redirect(self) -> str:
112        """Encode LogoutRequest for Redirect binding"""
113        logout_request = self.build()
114        # Note: For redirect binding, signatures are added as query parameters, not in XML
115        # Ensure proper XML serialization with encoding declaration
116        xml_str = etree.tostring(logout_request, encoding="UTF-8", xml_declaration=True)
117        return deflate_and_base64_encode(xml_str.decode("UTF-8"))
118
119    def get_redirect_url(self) -> str:
120        """Build complete logout URL for redirect binding with signature if needed"""
121        encoded_request = self.encode_redirect()
122        params = {
123            "SAMLRequest": encoded_request,
124        }
125
126        if self.relay_state:
127            params["RelayState"] = self.relay_state
128
129        if self.provider.signing_kp and self.provider.sign_logout_request:
130            sig_alg = self.provider.signature_algorithm
131            params["SigAlg"] = sig_alg
132
133            # Build the string to sign
134            query_string = self._build_signable_query_string(params)
135
136            signature = self._sign_query_string(query_string)
137            params["Signature"] = base64.b64encode(signature).decode()
138
139        # Some SP's use query params on their sls endpoint
140        separator = "&" if "?" in self.destination else "?"
141        return f"{self.destination}{separator}{urlencode(params)}"
142
143    def get_post_form_data(self) -> dict:
144        """Get form data for POST binding"""
145        return {
146            "SAMLRequest": self.encode_post(),
147            "RelayState": self.relay_state or "",
148        }
149
150    def _sign_logout_request(self, logout_request: _Element):
151        """Sign the LogoutRequest element"""
152        signature_algorithm_transform = SIGN_ALGORITHM_TRANSFORM_MAP.get(
153            self.provider.signature_algorithm, xmlsec.constants.TransformRsaSha1
154        )
155        signature = xmlsec.template.create(
156            logout_request,
157            xmlsec.constants.TransformExclC14N,
158            signature_algorithm_transform,
159            ns=xmlsec.constants.DSigNs,
160        )
161
162        issuer = logout_request.find(f"{{{NS_SAML_ASSERTION}}}Issuer")
163        if issuer is not None:
164            issuer.addnext(signature)
165        else:
166            logout_request.insert(0, signature)
167
168        self._sign(logout_request)
169
170    def _sign(self, element: _Element):
171        """Sign an XML element based on the providers' configured signing settings"""
172        digest_algorithm_transform = DIGEST_ALGORITHM_TRANSLATION_MAP.get(
173            self.provider.digest_algorithm, xmlsec.constants.TransformSha1
174        )
175        xmlsec.tree.add_ids(element, ["ID"])
176        signature_node = xmlsec.tree.find_node(element, xmlsec.constants.NodeSignature)
177        ref = xmlsec.template.add_reference(
178            signature_node,
179            digest_algorithm_transform,
180            uri="#" + element.attrib["ID"],
181        )
182        xmlsec.template.add_transform(ref, xmlsec.constants.TransformEnveloped)
183        xmlsec.template.add_transform(ref, xmlsec.constants.TransformExclC14N)
184        key_info = xmlsec.template.ensure_key_info(signature_node)
185        xmlsec.template.add_x509_data(key_info)
186
187        ctx = xmlsec.SignatureContext()
188
189        key = xmlsec.Key.from_memory(
190            self.provider.signing_kp.key_data,
191            xmlsec.constants.KeyDataFormatPem,
192            None,
193        )
194        key.load_cert_from_memory(
195            self.provider.signing_kp.certificate_data,
196            xmlsec.constants.KeyDataFormatCertPem,
197        )
198        ctx.key = key
199        ctx.sign(remove_xml_newlines(element, signature_node))
200
201    def _build_signable_query_string(self, params: dict) -> str:
202        """Build query string for signing (order matters per SAML spec)"""
203        # SAML spec requires specific order: SAMLRequest, RelayState, SigAlg
204        # Values must be URL-encoded individually before concatenation
205        ordered = []
206        if "SAMLRequest" in params:
207            ordered.append(f"SAMLRequest={quote(params['SAMLRequest'], safe='')}")
208        if "RelayState" in params:
209            ordered.append(f"RelayState={quote(params['RelayState'], safe='')}")
210        if "SigAlg" in params:
211            ordered.append(f"SigAlg={quote(params['SigAlg'], safe='')}")
212        return "&".join(ordered)
213
214    def _sign_query_string(self, query_string: str) -> bytes:
215        """Sign the query string for redirect binding"""
216        signature_algorithm_transform = SIGN_ALGORITHM_TRANSFORM_MAP.get(
217            self.provider.signature_algorithm, xmlsec.constants.TransformRsaSha256
218        )
219
220        key = xmlsec.Key.from_memory(
221            self.provider.signing_kp.key_data,
222            xmlsec.constants.KeyDataFormatPem,
223            None,
224        )
225
226        ctx = xmlsec.SignatureContext()
227        ctx.key = key
228
229        return ctx.sign_binary(query_string.encode("utf-8"), signature_algorithm_transform)