habsi/authentik
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

flows: complete migration to FlowExecutorView, fully use context

Browse Source
This commit is contained in:
Jens Langhammer
2020-05-08 16:10:27 +02:00
parent 114bb1b0bd
commit 2a85e5ae87
16 changed files with 180 additions and 416 deletions
Download Patch File Download Diff File Expand all files Collapse all files

288
passbook/flows/views.py
View File

@ -1,228 +1,24 @@
"""passbook multi-factor authentication engine"""
from typing import List, Optional, Tuple
from typing import Optional
from django.contrib.auth import login
from django.contrib.auth.mixins import UserPassesTestMixin
from django.http import HttpRequest, HttpResponse
from django.shortcuts import get_object_or_404, redirect, reverse
from django.utils.http import urlencode
from django.shortcuts import get_object_or_404, redirect
from django.views.generic import View
from structlog import get_logger
from passbook.core.models import Factor, User
from passbook.core.views.utils import PermissionDeniedView
from passbook.core.models import Factor
from passbook.flows.exceptions import FlowNonApplicableError
from passbook.flows.models import Flow
from passbook.flows.planner import FlowPlan, FlowPlanner
from passbook.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan, FlowPlanner
from passbook.lib.config import CONFIG
from passbook.lib.utils.reflection import class_to_path, path_to_class
from passbook.lib.utils.urls import is_url_absolute
from passbook.lib.utils.urls import is_url_absolute, redirect_with_qs
from passbook.lib.views import bad_request_message
from passbook.policies.engine import PolicyEngine
LOGGER = get_logger()
# Argument used to redirect user after login
NEXT_ARG_NAME = "next"
def _redirect_with_qs(view, get_query_set=None):
"""Wrapper to redirect whilst keeping GET Parameters"""
target = reverse(view)
if get_query_set:
target += "?" + urlencode(get_query_set.items())
return redirect(target)
class AuthenticationView(UserPassesTestMixin, View):
"""Wizard-like Multi-factor authenticator"""
SESSION_FACTOR = "passbook_factor"
SESSION_PENDING_FACTORS = "passbook_pending_factors"
SESSION_PENDING_USER = "passbook_pending_user"
SESSION_USER_BACKEND = "passbook_user_backend"
SESSION_IS_SSO_LOGIN = "passbook_sso_login"
pending_user: User
pending_factors: List[Tuple[str, str]] = []
_current_factor_class: Factor
current_factor: Factor
# Allow only not authenticated users to login
def test_func(self) -> bool:
return AuthenticationView.SESSION_PENDING_USER in self.request.session
def _check_config_domain(self) -> Optional[HttpResponse]:
"""Checks if current request's domain matches configured Domain, and
adds a warning if not."""
current_domain = self.request.get_host()
if ":" in current_domain:
current_domain, _ = current_domain.split(":")
config_domain = CONFIG.y("domain")
if current_domain != config_domain:
message = (
f"Current domain of '{current_domain}' doesn't "
f"match configured domain of '{config_domain}'."
)
LOGGER.warning(message)
return bad_request_message(self.request, message)
return None
def handle_no_permission(self) -> HttpResponse:
# Function from UserPassesTestMixin
if NEXT_ARG_NAME in self.request.GET:
return redirect(self.request.GET.get(NEXT_ARG_NAME))
if self.request.user.is_authenticated:
return _redirect_with_qs("passbook_core:overview", self.request.GET)
return _redirect_with_qs("passbook_core:auth-login", self.request.GET)
def get_pending_factors(self) -> List[Tuple[str, str]]:
"""Loading pending factors from Database or load from session variable"""
# Write pending factors to session
if AuthenticationView.SESSION_PENDING_FACTORS in self.request.session:
return self.request.session[AuthenticationView.SESSION_PENDING_FACTORS]
# Get an initial list of factors which are currently enabled
# and apply to the current user. We check policies here and block the request
_all_factors = (
Factor.objects.filter(enabled=True).order_by("order").select_subclasses()
)
pending_factors = []
for factor in _all_factors:
factor: Factor
LOGGER.debug(
"Checking if factor applies to user",
factor=factor,
user=self.pending_user,
)
policy_engine = PolicyEngine(
factor.policies.all(), self.pending_user, self.request
)
policy_engine.build()
if policy_engine.passing:
pending_factors.append((factor.uuid.hex, factor.type))
LOGGER.debug("Factor applies", factor=factor, user=self.pending_user)
return pending_factors
def dispatch(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
# Check if user passes test (i.e. SESSION_PENDING_USER is set)
user_test_result = self.get_test_func()()
if not user_test_result:
incorrect_domain_message = self._check_config_domain()
if incorrect_domain_message:
return incorrect_domain_message
return self.handle_no_permission()
# Extract pending user from session (only remember uid)
self.pending_user = get_object_or_404(
User, id=self.request.session[AuthenticationView.SESSION_PENDING_USER]
)
self.pending_factors = self.get_pending_factors()
# Read and instantiate factor from session
factor_uuid, factor_class = None, None
if AuthenticationView.SESSION_FACTOR not in request.session:
# Case when no factors apply to user, return error denied
if not self.pending_factors:
# Case when user logged in from SSO provider and no more factors apply
if AuthenticationView.SESSION_IS_SSO_LOGIN in request.session:
LOGGER.debug("User authenticated with SSO, logging in...")
return self._user_passed()
return self.user_invalid()
factor_uuid, factor_class = self.pending_factors[0]
else:
factor_uuid, factor_class = request.session[
AuthenticationView.SESSION_FACTOR
]
# Lookup current factor object
self.current_factor = (
Factor.objects.filter(uuid=factor_uuid).select_subclasses().first()
)
# Instantiate Next Factor and pass request
factor = path_to_class(factor_class)
self._current_factor_class = factor(self)
self._current_factor_class.pending_user = self.pending_user
self._current_factor_class.request = request
return super().dispatch(request, *args, **kwargs)
def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""pass get request to current factor"""
LOGGER.debug(
"Passing GET",
view_class=class_to_path(self._current_factor_class.__class__),
)
return self._current_factor_class.get(request, *args, **kwargs)
def post(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
"""pass post request to current factor"""
LOGGER.debug(
"Passing POST",
view_class=class_to_path(self._current_factor_class.__class__),
)
return self._current_factor_class.post(request, *args, **kwargs)
def user_ok(self) -> HttpResponse:
"""Redirect to next Factor"""
LOGGER.debug(
"Factor passed",
factor_class=class_to_path(self._current_factor_class.__class__),
)
# Remove passed factor from pending factors
current_factor_tuple = (
self.current_factor.uuid.hex,
class_to_path(self._current_factor_class.__class__),
)
if current_factor_tuple in self.pending_factors:
self.pending_factors.remove(current_factor_tuple)
next_factor = None
if self.pending_factors:
next_factor = self.pending_factors.pop()
# Save updated pening_factor list to session
self.request.session[
AuthenticationView.SESSION_PENDING_FACTORS
] = self.pending_factors
self.request.session[AuthenticationView.SESSION_FACTOR] = next_factor
LOGGER.debug("Rendering Factor", next_factor=next_factor)
return _redirect_with_qs("passbook_flows:auth-process", self.request.GET)
# User passed all factors
LOGGER.debug("User passed all factors, logging in", user=self.pending_user)
return self._user_passed()
def user_invalid(self) -> HttpResponse:
"""Show error message, user cannot login.
This should only be shown if user authenticated successfully, but is disabled/locked/etc"""
LOGGER.debug("User invalid")
self.cleanup()
return _redirect_with_qs("passbook_flows:auth-denied", self.request.GET)
def _user_passed(self) -> HttpResponse:
"""User Successfully passed all factors"""
backend = self.request.session[AuthenticationView.SESSION_USER_BACKEND]
login(self.request, self.pending_user, backend=backend)
LOGGER.debug("Logged in", user=self.pending_user)
# Cleanup
self.cleanup()
next_param = self.request.GET.get(NEXT_ARG_NAME, None)
if next_param and not is_url_absolute(next_param):
return redirect(next_param)
return _redirect_with_qs("passbook_core:overview")
def cleanup(self):
"""Remove temporary data from session"""
session_keys = [
self.SESSION_FACTOR,
self.SESSION_PENDING_FACTORS,
self.SESSION_PENDING_USER,
self.SESSION_USER_BACKEND,
]
for key in session_keys:
if key in self.request.session:
del self.request.session[key]
LOGGER.debug("Cleaned up sessions")
class FactorPermissionDeniedView(PermissionDeniedView):
"""User could not be authenticated"""
SESSION_KEY_PLAN = "passbook_flows_plan"
@ -240,6 +36,32 @@ class FlowExecutorView(View):
# TODO: Do we always need this?
self.flow = get_object_or_404(Flow, slug=flow_slug)
def _check_config_domain(self) -> Optional[HttpResponse]:
"""Checks if current request's domain matches configured Domain, and
adds a warning if not."""
current_domain = self.request.get_host()
if ":" in current_domain:
current_domain, _ = current_domain.split(":")
config_domain = CONFIG.y("domain")
if current_domain != config_domain:
message = (
f"Current domain of '{current_domain}' doesn't "
f"match configured domain of '{config_domain}'."
)
LOGGER.warning(message)
return bad_request_message(self.request, message)
return None
def handle_flow_non_applicable(self) -> HttpResponse:
"""When a flow is non-applicable check if user is on the correct domain"""
if NEXT_ARG_NAME in self.request.GET:
return redirect(self.request.GET.get(NEXT_ARG_NAME))
incorrect_domain_message = self._check_config_domain()
if incorrect_domain_message:
return incorrect_domain_message
# TODO: Add message
return redirect("passbook_core:index")
def dispatch(self, request: HttpRequest, flow_slug: str) -> HttpResponse:
# Early check if theres an active Plan for the current session
if SESSION_KEY_PLAN not in self.request.session:
@ -250,7 +72,7 @@ class FlowExecutorView(View):
self.plan = self._initiate_plan()
except FlowNonApplicableError as exc:
LOGGER.warning("Flow not applicable to current user", exc=exc)
return redirect("passbook_core:index")
return self.handle_flow_non_applicable()
else:
LOGGER.debug("Continuing existing plan", flow_slug=flow_slug)
self.plan = self.request.session[SESSION_KEY_PLAN]
@ -260,7 +82,6 @@ class FlowExecutorView(View):
LOGGER.debug("Current factor", current_factor=self.current_factor)
factor_cls = path_to_class(self.current_factor.type)
self.current_factor_view = factor_cls(self)
# self.current_factor_view.pending_user = self.pending_user
self.current_factor_view.request = request
return super().dispatch(request)
@ -284,3 +105,48 @@ class FlowExecutorView(View):
plan = planner.plan(self.request)
self.request.session[SESSION_KEY_PLAN] = plan
return plan
def _flow_done(self) -> HttpResponse:
"""User Successfully passed all factors"""
backend = self.plan.context[PLAN_CONTEXT_PENDING_USER].backend
login(
self.request, self.plan.context[PLAN_CONTEXT_PENDING_USER], backend=backend
)
LOGGER.debug("Logged in", user=self.plan.context[PLAN_CONTEXT_PENDING_USER])
self.cancel()
next_param = self.request.GET.get(NEXT_ARG_NAME, None)
if next_param and not is_url_absolute(next_param):
return redirect(next_param)
return redirect_with_qs("passbook_core:overview")
def factor_ok(self) -> HttpResponse:
"""Callback called by factors upon successful completion.
Persists updated plan and context to session."""
LOGGER.debug(
"Factor ok", factor_class=class_to_path(self.current_factor_view.__class__),
)
self.request.session[SESSION_KEY_PLAN] = self.plan
if self.plan.factors:
LOGGER.debug(
"Continuing with next factor", reamining=len(self.plan.factors)
)
return redirect_with_qs(
"passbook_flows:flow-executor", self.request.GET, **self.kwargs
)
# User passed all factors
LOGGER.debug(
"User passed all factors", user=self.plan.context[PLAN_CONTEXT_PENDING_USER]
)
return self._flow_done()
def factor_invalid(self) -> HttpResponse:
"""Callback used factor when data is correct but a policy denies access
or the user account is disabled."""
LOGGER.debug("User invalid")
self.cancel()
return redirect_with_qs("passbook_flows:auth-denied", self.request.GET)
def cancel(self) -> HttpResponse:
"""Cancel current execution and return a redirect"""
del self.request.session[SESSION_KEY_PLAN]
return redirect_with_qs("passbook_flows:auth-denied", self.request.GET)