flows: replace passbook_flows:denied with AccessDenied Reeponse

passbook/flows/tests/test_views.py

@@ -1,16 +1,19 @@
 """flow views tests"""
 from unittest.mock import MagicMock, PropertyMock, patch
 
+from django.http import HttpRequest, HttpResponse
 from django.shortcuts import reverse
 from django.test import Client, TestCase
 from django.utils.encoding import force_str
 
+from passbook.flows.exceptions import EmptyFlowException, FlowNonApplicableException
 from passbook.flows.markers import ReevaluateMarker, StageMarker
 from passbook.flows.models import Flow, FlowDesignation, FlowStageBinding
 from passbook.flows.planner import FlowPlan
 from passbook.flows.views import NEXT_ARG_NAME, SESSION_KEY_PLAN
 from passbook.lib.config import CONFIG
 from passbook.policies.dummy.models import DummyPolicy
+from passbook.policies.http import AccessDeniedResponse
 from passbook.policies.models import PolicyBinding
 from passbook.policies.types import PolicyResult
 from passbook.stages.dummy.models import DummyStage
@@ -19,6 +22,15 @@ POLICY_RETURN_FALSE = PropertyMock(return_value=PolicyResult(False))
 POLICY_RETURN_TRUE = MagicMock(return_value=PolicyResult(True))
 
 
+def to_stage_response(request: HttpRequest, source: HttpResponse):
+    """Mock for to_stage_response that returns the original response, so we can check
+    inheritance and member attributes"""
+    return source
+
+
+TO_STAGE_RESPONSE_MOCK = MagicMock(side_effect=to_stage_response)
+
+
 class TestFlowExecutor(TestCase):
     """Test views logic"""
 
@@ -50,6 +62,9 @@ class TestFlowExecutor(TestCase):
             self.assertEqual(response.status_code, 200)
             self.assertEqual(cancel_mock.call_count, 2)
 
+    @patch(
+        "passbook.flows.views.to_stage_response", TO_STAGE_RESPONSE_MOCK,
+    )
     @patch(
         "passbook.policies.engine.PolicyEngine.result", POLICY_RETURN_FALSE,
     )
@@ -66,11 +81,12 @@ class TestFlowExecutor(TestCase):
             reverse("passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}),
         )
         self.assertEqual(response.status_code, 200)
-        self.assertJSONEqual(
+        self.assertIsInstance(response, AccessDeniedResponse)
-            force_str(response.content),
+        self.assertInHTML(FlowNonApplicableException.__doc__, response.rendered_content)
-            {"type": "redirect", "to": reverse("passbook_flows:denied")},
-        )
 
+    @patch(
+        "passbook.flows.views.to_stage_response", TO_STAGE_RESPONSE_MOCK,
+    )
     def test_invalid_empty_flow(self):
         """Tests that an empty flow returns the correct error message"""
         flow = Flow.objects.create(
@@ -84,10 +100,8 @@ class TestFlowExecutor(TestCase):
             reverse("passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug}),
         )
         self.assertEqual(response.status_code, 200)
-        self.assertJSONEqual(
+        self.assertIsInstance(response, AccessDeniedResponse)
-            force_str(response.content),
+        self.assertInHTML(EmptyFlowException.__doc__, response.rendered_content)
-            {"type": "redirect", "to": reverse("passbook_flows:denied")},
-        )
 
     def test_invalid_flow_redirect(self):
         """Tests that an invalid flow still redirects"""
@@ -101,8 +115,10 @@ class TestFlowExecutor(TestCase):
         dest = "/unique-string"
         url = reverse("passbook_flows:flow-executor", kwargs={"flow_slug": flow.slug})
         response = self.client.get(url + f"?{NEXT_ARG_NAME}={dest}")
-        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.status_code, 200)
-        self.assertEqual(response.url, dest)
+        self.assertJSONEqual(
+            force_str(response.content), {"type": "redirect", "to": dest},
+        )
 
     def test_multi_stage_flow(self):
         """Test a full flow with multiple stages"""

passbook/flows/urls.py

@@ -6,12 +6,10 @@ from passbook.flows.views import (
     CancelView,
     FlowExecutorShellView,
     FlowExecutorView,
-    FlowPermissionDeniedView,
     ToDefaultFlow,
 )
 
 urlpatterns = [
-    path("-/denied/", FlowPermissionDeniedView.as_view(), name="denied"),
     path(
         "-/default/authentication/",
         ToDefaultFlow.as_view(designation=FlowDesignation.AUTHENTICATION),

passbook/flows/views.py

@@ -9,10 +9,9 @@ from django.http import (
     HttpResponseRedirect,
     JsonResponse,
 )
-from django.shortcuts import get_object_or_404, redirect, render, reverse
+from django.shortcuts import get_object_or_404, redirect, reverse
 from django.template.response import TemplateResponse
 from django.utils.decorators import method_decorator
-from django.utils.translation import gettext as _
 from django.views.decorators.clickjacking import xframe_options_sameorigin
 from django.views.generic import TemplateView, View
 from structlog import get_logger
@@ -24,6 +23,7 @@ from passbook.flows.models import Flow, FlowDesignation, Stage
 from passbook.flows.planner import FlowPlan, FlowPlanner
 from passbook.lib.utils.reflection import class_to_path
 from passbook.lib.utils.urls import is_url_absolute, redirect_with_qs
+from passbook.policies.http import AccessDeniedResponse
 
 LOGGER = get_logger()
 # Argument used to redirect user after login
@@ -31,8 +31,6 @@ NEXT_ARG_NAME = "next"
 SESSION_KEY_PLAN = "passbook_flows_plan"
 SESSION_KEY_APPLICATION_PRE = "passbook_flows_application_pre"
 SESSION_KEY_GET = "passbook_flows_get"
-SESSION_KEY_DENIED_ERROR = "passbook_flows_denied_error"
-SESSION_KEY_DENIED_POLICY_RESULT = "passbook_flows_denied_policy_result"
 
 
 @method_decorator(xframe_options_sameorigin, name="dispatch")
@@ -56,9 +54,7 @@ class FlowExecutorView(View):
                 LOGGER.debug("f(exec): Redirecting to next on fail")
                 return redirect(self.request.GET.get(NEXT_ARG_NAME))
         message = exc.__doc__ if exc.__doc__ else str(exc)
-        return to_stage_response(
+        return self.stage_invalid(error_message=message)
-            self.request, self.stage_invalid(error_message=message)
-        )
 
     def dispatch(self, request: HttpRequest, flow_slug: str) -> HttpResponse:
         # Early check if theres an active Plan for the current session
@@ -83,10 +79,10 @@ class FlowExecutorView(View):
                 self.plan = self._initiate_plan()
             except FlowNonApplicableException as exc:
                 LOGGER.warning("f(exec): Flow not applicable to current user", exc=exc)
-                return self.handle_invalid_flow(exc)
+                return to_stage_response(self.request, self.handle_invalid_flow(exc))
             except EmptyFlowException as exc:
                 LOGGER.warning("f(exec): Flow is empty", exc=exc)
-                return self.handle_invalid_flow(exc)
+                return to_stage_response(self.request, self.handle_invalid_flow(exc))
         # We don't save the Plan after getting the next stage
         # as it hasn't been successfully passed yet
         next_stage = self.plan.next()
@@ -119,14 +115,7 @@ class FlowExecutorView(View):
             return to_stage_response(request, stage_response)
         except Exception as exc:  # pylint: disable=broad-except
             LOGGER.exception(exc)
-            return to_stage_response(
+            return to_stage_response(request, FlowErrorResponse(request, exc))
-                request,
-                render(
-                    request,
-                    "flows/error.html",
-                    {"error": exc, "tb": "".join(format_tb(exc.__traceback__))},
-                ),
-            )
 
     def post(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
         """pass post request to current stage"""
@@ -141,14 +130,7 @@ class FlowExecutorView(View):
             return to_stage_response(request, stage_response)
         except Exception as exc:  # pylint: disable=broad-except
             LOGGER.exception(exc)
-            return to_stage_response(
+            return to_stage_response(request, FlowErrorResponse(request, exc))
-                request,
-                render(
-                    request,
-                    "flows/error.html",
-                    {"error": exc, "tb": "".join(format_tb(exc.__traceback__))},
-                ),
-            )
 
     def _initiate_plan(self) -> FlowPlan:
         planner = FlowPlanner(self.flow)
@@ -205,12 +187,9 @@ class FlowExecutorView(View):
         is a superuser."""
         LOGGER.debug("f(exec): Stage invalid", flow_slug=self.flow.slug)
         self.cancel()
-        if self.request.user and self.request.user.is_authenticated:
+        response = AccessDeniedResponse(self.request)
-            if self.request.user.is_superuser or self.request.user.attributes.get(
+        response.error_message = error_message
-                PASSBOOK_USER_DEBUG, False
+        return response
-            ):
-                self.request.session[SESSION_KEY_DENIED_ERROR] = error_message
-        return redirect_with_qs("passbook_flows:denied", self.request.GET)
 
     def cancel(self):
         """Cancel current execution and return a redirect"""
@@ -224,21 +203,30 @@ class FlowExecutorView(View):
                 del self.request.session[key]
 
 
-class FlowPermissionDeniedView(TemplateView):
+class FlowErrorResponse(TemplateResponse):
-    """User could not be authenticated"""
+    """Response class when an unhandled error occurs during a stage. Normal users
+    are shown an error message, superusers are shown a full stacktrace."""
 
-    template_name = "flows/denied.html"
+    error: Exception
-    title = _("Permission denied.")
 
-    def get_context_data(self, **kwargs):
+    def __init__(self, request: HttpRequest, error: Exception) -> None:
-        kwargs["title"] = self.title
+        # For some reason pyright complains about keyword argument usage here
-        if SESSION_KEY_DENIED_ERROR in self.request.session:
+        # pyright: reportGeneralTypeIssues=false
-            kwargs["error"] = self.request.session[SESSION_KEY_DENIED_ERROR]
+        super().__init__(request=request, template="flows/error.html")
-        if SESSION_KEY_DENIED_POLICY_RESULT in self.request.session:
+        self.error = error
-            kwargs["policy_result"] = self.request.session[
+
-                SESSION_KEY_DENIED_POLICY_RESULT
+    def resolve_context(
-            ]
+        self, context: Optional[Dict[str, Any]]
-        return super().get_context_data(**kwargs)
+    ) -> Optional[Dict[str, Any]]:
+        if not context:
+            context = {}
+        context["error"] = self.error
+        if self._request.user and self._request.user.is_authenticated:
+            if self._request.user.is_superuser or self._request.user.attributes.get(
+                PASSBOOK_USER_DEBUG, False
+            ):
+                context["tb"] = "".join(format_tb(self.error.__traceback__))
+        return context
 
 
 class FlowExecutorShellView(TemplateView):

passbook/stages/invitation/tests.py

@@ -10,7 +10,9 @@ from passbook.core.models import User
 from passbook.flows.markers import StageMarker
 from passbook.flows.models import Flow, FlowDesignation, FlowStageBinding
 from passbook.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
+from passbook.flows.tests.test_views import TO_STAGE_RESPONSE_MOCK
 from passbook.flows.views import SESSION_KEY_PLAN
+from passbook.policies.http import AccessDeniedResponse
 from passbook.stages.invitation.forms import InvitationStageForm
 from passbook.stages.invitation.models import Invitation, InvitationStage
 from passbook.stages.invitation.stage import INVITATION_TOKEN_KEY, PLAN_CONTEXT_PROMPT
@@ -38,6 +40,9 @@ class TestUserLoginStage(TestCase):
         data = {"name": "test"}
         self.assertEqual(InvitationStageForm(data).is_valid(), True)
 
+    @patch(
+        "passbook.flows.views.to_stage_response", TO_STAGE_RESPONSE_MOCK,
+    )
     def test_without_invitation_fail(self):
         """Test without any invitation, continue_flow_without_invitation not set."""
         plan = FlowPlan(
@@ -56,12 +61,8 @@ class TestUserLoginStage(TestCase):
                 "passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
             )
         )
-
         self.assertEqual(response.status_code, 200)
-        self.assertJSONEqual(
+        self.assertIsInstance(response, AccessDeniedResponse)
-            force_str(response.content),
-            {"type": "redirect", "to": reverse("passbook_flows:denied")},
-        )
 
     def test_without_invitation_continue(self):
         """Test without any invitation, continue_flow_without_invitation is set."""

passbook/stages/password/tests.py

@@ -12,7 +12,9 @@ from passbook.core.models import User
 from passbook.flows.markers import StageMarker
 from passbook.flows.models import Flow, FlowDesignation, FlowStageBinding
 from passbook.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
+from passbook.flows.tests.test_views import TO_STAGE_RESPONSE_MOCK
 from passbook.flows.views import SESSION_KEY_PLAN
+from passbook.policies.http import AccessDeniedResponse
 from passbook.stages.password.models import PasswordStage
 
 MOCK_BACKEND_AUTHENTICATE = MagicMock(side_effect=PermissionDenied("test"))
@@ -42,6 +44,9 @@ class TestPasswordStage(TestCase):
         )
         FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)
 
+    @patch(
+        "passbook.flows.views.to_stage_response", TO_STAGE_RESPONSE_MOCK,
+    )
     def test_without_user(self):
         """Test without user"""
         plan = FlowPlan(
@@ -60,10 +65,7 @@ class TestPasswordStage(TestCase):
         )
 
         self.assertEqual(response.status_code, 200)
-        self.assertJSONEqual(
+        self.assertIsInstance(response, AccessDeniedResponse)
-            force_str(response.content),
-            {"type": "redirect", "to": reverse("passbook_flows:denied")},
-        )
 
     def test_recovery_flow_link(self):
         """Test link to the default recovery flow"""
@@ -129,6 +131,9 @@ class TestPasswordStage(TestCase):
         )
         self.assertEqual(response.status_code, 200)
 
+    @patch(
+        "passbook.flows.views.to_stage_response", TO_STAGE_RESPONSE_MOCK,
+    )
     @patch(
         "django.contrib.auth.backends.ModelBackend.authenticate",
         MOCK_BACKEND_AUTHENTICATE,
@@ -153,7 +158,4 @@ class TestPasswordStage(TestCase):
         )
 
         self.assertEqual(response.status_code, 200)
-        self.assertJSONEqual(
+        self.assertIsInstance(response, AccessDeniedResponse)
-            force_str(response.content),
-            {"type": "redirect", "to": reverse("passbook_flows:denied")},
-        )

passbook/stages/user_delete/tests.py

@@ -1,4 +1,6 @@
 """delete tests"""
+from unittest.mock import patch
+
 from django.shortcuts import reverse
 from django.test import Client, TestCase
 from django.utils.encoding import force_str
@@ -7,7 +9,9 @@ from passbook.core.models import User
 from passbook.flows.markers import StageMarker
 from passbook.flows.models import Flow, FlowDesignation, FlowStageBinding
 from passbook.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
+from passbook.flows.tests.test_views import TO_STAGE_RESPONSE_MOCK
 from passbook.flows.views import SESSION_KEY_PLAN
+from passbook.policies.http import AccessDeniedResponse
 from passbook.stages.user_delete.models import UserDeleteStage
 
 
@@ -28,6 +32,9 @@ class TestUserDeleteStage(TestCase):
         self.stage = UserDeleteStage.objects.create(name="delete")
         FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)
 
+    @patch(
+        "passbook.flows.views.to_stage_response", TO_STAGE_RESPONSE_MOCK,
+    )
     def test_no_user(self):
         """Test without user set"""
         plan = FlowPlan(
@@ -43,10 +50,7 @@ class TestUserDeleteStage(TestCase):
             )
         )
         self.assertEqual(response.status_code, 200)
-        self.assertJSONEqual(
+        self.assertIsInstance(response, AccessDeniedResponse)
-            force_str(response.content),
-            {"type": "redirect", "to": reverse("passbook_flows:denied")},
-        )
 
     def test_user_delete_get(self):
         """Test Form render"""

passbook/stages/user_login/tests.py

@@ -1,4 +1,6 @@
 """login tests"""
+from unittest.mock import patch
+
 from django.shortcuts import reverse
 from django.test import Client, TestCase
 from django.utils.encoding import force_str
@@ -7,7 +9,9 @@ from passbook.core.models import User
 from passbook.flows.markers import StageMarker
 from passbook.flows.models import Flow, FlowDesignation, FlowStageBinding
 from passbook.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
+from passbook.flows.tests.test_views import TO_STAGE_RESPONSE_MOCK
 from passbook.flows.views import SESSION_KEY_PLAN
+from passbook.policies.http import AccessDeniedResponse
 from passbook.stages.password.stage import PLAN_CONTEXT_AUTHENTICATION_BACKEND
 from passbook.stages.user_login.forms import UserLoginStageForm
 from passbook.stages.user_login.models import UserLoginStage
@@ -54,6 +58,9 @@ class TestUserLoginStage(TestCase):
             {"type": "redirect", "to": reverse("passbook_core:overview")},
         )
 
+    @patch(
+        "passbook.flows.views.to_stage_response", TO_STAGE_RESPONSE_MOCK,
+    )
     def test_without_user(self):
         """Test a plan without any pending user, resulting in a denied"""
         plan = FlowPlan(
@@ -70,11 +77,11 @@ class TestUserLoginStage(TestCase):
         )
 
         self.assertEqual(response.status_code, 200)
-        self.assertJSONEqual(
+        self.assertIsInstance(response, AccessDeniedResponse)
-            force_str(response.content),
-            {"type": "redirect", "to": reverse("passbook_flows:denied")},
-        )
 
+    @patch(
+        "passbook.flows.views.to_stage_response", TO_STAGE_RESPONSE_MOCK,
+    )
     def test_without_backend(self):
         """Test a plan with pending user, without backend, resulting in a denied"""
         plan = FlowPlan(
@@ -92,10 +99,7 @@ class TestUserLoginStage(TestCase):
         )
 
         self.assertEqual(response.status_code, 200)
-        self.assertJSONEqual(
+        self.assertIsInstance(response, AccessDeniedResponse)
-            force_str(response.content),
-            {"type": "redirect", "to": reverse("passbook_flows:denied")},
-        )
 
     def test_form(self):
         """Test Form"""

passbook/stages/user_write/tests.py

@@ -1,6 +1,7 @@
 """write tests"""
 import string
 from random import SystemRandom
+from unittest.mock import patch
 
 from django.shortcuts import reverse
 from django.test import Client, TestCase
@@ -10,7 +11,9 @@ from passbook.core.models import User
 from passbook.flows.markers import StageMarker
 from passbook.flows.models import Flow, FlowDesignation, FlowStageBinding
 from passbook.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
+from passbook.flows.tests.test_views import TO_STAGE_RESPONSE_MOCK
 from passbook.flows.views import SESSION_KEY_PLAN
+from passbook.policies.http import AccessDeniedResponse
 from passbook.stages.prompt.stage import PLAN_CONTEXT_PROMPT
 from passbook.stages.user_write.forms import UserWriteStageForm
 from passbook.stages.user_write.models import UserWriteStage
@@ -107,6 +110,9 @@ class TestUserWriteStage(TestCase):
         self.assertTrue(user_qs.first().check_password(new_password))
         self.assertEqual(user_qs.first().attributes["some-custom-attribute"], "test")
 
+    @patch(
+        "passbook.flows.views.to_stage_response", TO_STAGE_RESPONSE_MOCK,
+    )
     def test_without_data(self):
         """Test without data results in error"""
         plan = FlowPlan(
@@ -123,10 +129,7 @@ class TestUserWriteStage(TestCase):
         )
 
         self.assertEqual(response.status_code, 200)
-        self.assertJSONEqual(
+        self.assertIsInstance(response, AccessDeniedResponse)
-            force_str(response.content),
-            {"type": "redirect", "to": reverse("passbook_flows:denied")},
-        )
 
     def test_form(self):
         """Test Form"""