We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Email stage links incorrectly encoded
**Describe the bug**
Links sent in emails are getting doubly encoded.

For example:
`http://localhost:9000/if/flow/signup/?query=&amp;flow_token=W2kRqeEmVIdh0h...`

The issue is the `&` between querystring parameters.

**To Reproduce**
Steps to reproduce the behavior:

I'm seeing these on all signup flows and reset password flows where an email is triggered. 

**Expected behavior**
I'd expect the above to instead be 
`http://localhost:9000/if/flow/signup/?query=&flow_token=W2kRqeEmVIdh0h...`

**Version and Deployment (please complete the following information):**

-   authentik version: 2023.10.2
-   Deployment: only tested so far in docker-compose



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `authentik/stages/email/stage.py`
Content:
```
1 """authentik multi-stage authentication engine"""
2 from datetime import timedelta
3 
4 from django.contrib import messages
5 from django.http import HttpRequest, HttpResponse
6 from django.http.request import QueryDict
7 from django.urls import reverse
8 from django.utils.text import slugify
9 from django.utils.timezone import now
10 from django.utils.translation import gettext as _
11 from rest_framework.fields import CharField
12 from rest_framework.serializers import ValidationError
13 
14 from authentik.flows.challenge import Challenge, ChallengeResponse, ChallengeTypes
15 from authentik.flows.models import FlowDesignation, FlowToken
16 from authentik.flows.planner import PLAN_CONTEXT_IS_RESTORED, PLAN_CONTEXT_PENDING_USER
17 from authentik.flows.stage import ChallengeStageView
18 from authentik.flows.views.executor import QS_KEY_TOKEN, QS_QUERY
19 from authentik.stages.email.models import EmailStage
20 from authentik.stages.email.tasks import send_mails
21 from authentik.stages.email.utils import TemplateEmailMessage
22 
23 PLAN_CONTEXT_EMAIL_SENT = "email_sent"
24 PLAN_CONTEXT_EMAIL_OVERRIDE = "email"
25 
26 
27 class EmailChallenge(Challenge):
28     """Email challenge"""
29 
30     component = CharField(default="ak-stage-email")
31 
32 
33 class EmailChallengeResponse(ChallengeResponse):
34     """Email challenge resposen. No fields. This challenge is
35     always declared invalid to give the user a chance to retry"""
36 
37     component = CharField(default="ak-stage-email")
38 
39     def validate(self, attrs):
40         raise ValidationError(detail="email-sent", code="email-sent")
41 
42 
43 class EmailStageView(ChallengeStageView):
44     """Email stage which sends Email for verification"""
45 
46     response_class = EmailChallengeResponse
47 
48     def get_full_url(self, **kwargs) -> str:
49         """Get full URL to be used in template"""
50         base_url = reverse(
51             "authentik_core:if-flow",
52             kwargs={"flow_slug": self.executor.flow.slug},
53         )
54         # Parse query string from current URL (full query string)
55         query_params = QueryDict(self.request.META.get("QUERY_STRING", ""), mutable=True)
56         query_params.pop(QS_KEY_TOKEN, None)
57 
58         # Check for nested query string used by flow executor, and remove any
59         # kind of flow token from that
60         if QS_QUERY in query_params:
61             inner_query_params = QueryDict(query_params.get(QS_QUERY), mutable=True)
62             inner_query_params.pop(QS_KEY_TOKEN, None)
63             query_params[QS_QUERY] = inner_query_params.urlencode()
64 
65         query_params.update(kwargs)
66         full_url = base_url
67         if len(query_params) > 0:
68             full_url = f"{full_url}?{query_params.urlencode()}"
69         return self.request.build_absolute_uri(full_url)
70 
71     def get_token(self) -> FlowToken:
72         """Get token"""
73         pending_user = self.get_pending_user()
74         current_stage: EmailStage = self.executor.current_stage
75         valid_delta = timedelta(
76             minutes=current_stage.token_expiry + 1
77         )  # + 1 because django timesince always rounds down
78         identifier = slugify(f"ak-email-stage-{current_stage.name}-{pending_user}")
79         # Don't check for validity here, we only care if the token exists
80         tokens = FlowToken.objects.filter(identifier=identifier)
81         if not tokens.exists():
82             return FlowToken.objects.create(
83                 expires=now() + valid_delta,
84                 user=pending_user,
85                 identifier=identifier,
86                 flow=self.executor.flow,
87                 _plan=FlowToken.pickle(self.executor.plan),
88             )
89         token = tokens.first()
90         # Check if token is expired and rotate key if so
91         if token.is_expired:
92             token.expire_action()
93         return token
94 
95     def send_email(self):
96         """Helper function that sends the actual email. Implies that you've
97         already checked that there is a pending user."""
98         pending_user = self.get_pending_user()
99         if not pending_user.pk and self.executor.flow.designation == FlowDesignation.RECOVERY:
100             # Pending user does not have a primary key, and we're in a recovery flow,
101             # which means the user entered an invalid identifier, so we pretend to send the
102             # email, to not disclose if the user exists
103             return
104         email = self.executor.plan.context.get(PLAN_CONTEXT_EMAIL_OVERRIDE, None)
105         if not email:
106             email = pending_user.email
107         current_stage: EmailStage = self.executor.current_stage
108         token = self.get_token()
109         # Send mail to user
110         message = TemplateEmailMessage(
111             subject=_(current_stage.subject),
112             to=[email],
113             language=pending_user.locale(self.request),
114             template_name=current_stage.template,
115             template_context={
116                 "url": self.get_full_url(**{QS_KEY_TOKEN: token.key}),
117                 "user": pending_user,
118                 "expires": token.expires,
119             },
120         )
121         send_mails(current_stage, message)
122 
123     def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
124         # Check if the user came back from the email link to verify
125         restore_token: FlowToken = self.executor.plan.context.get(PLAN_CONTEXT_IS_RESTORED, None)
126         user = self.get_pending_user()
127         if restore_token:
128             if restore_token.user != user:
129                 self.logger.warning("Flow token for non-matching user, denying request")
130                 return self.executor.stage_invalid()
131             messages.success(request, _("Successfully verified Email."))
132             if self.executor.current_stage.activate_user_on_success:
133                 user.is_active = True
134                 user.save()
135             return self.executor.stage_ok()
136         if PLAN_CONTEXT_PENDING_USER not in self.executor.plan.context:
137             self.logger.debug("No pending user")
138             messages.error(self.request, _("No pending user."))
139             return self.executor.stage_invalid()
140         # Check if we've already sent the initial e-mail
141         if PLAN_CONTEXT_EMAIL_SENT not in self.executor.plan.context:
142             self.send_email()
143             self.executor.plan.context[PLAN_CONTEXT_EMAIL_SENT] = True
144         return super().get(request, *args, **kwargs)
145 
146     def get_challenge(self) -> Challenge:
147         challenge = EmailChallenge(
148             data={
149                 "type": ChallengeTypes.NATIVE.value,
150                 "title": _("Email sent."),
151             }
152         )
153         return challenge
154 
155     def challenge_valid(self, response: ChallengeResponse) -> HttpResponse:
156         return super().challenge_invalid(response)
157 
158     def challenge_invalid(self, response: ChallengeResponse) -> HttpResponse:
159         if PLAN_CONTEXT_PENDING_USER not in self.executor.plan.context:
160             messages.error(self.request, _("No pending user."))
161             return super().challenge_invalid(response)
162         self.send_email()
163         messages.success(self.request, _("Email Successfully sent."))
164         # We can't call stage_ok yet, as we're still waiting
165         # for the user to click the link in the email
166         return super().challenge_invalid(response)
167
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/authentik/stages/email/stage.py b/authentik/stages/email/stage.py
--- a/authentik/stages/email/stage.py
+++ b/authentik/stages/email/stage.py
@@ -52,17 +52,13 @@
             kwargs={"flow_slug": self.executor.flow.slug},
         )
         # Parse query string from current URL (full query string)
-        query_params = QueryDict(self.request.META.get("QUERY_STRING", ""), mutable=True)
+        # this view is only run within a flow executor, where we need to get the query string
+        # from the query= parameter (double encoded); but for the redirect
+        # we need to expand it since it'll go through the flow interface
+        query_params = QueryDict(self.request.GET.get(QS_QUERY), mutable=True)
         query_params.pop(QS_KEY_TOKEN, None)
-
-        # Check for nested query string used by flow executor, and remove any
-        # kind of flow token from that
-        if QS_QUERY in query_params:
-            inner_query_params = QueryDict(query_params.get(QS_QUERY), mutable=True)
-            inner_query_params.pop(QS_KEY_TOKEN, None)
-            query_params[QS_QUERY] = inner_query_params.urlencode()
-
         query_params.update(kwargs)
+        print(query_params)
         full_url = base_url
         if len(query_params) > 0:
             full_url = f"{full_url}?{query_params.urlencode()}"

{"golden_diff": "diff --git a/authentik/stages/email/stage.py b/authentik/stages/email/stage.py\n--- a/authentik/stages/email/stage.py\n+++ b/authentik/stages/email/stage.py\n@@ -52,17 +52,13 @@\n             kwargs={\"flow_slug\": self.executor.flow.slug},\n         )\n         # Parse query string from current URL (full query string)\n-        query_params = QueryDict(self.request.META.get(\"QUERY_STRING\", \"\"), mutable=True)\n+        # this view is only run within a flow executor, where we need to get the query string\n+        # from the query= parameter (double encoded); but for the redirect\n+        # we need to expand it since it'll go through the flow interface\n+        query_params = QueryDict(self.request.GET.get(QS_QUERY), mutable=True)\n         query_params.pop(QS_KEY_TOKEN, None)\n-\n-        # Check for nested query string used by flow executor, and remove any\n-        # kind of flow token from that\n-        if QS_QUERY in query_params:\n-            inner_query_params = QueryDict(query_params.get(QS_QUERY), mutable=True)\n-            inner_query_params.pop(QS_KEY_TOKEN, None)\n-            query_params[QS_QUERY] = inner_query_params.urlencode()\n-\n         query_params.update(kwargs)\n+        print(query_params)\n         full_url = base_url\n         if len(query_params) > 0:\n             full_url = f\"{full_url}?{query_params.urlencode()}\"\n", "issue": "Email stage links incorrectly encoded\n**Describe the bug**\r\nLinks sent in emails are getting doubly encoded.\r\n\r\nFor example:\r\n`http://localhost:9000/if/flow/signup/?query=&amp;flow_token=W2kRqeEmVIdh0h...`\r\n\r\nThe issue is the `&` between querystring parameters.\r\n\r\n**To Reproduce**\r\nSteps to reproduce the behavior:\r\n\r\nI'm seeing these on all signup flows and reset password flows where an email is triggered. \r\n\r\n**Expected behavior**\r\nI'd expect the above to instead be \r\n`http://localhost:9000/if/flow/signup/?query=&flow_token=W2kRqeEmVIdh0h...`\r\n\r\n**Version and Deployment (please complete the following information):**\r\n\r\n-   authentik version: 2023.10.2\r\n-   Deployment: only tested so far in docker-compose\r\n\r\n\n", "before_files": [{"content": "\"\"\"authentik multi-stage authentication engine\"\"\"\nfrom datetime import timedelta\n\nfrom django.contrib import messages\nfrom django.http import HttpRequest, HttpResponse\nfrom django.http.request import QueryDict\nfrom django.urls import reverse\nfrom django.utils.text import slugify\nfrom django.utils.timezone import now\nfrom django.utils.translation import gettext as _\nfrom rest_framework.fields import CharField\nfrom rest_framework.serializers import ValidationError\n\nfrom authentik.flows.challenge import Challenge, ChallengeResponse, ChallengeTypes\nfrom authentik.flows.models import FlowDesignation, FlowToken\nfrom authentik.flows.planner import PLAN_CONTEXT_IS_RESTORED, PLAN_CONTEXT_PENDING_USER\nfrom authentik.flows.stage import ChallengeStageView\nfrom authentik.flows.views.executor import QS_KEY_TOKEN, QS_QUERY\nfrom authentik.stages.email.models import EmailStage\nfrom authentik.stages.email.tasks import send_mails\nfrom authentik.stages.email.utils import TemplateEmailMessage\n\nPLAN_CONTEXT_EMAIL_SENT = \"email_sent\"\nPLAN_CONTEXT_EMAIL_OVERRIDE = \"email\"\n\n\nclass EmailChallenge(Challenge):\n    \"\"\"Email challenge\"\"\"\n\n    component = CharField(default=\"ak-stage-email\")\n\n\nclass EmailChallengeResponse(ChallengeResponse):\n    \"\"\"Email challenge resposen. No fields. This challenge is\n    always declared invalid to give the user a chance to retry\"\"\"\n\n    component = CharField(default=\"ak-stage-email\")\n\n    def validate(self, attrs):\n        raise ValidationError(detail=\"email-sent\", code=\"email-sent\")\n\n\nclass EmailStageView(ChallengeStageView):\n    \"\"\"Email stage which sends Email for verification\"\"\"\n\n    response_class = EmailChallengeResponse\n\n    def get_full_url(self, **kwargs) -> str:\n        \"\"\"Get full URL to be used in template\"\"\"\n        base_url = reverse(\n            \"authentik_core:if-flow\",\n            kwargs={\"flow_slug\": self.executor.flow.slug},\n        )\n        # Parse query string from current URL (full query string)\n        query_params = QueryDict(self.request.META.get(\"QUERY_STRING\", \"\"), mutable=True)\n        query_params.pop(QS_KEY_TOKEN, None)\n\n        # Check for nested query string used by flow executor, and remove any\n        # kind of flow token from that\n        if QS_QUERY in query_params:\n            inner_query_params = QueryDict(query_params.get(QS_QUERY), mutable=True)\n            inner_query_params.pop(QS_KEY_TOKEN, None)\n            query_params[QS_QUERY] = inner_query_params.urlencode()\n\n        query_params.update(kwargs)\n        full_url = base_url\n        if len(query_params) > 0:\n            full_url = f\"{full_url}?{query_params.urlencode()}\"\n        return self.request.build_absolute_uri(full_url)\n\n    def get_token(self) -> FlowToken:\n        \"\"\"Get token\"\"\"\n        pending_user = self.get_pending_user()\n        current_stage: EmailStage = self.executor.current_stage\n        valid_delta = timedelta(\n            minutes=current_stage.token_expiry + 1\n        )  # + 1 because django timesince always rounds down\n        identifier = slugify(f\"ak-email-stage-{current_stage.name}-{pending_user}\")\n        # Don't check for validity here, we only care if the token exists\n        tokens = FlowToken.objects.filter(identifier=identifier)\n        if not tokens.exists():\n            return FlowToken.objects.create(\n                expires=now() + valid_delta,\n                user=pending_user,\n                identifier=identifier,\n                flow=self.executor.flow,\n                _plan=FlowToken.pickle(self.executor.plan),\n            )\n        token = tokens.first()\n        # Check if token is expired and rotate key if so\n        if token.is_expired:\n            token.expire_action()\n        return token\n\n    def send_email(self):\n        \"\"\"Helper function that sends the actual email. Implies that you've\n        already checked that there is a pending user.\"\"\"\n        pending_user = self.get_pending_user()\n        if not pending_user.pk and self.executor.flow.designation == FlowDesignation.RECOVERY:\n            # Pending user does not have a primary key, and we're in a recovery flow,\n            # which means the user entered an invalid identifier, so we pretend to send the\n            # email, to not disclose if the user exists\n            return\n        email = self.executor.plan.context.get(PLAN_CONTEXT_EMAIL_OVERRIDE, None)\n        if not email:\n            email = pending_user.email\n        current_stage: EmailStage = self.executor.current_stage\n        token = self.get_token()\n        # Send mail to user\n        message = TemplateEmailMessage(\n            subject=_(current_stage.subject),\n            to=[email],\n            language=pending_user.locale(self.request),\n            template_name=current_stage.template,\n            template_context={\n                \"url\": self.get_full_url(**{QS_KEY_TOKEN: token.key}),\n                \"user\": pending_user,\n                \"expires\": token.expires,\n            },\n        )\n        send_mails(current_stage, message)\n\n    def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:\n        # Check if the user came back from the email link to verify\n        restore_token: FlowToken = self.executor.plan.context.get(PLAN_CONTEXT_IS_RESTORED, None)\n        user = self.get_pending_user()\n        if restore_token:\n            if restore_token.user != user:\n                self.logger.warning(\"Flow token for non-matching user, denying request\")\n                return self.executor.stage_invalid()\n            messages.success(request, _(\"Successfully verified Email.\"))\n            if self.executor.current_stage.activate_user_on_success:\n                user.is_active = True\n                user.save()\n            return self.executor.stage_ok()\n        if PLAN_CONTEXT_PENDING_USER not in self.executor.plan.context:\n            self.logger.debug(\"No pending user\")\n            messages.error(self.request, _(\"No pending user.\"))\n            return self.executor.stage_invalid()\n        # Check if we've already sent the initial e-mail\n        if PLAN_CONTEXT_EMAIL_SENT not in self.executor.plan.context:\n            self.send_email()\n            self.executor.plan.context[PLAN_CONTEXT_EMAIL_SENT] = True\n        return super().get(request, *args, **kwargs)\n\n    def get_challenge(self) -> Challenge:\n        challenge = EmailChallenge(\n            data={\n                \"type\": ChallengeTypes.NATIVE.value,\n                \"title\": _(\"Email sent.\"),\n            }\n        )\n        return challenge\n\n    def challenge_valid(self, response: ChallengeResponse) -> HttpResponse:\n        return super().challenge_invalid(response)\n\n    def challenge_invalid(self, response: ChallengeResponse) -> HttpResponse:\n        if PLAN_CONTEXT_PENDING_USER not in self.executor.plan.context:\n            messages.error(self.request, _(\"No pending user.\"))\n            return super().challenge_invalid(response)\n        self.send_email()\n        messages.success(self.request, _(\"Email Successfully sent.\"))\n        # We can't call stage_ok yet, as we're still waiting\n        # for the user to click the link in the email\n        return super().challenge_invalid(response)\n", "path": "authentik/stages/email/stage.py"}], "after_files": [{"content": "\"\"\"authentik multi-stage authentication engine\"\"\"\nfrom datetime import timedelta\n\nfrom django.contrib import messages\nfrom django.http import HttpRequest, HttpResponse\nfrom django.http.request import QueryDict\nfrom django.urls import reverse\nfrom django.utils.text import slugify\nfrom django.utils.timezone import now\nfrom django.utils.translation import gettext as _\nfrom rest_framework.fields import CharField\nfrom rest_framework.serializers import ValidationError\n\nfrom authentik.flows.challenge import Challenge, ChallengeResponse, ChallengeTypes\nfrom authentik.flows.models import FlowDesignation, FlowToken\nfrom authentik.flows.planner import PLAN_CONTEXT_IS_RESTORED, PLAN_CONTEXT_PENDING_USER\nfrom authentik.flows.stage import ChallengeStageView\nfrom authentik.flows.views.executor import QS_KEY_TOKEN, QS_QUERY\nfrom authentik.stages.email.models import EmailStage\nfrom authentik.stages.email.tasks import send_mails\nfrom authentik.stages.email.utils import TemplateEmailMessage\n\nPLAN_CONTEXT_EMAIL_SENT = \"email_sent\"\nPLAN_CONTEXT_EMAIL_OVERRIDE = \"email\"\n\n\nclass EmailChallenge(Challenge):\n    \"\"\"Email challenge\"\"\"\n\n    component = CharField(default=\"ak-stage-email\")\n\n\nclass EmailChallengeResponse(ChallengeResponse):\n    \"\"\"Email challenge resposen. No fields. This challenge is\n    always declared invalid to give the user a chance to retry\"\"\"\n\n    component = CharField(default=\"ak-stage-email\")\n\n    def validate(self, attrs):\n        raise ValidationError(detail=\"email-sent\", code=\"email-sent\")\n\n\nclass EmailStageView(ChallengeStageView):\n    \"\"\"Email stage which sends Email for verification\"\"\"\n\n    response_class = EmailChallengeResponse\n\n    def get_full_url(self, **kwargs) -> str:\n        \"\"\"Get full URL to be used in template\"\"\"\n        base_url = reverse(\n            \"authentik_core:if-flow\",\n            kwargs={\"flow_slug\": self.executor.flow.slug},\n        )\n        # Parse query string from current URL (full query string)\n        # this view is only run within a flow executor, where we need to get the query string\n        # from the query= parameter (double encoded); but for the redirect\n        # we need to expand it since it'll go through the flow interface\n        query_params = QueryDict(self.request.GET.get(QS_QUERY), mutable=True)\n        query_params.pop(QS_KEY_TOKEN, None)\n        query_params.update(kwargs)\n        print(query_params)\n        full_url = base_url\n        if len(query_params) > 0:\n            full_url = f\"{full_url}?{query_params.urlencode()}\"\n        return self.request.build_absolute_uri(full_url)\n\n    def get_token(self) -> FlowToken:\n        \"\"\"Get token\"\"\"\n        pending_user = self.get_pending_user()\n        current_stage: EmailStage = self.executor.current_stage\n        valid_delta = timedelta(\n            minutes=current_stage.token_expiry + 1\n        )  # + 1 because django timesince always rounds down\n        identifier = slugify(f\"ak-email-stage-{current_stage.name}-{pending_user}\")\n        # Don't check for validity here, we only care if the token exists\n        tokens = FlowToken.objects.filter(identifier=identifier)\n        if not tokens.exists():\n            return FlowToken.objects.create(\n                expires=now() + valid_delta,\n                user=pending_user,\n                identifier=identifier,\n                flow=self.executor.flow,\n                _plan=FlowToken.pickle(self.executor.plan),\n            )\n        token = tokens.first()\n        # Check if token is expired and rotate key if so\n        if token.is_expired:\n            token.expire_action()\n        return token\n\n    def send_email(self):\n        \"\"\"Helper function that sends the actual email. Implies that you've\n        already checked that there is a pending user.\"\"\"\n        pending_user = self.get_pending_user()\n        if not pending_user.pk and self.executor.flow.designation == FlowDesignation.RECOVERY:\n            # Pending user does not have a primary key, and we're in a recovery flow,\n            # which means the user entered an invalid identifier, so we pretend to send the\n            # email, to not disclose if the user exists\n            return\n        email = self.executor.plan.context.get(PLAN_CONTEXT_EMAIL_OVERRIDE, None)\n        if not email:\n            email = pending_user.email\n        current_stage: EmailStage = self.executor.current_stage\n        token = self.get_token()\n        # Send mail to user\n        message = TemplateEmailMessage(\n            subject=_(current_stage.subject),\n            to=[email],\n            language=pending_user.locale(self.request),\n            template_name=current_stage.template,\n            template_context={\n                \"url\": self.get_full_url(**{QS_KEY_TOKEN: token.key}),\n                \"user\": pending_user,\n                \"expires\": token.expires,\n            },\n        )\n        send_mails(current_stage, message)\n\n    def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:\n        # Check if the user came back from the email link to verify\n        restore_token: FlowToken = self.executor.plan.context.get(PLAN_CONTEXT_IS_RESTORED, None)\n        user = self.get_pending_user()\n        if restore_token:\n            if restore_token.user != user:\n                self.logger.warning(\"Flow token for non-matching user, denying request\")\n                return self.executor.stage_invalid()\n            messages.success(request, _(\"Successfully verified Email.\"))\n            if self.executor.current_stage.activate_user_on_success:\n                user.is_active = True\n                user.save()\n            return self.executor.stage_ok()\n        if PLAN_CONTEXT_PENDING_USER not in self.executor.plan.context:\n            self.logger.debug(\"No pending user\")\n            messages.error(self.request, _(\"No pending user.\"))\n            return self.executor.stage_invalid()\n        # Check if we've already sent the initial e-mail\n        if PLAN_CONTEXT_EMAIL_SENT not in self.executor.plan.context:\n            self.send_email()\n            self.executor.plan.context[PLAN_CONTEXT_EMAIL_SENT] = True\n        return super().get(request, *args, **kwargs)\n\n    def get_challenge(self) -> Challenge:\n        challenge = EmailChallenge(\n            data={\n                \"type\": ChallengeTypes.NATIVE.value,\n                \"title\": _(\"Email sent.\"),\n            }\n        )\n        return challenge\n\n    def challenge_valid(self, response: ChallengeResponse) -> HttpResponse:\n        return super().challenge_invalid(response)\n\n    def challenge_invalid(self, response: ChallengeResponse) -> HttpResponse:\n        if PLAN_CONTEXT_PENDING_USER not in self.executor.plan.context:\n            messages.error(self.request, _(\"No pending user.\"))\n            return super().challenge_invalid(response)\n        self.send_email()\n        messages.success(self.request, _(\"Email Successfully sent.\"))\n        # We can't call stage_ok yet, as we're still waiting\n        # for the user to click the link in the email\n        return super().challenge_invalid(response)\n", "path": "authentik/stages/email/stage.py"}]}