173 lines
6.2 KiB
Python
173 lines
6.2 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 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 (
|
|
DIGEST_ALGORITHM_TRANSLATION_MAP,
|
|
NS_MAP,
|
|
NS_SAML_ASSERTION,
|
|
NS_SAML_PROTOCOL,
|
|
SIGN_ALGORITHM_TRANSFORM_MAP,
|
|
)
|
|
|
|
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())
|
|
|
|
if self.source.signing_kp:
|
|
sign_algorithm_transform = SIGN_ALGORITHM_TRANSFORM_MAP.get(
|
|
self.source.signature_algorithm, xmlsec.constants.TransformRsaSha1
|
|
)
|
|
signature = xmlsec.template.create(
|
|
auth_n_request,
|
|
xmlsec.constants.TransformExclC14N,
|
|
sign_algorithm_transform,
|
|
ns="ds", # type: ignore
|
|
)
|
|
auth_n_request.append(signature)
|
|
|
|
# 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:
|
|
xmlsec.tree.add_ids(auth_n_request, ["ID"])
|
|
|
|
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.KeyDataFormatCertPem,
|
|
)
|
|
ctx.key = key
|
|
|
|
digest_algorithm_transform = DIGEST_ALGORITHM_TRANSLATION_MAP.get(
|
|
self.source.digest_algorithm, xmlsec.constants.TransformSha1
|
|
)
|
|
|
|
signature_node = xmlsec.tree.find_node(
|
|
auth_n_request, xmlsec.constants.NodeSignature
|
|
)
|
|
|
|
ref = xmlsec.template.add_reference(
|
|
signature_node,
|
|
digest_algorithm_transform,
|
|
uri="#" + auth_n_request.attrib["ID"],
|
|
)
|
|
xmlsec.template.add_transform(ref, xmlsec.constants.TransformEnveloped)
|
|
xmlsec.template.add_transform(ref, xmlsec.constants.TransformExclC14N)
|
|
key_info = xmlsec.template.ensure_key_info(signature_node)
|
|
xmlsec.template.add_x509_data(key_info)
|
|
|
|
ctx.sign(signature_node)
|
|
|
|
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 = 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
|