authentik.flows.api.flows_diagram

  1"""Flows Diagram API"""
  2
  3from dataclasses import dataclass, field
  4
  5from django.utils.translation import gettext as _
  6from guardian.shortcuts import get_objects_for_user
  7from rest_framework.serializers import CharField
  8
  9from authentik.core.api.utils import PassiveSerializer
 10from authentik.core.models import User
 11from authentik.flows.models import Flow, FlowAuthenticationRequirement, FlowStageBinding
 12
 13
 14@dataclass
 15class DiagramElement:
 16    """Single element used in a diagram"""
 17
 18    identifier: str
 19    description: str
 20    action: str | None = None
 21    source: list[DiagramElement] | None = None
 22
 23    style: list[str] = field(default_factory=lambda: ["[", "]"])
 24
 25    def __str__(self) -> str:
 26        description = self.description.replace('"', "#quot;")
 27        element = f'{self.identifier}{self.style[0]}"{description}"{self.style[1]}'
 28        if self.action is not None:
 29            if self.action != "":
 30                element = f"--{self.action}--> {element}"
 31            else:
 32                element = f"--> {element}"
 33        if self.source:
 34            source_element = []
 35            for source in self.source:
 36                source_element.append(f"{source.identifier} {element}")
 37            return "\n".join(source_element)
 38        return element
 39
 40
 41class FlowDiagramSerializer(PassiveSerializer):
 42    """response of the flow's diagram action"""
 43
 44    diagram = CharField(read_only=True)
 45
 46
 47class FlowDiagram:
 48    """Generate flow chart fow a flow"""
 49
 50    flow: Flow
 51    user: User
 52
 53    def __init__(self, flow: Flow, user: User) -> None:
 54        self.flow = flow
 55        self.user = user
 56
 57    def get_flow_policies(self, parent_elements: list[DiagramElement]) -> list[DiagramElement]:
 58        """Collect all policies bound to the flow"""
 59        elements = []
 60        for p_index, policy_binding in enumerate(
 61            get_objects_for_user(self.user, "authentik_policies.view_policybinding")
 62            .filter(target=self.flow)
 63            .exclude(policy__isnull=True)
 64            .order_by("order")
 65        ):
 66            element = DiagramElement(
 67                f"flow_policy_{p_index}",
 68                _("Policy ({type})".format_map({"type": policy_binding.policy._meta.verbose_name}))
 69                + "\n"
 70                + policy_binding.policy.name,
 71                _("Binding {order}".format_map({"order": policy_binding.order})),
 72                parent_elements,
 73                style=["{{", "}}"],
 74            )
 75            elements.append(element)
 76        return elements
 77
 78    def get_stage_policies(
 79        self,
 80        stage_index: int,
 81        stage_binding: FlowStageBinding,
 82        parent_elements: list[DiagramElement],
 83    ) -> list[DiagramElement]:
 84        """First all policies bound to stages since they execute before stages"""
 85        elements = []
 86        for p_index, policy_binding in enumerate(
 87            get_objects_for_user(self.user, "authentik_policies.view_policybinding")
 88            .filter(target=stage_binding)
 89            .exclude(policy__isnull=True)
 90            .order_by("order")
 91        ):
 92            element = DiagramElement(
 93                f"stage_{stage_index}_policy_{p_index}",
 94                _("Policy ({type})".format_map({"type": policy_binding.policy._meta.verbose_name}))
 95                + "\n"
 96                + policy_binding.policy.name,
 97                "",
 98                parent_elements,
 99                style=["{{", "}}"],
100            )
101            elements.append(element)
102        return elements
103
104    def get_stages(self, parent_elements: list[DiagramElement]) -> list[str | DiagramElement]:
105        """Collect all stages"""
106        elements = []
107        stages = []
108        for s_index, stage_binding in enumerate(
109            get_objects_for_user(self.user, "authentik_flows.view_flowstagebinding")
110            .filter(target=self.flow)
111            .order_by("order")
112        ):
113            stage_policies = self.get_stage_policies(s_index, stage_binding, parent_elements)
114            elements.extend(stage_policies)
115
116            action = ""
117            if len(stage_policies) > 0:
118                action = _("Policy passed")
119
120            element = DiagramElement(
121                f"stage_{s_index}",
122                _("Stage ({type})".format_map({"type": stage_binding.stage._meta.verbose_name}))
123                + "\n"
124                + stage_binding.stage.name,
125                action,
126                stage_policies,
127                style=["([", "])"],
128            )
129            stages.append(element)
130
131            parent_elements = [element]
132
133            # This adds connections for policy denies, but retroactively, as we can't really
134            # look ahead
135            # Check if we have a stage behind us and if it has any sources
136            if s_index > 0:
137                last_stage: DiagramElement = stages[s_index - 1]
138                if last_stage.source and len(last_stage.source) > 0:
139                    # If it has any sources, add a connection from each of that stage's sources
140                    # to this stage
141                    for source in last_stage.source:
142                        elements.append(
143                            DiagramElement(
144                                element.identifier,
145                                element.description,
146                                _("Policy denied"),
147                                [source],
148                                style=element.style,
149                            )
150                        )
151
152        if len(stages) > 0:
153            elements.append(
154                DiagramElement(
155                    "done",
156                    _("End of the flow"),
157                    "",
158                    [stages[-1]],
159                    style=["[[", "]]"],
160                ),
161            )
162        return stages + elements
163
164    def get_flow_auth_requirement(self) -> list[DiagramElement]:
165        """Get flow authentication requirement"""
166        end_el = DiagramElement(
167            "done",
168            _("End of the flow"),
169            _("Requirement not fulfilled"),
170            style=["[[", "]]"],
171        )
172        elements = []
173        if self.flow.authentication == FlowAuthenticationRequirement.NONE:
174            return []
175        auth = DiagramElement(
176            "flow_auth_requirement",
177            _("Flow authentication requirement") + "\n" + self.flow.authentication,
178        )
179        elements.append(auth)
180        end_el.source = [auth]
181        elements.append(end_el)
182        elements.append(
183            DiagramElement("flow_start", "placeholder", _("Requirement fulfilled"), source=[auth])
184        )
185        return elements
186
187    def build(self) -> str:
188        """Build flowchart"""
189        all_elements = [
190            "graph TD",
191        ]
192
193        all_elements.extend(self.get_flow_auth_requirement())
194
195        pre_flow_policies_element = DiagramElement(
196            "flow_pre", _("Pre-flow policies"), style=["[[", "]]"]
197        )
198        flow_policies = self.get_flow_policies([pre_flow_policies_element])
199        if len(flow_policies) > 0:
200            all_elements.append(pre_flow_policies_element)
201            all_elements.extend(flow_policies)
202            all_elements.append(
203                DiagramElement(
204                    "done",
205                    _("End of the flow"),
206                    _("Policy denied"),
207                    flow_policies,
208                    style=["[[", "]]"],
209                )
210            )
211
212        flow_element = DiagramElement(
213            "flow_start",
214            _("Flow") + "\n" + self.flow.name,
215            "" if len(flow_policies) > 0 else None,
216            source=flow_policies,
217            style=["[[", "]]"],
218        )
219        all_elements.append(flow_element)
220
221        stages = self.get_stages([flow_element])
222        all_elements.extend(stages)
223        if len(stages) < 1:
224            all_elements.append(
225                DiagramElement(
226                    "done",
227                    _("End of the flow"),
228                    "",
229                    [flow_element],
230                    style=["[[", "]]"],
231                ),
232            )
233        return "\n".join([str(x) for x in all_elements])