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/providers/saml/views.py

264 lines
11 KiB
Python

"""passbook SAML IDP Views"""
from typing import Optional
from django.contrib.auth.mixins import LoginRequiredMixin
from django.core.exceptions import PermissionDenied
from django.core.validators import URLValidator
from django.http import HttpRequest, HttpResponse
from django.shortcuts import get_object_or_404, redirect, render, reverse
from django.utils.decorators import method_decorator
from django.utils.http import urlencode
from django.views import View
from django.views.decorators.csrf import csrf_exempt
from signxml.util import strip_pem_header
from structlog import get_logger
from passbook.audit.models import Event, EventAction
from passbook.core.models import Application, Provider
from passbook.flows.models import in_memory_stage
from passbook.flows.planner import (
PLAN_CONTEXT_APPLICATION,
PLAN_CONTEXT_SSO,
FlowPlanner,
)
from passbook.flows.stage import StageView
from passbook.flows.views import SESSION_KEY_PLAN
from passbook.lib.utils.template import render_to_string
from passbook.lib.utils.urls import redirect_with_qs
from passbook.lib.views import bad_request_message
from passbook.policies.mixins import PolicyAccessMixin
from passbook.providers.saml.exceptions import CannotHandleAssertion
from passbook.providers.saml.models import SAMLBindings, SAMLProvider
from passbook.providers.saml.processors.types import SAMLResponseParams
from passbook.stages.consent.stage import PLAN_CONTEXT_CONSENT_TEMPLATE
LOGGER = get_logger()
URL_VALIDATOR = URLValidator(schemes=("http", "https"))
SESSION_KEY_SAML_REQUEST = "SAMLRequest"
SESSION_KEY_SAML_RESPONSE = "SAMLResponse"
SESSION_KEY_RELAY_STATE = "RelayState"
SESSION_KEY_PARAMS = "SAMLParams"
class SAMLSSOView(LoginRequiredMixin, PolicyAccessMixin, View):
""""SAML SSO Base View, which plans a flow and injects our final stage.
Calls get/post handler."""
application: Application
provider: SAMLProvider
def dispatch(
self, request: HttpRequest, *args, application_slug: str, **kwargs
) -> HttpResponse:
self.application = get_object_or_404(Application, slug=application_slug)
self.provider: SAMLProvider = get_object_or_404(
SAMLProvider, pk=self.application.provider_id
)
if not self.user_has_access(self.application).passing:
raise PermissionDenied()
# Call the method handler, which checks the SAML Request
method_response = super().dispatch(request, *args, application_slug, **kwargs)
if method_response:
return method_response
# Regardless, we start the planner and return to it
planner = FlowPlanner(self.provider.authorization_flow)
planner.allow_empty_flows = True
plan = planner.plan(
self.request,
{
PLAN_CONTEXT_SSO: True,
PLAN_CONTEXT_APPLICATION: self.application,
PLAN_CONTEXT_CONSENT_TEMPLATE: "providers/saml/consent.html",
},
)
plan.append(in_memory_stage(SAMLFlowFinalView))
self.request.session[SESSION_KEY_PLAN] = plan
return redirect_with_qs(
"passbook_flows:flow-executor-shell",
self.request.GET,
flow_slug=self.provider.authorization_flow.slug,
)
class SAMLSSOBindingRedirectView(SAMLSSOView):
"""SAML Handler for SSO/Redirect bindings, which are sent via GET"""
# pylint: disable=unused-argument
def get( # lgtm [py/similar-function]
self, request: HttpRequest, application_slug: str
) -> Optional[HttpResponse]:
"""Handle REDIRECT bindings"""
# Store these values now, because Django's login cycle won't preserve them.
if SESSION_KEY_SAML_REQUEST not in request.GET:
LOGGER.info("handle_saml_request: SAML payload missing")
return bad_request_message(
self.request, "The SAML request payload is missing."
)
self.request.session[SESSION_KEY_SAML_REQUEST] = request.GET[
SESSION_KEY_SAML_REQUEST
]
self.request.session[SESSION_KEY_RELAY_STATE] = request.GET.get(
SESSION_KEY_RELAY_STATE, ""
)
try:
self.provider.processor.can_handle(self.request)
params = self.provider.processor.generate_response()
self.request.session[SESSION_KEY_PARAMS] = params
except CannotHandleAssertion as exc:
LOGGER.info(exc)
return bad_request_message(self.request, str(exc))
return None
@method_decorator(csrf_exempt, name="dispatch")
class SAMLSSOBindingPOSTView(SAMLSSOView):
"""SAML Handler for SSO/POST bindings"""
# pylint: disable=unused-argument
def post(
self, request: HttpRequest, application_slug: str
) -> Optional[HttpResponse]:
"""Handle POST bindings"""
# Store these values now, because Django's login cycle won't preserve them.
if SESSION_KEY_SAML_REQUEST not in request.POST:
LOGGER.info("handle_saml_request: SAML payload missing")
return bad_request_message(
self.request, "The SAML request payload is missing."
)
self.request.session[SESSION_KEY_SAML_REQUEST] = request.POST[
SESSION_KEY_SAML_REQUEST
]
self.request.session[SESSION_KEY_RELAY_STATE] = request.POST.get(
SESSION_KEY_RELAY_STATE, ""
)
try:
self.provider.processor.can_handle(self.request)
params = self.provider.processor.generate_response()
self.request.session[SESSION_KEY_PARAMS] = params
except CannotHandleAssertion as exc:
LOGGER.info(exc)
return bad_request_message(self.request, str(exc))
return None
class SAMLSSOBindingInitView(SAMLSSOView):
"""SAML Handler for for IdP Initiated login flows"""
# pylint: disable=unused-argument
def get(
self, request: HttpRequest, application_slug: str
) -> Optional[HttpResponse]:
"""Create saml params from scratch"""
LOGGER.debug(
"handle_saml_no_request: No SAML Request, using IdP-initiated flow."
)
self.provider.processor.is_idp_initiated = True
self.provider.processor.init_deep_link(self.request)
params = self.provider.processor.generate_response()
self.request.session[SESSION_KEY_PARAMS] = params
# This View doesn't have a URL on purpose, as its called by the FlowExecutor
class SAMLFlowFinalView(StageView):
"""View used by FlowExecutor after all stages have passed. Logs the authorization,
and redirects to the SP (if REDIRECT is configured) or shows and auto-submit for
(if POST is configured)."""
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
application: Application = self.executor.plan.context[PLAN_CONTEXT_APPLICATION]
provider: SAMLProvider = get_object_or_404(
SAMLProvider, pk=application.provider_id
)
# Log Application Authorization
Event.new(
EventAction.AUTHORIZE_APPLICATION,
authorized_application=application,
flow=self.executor.plan.flow_pk,
).from_http(self.request)
self.request.session.pop(SESSION_KEY_SAML_REQUEST, None)
self.request.session.pop(SESSION_KEY_SAML_RESPONSE, None)
self.request.session.pop(SESSION_KEY_RELAY_STATE, None)
if SESSION_KEY_PARAMS not in self.request.session:
return self.executor.stage_invalid()
response: SAMLResponseParams = self.request.session.pop(SESSION_KEY_PARAMS)
if provider.sp_binding == SAMLBindings.POST:
return render(
self.request,
"providers/saml/autosubmit_form.html",
{
"url": response.acs_url,
"application": application,
"attrs": {
"ACSUrl": response.acs_url,
SESSION_KEY_SAML_RESPONSE: response.saml_response,
SESSION_KEY_RELAY_STATE: response.relay_state,
},
},
)
if provider.sp_binding == SAMLBindings.REDIRECT:
querystring = urlencode(
{
SESSION_KEY_SAML_RESPONSE: response.saml_response,
SESSION_KEY_RELAY_STATE: response.relay_state,
}
)
return redirect(f"{response.acs_url}?{querystring}")
return bad_request_message(request, "Invalid sp_binding specified")
class DescriptorDownloadView(View):
"""Replies with the XML Metadata IDSSODescriptor."""
@staticmethod
def get_metadata(request: HttpRequest, provider: SAMLProvider) -> str:
"""Return rendered XML Metadata"""
entity_id = provider.issuer
saml_sso_binding_post = request.build_absolute_uri(
reverse(
"passbook_providers_saml:sso-post",
kwargs={"application_slug": provider.application.slug},
)
)
saml_sso_binding_redirect = request.build_absolute_uri(
reverse(
"passbook_providers_saml:sso-redirect",
kwargs={"application_slug": provider.application.slug},
)
)
subject_format = provider.processor.subject_format
ctx = {
"saml_sso_binding_post": saml_sso_binding_post,
"saml_sso_binding_redirect": saml_sso_binding_redirect,
"entity_id": entity_id,
"subject_format": subject_format,
}
if provider.signing_kp:
ctx["cert_public_key"] = strip_pem_header(
provider.signing_kp.certificate_data.replace("\r", "")
).replace("\n", "")
return render_to_string("providers/saml/xml/metadata.xml", ctx)
def get(self, request: HttpRequest, application_slug: str) -> HttpResponse:
"""Replies with the XML Metadata IDSSODescriptor."""
application = get_object_or_404(Application, slug=application_slug)
provider: SAMLProvider = get_object_or_404(
SAMLProvider, pk=application.provider_id
)
try:
metadata = DescriptorDownloadView.get_metadata(request, provider)
except Provider.application.RelatedObjectDoesNotExist: # pylint: disable=no-member
return bad_request_message(
request, "Provider is not assigned to an application."
)
else:
response = HttpResponse(metadata, content_type="application/xml")
response[
"Content-Disposition"
] = f'attachment; filename="{provider.name}_passbook_meta.xml"'
return response