authentik.providers.saml.processors.metadata_parser

  1"""SAML ServiceProvider Metadata Parser and dataclass"""
  2
  3from dataclasses import dataclass
  4
  5import xmlsec
  6from cryptography.hazmat.backends import default_backend
  7from cryptography.x509 import load_pem_x509_certificate
  8from defusedxml.lxml import fromstring
  9from lxml import etree  # nosec
 10from structlog.stdlib import get_logger
 11
 12from authentik.common.saml.constants import (
 13    NS_MAP,
 14    NS_SAML_METADATA,
 15    SAML_BINDING_POST,
 16    SAML_BINDING_REDIRECT,
 17)
 18from authentik.crypto.models import CertificateKeyPair, format_cert
 19from authentik.flows.models import Flow
 20from authentik.providers.saml.models import SAMLBindings, SAMLPropertyMapping, SAMLProvider
 21from authentik.sources.saml.models import SAMLNameIDPolicy
 22
 23LOGGER = get_logger()
 24
 25
 26@dataclass(slots=True)
 27class ServiceProviderMetadata:
 28    """SP Metadata Dataclass"""
 29
 30    entity_id: str
 31
 32    acs_binding: str
 33    acs_location: str
 34
 35    auth_n_request_signed: bool
 36    assertion_signed: bool
 37    name_id_policy: SAMLNameIDPolicy
 38
 39    signing_keypair: CertificateKeyPair | None = None
 40    encryption_keypair: CertificateKeyPair | None = None
 41
 42    # Single Logout Service (optional)
 43    sls_binding: str | None = None
 44    sls_location: str | None = None
 45
 46    def to_provider(
 47        self, name: str, authorization_flow: Flow, invalidation_flow: Flow
 48    ) -> SAMLProvider:
 49        """Create a SAMLProvider instance from the details. `name` is required,
 50        as depending on the metadata CertificateKeypairs might have to be created."""
 51        provider = SAMLProvider.objects.create(
 52            name=name, authorization_flow=authorization_flow, invalidation_flow=invalidation_flow
 53        )
 54        provider.issuer = self.entity_id
 55        provider.sp_binding = self.acs_binding
 56        provider.acs_url = self.acs_location
 57        provider.default_name_id_policy = self.name_id_policy
 58        # Single Logout Service
 59        if self.sls_location:
 60            provider.sls_url = self.sls_location
 61        if self.sls_binding:
 62            provider.sls_binding = self.sls_binding
 63        if self.signing_keypair and self.auth_n_request_signed:
 64            self.signing_keypair.name = f"Provider {name} - SAML Signing Certificate"
 65            self.signing_keypair.save()
 66            provider.verification_kp = self.signing_keypair
 67        if self.encryption_keypair:
 68            self.encryption_keypair.name = f"Provider {name} - SAML Encryption Certificate"
 69            self.encryption_keypair.save()
 70            provider.encryption_kp = self.encryption_keypair
 71        if self.assertion_signed:
 72            provider.signing_kp = CertificateKeyPair.objects.exclude(key_data__iexact="").first()
 73        # Set all auto-generated Property-mappings as defaults
 74        # They should provide a sane default for most applications:
 75        provider.property_mappings.set(SAMLPropertyMapping.objects.exclude(managed__isnull=True))
 76        provider.save()
 77        return provider
 78
 79
 80class ServiceProviderMetadataParser:
 81    """Service-Provider Metadata Parser"""
 82
 83    def get_signing_cert(self, root: etree.Element) -> CertificateKeyPair | None:
 84        """Extract signing X509Certificate from metadata, when given."""
 85        signing_certs = root.xpath(
 86            '//md:SPSSODescriptor/md:KeyDescriptor[@use="signing"]//ds:X509Certificate/text()',
 87            namespaces=NS_MAP,
 88        )
 89        if len(signing_certs) < 1:
 90            return None
 91        raw_cert = format_cert(signing_certs[0])
 92        # sanity check, make sure the certificate is valid.
 93        load_pem_x509_certificate(raw_cert.encode("utf-8"), default_backend())
 94        return CertificateKeyPair(
 95            certificate_data=raw_cert,
 96        )
 97
 98    def get_encryption_cert(self, root: etree.Element) -> CertificateKeyPair | None:
 99        """Extract encryption X509Certificate from metadata, when given."""
100        encryption_certs = root.xpath(
101            '//md:SPSSODescriptor/md:KeyDescriptor[@use="encryption"]//ds:X509Certificate/text()',
102            namespaces=NS_MAP,
103        )
104        if len(encryption_certs) < 1:
105            return None
106        raw_cert = format_cert(encryption_certs[0])
107        # sanity check, make sure the certificate is valid.
108        load_pem_x509_certificate(raw_cert.encode("utf-8"), default_backend())
109        return CertificateKeyPair(
110            certificate_data=raw_cert,
111        )
112
113    def check_signature(self, root: etree.Element, keypair: CertificateKeyPair):
114        """If Metadata is signed, check validity of signature"""
115        xmlsec.tree.add_ids(root, ["ID"])
116        signature_nodes = root.xpath("/md:EntityDescriptor/ds:Signature", namespaces=NS_MAP)
117        if len(signature_nodes) != 1:
118            # No Signature
119            return
120
121        signature_node = signature_nodes[0]
122
123        if signature_node is not None:
124            try:
125                ctx = xmlsec.SignatureContext()
126                key = xmlsec.Key.from_memory(
127                    keypair.certificate_data,
128                    xmlsec.constants.KeyDataFormatCertPem,
129                    None,
130                )
131                ctx.key = key
132                ctx.verify(signature_node)
133            except xmlsec.Error as exc:
134                raise ValueError("Failed to verify Metadata signature") from exc
135
136    def parse(self, raw_xml: str) -> ServiceProviderMetadata:
137        """Parse raw XML to ServiceProviderMetadata"""
138        root = fromstring(raw_xml.encode())
139
140        entity_id = root.attrib["entityID"]
141        sp_sso_descriptors = root.findall(f"{{{NS_SAML_METADATA}}}SPSSODescriptor")
142        if len(sp_sso_descriptors) < 1:
143            raise ValueError("no SPSSODescriptor objects found.")
144        # For now we'll only look at the first descriptor.
145        # Even if multiple descriptors exist, we can only configure one
146        descriptor = sp_sso_descriptors[0]
147        auth_n_request_signed = False
148        if "AuthnRequestsSigned" in descriptor.attrib:
149            auth_n_request_signed = descriptor.attrib["AuthnRequestsSigned"].lower() == "true"
150
151        assertion_signed = False
152        if "WantAssertionsSigned" in descriptor.attrib:
153            assertion_signed = descriptor.attrib["WantAssertionsSigned"].lower() == "true"
154
155        acs_services = descriptor.findall(f"{{{NS_SAML_METADATA}}}AssertionConsumerService")
156        if len(acs_services) < 1:
157            raise ValueError("No AssertionConsumerService found.")
158
159        acs_service = acs_services[0]
160        acs_binding = {
161            SAML_BINDING_REDIRECT: SAMLBindings.REDIRECT,
162            SAML_BINDING_POST: SAMLBindings.POST,
163        }[acs_service.attrib["Binding"]]
164        acs_location = acs_service.attrib["Location"]
165
166        signing_keypair = self.get_signing_cert(root)
167        if signing_keypair:
168            self.check_signature(root, signing_keypair)
169        encryption_keypair = self.get_encryption_cert(root)
170
171        name_id_format = descriptor.findall(f"{{{NS_SAML_METADATA}}}NameIDFormat")
172        name_id_policy = SAMLNameIDPolicy.UNSPECIFIED
173        if len(name_id_format) > 0:
174            name_id_policy = SAMLNameIDPolicy(name_id_format[0].text)
175
176        # Parse SingleLogoutService (optional)
177        sls_binding = None
178        sls_location = None
179        sls_services = descriptor.findall(f"{{{NS_SAML_METADATA}}}SingleLogoutService")
180        if len(sls_services) > 0:
181            sls_service = sls_services[0]
182            sls_binding = {
183                SAML_BINDING_REDIRECT: SAMLBindings.REDIRECT,
184                SAML_BINDING_POST: SAMLBindings.POST,
185            }.get(sls_service.attrib.get("Binding"))
186            sls_location = sls_service.attrib.get("Location")
187
188        return ServiceProviderMetadata(
189            entity_id=entity_id,
190            acs_binding=acs_binding,
191            acs_location=acs_location,
192            auth_n_request_signed=auth_n_request_signed,
193            assertion_signed=assertion_signed,
194            signing_keypair=signing_keypair,
195            encryption_keypair=encryption_keypair,
196            name_id_policy=name_id_policy,
197            sls_binding=sls_binding,
198            sls_location=sls_location,
199        )