authentik.providers.saml.processors.assertion

  1"""SAML Assertion generator"""
  2
  3from datetime import datetime
  4from hashlib import sha256
  5from types import GeneratorType
  6
  7import xmlsec
  8from django.http import HttpRequest
  9from django.urls import reverse
 10from django.utils.timezone import now
 11from lxml import etree  # nosec
 12from lxml.etree import Element, SubElement, _Element  # nosec
 13from structlog.stdlib import get_logger
 14
 15from authentik.common.saml.constants import (
 16    DIGEST_ALGORITHM_TRANSLATION_MAP,
 17    NS_MAP,
 18    NS_SAML_ASSERTION,
 19    NS_SAML_PROTOCOL,
 20    SAML_NAME_ID_FORMAT_EMAIL,
 21    SAML_NAME_ID_FORMAT_PERSISTENT,
 22    SAML_NAME_ID_FORMAT_TRANSIENT,
 23    SAML_NAME_ID_FORMAT_UNSPECIFIED,
 24    SAML_NAME_ID_FORMAT_WINDOWS,
 25    SAML_NAME_ID_FORMAT_X509,
 26    SIGN_ALGORITHM_TRANSFORM_MAP,
 27)
 28from authentik.core.expression.exceptions import PropertyMappingExpressionException
 29from authentik.events.models import Event, EventAction
 30from authentik.events.signals import get_login_event
 31from authentik.lib.utils.time import timedelta_from_string
 32from authentik.lib.xml import remove_xml_newlines
 33from authentik.providers.saml.models import SAMLPropertyMapping, SAMLProvider
 34from authentik.providers.saml.processors.authn_request_parser import AuthNRequest
 35from authentik.providers.saml.utils import get_random_id
 36from authentik.providers.saml.utils.time import get_time_string
 37from authentik.sources.ldap.auth import LDAP_DISTINGUISHED_NAME
 38from authentik.sources.saml.exceptions import (
 39    InvalidEncryption,
 40    InvalidSignature,
 41    UnsupportedNameIDFormat,
 42)
 43from authentik.stages.password.stage import PLAN_CONTEXT_METHOD, PLAN_CONTEXT_METHOD_ARGS
 44
 45LOGGER = get_logger()
 46
 47
 48class AssertionProcessor:
 49    """Generate a SAML Response from an AuthNRequest"""
 50
 51    provider: SAMLProvider
 52    http_request: HttpRequest
 53    auth_n_request: AuthNRequest
 54
 55    _issue_instant: str
 56    _assertion_id: str
 57    _response_id: str
 58
 59    _auth_instant: str
 60    _valid_not_before: str
 61    _session_not_on_or_after: str
 62    _valid_not_on_or_after: str
 63
 64    session_index: str
 65    name_id: str
 66    name_id_format: str
 67    issuer: str
 68    session_not_on_or_after_datetime: datetime
 69
 70    def __init__(self, provider: SAMLProvider, request: HttpRequest, auth_n_request: AuthNRequest):
 71        self.provider = provider
 72        self.http_request = request
 73        self.auth_n_request = auth_n_request
 74
 75        self._issue_instant = get_time_string()
 76        self._assertion_id = get_random_id()
 77        self._response_id = get_random_id()
 78
 79        _login_event = get_login_event(self.http_request)
 80        _login_time = now()
 81        if _login_event:
 82            _login_time = _login_event.created
 83        self._auth_instant = get_time_string(_login_time)
 84        self._valid_not_before = get_time_string(
 85            timedelta_from_string(self.provider.assertion_valid_not_before)
 86        )
 87        self.session_not_on_or_after_datetime = now() + timedelta_from_string(
 88            self.provider.session_valid_not_on_or_after
 89        )
 90        self._session_not_on_or_after = get_time_string(self.session_not_on_or_after_datetime)
 91        self._valid_not_on_or_after = get_time_string(
 92            timedelta_from_string(self.provider.assertion_valid_not_on_or_after)
 93        )
 94
 95    def get_attributes(self) -> Element:
 96        """Get AttributeStatement Element with Attributes from Property Mappings."""
 97        # https://commons.lbl.gov/display/IDMgmt/Attribute+Definitions
 98        attribute_statement = Element(f"{{{NS_SAML_ASSERTION}}}AttributeStatement")
 99        user = self.http_request.user
100        for mapping in SAMLPropertyMapping.objects.filter(provider=self.provider).order_by(
101            "saml_name"
102        ):
103            try:
104                mapping: SAMLPropertyMapping
105                value = mapping.evaluate(
106                    user=user,
107                    request=self.http_request,
108                    provider=self.provider,
109                )
110                if value is None:
111                    continue
112
113                attribute = Element(f"{{{NS_SAML_ASSERTION}}}Attribute")
114                if mapping.friendly_name and mapping.friendly_name != "":
115                    attribute.attrib["FriendlyName"] = mapping.friendly_name
116                attribute.attrib["Name"] = mapping.saml_name
117
118                if not isinstance(value, list | GeneratorType):
119                    value = [value]
120
121                for value_item in value:
122                    attribute_value = SubElement(
123                        attribute, f"{{{NS_SAML_ASSERTION}}}AttributeValue"
124                    )
125                    str_value = str(value_item) if not isinstance(value_item, str) else value_item
126                    attribute_value.text = str_value
127
128                attribute_statement.append(attribute)
129
130            except (PropertyMappingExpressionException, ValueError) as exc:
131                # Value error can be raised when assigning invalid data to an attribute
132                Event.new(
133                    EventAction.CONFIGURATION_ERROR,
134                    message=f"Failed to evaluate property-mapping: '{mapping.name}'",
135                    provider=self.provider,
136                    mapping=mapping,
137                ).from_http(self.http_request)
138                LOGGER.warning("Failed to evaluate property mapping", exc=exc)
139                continue
140        return attribute_statement
141
142    def _get_issuer_value(self) -> str:
143        """Get issuer value, with fallback to generated URL if empty"""
144        # If user has set an override issuer, use it
145        if self.provider.issuer_override:
146            return self.provider.issuer_override
147
148        return self.http_request.build_absolute_uri(
149            reverse(
150                "authentik_providers_saml:metadata-download",
151                kwargs={"application_slug": self.provider.application.slug},
152            )
153        )
154
155    def get_issuer(self) -> Element:
156        """Get Issuer Element"""
157        issuer = Element(f"{{{NS_SAML_ASSERTION}}}Issuer", nsmap=NS_MAP)
158        self.issuer = self._get_issuer_value()
159        issuer.text = self.issuer
160        return issuer
161
162    def get_assertion_auth_n_statement(self) -> Element:
163        """Generate AuthnStatement with AuthnContext and ContextClassRef Elements."""
164        auth_n_statement = Element(f"{{{NS_SAML_ASSERTION}}}AuthnStatement")
165        auth_n_statement.attrib["AuthnInstant"] = self._auth_instant
166        self.session_index = sha256(
167            self.http_request.session.session_key.encode("ascii")
168        ).hexdigest()
169        auth_n_statement.attrib["SessionIndex"] = self.session_index
170        auth_n_statement.attrib["SessionNotOnOrAfter"] = self._session_not_on_or_after
171
172        auth_n_context = SubElement(auth_n_statement, f"{{{NS_SAML_ASSERTION}}}AuthnContext")
173        auth_n_context_class_ref = SubElement(
174            auth_n_context, f"{{{NS_SAML_ASSERTION}}}AuthnContextClassRef"
175        )
176        auth_n_context_class_ref.text = "urn:oasis:names:tc:SAML:2.0:ac:classes:unspecified"
177        event = get_login_event(self.http_request)
178        if event:
179            method = event.context.get(PLAN_CONTEXT_METHOD, "")
180            method_args = event.context.get(PLAN_CONTEXT_METHOD_ARGS, {})
181            if method == "password":
182                auth_n_context_class_ref.text = (
183                    "urn:oasis:names:tc:SAML:2.0:ac:classes:PasswordProtectedTransport"
184                )
185            if "mfa_devices" in method_args:
186                auth_n_context_class_ref.text = (
187                    "urn:oasis:names:tc:SAML:2.0:ac:classes:MobileTwoFactorContract"
188                )
189            if method in ["auth_mfa", "auth_webauthn_pwl"]:
190                auth_n_context_class_ref.text = (
191                    "urn:oasis:names:tc:SAML:2.0:ac:classes:MobileOneFactorContract"
192                )
193        if self.provider.authn_context_class_ref_mapping:
194            try:
195                value = self.provider.authn_context_class_ref_mapping.evaluate(
196                    user=self.http_request.user,
197                    request=self.http_request,
198                    provider=self.provider,
199                )
200                if value is not None:
201                    auth_n_context_class_ref.text = str(value)
202                return auth_n_statement
203            except PropertyMappingExpressionException as exc:
204                Event.new(
205                    EventAction.CONFIGURATION_ERROR,
206                    message=(
207                        "Failed to evaluate property-mapping: "
208                        f"'{self.provider.authn_context_class_ref_mapping.name}'"
209                    ),
210                    provider=self.provider,
211                    mapping=self.provider.authn_context_class_ref_mapping,
212                ).from_http(self.http_request)
213                LOGGER.warning("Failed to evaluate property mapping", exc=exc)
214                return auth_n_statement
215        return auth_n_statement
216
217    def get_assertion_conditions(self) -> Element:
218        """Generate Conditions with AudienceRestriction and Audience Elements."""
219        conditions = Element(f"{{{NS_SAML_ASSERTION}}}Conditions")
220        conditions.attrib["NotBefore"] = self._valid_not_before
221        conditions.attrib["NotOnOrAfter"] = self._valid_not_on_or_after
222        if self.provider.audience != "":
223            audience_restriction = SubElement(
224                conditions, f"{{{NS_SAML_ASSERTION}}}AudienceRestriction"
225            )
226            audience = SubElement(audience_restriction, f"{{{NS_SAML_ASSERTION}}}Audience")
227            audience.text = self.provider.audience
228        return conditions
229
230    def get_name_id(self) -> Element:
231        """Get NameID Element"""
232        name_id = Element(f"{{{NS_SAML_ASSERTION}}}NameID")
233        # For requests that don't specify a NameIDPolicy, check if we
234        # can fall back to the provider default
235        if (
236            self.auth_n_request.name_id_policy == SAML_NAME_ID_FORMAT_UNSPECIFIED
237            and self.provider.default_name_id_policy != SAML_NAME_ID_FORMAT_UNSPECIFIED
238        ):
239            self.auth_n_request.name_id_policy = self.provider.default_name_id_policy
240        name_id.attrib["Format"] = self.auth_n_request.name_id_policy
241        self.name_id_format = self.auth_n_request.name_id_policy
242        # persistent is used as a fallback, so always generate it
243        persistent = self.http_request.user.uid
244        name_id.text = persistent
245        self.name_id = persistent
246        # If name_id_mapping is set, we override the value, regardless of what the SP asks for
247        if self.provider.name_id_mapping:
248            try:
249                value = self.provider.name_id_mapping.evaluate(
250                    user=self.http_request.user,
251                    request=self.http_request,
252                    provider=self.provider,
253                )
254                if value is not None:
255                    name_id.text = str(value)
256                    self.name_id = str(value)
257                return name_id
258            except PropertyMappingExpressionException as exc:
259                Event.new(
260                    EventAction.CONFIGURATION_ERROR,
261                    message=(
262                        "Failed to evaluate property-mapping: "
263                        f"'{self.provider.name_id_mapping.name}'",
264                    ),
265                    provider=self.provider,
266                    mapping=self.provider.name_id_mapping,
267                ).from_http(self.http_request)
268                LOGGER.warning("Failed to evaluate property mapping", exc=exc)
269                return name_id
270        if self.auth_n_request.name_id_policy == SAML_NAME_ID_FORMAT_EMAIL:
271            name_id.text = self.http_request.user.email
272            self.name_id = self.http_request.user.email
273            return name_id
274        if self.auth_n_request.name_id_policy in [
275            SAML_NAME_ID_FORMAT_PERSISTENT,
276            SAML_NAME_ID_FORMAT_UNSPECIFIED,
277        ]:
278            name_id.text = persistent
279            self.name_id = persistent
280            return name_id
281        if self.auth_n_request.name_id_policy == SAML_NAME_ID_FORMAT_X509:
282            # This attribute is statically set by the LDAP source
283            name_id.text = self.http_request.user.attributes.get(
284                LDAP_DISTINGUISHED_NAME, persistent
285            )
286            self.name_id = name_id.text
287            return name_id
288        if self.auth_n_request.name_id_policy == SAML_NAME_ID_FORMAT_WINDOWS:
289            # This attribute is statically set by the LDAP source
290            name_id.text = self.http_request.user.attributes.get("upn", persistent)
291            self.name_id = name_id.text
292            return name_id
293        if self.auth_n_request.name_id_policy == SAML_NAME_ID_FORMAT_TRANSIENT:
294            # Use the hash of the user's session, which changes every session
295            session_key: str = self.http_request.session.session_key
296            name_id.text = sha256(session_key.encode()).hexdigest()
297            self.name_id = name_id.text
298            return name_id
299        raise UnsupportedNameIDFormat(
300            "Assertion contains NameID with unsupported "
301            f"format {self.auth_n_request.name_id_policy}."
302        )
303
304    def get_assertion_subject(self) -> Element:
305        """Generate Subject Element with NameID and SubjectConfirmation Objects"""
306        subject = Element(f"{{{NS_SAML_ASSERTION}}}Subject")
307        subject.append(self.get_name_id())
308
309        subject_confirmation = SubElement(subject, f"{{{NS_SAML_ASSERTION}}}SubjectConfirmation")
310        subject_confirmation.attrib["Method"] = "urn:oasis:names:tc:SAML:2.0:cm:bearer"
311
312        subject_confirmation_data = SubElement(
313            subject_confirmation, f"{{{NS_SAML_ASSERTION}}}SubjectConfirmationData"
314        )
315        if self.auth_n_request.id:
316            subject_confirmation_data.attrib["InResponseTo"] = self.auth_n_request.id
317        subject_confirmation_data.attrib["NotOnOrAfter"] = self._valid_not_on_or_after
318        subject_confirmation_data.attrib["Recipient"] = self.provider.acs_url
319        return subject
320
321    def get_assertion(self) -> Element:
322        """Generate Main Assertion Element"""
323        assertion = Element(f"{{{NS_SAML_ASSERTION}}}Assertion", nsmap=NS_MAP)
324        assertion.attrib["Version"] = "2.0"
325        assertion.attrib["ID"] = self._assertion_id
326        assertion.attrib["IssueInstant"] = self._issue_instant
327        assertion.append(self.get_issuer())
328
329        if self.provider.signing_kp and self.provider.sign_assertion:
330            sign_algorithm_transform = SIGN_ALGORITHM_TRANSFORM_MAP.get(
331                self.provider.signature_algorithm, xmlsec.constants.TransformRsaSha1
332            )
333            signature = xmlsec.template.create(
334                assertion,
335                xmlsec.constants.TransformExclC14N,
336                sign_algorithm_transform,
337                ns=xmlsec.constants.DSigNs,
338            )
339            assertion.append(signature)
340
341        assertion.append(self.get_assertion_subject())
342        assertion.append(self.get_assertion_conditions())
343        assertion.append(self.get_assertion_auth_n_statement())
344
345        assertion.append(self.get_attributes())
346        return assertion
347
348    def get_response(self) -> Element:
349        """Generate Root response element"""
350        response = Element(f"{{{NS_SAML_PROTOCOL}}}Response", nsmap=NS_MAP)
351        response.attrib["Version"] = "2.0"
352        response.attrib["IssueInstant"] = self._issue_instant
353        response.attrib["Destination"] = self.provider.acs_url
354        response.attrib["ID"] = self._response_id
355        if self.auth_n_request.id:
356            response.attrib["InResponseTo"] = self.auth_n_request.id
357
358        response.append(self.get_issuer())
359
360        if self.provider.signing_kp and self.provider.sign_response:
361            sign_algorithm_transform = SIGN_ALGORITHM_TRANSFORM_MAP.get(
362                self.provider.signature_algorithm, xmlsec.constants.TransformRsaSha1
363            )
364            signature = xmlsec.template.create(
365                response,
366                xmlsec.constants.TransformExclC14N,
367                sign_algorithm_transform,
368                ns=xmlsec.constants.DSigNs,
369            )
370            response.append(signature)
371
372        status = SubElement(response, f"{{{NS_SAML_PROTOCOL}}}Status")
373        status_code = SubElement(status, f"{{{NS_SAML_PROTOCOL}}}StatusCode")
374        status_code.attrib["Value"] = "urn:oasis:names:tc:SAML:2.0:status:Success"
375
376        response.append(self.get_assertion())
377        return response
378
379    def _sign(self, element: _Element):
380        """Sign an XML element based on the providers' configured signing settings"""
381        digest_algorithm_transform = DIGEST_ALGORITHM_TRANSLATION_MAP.get(
382            self.provider.digest_algorithm, xmlsec.constants.TransformSha1
383        )
384        xmlsec.tree.add_ids(element, ["ID"])
385        signature_node = xmlsec.tree.find_node(element, xmlsec.constants.NodeSignature)
386        ref = xmlsec.template.add_reference(
387            signature_node,
388            digest_algorithm_transform,
389            uri="#" + element.attrib["ID"],
390        )
391        xmlsec.template.add_transform(ref, xmlsec.constants.TransformEnveloped)
392        xmlsec.template.add_transform(ref, xmlsec.constants.TransformExclC14N)
393        key_info = xmlsec.template.ensure_key_info(signature_node)
394        xmlsec.template.add_x509_data(key_info)
395
396        ctx = xmlsec.SignatureContext()
397
398        key = xmlsec.Key.from_memory(
399            self.provider.signing_kp.key_data,
400            xmlsec.constants.KeyDataFormatPem,
401            None,
402        )
403        key.load_cert_from_memory(
404            self.provider.signing_kp.certificate_data,
405            xmlsec.constants.KeyDataFormatCertPem,
406        )
407        ctx.key = key
408        try:
409            ctx.sign(remove_xml_newlines(element, signature_node))
410        except xmlsec.Error as exc:
411            raise InvalidSignature() from exc
412
413    def _encrypt(self, element: _Element, parent: _Element):
414        """Encrypt SAMLResponse EncryptedAssertion Element"""
415        # Create a standalone copy so namespace declarations are included in the encrypted content
416        element_xml = etree.tostring(element)
417        standalone_element = etree.fromstring(element_xml)
418
419        # Remove the original element from the tree since we're replacing it with encrypted version
420        parent.remove(element)
421
422        manager = xmlsec.KeysManager()
423        key = xmlsec.Key.from_memory(
424            self.provider.encryption_kp.certificate_data,
425            xmlsec.constants.KeyDataFormatCertPem,
426        )
427
428        manager.add_key(key)
429        encryption_context = xmlsec.EncryptionContext(manager)
430        encryption_context.key = xmlsec.Key.generate(
431            xmlsec.constants.KeyDataAes, 128, xmlsec.constants.KeyDataTypeSession
432        )
433
434        container = SubElement(parent, f"{{{NS_SAML_ASSERTION}}}EncryptedAssertion")
435        enc_data = xmlsec.template.encrypted_data_create(
436            container, xmlsec.Transform.AES128, type=xmlsec.EncryptionType.ELEMENT, ns="xenc"
437        )
438        xmlsec.template.encrypted_data_ensure_cipher_value(enc_data)
439        key_info = xmlsec.template.encrypted_data_ensure_key_info(enc_data, ns="ds")
440        enc_key = xmlsec.template.add_encrypted_key(key_info, xmlsec.Transform.RSA_OAEP)
441        xmlsec.template.encrypted_data_ensure_cipher_value(enc_key)
442
443        try:
444            enc_data = encryption_context.encrypt_xml(enc_data, standalone_element)
445        except xmlsec.Error as exc:
446            raise InvalidEncryption() from exc
447
448        container.append(enc_data)
449
450    def build_response(self) -> str:
451        """Build string XML Response and sign if signing is enabled."""
452        root_response = self.get_response()
453        # Sign assertion first (before encryption)
454        if self.provider.signing_kp and self.provider.sign_assertion:
455            assertion = root_response.xpath("//saml:Assertion", namespaces=NS_MAP)[0]
456            self._sign(assertion)
457        # Encrypt assertion (this replaces Assertion with EncryptedAssertion)
458        if self.provider.encryption_kp:
459            assertion = root_response.xpath("//saml:Assertion", namespaces=NS_MAP)[0]
460            self._encrypt(assertion, root_response)
461        # Sign response AFTER encryption so signature covers the encrypted content
462        if self.provider.signing_kp and self.provider.sign_response:
463            response = root_response.xpath("//samlp:Response", namespaces=NS_MAP)[0]
464            self._sign(response)
465        return etree.tostring(root_response).decode("utf-8")  # nosec