stages/user_login: require next action flows before login

authentik/core/api/users.py

@@ -80,6 +80,7 @@
     SESSION_KEY_IMPERSONATE_USER,
 )
 from authentik.core.models import (
+    USER_ATTRIBUTE_NEXT_ACTIONS,
     USER_ATTRIBUTE_TOKEN_EXPIRING,
     USER_PATH_SERVICE_ACCOUNT,
     USERNAME_MAX_LENGTH,
@@ -93,6 +94,7 @@
 )
 from authentik.core.views.user_switch import start_user_switch_flow
 from authentik.endpoints.connectors.agent.auth import AgentAuth
+from authentik.events.middleware import audit_ignore
 from authentik.events.models import Event, EventAction
 from authentik.flows.exceptions import FlowNonApplicableException
 from authentik.flows.models import FlowToken
@@ -109,6 +111,7 @@
 from authentik.stages.email.models import EmailStage
 from authentik.stages.email.tasks import send_mails
 from authentik.stages.email.utils import TemplateEmailMessage
+from authentik.stages.user_login.next_actions import next_action_slugs, resolve_next_actions
 
 LOGGER = get_logger()
 
@@ -215,6 +218,7 @@
             permissions = validated_data.pop("permissions", [])
 
         instance: User = super().create(validated_data)
+        self._log_next_action_changes([], instance)
         if is_blueprint:
             self._set_password(instance, password, password_hash)
             perms_qs = Permission.objects.filter(
@@ -233,7 +237,15 @@
             password_hash = validated_data.pop("password_hash", None)
             permissions = validated_data.pop("permissions", [])
 
-        instance = super().update(instance, validated_data)
+        previous_actions = next_action_slugs(instance.attributes.get(USER_ATTRIBUTE_NEXT_ACTIONS))
+        # When only the next-actions attribute changes, the dedicated events below
+        # replace the generic model update event
+        if self._is_next_actions_only_change(instance, validated_data):
+            with audit_ignore():
+                instance = super().update(instance, validated_data)
+        else:
+            instance = super().update(instance, validated_data)
+        self._log_next_action_changes(previous_actions, instance)
         if is_blueprint:
             self._set_password(instance, password, password_hash)
             perms_qs = Permission.objects.filter(
@@ -244,6 +256,43 @@
         self._ensure_password_not_empty(instance)
         return instance
 
+    def _is_next_actions_only_change(self, instance: User, validated_data: dict) -> bool:
+        """Check whether the update only changes the next-actions attribute."""
+        if set(validated_data.keys()) != {"attributes"}:
+            return False
+        previous = {
+            k: v for k, v in instance.attributes.items() if k != USER_ATTRIBUTE_NEXT_ACTIONS
+        }
+        updated = {
+            k: v
+            for k, v in validated_data["attributes"].items()
+            if k != USER_ATTRIBUTE_NEXT_ACTIONS
+        }
+        return previous == updated
+
+    def _log_next_action_changes(self, previous_actions: list[str], instance: User):
+        """Create events for next actions added to or removed from the user."""
+        current_actions = next_action_slugs(instance.attributes.get(USER_ATTRIBUTE_NEXT_ACTIONS))
+        request = self.context.get("request")
+        changes = (
+            (
+                EventAction.NEXT_ACTION_SET,
+                [s for s in current_actions if s not in previous_actions],
+            ),
+            (
+                EventAction.NEXT_ACTION_REMOVED,
+                [s for s in previous_actions if s not in current_actions],
+            ),
+        )
+        for event_action, slugs in changes:
+            for slug in slugs:
+                # `username` in the context makes the event visible on the user's events tab
+                event = Event.new(event_action, flow_slug=slug, username=instance.username)
+                if request:
+                    event.from_http(request)
+                else:
+                    event.save()
+
     def _set_password(self, instance: User, password: str | None, password_hash: str | None = None):
         """Set password from plain text or hash."""
         if password_hash is not None:
@@ -289,6 +338,15 @@
                 )
         return user_type
 
+    def validate_attributes(self, attributes: dict) -> dict:
+        """Validate that the next-actions attribute only holds usable flows."""
+        if USER_ATTRIBUTE_NEXT_ACTIONS in attributes:
+            try:
+                resolve_next_actions(attributes[USER_ATTRIBUTE_NEXT_ACTIONS])
+            except ValueError as exc:
+                raise ValidationError(str(exc)) from exc
+        return attributes
+
     def validate_groups(self, groups: list) -> list:
         """Require enable_group_superuser permission when adding a user to a superuser group."""
         request: Request = self.context.get("request", None)

authentik/core/models.py

@@ -70,6 +70,7 @@
 USER_ATTRIBUTE_CHANGE_USERNAME = f"{_USER_ATTR_PREFIX}/can-change-username"
 USER_ATTRIBUTE_CHANGE_NAME = f"{_USER_ATTR_PREFIX}/can-change-name"
 USER_ATTRIBUTE_CHANGE_EMAIL = f"{_USER_ATTR_PREFIX}/can-change-email"
+USER_ATTRIBUTE_NEXT_ACTIONS = f"{_USER_ATTR_PREFIX}/next-actions"
 USER_PATH_SERVICE_ACCOUNT = f"{USER_PATH_SYSTEM_PREFIX}/service-accounts"
 
 options.DEFAULT_NAMES = options.DEFAULT_NAMES + (

authentik/core/tests/test_users_api.py

@@ -10,6 +10,7 @@
 
 from authentik.brands.models import Brand
 from authentik.core.models import (
+    USER_ATTRIBUTE_NEXT_ACTIONS,
     USER_ATTRIBUTE_TOKEN_EXPIRING,
     AuthenticatedSession,
     Group,
@@ -24,6 +25,7 @@
     create_test_flow,
     create_test_user,
 )
+from authentik.events.models import Event, EventAction
 from authentik.flows.models import FlowAuthenticationRequirement, FlowDesignation
 from authentik.lib.generators import generate_id, generate_key
 from authentik.rbac.models import Role
@@ -135,6 +137,59 @@ def test_set_type(self):
             {"type": ["Can't change internal service account to other user type."]},
         )
 
+    def test_set_next_actions(self):
+        """Test setting next action flows on a user"""
+        self.client.force_login(self.admin)
+        flow = create_test_flow(FlowDesignation.STAGE_CONFIGURATION)
+        for value in [flow.slug, [flow.slug]]:
+            response = self.client.patch(
+                reverse("authentik_api:user-detail", kwargs={"pk": self.user.pk}),
+                data={"attributes": {USER_ATTRIBUTE_NEXT_ACTIONS: value}},
+                format="json",
+            )
+            self.assertEqual(response.status_code, 200)
+            self.user.refresh_from_db()
+            self.assertEqual(self.user.attributes[USER_ATTRIBUTE_NEXT_ACTIONS], value)
+        # Only the first patch changes the set of actions
+        self.assertEqual(
+            Event.objects.filter(
+                action=EventAction.NEXT_ACTION_SET, context__flow_slug=flow.slug
+            ).count(),
+            1,
+        )
+
+        response = self.client.patch(
+            reverse("authentik_api:user-detail", kwargs={"pk": self.user.pk}),
+            data={"attributes": {}},
+            format="json",
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(
+            Event.objects.filter(
+                action=EventAction.NEXT_ACTION_REMOVED, context__flow_slug=flow.slug
+            ).count(),
+            1,
+        )
+        # Next-action-only updates don't additionally log a model update
+        self.assertFalse(
+            Event.objects.filter(
+                action=EventAction.MODEL_UPDATED,
+                context__model__pk=self.user.pk,
+            ).exists()
+        )
+
+    def test_set_next_actions_invalid(self):
+        """Test that unknown flows and disallowed designations are rejected"""
+        self.client.force_login(self.admin)
+        authentication_flow = create_test_flow(FlowDesignation.AUTHENTICATION)
+        for value in ["does-not-exist", [authentication_flow.slug], [42]]:
+            response = self.client.patch(
+                reverse("authentik_api:user-detail", kwargs={"pk": self.user.pk}),
+                data={"attributes": {USER_ATTRIBUTE_NEXT_ACTIONS: value}},
+                format="json",
+            )
+            self.assertEqual(response.status_code, 400)
+
     def test_set_password(self):
         """Test Direct password set"""
         self.client.force_login(self.admin)

authentik/events/models.py

@@ -99,6 +99,10 @@ class EventAction(models.TextChoices):
     SUSPICIOUS_REQUEST = "suspicious_request"
     PASSWORD_SET = "password_set"  # noqa # nosec
 
+    NEXT_ACTION_SET = "next_action_set"
+    NEXT_ACTION_REMOVED = "next_action_removed"
+    NEXT_ACTION_COMPLETED = "next_action_completed"
+
     SECRET_VIEW = "secret_view"  # noqa # nosec
     SECRET_ROTATE = "secret_rotate"  # noqa # nosec
 

authentik/stages/user_login/next_actions.py

@@ -0,0 +1,35 @@
+"""Next action flows, required before a user can log in"""
+
+from typing import Any
+
+from authentik.flows.models import Flow, FlowDesignation
+
+# Flows that create or end a session cannot run as a next action inside a login
+NEXT_ACTION_DISALLOWED_DESIGNATIONS = [
+    FlowDesignation.AUTHENTICATION,
+    FlowDesignation.INVALIDATION,
+]
+
+
+def next_action_slugs(value: Any) -> list[str]:
+    """Normalize the next-actions attribute value to a list of slugs, without validation"""
+    slugs = value if isinstance(value, list) else [value]
+    return [slug for slug in slugs if isinstance(slug, str)]
+
+
+def resolve_next_actions(value: Any) -> list[Flow]:
+    """Resolve the value of the next-actions user attribute (a flow slug or
+    a list of flow slugs) to flows. Raises ValueError for entries that don't
+    resolve to a usable flow."""
+    slugs = value if isinstance(value, list) else [value]
+    flows = []
+    for slug in slugs:
+        if not isinstance(slug, str):
+            raise ValueError(f"Invalid next action entry: {slug!r}")
+        flow = Flow.objects.filter(slug=slug).first()
+        if not flow:
+            raise ValueError(f"Next action flow does not exist: {slug}")
+        if flow.designation in NEXT_ACTION_DISALLOWED_DESIGNATIONS:
+            raise ValueError(f"Flow cannot be used as a next action: {slug}")
+        flows.append(flow)
+    return flows

authentik/stages/user_login/stage.py

@@ -12,16 +12,26 @@
 from rest_framework.fields import BooleanField, CharField
 
 from authentik.core import user_switching
-from authentik.core.models import AuthenticatedSession, Session, User
+from authentik.core.models import (
+    USER_ATTRIBUTE_NEXT_ACTIONS,
+    AuthenticatedSession,
+    Session,
+    User,
+)
 from authentik.core.sessions import SessionStore
 from authentik.events.middleware import audit_ignore
+from authentik.events.models import Event, EventAction
 from authentik.flows.challenge import ChallengeResponse, WithUserInfoChallenge
+from authentik.flows.exceptions import FlowNonApplicableException
+from authentik.flows.models import in_memory_stage
 from authentik.flows.planner import (
     PLAN_CONTEXT_PENDING_USER,
     PLAN_CONTEXT_USER_SWITCH_ADD_USER,
     PLAN_CONTEXT_USER_SWITCH_TARGET_SESSION,
+    FlowPlan,
+    FlowPlanner,
 )
-from authentik.flows.stage import ChallengeStageView
+from authentik.flows.stage import ChallengeStageView, StageView
 from authentik.flows.views.executor import SESSION_KEY_GET, SESSION_KEY_PLAN
 from authentik.lib.utils.time import timedelta_from_string
 from authentik.root.middleware import ClientIPMiddleware
@@ -35,6 +45,7 @@
     SESSION_KEY_BINDING_NET,
 )
 from authentik.stages.user_login.models import UserLoginStage
+from authentik.stages.user_login.next_actions import resolve_next_actions
 from authentik.tenants.utils import get_unique_identifier
 
 COOKIE_NAME_KNOWN_DEVICE = "authentik_device"
@@ -56,6 +67,30 @@ class UserLoginChallengeResponse(ChallengeResponse):
     remember_me = BooleanField(required=True)
 
 
+class NextActionDoneStageView(StageView):
+    """Remove a completed next action flow from the pending user's attributes"""
+
+    def dispatch(self, request: HttpRequest) -> HttpResponse:
+        user: User | None = self.executor.plan.context.get(PLAN_CONTEXT_PENDING_USER)
+        slug = self.executor.current_stage.flow_slug
+        if not user:
+            return self.executor.stage_ok()
+        value = user.attributes.get(USER_ATTRIBUTE_NEXT_ACTIONS)
+        if isinstance(value, list):
+            if slug in value:
+                value.remove(slug)
+            if not value:
+                user.attributes.pop(USER_ATTRIBUTE_NEXT_ACTIONS, None)
+        elif value == slug:
+            user.attributes.pop(USER_ATTRIBUTE_NEXT_ACTIONS, None)
+        with audit_ignore():
+            user.save(update_fields=["attributes"])
+        Event.new(EventAction.NEXT_ACTION_COMPLETED, flow_slug=slug).from_http(
+            self.request, user=user
+        )
+        return self.executor.stage_ok()
+
+
 class UserLoginStageView(ChallengeStageView):
     """Finalize Authentication flow by logging the user in"""
 
@@ -64,8 +99,61 @@ class UserLoginStageView(ChallengeStageView):
     def get_challenge(self, *args, **kwargs) -> UserLoginChallenge:
         return UserLoginChallenge(data={})
 
+    def enforce_next_actions(self) -> HttpResponse | None:
+        """Splice the pending user's next action flows into the plan, ahead of this
+        login stage. Returns None when there is nothing to enforce."""
+        context = self.executor.plan.context
+        user: User | None = context.get(PLAN_CONTEXT_PENDING_USER)
+        if not user or not user.pk:
+            return None
+        if (
+            PLAN_CONTEXT_USER_SWITCH_ADD_USER in context
+            or PLAN_CONTEXT_USER_SWITCH_TARGET_SESSION in context
+        ):
+            return None
+        value = user.attributes.get(USER_ATTRIBUTE_NEXT_ACTIONS)
+        if not value:
+            return None
+        from authentik.enterprise.license import LicenseKey
+
+        if not LicenseKey.cached_summary().status.is_valid:
+            return None
+        error_message = _(
+            "Actions required for this login are invalid. Please contact your administrator."
+        )
+        try:
+            flows = resolve_next_actions(value)
+        except ValueError as exc:
+            self.logger.warning(
+                "Failed to resolve next actions", user=user.username, error=str(exc)
+            )
+            return self.executor.stage_invalid(error_message)
+        splice = FlowPlan(flow_pk=self.executor.plan.flow_pk)
+        for flow in flows:
+            planner = FlowPlanner(flow)
+            planner.use_cache = False
+            planner.allow_empty_flows = True
+            # The pending user has already passed this flow's authentication requirements
+            planner.check_authentication = False
+            try:
+                action_plan = planner.plan(self.request, context)
+            except FlowNonApplicableException:
+                self.logger.warning(
+                    "Next action flow not applicable to user", user=user.username, flow=flow.slug
+                )
+                return self.executor.stage_invalid(error_message)
+            splice.bindings.extend(action_plan.bindings)
+            splice.markers.extend(action_plan.markers)
+            splice.append_stage(in_memory_stage(NextActionDoneStageView, flow_slug=flow.slug))
+        # Run this login stage again once all actions are completed
+        splice.append(self.executor.plan.bindings[0], self.executor.plan.markers[0])
+        self.executor.plan.insert_plan(splice)
+        return self.executor.stage_ok()
+
     def dispatch(self, request: HttpRequest) -> HttpResponse:
-        """Check for remember_me, and do login"""
+        """Check for pending next actions and remember_me, and do login"""
+        if next_actions_response := self.enforce_next_actions():
+            return next_actions_response
         stage: UserLoginStage = self.executor.current_stage
         if timedelta_from_string(stage.remember_me_offset).total_seconds() > 0:
             return super().dispatch(request)