sources/saml: rewrite Processors and Views to directly build XML without templates

passbook/providers/saml/utils/encoding.py

@@ -12,9 +12,9 @@ def decode_base64_and_inflate(encoded: str, encoding="utf-8") -> str:
         return decoded_data.decode(encoding)
 
 
-def deflate_and_base64_encode(inflated: bytes, encoding="utf-8"):
+def deflate_and_base64_encode(inflated: str, encoding="utf-8"):
     """Base64 and ZLib Compress b64string"""
-    zlibbed_str = zlib.compress(inflated)
+    zlibbed_str = zlib.compress(inflated.encode())
     compressed_string = zlibbed_str[2:-4]
     return base64.b64encode(compressed_string).decode(encoding)
 

passbook/sources/saml/models.py

@@ -1,5 +1,7 @@
 """saml sp models"""
 from django.db import models
+from django.http import HttpRequest
+from django.shortcuts import reverse
 from django.urls import reverse_lazy
 from django.utils.translation import gettext_lazy as _
 
@@ -93,6 +95,18 @@ class SAMLSource(Source):
 
     form = "passbook.sources.saml.forms.SAMLSourceForm"
 
+    def get_issuer(self, request: HttpRequest) -> str:
+        """Get Source's Issuer, falling back to our Metadata URL if none is set"""
+        if self.issuer is None:
+            return self.build_full_url(request, view="metadata")
+        return self.issuer
+
+    def build_full_url(self, request: HttpRequest, view: str = "acs") -> str:
+        """Build Full ACS URL to be used in IDP"""
+        return request.build_absolute_uri(
+            reverse(f"passbook_sources_saml:{view}", kwargs={"source_slug": self.slug})
+        )
+
     @property
     def ui_login_button(self) -> UILoginButton:
         return UILoginButton(

passbook/sources/saml/processors/constants.py

@@ -1,4 +1,16 @@
 """SAML Source processor constants"""
+NS_SAML_PROTOCOL = "urn:oasis:names:tc:SAML:2.0:protocol"
+NS_SAML_ASSERTION = "urn:oasis:names:tc:SAML:2.0:assertion"
+NS_SAML_METADATA = "urn:oasis:names:tc:SAML:2.0:metadata"
+NS_SIGNATURE = "http://www.w3.org/2000/09/xmldsig#"
+
+NS_MAP = {
+    "samlp": NS_SAML_PROTOCOL,
+    "saml": NS_SAML_ASSERTION,
+    "ds": NS_SIGNATURE,
+    "md": NS_SAML_METADATA,
+}
+
 SAML_NAME_ID_FORMAT_EMAIL = "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress"
 SAML_NAME_ID_FORMAT_PRESISTENT = "urn:oasis:names:tc:SAML:2.0:nameid-format:persistent"
 SAML_NAME_ID_FORMAT_X509 = "urn:oasis:names:tc:SAML:2.0:nameid-format:X509SubjectName"
@@ -6,3 +18,6 @@ SAML_NAME_ID_FORMAT_WINDOWS = (
     "urn:oasis:names:tc:SAML:2.0:nameid-format:WindowsDomainQualifiedName"
 )
 SAML_NAME_ID_FORMAT_TRANSIENT = "urn:oasis:names:tc:SAML:2.0:nameid-format:transient"
+
+SAML_BINDING_POST = "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST"
+SAML_BINDING_REDIRECT = "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"

passbook/sources/saml/processors/metadata.py

@@ -0,0 +1,94 @@
+"""SAML Service Provider Metadata Processor"""
+from typing import Iterator, Optional
+
+from defusedxml import ElementTree
+from django.http import HttpRequest
+from lxml.etree import Element, SubElement  # nosec
+from signxml.util import strip_pem_header
+
+from passbook.sources.saml.models import SAMLSource
+from passbook.sources.saml.processors.constants import (
+    NS_MAP,
+    NS_SAML_METADATA,
+    NS_SIGNATURE,
+    SAML_BINDING_POST,
+    SAML_NAME_ID_FORMAT_EMAIL,
+    SAML_NAME_ID_FORMAT_PRESISTENT,
+    SAML_NAME_ID_FORMAT_TRANSIENT,
+    SAML_NAME_ID_FORMAT_WINDOWS,
+    SAML_NAME_ID_FORMAT_X509,
+)
+
+
+class MetadataProcessor:
+    """SAML Service Provider Metadata Processor"""
+
+    source: SAMLSource
+    http_request: HttpRequest
+
+    def __init__(self, source: SAMLSource, request: HttpRequest):
+        self.source = source
+        self.http_request = request
+
+    def get_signing_key_descriptor(self) -> Optional[Element]:
+        """Get Singing KeyDescriptor, if enabled for the source"""
+        if self.source.signing_kp:
+            key_descriptor = Element(f"{{{NS_SAML_METADATA}}}KeyDescriptor")
+            key_descriptor.attrib["use"] = "signing"
+            key_info = SubElement(key_descriptor, f"{{{NS_SIGNATURE}}}KeyInfo")
+            x509_data = SubElement(key_info, f"{{{NS_SIGNATURE}}}X509Data")
+            x509_certificate = SubElement(
+                x509_data, f"{{{NS_SIGNATURE}}}X509Certificate"
+            )
+            x509_certificate.text = strip_pem_header(
+                self.source.signing_kp.certificate_data.replace("\r", "")
+            ).replace("\n", "")
+            return key_descriptor
+        return None
+
+    def get_name_id_formats(self) -> Iterator[Element]:
+        """Get compatible NameID Formats"""
+        formats = [
+            SAML_NAME_ID_FORMAT_EMAIL,
+            SAML_NAME_ID_FORMAT_PRESISTENT,
+            SAML_NAME_ID_FORMAT_X509,
+            SAML_NAME_ID_FORMAT_WINDOWS,
+            SAML_NAME_ID_FORMAT_TRANSIENT,
+        ]
+        for name_id_format in formats:
+            element = Element(f"{{{NS_SAML_METADATA}}}NameIDFormat")
+            element.text = name_id_format
+            yield element
+
+    def build_entity_descriptor(self) -> str:
+        """Build full EntityDescriptor"""
+        entity_descriptor = Element(
+            f"{{{NS_SAML_METADATA}}}EntityDescriptor", nsmap=NS_MAP
+        )
+        entity_descriptor.attrib["entityID"] = self.source.get_issuer(self.http_request)
+
+        sp_sso_descriptor = SubElement(
+            entity_descriptor, f"{{{NS_SAML_METADATA}}}SPSSODescriptor"
+        )
+        sp_sso_descriptor.attrib[
+            "protocolSupportEnumeration"
+        ] = "urn:oasis:names:tc:SAML:2.0:protocol"
+
+        signing_descriptor = self.get_signing_key_descriptor()
+        if signing_descriptor:
+            sp_sso_descriptor.append(signing_descriptor)
+
+        for name_id_format in self.get_name_id_formats():
+            sp_sso_descriptor.append(name_id_format)
+
+        assertion_consumer_service = SubElement(
+            sp_sso_descriptor, f"{{{NS_SAML_METADATA}}}"
+        )
+        assertion_consumer_service.attrib["isDefault"] = True
+        assertion_consumer_service.attrib["index"] = 0
+        assertion_consumer_service.attrib["Binding"] = SAML_BINDING_POST
+        assertion_consumer_service.attrib["Location"] = self.source.build_full_url(
+            self.http_request
+        )
+
+        return ElementTree.tostring(entity_descriptor).decode()

passbook/sources/saml/processors/request.py

@@ -0,0 +1,53 @@
+"""SAML AuthnRequest Processor"""
+from defusedxml import ElementTree
+from django.http import HttpRequest
+from lxml.etree import Element  # nosec
+
+from passbook.providers.saml.utils import get_random_id
+from passbook.providers.saml.utils.time import get_time_string
+from passbook.sources.saml.models import SAMLSource
+from passbook.sources.saml.processors.constants import (
+    NS_MAP,
+    NS_SAML_ASSERTION,
+    NS_SAML_PROTOCOL,
+)
+
+
+class RequestProcessor:
+    """SAML AuthnRequest Processor"""
+
+    source: SAMLSource
+    http_request: HttpRequest
+
+    def __init__(self, source: SAMLSource, request: HttpRequest):
+        self.source = source
+        self.http_request = request
+
+    def get_issuer(self) -> Element:
+        """Get Issuer Element"""
+        issuer = Element(f"{{{NS_SAML_ASSERTION}}}Issuer")
+        issuer.text = self.source.get_issuer(self.http_request)
+        return issuer
+
+    def get_name_id_policy(self) -> Element:
+        """Get NameID Policy Element"""
+        name_id_policy = Element(f"{{{NS_SAML_PROTOCOL}}}NameIDPolicy")
+        name_id_policy.text = self.source.name_id_policy
+        return name_id_policy
+
+    def build_auth_n(self) -> str:
+        """Get full AuthnRequest"""
+        auth_n_request = Element(f"{{{NS_SAML_PROTOCOL}}}AuthnRequest", nsmap=NS_MAP)
+        auth_n_request.attrib[
+            "AssertionConsumerServiceURL"
+        ] = self.source.build_full_url(self.http_request)
+        auth_n_request.attrib["Destination"] = self.source.sso_url
+        auth_n_request.attrib["ID"] = get_random_id()
+        auth_n_request.attrib["IssueInstant"] = get_time_string()
+        auth_n_request.attrib["ProtocolBinding"] = self.source.binding_type
+        auth_n_request.attrib["Version"] = "2.0"
+        # Create issuer object
+        auth_n_request.append(self.get_issuer())
+        # Create NameID Policy Object
+        auth_n_request.append(self.get_name_id_policy())
+        return ElementTree.tostring(auth_n_request).decode()

passbook/sources/saml/processors/response.py

@@ -38,7 +38,7 @@ if TYPE_CHECKING:
 DEFAULT_BACKEND = "django.contrib.auth.backends.ModelBackend"
 
 
-class Processor:
+class ResponseProcessor:
     """SAML Response Processor"""
 
     _source: SAMLSource

passbook/sources/saml/templates/saml/sp/xml/authn_request.xml

@@ -1,12 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<samlp:AuthnRequest AssertionConsumerServiceURL="{{ ACS_URL }}"
-        Destination="{{ DESTINATION }}"
-        ID="{{ AUTHN_REQUEST_ID }}"
-        IssueInstant="{{ ISSUE_INSTANT }}"
-        ProtocolBinding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST"
-        Version="2.0"
-        xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol">
-    <saml:Issuer xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion">{{ ISSUER }}</saml:Issuer>
-    {{ AUTHN_REQUEST_SIGNATURE }}
-    <samlp:NameIDPolicy Format="{{ NAME_ID_POLICY }}"></samlp:NameIDPolicy>
-</samlp:AuthnRequest>

passbook/sources/saml/templates/saml/sp/xml/signature.xml

@@ -1,9 +0,0 @@
-<ds:Signature xmlns:ds="http://www.w3.org/2000/09/xmldsig#">
-    {{ SIGNED_INFO }}
-    <ds:SignatureValue>{{ RSA_SIGNATURE }}</ds:SignatureValue>
-    <ds:KeyInfo>
-        <ds:X509Data>
-            <ds:X509Certificate>{{ CERTIFICATE }}</ds:X509Certificate>
-        </ds:X509Data>
-    </ds:KeyInfo>
-</ds:Signature>

passbook/sources/saml/templates/saml/sp/xml/signed_info.xml

@@ -1,12 +0,0 @@
-<ds:SignedInfo xmlns:ds="http://www.w3.org/2000/09/xmldsig#">
-    <ds:CanonicalizationMethod Algorithm="http://www.w3.org/2001/10/xml-exc-c14n#"></ds:CanonicalizationMethod>
-    <ds:SignatureMethod Algorithm="http://www.w3.org/2000/09/xmldsig#rsa-sha1"></ds:SignatureMethod>
-    <ds:Reference URI="#${REFERENCE_URI}">
-        <ds:Transforms>
-            <ds:Transform Algorithm="http://www.w3.org/2000/09/xmldsig#enveloped-signature"></ds:Transform>
-            <ds:Transform Algorithm="http://www.w3.org/2001/10/xml-exc-c14n#"></ds:Transform>
-        </ds:Transforms>
-        <ds:DigestMethod Algorithm="http://www.w3.org/2000/09/xmldsig#sha1"></ds:DigestMethod>
-        <ds:DigestValue>{{ SUBJECT_DIGEST }}</ds:DigestValue>
-    </ds:Reference>
-</ds:SignedInfo>

passbook/sources/saml/templates/saml/sp/xml/sp_sso_descriptor.xml

@@ -1,22 +0,0 @@
-<md:EntityDescriptor xmlns:md="urn:oasis:names:tc:SAML:2.0:metadata"
-  xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion"
-  xmlns:ds="http://www.w3.org/2000/09/xmldsig#" entityID="{{ issuer }}">
-  <md:SPSSODescriptor protocolSupportEnumeration="urn:oasis:names:tc:SAML:2.0:protocol">
-    <md:KeyDescriptor use="signing">
-      <ds:KeyInfo xmlns:ds="http://www.w3.org/2000/09/xmldsig#">
-        <ds:X509Data>
-          <ds:X509Certificate>{{ cert_public_key }}</ds:X509Certificate>
-        </ds:X509Data>
-      </ds:KeyInfo>
-    </md:KeyDescriptor>
-    <md:KeyDescriptor use="encryption">
-      <ds:KeyInfo xmlns:ds="http://www.w3.org/2000/09/xmldsig#">
-        <ds:X509Data>
-          <ds:X509Certificate>{{ cert_public_key }}</ds:X509Certificate>
-        </ds:X509Data>
-      </ds:KeyInfo>
-    </md:KeyDescriptor>
-    <md:NameIDFormat>urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress</md:NameIDFormat>
-    <md:AssertionConsumerService isDefault="true" index="0" Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST" Location="{{ acs_url }}"/>
-  </md:SPSSODescriptor>
-</md:EntityDescriptor>

passbook/sources/saml/utils.py

@@ -1,20 +0,0 @@
-"""saml sp helpers"""
-from django.http import HttpRequest
-from django.shortcuts import reverse
-
-from passbook.sources.saml.models import SAMLSource
-
-
-def get_issuer(request: HttpRequest, source: SAMLSource) -> str:
-    """Get Source's Issuer, falling back to our Metadata URL if none is set"""
-    issuer = source.issuer
-    if issuer is None:
-        return build_full_url("metadata", request, source)
-    return issuer
-
-
-def build_full_url(view: str, request: HttpRequest, source: SAMLSource) -> str:
-    """Build Full ACS URL to be used in IDP"""
-    return request.build_absolute_uri(
-        reverse(f"passbook_sources_saml:{view}", kwargs={"source_slug": source.slug})
-    )

passbook/sources/saml/views.py

@@ -9,20 +9,17 @@ from django.utils.translation import gettext_lazy as _
 from django.views import View
 from django.views.decorators.csrf import csrf_exempt
 from signxml import InvalidSignature
-from signxml.util import strip_pem_header
 
 from passbook.lib.views import bad_request_message
-from passbook.providers.saml.utils import get_random_id, render_xml
 from passbook.providers.saml.utils.encoding import deflate_and_base64_encode, nice64
-from passbook.providers.saml.utils.time import get_time_string
 from passbook.sources.saml.exceptions import (
     MissingSAMLResponse,
     UnsupportedNameIDFormat,
 )
 from passbook.sources.saml.models import SAMLBindingTypes, SAMLSource
-from passbook.sources.saml.processors.base import Processor
+from passbook.sources.saml.processors.metadata import MetadataProcessor
-from passbook.sources.saml.utils import build_full_url, get_issuer
+from passbook.sources.saml.processors.request import RequestProcessor
-from passbook.sources.saml.xml_render import get_authnrequest_xml
+from passbook.sources.saml.processors.response import ResponseProcessor
 
 
 class InitiateView(View):
@@ -35,29 +32,23 @@ class InitiateView(View):
             raise Http404
         relay_state = request.GET.get("next", "")
         request.session["sso_destination"] = relay_state
-        parameters = {
+        auth_n_req = RequestProcessor(source, request).build_auth_n()
-            "ACS_URL": build_full_url("acs", request, source),
-            "DESTINATION": source.sso_url,
-            "AUTHN_REQUEST_ID": get_random_id(),
-            "ISSUE_INSTANT": get_time_string(),
-            "ISSUER": get_issuer(request, source),
-            "NAME_ID_POLICY": source.name_id_policy,
-        }
-        authn_req = get_authnrequest_xml(parameters, signed=False)
         # If the source is configured for Redirect bindings, we can just redirect there
         if source.binding_type == SAMLBindingTypes.Redirect:
-            _request = deflate_and_base64_encode(authn_req.encode())
+            saml_request = deflate_and_base64_encode(auth_n_req)
-            url_args = urlencode({"SAMLRequest": _request, "RelayState": relay_state})
+            url_args = urlencode(
+                {"SAMLRequest": saml_request, "RelayState": relay_state}
+            )
             return redirect(f"{source.sso_url}?{url_args}")
         # As POST Binding we show a form
-        _request = nice64(authn_req.encode())
+        saml_request = nice64(auth_n_req)
         if source.binding_type == SAMLBindingTypes.POST:
             return render(
                 request,
                 "saml/sp/login.html",
                 {
                     "request_url": source.sso_url,
-                    "request": _request,
+                    "request": saml_request,
                     "relay_state": relay_state,
                     "source": source,
                 },
@@ -69,7 +60,7 @@ class InitiateView(View):
                 "generic/autosubmit_form.html",
                 {
                     "title": _("Redirecting to %(app)s..." % {"app": source.name}),
-                    "attrs": {"SAMLRequest": _request, "RelayState": relay_state},
+                    "attrs": {"SAMLRequest": saml_request, "RelayState": relay_state},
                     "url": source.sso_url,
                 },
             )
@@ -85,7 +76,7 @@ class ACSView(View):
         source: SAMLSource = get_object_or_404(SAMLSource, slug=source_slug)
         if not source.enabled:
             raise Http404
-        processor = Processor(source)
+        processor = ResponseProcessor(source)
         try:
             processor.parse(request)
         except MissingSAMLResponse as exc:
@@ -122,16 +113,5 @@ class MetadataView(View):
     def dispatch(self, request: HttpRequest, source_slug: str) -> HttpResponse:
         """Replies with the XML Metadata SPSSODescriptor."""
         source: SAMLSource = get_object_or_404(SAMLSource, slug=source_slug)
-        issuer = get_issuer(request, source)
+        metadata = MetadataProcessor(source, request).build_entity_descriptor()
-        cert_stripped = strip_pem_header(
+        return HttpResponse(metadata, content_type="text/xml")
-            source.signing_kp.certificate_data.replace("\r", "")
-        ).replace("\n", "")
-        return render_xml(
-            request,
-            "saml/sp/xml/sp_sso_descriptor.xml",
-            {
-                "acs_url": build_full_url("acs", request, source),
-                "issuer": issuer,
-                "cert_public_key": cert_stripped,
-            },
-        )

passbook/sources/saml/xml_render.py

@@ -1,26 +0,0 @@
-"""Functions for creating XML output."""
-from structlog import get_logger
-
-from passbook.lib.utils.template import render_to_string
-from passbook.providers.saml.utils.xml_signing import get_signature_xml
-
-LOGGER = get_logger()
-
-
-def get_authnrequest_xml(parameters, signed=False):
-    """Get AuthN Request XML"""
-    # Reset signature.
-    params = {}
-    params.update(parameters)
-    params["AUTHN_REQUEST_SIGNATURE"] = ""
-
-    unsigned = render_to_string("saml/sp/xml/authn_request.xml", params)
-    if not signed:
-        return unsigned
-
-    # Sign it.
-    signature_xml = get_signature_xml()
-    params["AUTHN_REQUEST_SIGNATURE"] = signature_xml
-    signed = render_to_string("saml/sp/xml/authn_request.xml", params)
-
-    return signed