authentik.policies.event_matcher.models

  1"""Event Matcher models"""
  2
  3from itertools import chain
  4
  5from django.db import models
  6from django.utils.translation import gettext as _
  7from rest_framework.serializers import BaseSerializer
  8from structlog.stdlib import get_logger
  9
 10from authentik.events.models import Event, EventAction
 11from authentik.policies.models import Policy
 12from authentik.policies.types import PolicyRequest, PolicyResult
 13
 14LOGGER = get_logger()
 15
 16
 17class EventMatcherPolicy(Policy):
 18    """Passes when Event matches selected criteria."""
 19
 20    action = models.TextField(
 21        choices=EventAction.choices,
 22        null=True,
 23        default=None,
 24        help_text=_(
 25            "Match created events with this action type. "
 26            "When left empty, all action types will be matched."
 27        ),
 28    )
 29    app = models.TextField(
 30        null=True,
 31        default=None,
 32        help_text=_(
 33            "Match events created by selected application. "
 34            "When left empty, all applications are matched."
 35        ),
 36    )
 37    model = models.TextField(
 38        null=True,
 39        default=None,
 40        help_text=_(
 41            "Match events created by selected model. "
 42            "When left empty, all models are matched. When an app is selected, "
 43            "all the application's models are matched."
 44        ),
 45    )
 46    client_ip = models.TextField(
 47        null=True,
 48        default=None,
 49        help_text=_(
 50            "Matches Event's Client IP (strict matching, "
 51            "for network matching use an Expression Policy)"
 52        ),
 53    )
 54
 55    @property
 56    def serializer(self) -> type[BaseSerializer]:
 57        from authentik.policies.event_matcher.api import EventMatcherPolicySerializer
 58
 59        return EventMatcherPolicySerializer
 60
 61    @property
 62    def component(self) -> str:
 63        return "ak-policy-event-matcher-form"
 64
 65    def passes(self, request: PolicyRequest) -> PolicyResult:
 66        if "event" not in request.context:
 67            return PolicyResult(False)
 68        event: Event = request.context["event"]
 69        matches: list[PolicyResult] = []
 70        messages = []
 71        checks = [
 72            self.passes_action,
 73            self.passes_client_ip,
 74            self.passes_app,
 75            self.passes_model,
 76        ]
 77        for checker in checks:
 78            result = checker(request, event)
 79            if result is None:
 80                continue
 81            LOGGER.debug(
 82                "Event matcher check result",
 83                checker=checker.__name__,
 84                result=result,
 85            )
 86            matches.append(result)
 87        passing = all(x.passing for x in matches)
 88        messages = chain(*[x.messages for x in matches])
 89        result = PolicyResult(passing, *messages)
 90        result.source_results = matches
 91        return result
 92
 93    def passes_action(self, request: PolicyRequest, event: Event) -> PolicyResult | None:
 94        """Check if `self.action` matches"""
 95        if self.action is None:
 96            return None
 97        return PolicyResult(self.action == event.action, "Action matched.")
 98
 99    def passes_client_ip(self, request: PolicyRequest, event: Event) -> PolicyResult | None:
100        """Check if `self.client_ip` matches"""
101        if self.client_ip is None:
102            return None
103        return PolicyResult(self.client_ip == event.client_ip, "Client IP matched.")
104
105    def passes_app(self, request: PolicyRequest, event: Event) -> PolicyResult | None:
106        """Check if `self.app` matches"""
107        if self.app is None:
108            return None
109        return PolicyResult(self.app == event.app, "App matched.")
110
111    def passes_model(self, request: PolicyRequest, event: Event) -> PolicyResult | None:
112        """Check if `self.model` is set, and pass if it matches the event's model"""
113        if self.model is None:
114            return None
115        event_model_info = event.context.get("model", {})
116        event_model = f"{event_model_info.get('app')}.{event_model_info.get('model_name')}"
117        return PolicyResult(event_model == self.model, "Model matched.")
118
119    class Meta(Policy.PolicyMeta):
120        verbose_name = _("Event Matcher Policy")
121        verbose_name_plural = _("Event Matcher Policies")