This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
authentik/passbook/sources/saml/processors/request.py

149 lines
5.4 KiB
Python

"""SAML AuthnRequest Processor"""
from base64 import b64encode
from typing import Dict
from urllib.parse import quote_plus
import xmlsec
from django.http import HttpRequest
from lxml import etree # nosec
from lxml.etree import Element # nosec
from signxml import XMLSigner
from passbook.providers.saml.utils import get_random_id
from passbook.providers.saml.utils.encoding import deflate_and_base64_encode
from passbook.providers.saml.utils.time import get_time_string
from passbook.sources.saml.models import SAMLSource
from passbook.sources.saml.processors.constants import (
DSA_SHA1,
NS_MAP,
NS_SAML_ASSERTION,
NS_SAML_PROTOCOL,
RSA_SHA1,
RSA_SHA256,
RSA_SHA384,
RSA_SHA512,
)
SESSION_REQUEST_ID = "passbook_source_saml_request_id"
class RequestProcessor:
"""SAML AuthnRequest Processor"""
source: SAMLSource
http_request: HttpRequest
relay_state: str
request_id: str
issue_instant: str
def __init__(self, source: SAMLSource, request: HttpRequest, relay_state: str):
self.source = source
self.http_request = request
self.relay_state = relay_state
self.request_id = get_random_id()
self.http_request.session[SESSION_REQUEST_ID] = self.request_id
self.issue_instant = get_time_string()
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.attrib["Format"] = self.source.name_id_policy
return name_id_policy
def get_auth_n(self) -> Element:
"""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"] = self.request_id
auth_n_request.attrib["IssueInstant"] = self.issue_instant
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 auth_n_request
def build_auth_n(self) -> str:
"""Get Signed string representation of AuthN Request
(used for POST Bindings)"""
auth_n_request = self.get_auth_n()
if self.source.signing_kp:
signed_request = XMLSigner(
c14n_algorithm="http://www.w3.org/2001/10/xml-exc-c14n#",
signature_algorithm=self.source.signature_algorithm,
digest_algorithm=self.source.digest_algorithm,
).sign(
auth_n_request,
cert=self.source.signing_kp.certificate_data,
key=self.source.signing_kp.key_data,
)
return etree.tostring(signed_request).decode()
return etree.tostring(auth_n_request).decode()
def build_auth_n_detached(self) -> Dict[str, str]:
"""Get Dict AuthN Request for Redirect bindings, with detached
Signature. See https://docs.oasis-open.org/security/saml/v2.0/saml-bindings-2.0-os.pdf"""
auth_n_request = self.get_auth_n()
saml_request = deflate_and_base64_encode(
etree.tostring(auth_n_request).decode()
)
response_dict = {
"SAMLRequest": saml_request,
}
if self.relay_state != "":
response_dict["RelayState"] = self.relay_state
if self.source.signing_kp:
sign_algorithm_transform_map = {
DSA_SHA1: xmlsec.constants.TransformDsaSha1,
RSA_SHA1: xmlsec.constants.TransformRsaSha1,
RSA_SHA256: xmlsec.constants.TransformRsaSha256,
RSA_SHA384: xmlsec.constants.TransformRsaSha384,
RSA_SHA512: xmlsec.constants.TransformRsaSha512,
}
sign_algorithm_transform = sign_algorithm_transform_map.get(
self.source.signature_algorithm, xmlsec.constants.TransformRsaSha1
)
# Create the full querystring in the correct order to be signed
querystring = f"SAMLRequest={quote_plus(saml_request)}&"
if "RelayState" in response_dict:
querystring += f"RelayState={quote_plus(response_dict['RelayState'])}&"
querystring += f"SigAlg={quote_plus(self.source.signature_algorithm)}"
ctx = xmlsec.SignatureContext()
key = xmlsec.Key.from_memory(
self.source.signing_kp.key_data, xmlsec.constants.KeyDataFormatPem, None
)
key.load_cert_from_memory(
self.source.signing_kp.certificate_data,
xmlsec.constants.KeyDataFormatPem,
)
ctx.key = key
signature = ctx.sign_binary(
querystring.encode("utf-8"), sign_algorithm_transform
)
response_dict["Signature"] = b64encode(signature).decode()
response_dict["SigAlg"] = self.source.signature_algorithm
return response_dict