diff --git a/authentik/flows/api/bindings.py b/authentik/flows/api/bindings.py
index 74b97ca99..13fd04887 100644
--- a/authentik/flows/api/bindings.py
+++ b/authentik/flows/api/bindings.py
@@ -25,6 +25,7 @@ class FlowStageBindingSerializer(ModelSerializer):
             "re_evaluate_policies",
             "order",
             "policy_engine_mode",
+            "invalid_response_action",
         ]
 
 
diff --git a/authentik/flows/markers.py b/authentik/flows/markers.py
index d5bc297f1..e545cf89b 100644
--- a/authentik/flows/markers.py
+++ b/authentik/flows/markers.py
@@ -5,7 +5,6 @@ from typing import TYPE_CHECKING, Optional
 from django.http.request import HttpRequest
 from structlog.stdlib import get_logger
 
-from authentik.core.models import User
 from authentik.flows.models import FlowStageBinding
 from authentik.policies.engine import PolicyEngine
 from authentik.policies.models import PolicyBinding
@@ -25,7 +24,7 @@ class StageMarker:
         self,
         plan: "FlowPlan",
         binding: FlowStageBinding,
-        http_request: Optional[HttpRequest],
+        http_request: HttpRequest,
     ) -> Optional[FlowStageBinding]:
         """Process callback for this marker. This should be overridden by sub-classes.
         If a stage should be removed, return None."""
@@ -37,24 +36,26 @@ class ReevaluateMarker(StageMarker):
     """Reevaluate Marker, forces stage's policies to be evaluated again."""
 
     binding: PolicyBinding
-    user: User
 
     def process(
         self,
         plan: "FlowPlan",
         binding: FlowStageBinding,
-        http_request: Optional[HttpRequest],
+        http_request: HttpRequest,
     ) -> Optional[FlowStageBinding]:
         """Re-evaluate policies bound to stage, and if they fail, remove from plan"""
+        from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER
+
         LOGGER.debug(
             "f(plan_inst)[re-eval marker]: running re-evaluation",
             binding=binding,
             policy_binding=self.binding,
         )
-        engine = PolicyEngine(self.binding, self.user)
+        engine = PolicyEngine(
+            self.binding, plan.context.get(PLAN_CONTEXT_PENDING_USER, http_request.user)
+        )
         engine.use_cache = False
-        if http_request:
-            engine.request.set_http_request(http_request)
+        engine.request.set_http_request(http_request)
         engine.request.context = plan.context
         engine.build()
         result = engine.result
diff --git a/authentik/flows/migrations/0021_flowstagebinding_invalid_response_action.py b/authentik/flows/migrations/0021_flowstagebinding_invalid_response_action.py
new file mode 100644
index 000000000..1c0add77f
--- /dev/null
+++ b/authentik/flows/migrations/0021_flowstagebinding_invalid_response_action.py
@@ -0,0 +1,22 @@
+# Generated by Django 3.2.4 on 2021-06-27 16:20
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_flows", "0020_flow_compatibility_mode"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="flowstagebinding",
+            name="invalid_response_action",
+            field=models.TextField(
+                choices=[("retry", "Retry"), ("continue", "Continue")],
+                default="retry",
+                help_text="Configure how the flow executor should handle an invalid response to a challenge. RETRY returns the error message and a similar challenge to the executor while CONTINUE continues with the next stage.",
+            ),
+        ),
+    ]
diff --git a/authentik/flows/models.py b/authentik/flows/models.py
index 6e9663882..df713dba2 100644
--- a/authentik/flows/models.py
+++ b/authentik/flows/models.py
@@ -27,6 +27,14 @@ class NotConfiguredAction(models.TextChoices):
     CONFIGURE = "configure"
 
 
+class InvalidResponseAction(models.TextChoices):
+    """Configure how the flow executor should handle invalid responses to challenges"""
+
+    RETRY = "retry"
+    RESTART = "restart"
+    RESTART_WITH_CONTEXT = "restart_with_context"
+
+
 class FlowDesignation(models.TextChoices):
     """Designation of what a Flow should be used for. At a later point, this
     should be replaced by a database entry."""
@@ -201,6 +209,17 @@ class FlowStageBinding(SerializerModel, PolicyBindingModel):
         help_text=_("Evaluate policies when the Stage is present to the user."),
     )
 
+    invalid_response_action = models.TextField(
+        choices=InvalidResponseAction.choices,
+        default=InvalidResponseAction.RETRY,
+        help_text=_(
+            "Configure how the flow executor should handle an invalid response to a "
+            "challenge. RETRY returns the error message and a similar challenge to the "
+            "executor. RESTART restarts the flow from the beginning, and RESTART_WITH_CONTEXT "
+            "restarts the flow while keeping the current context."
+        ),
+    )
+
     order = models.IntegerField()
 
     objects = InheritanceManager()
diff --git a/authentik/flows/planner.py b/authentik/flows/planner.py
index 269ae2612..ed1373601 100644
--- a/authentik/flows/planner.py
+++ b/authentik/flows/planner.py
@@ -224,7 +224,7 @@ class FlowPlanner:
                         "f(plan): stage has re-evaluate marker",
                         stage=binding.stage,
                     )
-                    marker = ReevaluateMarker(binding=binding, user=user)
+                    marker = ReevaluateMarker(binding=binding)
                 if stage:
                     plan.append(binding, marker)
             HIST_FLOWS_PLAN_TIME.labels(flow_slug=self.flow.slug)
diff --git a/authentik/flows/stage.py b/authentik/flows/stage.py
index 8502a42c6..93461ce42 100644
--- a/authentik/flows/stage.py
+++ b/authentik/flows/stage.py
@@ -16,6 +16,7 @@ from authentik.flows.challenge import (
     HttpChallengeResponse,
     WithUserInfoChallenge,
 )
+from authentik.flows.models import InvalidResponseAction
 from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER
 from authentik.flows.views import FlowExecutorView
 
@@ -69,7 +70,13 @@ class ChallengeStageView(StageView):
         """Return a challenge for the frontend to solve"""
         challenge = self._get_challenge(*args, **kwargs)
         if not challenge.is_valid():
-            LOGGER.warning(challenge.errors, stage_view=self, challenge=challenge)
+            LOGGER.warning(
+                "f(ch): Invalid challenge",
+                binding=self.executor.current_binding,
+                errors=challenge.errors,
+                stage_view=self,
+                challenge=challenge,
+            )
         return HttpChallengeResponse(challenge)
 
     # pylint: disable=unused-argument
@@ -77,6 +84,21 @@ class ChallengeStageView(StageView):
         """Handle challenge response"""
         challenge: ChallengeResponse = self.get_response_instance(data=request.data)
         if not challenge.is_valid():
+            if self.executor.current_binding.invalid_response_action in [
+                InvalidResponseAction.RESTART,
+                InvalidResponseAction.RESTART_WITH_CONTEXT,
+            ]:
+                keep_context = (
+                    self.executor.current_binding.invalid_response_action
+                    == InvalidResponseAction.RESTART_WITH_CONTEXT
+                )
+                LOGGER.debug(
+                    "f(ch): Invalid response, restarting flow",
+                    binding=self.executor.current_binding,
+                    stage_view=self,
+                    keep_context=keep_context,
+                )
+                return self.executor.restart_flow(keep_context)
             return self.challenge_invalid(challenge)
         return self.challenge_valid(challenge)
 
@@ -126,5 +148,10 @@ class ChallengeStageView(StageView):
                 )
         challenge_response.initial_data["response_errors"] = full_errors
         if not challenge_response.is_valid():
-            LOGGER.warning(challenge_response.errors)
+            LOGGER.warning(
+                "f(ch): invalid challenge response",
+                binding=self.executor.current_binding,
+                errors=challenge_response.errors,
+                stage_view=self,
+            )
         return HttpChallengeResponse(challenge_response)
diff --git a/authentik/flows/tests/test_views.py b/authentik/flows/tests/test_views.py
index 4b15726c0..3ccf58f0d 100644
--- a/authentik/flows/tests/test_views.py
+++ b/authentik/flows/tests/test_views.py
@@ -11,15 +11,23 @@ from authentik.core.models import User
 from authentik.flows.challenge import ChallengeTypes
 from authentik.flows.exceptions import FlowNonApplicableException
 from authentik.flows.markers import ReevaluateMarker, StageMarker
-from authentik.flows.models import Flow, FlowDesignation, FlowStageBinding
+from authentik.flows.models import (
+    Flow,
+    FlowDesignation,
+    FlowStageBinding,
+    InvalidResponseAction,
+)
 from authentik.flows.planner import FlowPlan, FlowPlanner
 from authentik.flows.stage import PLAN_CONTEXT_PENDING_USER_IDENTIFIER, StageView
 from authentik.flows.views import NEXT_ARG_NAME, SESSION_KEY_PLAN, FlowExecutorView
 from authentik.lib.config import CONFIG
 from authentik.policies.dummy.models import DummyPolicy
 from authentik.policies.models import PolicyBinding
+from authentik.policies.reputation.models import ReputationPolicy
 from authentik.policies.types import PolicyResult
+from authentik.stages.deny.models import DenyStage
 from authentik.stages.dummy.models import DummyStage
+from authentik.stages.identification.models import IdentificationStage, UserFields
 
 POLICY_RETURN_FALSE = PropertyMock(return_value=PolicyResult(False))
 POLICY_RETURN_TRUE = MagicMock(return_value=PolicyResult(True))
@@ -513,3 +521,78 @@ class TestFlowExecutor(TestCase):
 
         stage_view = StageView(executor)
         self.assertEqual(ident, stage_view.get_pending_user(for_display=True).username)
+
+    def test_invalid_restart(self):
+        """Test flow that restarts on invalid entry"""
+        flow = Flow.objects.create(
+            name="restart-on-invalid",
+            slug="restart-on-invalid",
+            designation=FlowDesignation.AUTHENTICATION,
+        )
+        # Stage 0 is a deny stage that is added dynamically
+        # when the reputation policy says so
+        deny_stage = DenyStage.objects.create(name="deny")
+        reputation_policy = ReputationPolicy.objects.create(
+            name="reputation", threshold=-1, check_ip=False
+        )
+        deny_binding = FlowStageBinding.objects.create(
+            target=flow,
+            stage=deny_stage,
+            order=0,
+            evaluate_on_plan=False,
+            re_evaluate_policies=True,
+        )
+        PolicyBinding.objects.create(
+            policy=reputation_policy, target=deny_binding, order=0
+        )
+
+        # Stage 1 is an identification stage
+        ident_stage = IdentificationStage.objects.create(
+            name="ident",
+            user_fields=[UserFields.E_MAIL],
+        )
+        FlowStageBinding.objects.create(
+            target=flow,
+            stage=ident_stage,
+            order=1,
+            invalid_response_action=InvalidResponseAction.RESTART_WITH_CONTEXT,
+        )
+        exec_url = reverse(
+            "authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}
+        )
+        # First request, run the planner
+        response = self.client.get(exec_url)
+        self.assertEqual(response.status_code, 200)
+        self.assertJSONEqual(
+            force_str(response.content),
+            {
+                "type": ChallengeTypes.NATIVE.value,
+                "component": "ak-stage-identification",
+                "flow_info": {
+                    "background": flow.background_url,
+                    "cancel_url": reverse("authentik_flows:cancel"),
+                    "title": "",
+                },
+                "password_fields": False,
+                "primary_action": "Log in",
+                "sources": [],
+                "user_fields": [UserFields.E_MAIL],
+            },
+        )
+        response = self.client.post(
+            exec_url, {"uid_field": "invalid-string"}, follow=True
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertJSONEqual(
+            force_str(response.content),
+            {
+                "component": "ak-stage-access-denied",
+                "error_message": None,
+                "flow_info": {
+                    "background": flow.background_url,
+                    "cancel_url": reverse("authentik_flows:cancel"),
+                    "title": "",
+                },
+                "type": ChallengeTypes.NATIVE.value,
+            },
+        )
diff --git a/authentik/flows/views.py b/authentik/flows/views.py
index 34c14ff07..39c9a1662 100644
--- a/authentik/flows/views.py
+++ b/authentik/flows/views.py
@@ -288,6 +288,20 @@ class FlowExecutorView(APIView):
             return self._initiate_plan()
         return plan
 
+    def restart_flow(self, keep_context=False) -> HttpResponse:
+        """Restart the currently active flow, optionally keeping the current context"""
+        planner = FlowPlanner(self.flow)
+        default_context = None
+        if keep_context:
+            default_context = self.plan.context
+        plan = planner.plan(self.request, default_context)
+        self.request.session[SESSION_KEY_PLAN] = plan
+        kwargs = self.kwargs
+        kwargs.update({"flow_slug": self.flow.slug})
+        return redirect_with_qs(
+            "authentik_api:flow-executor", self.request.GET, **kwargs
+        )
+
     def _flow_done(self) -> HttpResponse:
         """User Successfully passed all stages"""
         # Since this is wrapped by the ExecutorShell, the next argument is saved in the session
diff --git a/authentik/policies/reputation/models.py b/authentik/policies/reputation/models.py
index 305a33832..5a52241e2 100644
--- a/authentik/policies/reputation/models.py
+++ b/authentik/policies/reputation/models.py
@@ -33,21 +33,21 @@ class ReputationPolicy(Policy):
 
     def passes(self, request: PolicyRequest) -> PolicyResult:
         remote_ip = get_client_ip(request.http_request)
-        passing = True
+        passing = False
         if self.check_ip:
             score = cache.get_or_set(CACHE_KEY_IP_PREFIX + remote_ip, 0)
-            passing = passing and score <= self.threshold
+            passing += passing or score <= self.threshold
             LOGGER.debug("Score for IP", ip=remote_ip, score=score, passing=passing)
         if self.check_username:
             score = cache.get_or_set(CACHE_KEY_USER_PREFIX + request.user.username, 0)
-            passing = passing and score <= self.threshold
+            passing += passing or score <= self.threshold
             LOGGER.debug(
                 "Score for Username",
                 username=request.user.username,
                 score=score,
                 passing=passing,
             )
-        return PolicyResult(passing)
+        return PolicyResult(bool(passing))
 
     class Meta:
 
diff --git a/authentik/stages/identification/stage.py b/authentik/stages/identification/stage.py
index dc50b4624..88248e76c 100644
--- a/authentik/stages/identification/stage.py
+++ b/authentik/stages/identification/stage.py
@@ -85,6 +85,18 @@ class IdentificationChallengeResponse(ChallengeResponse):
             identification_failed.send(
                 sender=self, request=self.stage.request, uid_field=uid_field
             )
+            # We set the pending_user even on failure so it's part of the context, even
+            # when the input is invalid
+            # This is so its part of the current flow plan, and on flow restart can be kept, and
+            # policies can be applied.
+            self.stage.executor.plan.context[PLAN_CONTEXT_PENDING_USER] = User(
+                username=uid_field,
+                email=uid_field,
+            )
+            if not current_stage.show_matched_user:
+                self.stage.executor.plan.context[
+                    PLAN_CONTEXT_PENDING_USER_IDENTIFIER
+                ] = uid_field
             raise ValidationError("Failed to authenticate.")
         self.pre_user = pre_user
         if not current_stage.password_stage:
diff --git a/schema.yml b/schema.yml
index 761f424f9..40a26b97d 100644
--- a/schema.yml
+++ b/schema.yml
@@ -4572,6 +4572,18 @@ paths:
         schema:
           type: string
           format: uuid
+      - in: query
+        name: invalid_response_action
+        schema:
+          type: string
+          enum:
+          - restart
+          - restart_with_context
+          - retry
+        description: Configure how the flow executor should handle an invalid response
+          to a challenge. RETRY returns the error message and a similar challenge
+          to the executor. RESTART restarts the flow from the beginning, and RESTART_WITH_CONTEXT
+          restarts the flow while keeping the current context.
       - in: query
         name: order
         schema:
@@ -19810,6 +19822,13 @@ components:
           minimum: -2147483648
         policy_engine_mode:
           $ref: '#/components/schemas/PolicyEngineMode'
+        invalid_response_action:
+          allOf:
+          - $ref: '#/components/schemas/InvalidResponseActionEnum'
+          description: Configure how the flow executor should handle an invalid response
+            to a challenge. RETRY returns the error message and a similar challenge
+            to the executor. RESTART restarts the flow from the beginning, and RESTART_WITH_CONTEXT
+            restarts the flow while keeping the current context.
       required:
       - order
       - pk
@@ -19840,6 +19859,13 @@ components:
           minimum: -2147483648
         policy_engine_mode:
           $ref: '#/components/schemas/PolicyEngineMode'
+        invalid_response_action:
+          allOf:
+          - $ref: '#/components/schemas/InvalidResponseActionEnum'
+          description: Configure how the flow executor should handle an invalid response
+            to a challenge. RETRY returns the error message and a similar challenge
+            to the executor. RESTART restarts the flow from the beginning, and RESTART_WITH_CONTEXT
+            restarts the flow while keeping the current context.
       required:
       - order
       - stage
@@ -20185,6 +20211,12 @@ components:
       - api
       - recovery
       type: string
+    InvalidResponseActionEnum:
+      enum:
+      - retry
+      - restart
+      - restart_with_context
+      type: string
     Invitation:
       type: object
       description: Invitation Serializer
@@ -24662,6 +24694,13 @@ components:
           minimum: -2147483648
         policy_engine_mode:
           $ref: '#/components/schemas/PolicyEngineMode'
+        invalid_response_action:
+          allOf:
+          - $ref: '#/components/schemas/InvalidResponseActionEnum'
+          description: Configure how the flow executor should handle an invalid response
+            to a challenge. RETRY returns the error message and a similar challenge
+            to the executor. RESTART restarts the flow from the beginning, and RESTART_WITH_CONTEXT
+            restarts the flow while keeping the current context.
     PatchedGroupRequest:
       type: object
       description: Group Serializer
diff --git a/web/src/locales/en.po b/web/src/locales/en.po
index d53b44033..406594bce 100644
--- a/web/src/locales/en.po
+++ b/web/src/locales/en.po
@@ -698,6 +698,10 @@ msgstr "Configure how long refresh tokens and their id_tokens are valid for."
 msgid "Configure how the NameID value will be created. When left empty, the NameIDPolicy of the incoming request will be respected."
 msgstr "Configure how the NameID value will be created. When left empty, the NameIDPolicy of the incoming request will be respected."
 
+#: src/pages/flows/StageBindingForm.ts
+msgid "Configure how the flow executor should handle an invalid response to a challenge."
+msgstr "Configure how the flow executor should handle an invalid response to a challenge."
+
 #: src/pages/providers/oauth2/OAuth2ProviderForm.ts
 msgid "Configure how the issuer field of the ID Token should be filled."
 msgstr "Configure how the issuer field of the ID Token should be filled."
@@ -1881,6 +1885,10 @@ msgstr "Internal host"
 msgid "Internal host SSL Validation"
 msgstr "Internal host SSL Validation"
 
+#: src/pages/flows/StageBindingForm.ts
+msgid "Invalid response action"
+msgstr "Invalid response action"
+
 #: src/pages/flows/FlowForm.ts
 msgid "Invalidation"
 msgstr "Invalidation"
@@ -2847,6 +2855,18 @@ msgstr "Public key, acquired from https://www.google.com/recaptcha/intro/v3.html
 msgid "Publisher"
 msgstr "Publisher"
 
+#: src/pages/flows/StageBindingForm.ts
+msgid "RESTART restarts the flow from the beginning, while keeping the flow context."
+msgstr "RESTART restarts the flow from the beginning, while keeping the flow context."
+
+#: src/pages/flows/StageBindingForm.ts
+msgid "RESTART restarts the flow from the beginning."
+msgstr "RESTART restarts the flow from the beginning."
+
+#: src/pages/flows/StageBindingForm.ts
+msgid "RETRY returns the error message and a similar challenge to the executor."
+msgstr "RETRY returns the error message and a similar challenge to the executor."
+
 #: src/pages/providers/oauth2/OAuth2ProviderForm.ts
 msgid "RS256 (Asymmetric Encryption)"
 msgstr "RS256 (Asymmetric Encryption)"
diff --git a/web/src/locales/pseudo-LOCALE.po b/web/src/locales/pseudo-LOCALE.po
index 5d623fb18..15917a8a8 100644
--- a/web/src/locales/pseudo-LOCALE.po
+++ b/web/src/locales/pseudo-LOCALE.po
@@ -692,6 +692,10 @@ msgstr ""
 msgid "Configure how the NameID value will be created. When left empty, the NameIDPolicy of the incoming request will be respected."
 msgstr ""
 
+#: 
+msgid "Configure how the flow executor should handle an invalid response to a challenge."
+msgstr ""
+
 #: 
 msgid "Configure how the issuer field of the ID Token should be filled."
 msgstr ""
@@ -1873,6 +1877,10 @@ msgstr ""
 msgid "Internal host SSL Validation"
 msgstr ""
 
+#: 
+msgid "Invalid response action"
+msgstr ""
+
 #: 
 msgid "Invalidation"
 msgstr ""
@@ -2839,6 +2847,18 @@ msgstr ""
 msgid "Publisher"
 msgstr ""
 
+#: 
+msgid "RESTART restarts the flow from the beginning, while keeping the flow context."
+msgstr ""
+
+#: 
+msgid "RESTART restarts the flow from the beginning."
+msgstr ""
+
+#: 
+msgid "RETRY returns the error message and a similar challenge to the executor."
+msgstr ""
+
 #: 
 msgid "RS256 (Asymmetric Encryption)"
 msgstr ""
diff --git a/web/src/pages/flows/StageBindingForm.ts b/web/src/pages/flows/StageBindingForm.ts
index b669752c9..81a4563cc 100644
--- a/web/src/pages/flows/StageBindingForm.ts
+++ b/web/src/pages/flows/StageBindingForm.ts
@@ -1,4 +1,4 @@
-import { FlowsApi, FlowStageBinding, PolicyEngineMode, Stage, StagesApi } from "authentik-api";
+import { FlowsApi, FlowStageBinding, InvalidResponseActionEnum, PolicyEngineMode, Stage, StagesApi } from "authentik-api";
 import { t } from "@lingui/macro";
 import { customElement, property } from "lit-element";
 import { html, TemplateResult } from "lit-html";
@@ -135,6 +135,23 @@ export class StageBindingForm extends ModelForm<FlowStageBinding, string> {
                 </div>
                 <p class="pf-c-form__helper-text">${t`Evaluate policies before the Stage is present to the user.`}</p>
             </ak-form-element-horizontal>
+            <ak-form-element-horizontal
+                label=${t`Invalid response action`}
+                ?required=${true}
+                name="invalidResponseAction">
+                <select class="pf-c-form-control">
+                    <option value=${InvalidResponseActionEnum.Retry} ?selected=${this.instance?.invalidResponseAction === InvalidResponseActionEnum.Retry}>
+                        ${t`RETRY returns the error message and a similar challenge to the executor.`}
+                    </option>
+                    <option value=${InvalidResponseActionEnum.Restart} ?selected=${this.instance?.invalidResponseAction === InvalidResponseActionEnum.Restart}>
+                        ${t`RESTART restarts the flow from the beginning.`}
+                    </option>
+                    <option value=${InvalidResponseActionEnum.RestartWithContext} ?selected=${this.instance?.invalidResponseAction === InvalidResponseActionEnum.RestartWithContext}>
+                        ${t`RESTART restarts the flow from the beginning, while keeping the flow context.`}
+                    </option>
+                </select>
+                <p class="pf-c-form__helper-text">${t`Configure how the flow executor should handle an invalid response to a challenge.`}</p>
+            </ak-form-element-horizontal>
             <ak-form-element-horizontal
                 label=${t`Policy engine mode`}
                 ?required=${true}