diff --git a/passbook/sources/saml/apps.py b/passbook/sources/saml/apps.py index d133474cc..39afcbc76 100644 --- a/passbook/sources/saml/apps.py +++ b/passbook/sources/saml/apps.py @@ -1,5 +1,7 @@ """Passbook SAML app config""" +from importlib import import_module + from django.apps import AppConfig @@ -10,3 +12,6 @@ class PassbookSourceSAMLConfig(AppConfig): label = "passbook_sources_saml" verbose_name = "passbook Sources.SAML" mountpoint = "source/saml/" + + def ready(self): + import_module("passbook.sources.saml.signals") diff --git a/passbook/sources/saml/processors/base.py b/passbook/sources/saml/processors/base.py index f0cf7ff94..163f5a588 100644 --- a/passbook/sources/saml/processors/base.py +++ b/passbook/sources/saml/processors/base.py @@ -1,5 +1,5 @@ """passbook saml source processor""" -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, Dict, Optional from defusedxml import ElementTree from django.http import HttpRequest, HttpResponse @@ -7,6 +7,7 @@ from signxml import XMLVerifier from structlog import get_logger from passbook.core.models import User +from passbook.flows.models import Flow from passbook.flows.planner import ( PLAN_CONTEXT_PENDING_USER, PLAN_CONTEXT_SSO, @@ -60,54 +61,98 @@ class Processor: self._root_xml, x509_cert=self._source.signing_kp.certificate_data ) - def _get_email(self) -> Optional[str]: - """ - Returns the email out of the response. + def _handle_name_id_transient(self, request: HttpRequest) -> HttpResponse: + """Handle a NameID with the Format of Transient. This is a bit more complex than other + formats, as we need to create a temporary User that is used in the session. This + user has an attribute that refers to our Source for cleanup. The user is also deleted + on logout and periodically.""" + # Create a temporary User + name_id = self._get_name_id().text + user: User = User.objects.create( + username=name_id, + attributes={ + "saml": {"source": self._source.pk.hex, "delete_on_logout": True} + }, + ) + LOGGER.debug("Created temporary user for NameID Transient", username=name_id) + user.set_unusable_password() + user.save() + return self._flow_response( + request, + self._source.authentication_flow, + **{ + PLAN_CONTEXT_PENDING_USER: user, + PLAN_CONTEXT_AUTHENTICATION_BACKEND: DEFAULT_BACKEND, + }, + ) - At present, response must pass the email address as the Subject, eg.: - - - - email@example.com - - """ + def _get_name_id(self) -> "Element": + """Get NameID Element""" assertion = self._root.find("{urn:oasis:names:tc:SAML:2.0:assertion}Assertion") subject = assertion.find("{urn:oasis:names:tc:SAML:2.0:assertion}Subject") name_id = subject.find("{urn:oasis:names:tc:SAML:2.0:assertion}NameID") - name_id_format = name_id.attrib["Format"] - if name_id_format != "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress": - raise UnsupportedNameIDFormat( - f"Assertion contains NameID with unsupported format {name_id_format}." - ) - return name_id.text + if name_id is None: + raise ValueError("NameID Element not found!") + return name_id + + def _get_name_id_filter(self) -> Dict[str, str]: + """Returns the subject's NameID as a Filter for the `User`""" + name_id_el = self._get_name_id() + name_id = name_id_el.text + if not name_id: + raise UnsupportedNameIDFormat(f"Subject's NameID is empty.") + _format = name_id_el.attrib["Format"] + if _format == "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress": + return {"email": name_id} + if _format == "urn:oasis:names:tc:SAML:2.0:nameid-format:persistent": + return {"username": name_id} + if _format == "urn:oasis:names:tc:SAML:2.0:nameid-format:X509SubjectName": + # This attribute is statically set by the LDAP source + return {"attributes__distinguishedName": name_id} + if ( + _format + == "urn:oasis:names:tc:SAML:2.0:nameid-format:WindowsDomainQualifiedName" + ): + if "\\" in name_id: + name_id = name_id.split("\\")[1] + return {"username": name_id} + raise UnsupportedNameIDFormat( + f"Assertion contains NameID with unsupported format {_format}." + ) def prepare_flow(self, request: HttpRequest) -> HttpResponse: """Prepare flow plan depending on whether or not the user exists""" - email = self._get_email() - matching_users = User.objects.filter(email=email) + name_id = self._get_name_id() + # transient NameIDs are handeled seperately as they don't have to go through flows. + if ( + name_id.attrib["Format"] + == "urn:oasis:names:tc:SAML:2.0:nameid-format:transient" + ): + return self._handle_name_id_transient(request) + + name_id_filter = self._get_name_id_filter() + matching_users = User.objects.filter(**name_id_filter) if matching_users.exists(): # User exists already, switch to authentication flow - flow = self._source.authentication_flow - request.session[SESSION_KEY_PLAN] = FlowPlanner(flow).plan( + return self._flow_response( request, - { - # Data for authentication + self._source.authentication_flow, + **{ PLAN_CONTEXT_PENDING_USER: matching_users.first(), PLAN_CONTEXT_AUTHENTICATION_BACKEND: DEFAULT_BACKEND, - PLAN_CONTEXT_SSO: True, - }, - ) - else: - flow = self._source.enrollment_flow - request.session[SESSION_KEY_PLAN] = FlowPlanner(flow).plan( - request, - { - # Data for enrollment - PLAN_CONTEXT_PROMPT: {"username": email, "email": email}, - PLAN_CONTEXT_SSO: True, }, ) + return self._flow_response( + request, + self._source.enrollment_flow, + **{PLAN_CONTEXT_PROMPT: name_id_filter,}, + ) + + def _flow_response( + self, request: HttpRequest, flow: Flow, **kwargs + ) -> HttpResponse: + kwargs[PLAN_CONTEXT_SSO] = True + request.session[SESSION_KEY_PLAN] = FlowPlanner(flow).plan(request, kwargs,) return redirect_with_qs( "passbook_flows:flow-executor-shell", request.GET, flow_slug=flow.slug, ) diff --git a/passbook/sources/saml/signals.py b/passbook/sources/saml/signals.py new file mode 100644 index 000000000..15a535704 --- /dev/null +++ b/passbook/sources/saml/signals.py @@ -0,0 +1,20 @@ +"""passbook saml source signal listener""" +from django.contrib.auth.signals import user_logged_out +from django.dispatch import receiver +from django.http import HttpRequest +from structlog import get_logger + +from passbook.core.models import User + +LOGGER = get_logger() + + +@receiver(user_logged_out) +# pylint: disable=unused-argument +def on_user_logged_out(sender, request: HttpRequest, user: User, **_): + """Delete temporary user if the `delete_on_logout` flag is enabled""" + if "saml" in user.attributes: + if "delete_on_logout" in user.attributes["saml"]: + if user.attributes["saml"]["delete_on_logout"]: + LOGGER.debug("Deleted temporary user", user=user) + user.delete() diff --git a/passbook/sources/saml/views.py b/passbook/sources/saml/views.py index 517090e10..21f6d1e42 100644 --- a/passbook/sources/saml/views.py +++ b/passbook/sources/saml/views.py @@ -31,7 +31,7 @@ class InitiateView(View): source: SAMLSource = get_object_or_404(SAMLSource, slug=source_slug) if not source.enabled: raise Http404 - relay_state = request.GET.get("next", None) + relay_state = request.GET.get("next", "") request.session["sso_destination"] = relay_state parameters = { "ACS_URL": build_full_url("acs", request, source), diff --git a/passbook/stages/identification/stage.py b/passbook/stages/identification/stage.py index 38189bf1b..bc3375843 100644 --- a/passbook/stages/identification/stage.py +++ b/passbook/stages/identification/stage.py @@ -49,7 +49,7 @@ class IdentificationStageView(FormView, StageView): # Check all enabled source, add them if they have a UI Login button. kwargs["sources"] = [] - sources = ( + sources: List[Source] = ( Source.objects.filter(enabled=True).order_by("name").select_subclasses() ) for source in sources: