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