authentik.providers.saml.processors.metadata

  1"""SAML Identity Provider Metadata Processor"""
  2
  3from collections.abc import Iterator
  4from hashlib import sha256
  5
  6import xmlsec  # nosec
  7from django.http import HttpRequest
  8from django.urls import reverse
  9from lxml.etree import Element, SubElement, _Element, tostring  # nosec
 10
 11from authentik.common.saml.constants import (
 12    DIGEST_ALGORITHM_TRANSLATION_MAP,
 13    NS_MAP,
 14    NS_SAML_METADATA,
 15    NS_SAML_PROTOCOL,
 16    NS_SIGNATURE,
 17    SAML_BINDING_POST,
 18    SAML_BINDING_REDIRECT,
 19    SAML_NAME_ID_FORMAT_EMAIL,
 20    SAML_NAME_ID_FORMAT_PERSISTENT,
 21    SAML_NAME_ID_FORMAT_TRANSIENT,
 22    SAML_NAME_ID_FORMAT_X509,
 23    SIGN_ALGORITHM_TRANSFORM_MAP,
 24)
 25from authentik.lib.xml import remove_xml_newlines
 26from authentik.providers.saml.models import SAMLProvider
 27from authentik.providers.saml.utils.encoding import strip_pem_header
 28
 29
 30class MetadataProcessor:
 31    """SAML Identity Provider Metadata Processor"""
 32
 33    provider: SAMLProvider
 34    http_request: HttpRequest
 35    force_binding: str | None
 36
 37    def __init__(self, provider: SAMLProvider, request: HttpRequest):
 38        self.provider = provider
 39        self.http_request = request
 40        self.force_binding = None
 41        self.xml_id = "_" + sha256(f"{provider.name}-{provider.pk}".encode("ascii")).hexdigest()
 42
 43    def _get_issuer_value(self) -> str:
 44        """Get issuer value, with fallback to generated URL if empty"""
 45        # If user has set an override issuer, use it
 46        if self.provider.issuer_override:
 47            return self.provider.issuer_override
 48
 49        return self.http_request.build_absolute_uri(
 50            reverse(
 51                "authentik_providers_saml:metadata-download",
 52                kwargs={"application_slug": self.provider.application.slug},
 53            )
 54        )
 55
 56    # Using type unions doesn't work with cython types (which is what lxml is)
 57    def get_signing_key_descriptor(self) -> Element | None:
 58        """Get Signing KeyDescriptor, if enabled for the provider"""
 59        if not self.provider.signing_kp:
 60            return None
 61        key_descriptor = Element(f"{{{NS_SAML_METADATA}}}KeyDescriptor")
 62        key_descriptor.attrib["use"] = "signing"
 63        key_info = SubElement(key_descriptor, f"{{{NS_SIGNATURE}}}KeyInfo")
 64        x509_data = SubElement(key_info, f"{{{NS_SIGNATURE}}}X509Data")
 65        x509_certificate = SubElement(x509_data, f"{{{NS_SIGNATURE}}}X509Certificate")
 66        x509_certificate.text = strip_pem_header(
 67            self.provider.signing_kp.certificate_data.replace("\r", "")
 68        )
 69        return key_descriptor
 70
 71    def get_name_id_formats(self) -> Iterator[Element]:
 72        """Get compatible NameID Formats"""
 73        formats = [
 74            SAML_NAME_ID_FORMAT_EMAIL,
 75            SAML_NAME_ID_FORMAT_PERSISTENT,
 76            SAML_NAME_ID_FORMAT_X509,
 77            SAML_NAME_ID_FORMAT_TRANSIENT,
 78        ]
 79        for name_id_format in formats:
 80            element = Element(f"{{{NS_SAML_METADATA}}}NameIDFormat")
 81            element.text = name_id_format
 82            yield element
 83
 84    def _get_unified_url(self) -> str:
 85        """Get the unified SAML endpoint URL"""
 86        return self.http_request.build_absolute_uri(
 87            reverse(
 88                "authentik_providers_saml:base",
 89                kwargs={"application_slug": self.provider.application.slug},
 90            )
 91        )
 92
 93    def get_sso_bindings(self) -> Iterator[Element]:
 94        """Get all SSO Bindings - both point to unified endpoint"""
 95        unified_url = self._get_unified_url()
 96        for binding in [SAML_BINDING_REDIRECT, SAML_BINDING_POST]:
 97            if self.force_binding and self.force_binding != binding:
 98                continue
 99            element = Element(f"{{{NS_SAML_METADATA}}}SingleSignOnService")
100            element.attrib["Binding"] = binding
101            element.attrib["Location"] = unified_url
102            yield element
103
104    def get_slo_bindings(self) -> Iterator[Element]:
105        """Get all SLO Bindings - both point to unified endpoint"""
106        unified_url = self._get_unified_url()
107        for binding in [SAML_BINDING_REDIRECT, SAML_BINDING_POST]:
108            if self.force_binding and self.force_binding != binding:
109                continue
110            element = Element(f"{{{NS_SAML_METADATA}}}SingleLogoutService")
111            element.attrib["Binding"] = binding
112            element.attrib["Location"] = unified_url
113            yield element
114
115    def _prepare_signature(self, entity_descriptor: _Element):
116        sign_algorithm_transform = SIGN_ALGORITHM_TRANSFORM_MAP.get(
117            self.provider.signature_algorithm, xmlsec.constants.TransformRsaSha1
118        )
119        signature = xmlsec.template.create(
120            entity_descriptor,
121            xmlsec.constants.TransformExclC14N,
122            sign_algorithm_transform,
123            ns=xmlsec.constants.DSigNs,
124        )
125        entity_descriptor.append(signature)
126
127    def _sign(self, entity_descriptor: _Element):
128        digest_algorithm_transform = DIGEST_ALGORITHM_TRANSLATION_MAP.get(
129            self.provider.digest_algorithm, xmlsec.constants.TransformSha1
130        )
131        assertion = entity_descriptor.xpath("//md:EntityDescriptor", namespaces=NS_MAP)[0]
132        xmlsec.tree.add_ids(assertion, ["ID"])
133        signature_node = xmlsec.tree.find_node(assertion, xmlsec.constants.NodeSignature)
134        ref = xmlsec.template.add_reference(
135            signature_node,
136            digest_algorithm_transform,
137            uri="#" + self.xml_id,
138        )
139        xmlsec.template.add_transform(ref, xmlsec.constants.TransformEnveloped)
140        xmlsec.template.add_transform(ref, xmlsec.constants.TransformExclC14N)
141        key_info = xmlsec.template.ensure_key_info(signature_node)
142        xmlsec.template.add_x509_data(key_info)
143
144        ctx = xmlsec.SignatureContext()
145
146        key = xmlsec.Key.from_memory(
147            self.provider.signing_kp.key_data,
148            xmlsec.constants.KeyDataFormatPem,
149            None,
150        )
151        key.load_cert_from_memory(
152            self.provider.signing_kp.certificate_data,
153            xmlsec.constants.KeyDataFormatCertPem,
154        )
155        ctx.key = key
156        ctx.sign(remove_xml_newlines(assertion, signature_node))
157
158    def add_children(self, entity_descriptor: _Element):
159        self.add_idp_sso(entity_descriptor)
160
161    def add_idp_sso(self, entity_descriptor: _Element):
162        idp_sso_descriptor = SubElement(
163            entity_descriptor, f"{{{NS_SAML_METADATA}}}IDPSSODescriptor"
164        )
165        idp_sso_descriptor.attrib["protocolSupportEnumeration"] = NS_SAML_PROTOCOL
166        if self.provider.verification_kp:
167            idp_sso_descriptor.attrib["WantAuthnRequestsSigned"] = "true"
168
169        signing_descriptor = self.get_signing_key_descriptor()
170        if signing_descriptor is not None:
171            idp_sso_descriptor.append(signing_descriptor)
172
173        for binding in self.get_slo_bindings():
174            idp_sso_descriptor.append(binding)
175
176        for name_id_format in self.get_name_id_formats():
177            idp_sso_descriptor.append(name_id_format)
178
179        for binding in self.get_sso_bindings():
180            idp_sso_descriptor.append(binding)
181
182    def build_entity_descriptor(self) -> str:
183        """Build full EntityDescriptor"""
184        entity_descriptor = Element(f"{{{NS_SAML_METADATA}}}EntityDescriptor", nsmap=NS_MAP)
185        entity_descriptor.attrib["ID"] = self.xml_id
186        entity_descriptor.attrib["entityID"] = self._get_issuer_value()
187
188        if self.provider.signing_kp:
189            self._prepare_signature(entity_descriptor)
190
191        self.add_children(entity_descriptor)
192
193        if self.provider.signing_kp:
194            self._sign(entity_descriptor)
195
196        return tostring(entity_descriptor).decode()