authentik.lib.expression.evaluator

  1"""authentik expression policy evaluator"""
  2
  3import re
  4import socket
  5from ipaddress import ip_address, ip_network
  6from smtplib import SMTPException
  7from textwrap import indent
  8from types import CodeType
  9from typing import TYPE_CHECKING, Any
 10
 11from cachetools import TLRUCache, cached
 12from django.core.exceptions import FieldError
 13from django.http import HttpRequest
 14from django.utils.text import slugify
 15from django.utils.timezone import now
 16from guardian.shortcuts import get_anonymous_user
 17from rest_framework.serializers import ValidationError
 18from sentry_sdk import start_span
 19from sentry_sdk.tracing import Span
 20from structlog.stdlib import get_logger
 21
 22from authentik.core.models import User
 23from authentik.events.models import Event
 24from authentik.lib.expression.exceptions import ControlFlowException
 25from authentik.lib.utils.email import normalize_addresses
 26from authentik.lib.utils.http import get_http_session
 27from authentik.lib.utils.time import timedelta_from_string
 28from authentik.policies.models import Policy, PolicyBinding
 29from authentik.policies.process import PolicyProcess
 30from authentik.policies.types import PolicyRequest, PolicyResult
 31from authentik.providers.oauth2.id_token import IDToken
 32from authentik.providers.oauth2.models import AccessToken, OAuth2Provider
 33from authentik.stages.authenticator import devices_for_user
 34from authentik.stages.email.utils import TemplateEmailMessage
 35
 36if TYPE_CHECKING:
 37    from authentik.stages.email.models import EmailStage
 38
 39LOGGER = get_logger()
 40
 41ARG_SANITIZE = re.compile(r"[:.-]")
 42
 43
 44def sanitize_arg(arg_name: str) -> str:
 45    return re.sub(ARG_SANITIZE, "_", slugify(arg_name))
 46
 47
 48class BaseEvaluator:
 49    """Validate and evaluate python-based expressions"""
 50
 51    # Globals that can be used by function
 52    _globals: dict[str, Any]
 53    # Context passed as locals to exec()
 54    _context: dict[str, Any]
 55
 56    # Filename used for exec
 57    _filename: str
 58
 59    def __init__(self, filename: str | None = None):
 60        self._filename = filename if filename else "BaseEvaluator"
 61        # update website/docs/expressions/_objects.md
 62        # update website/docs/expressions/_functions.md
 63        self._globals = {
 64            "ak_call_policy": self.expr_func_call_policy,
 65            "ak_create_event": self.expr_event_create,
 66            "ak_create_jwt": self.expr_create_jwt,
 67            "ak_create_jwt_raw": self.expr_create_jwt_raw,
 68            "ak_is_group_member": BaseEvaluator.expr_is_group_member,
 69            "ak_logger": get_logger(self._filename).bind(),
 70            "ak_send_email": self.expr_send_email,
 71            "ak_user_by": BaseEvaluator.expr_user_by,
 72            "ak_user_has_authenticator": BaseEvaluator.expr_func_user_has_authenticator,
 73            "ip_address": ip_address,
 74            "ip_network": ip_network,
 75            "list_flatten": BaseEvaluator.expr_flatten,
 76            "regex_match": BaseEvaluator.expr_regex_match,
 77            "regex_replace": BaseEvaluator.expr_regex_replace,
 78            "requests": get_http_session(),
 79            "resolve_dns": BaseEvaluator.expr_resolve_dns,
 80            "reverse_dns": BaseEvaluator.expr_reverse_dns,
 81            "slugify": slugify,
 82        }
 83        self._context = {}
 84
 85    @cached(cache=TLRUCache(maxsize=32, ttu=lambda key, value, now: now + 180))
 86    @staticmethod
 87    def expr_resolve_dns(host: str, ip_version: int | None = None) -> list[str]:
 88        """Resolve host to a list of IPv4 and/or IPv6 addresses."""
 89        # Although it seems to be fine (raising OSError), docs warn
 90        # against passing `None` for both the host and the port
 91        # https://docs.python.org/3/library/socket.html#socket.getaddrinfo
 92        host = host or ""
 93
 94        ip_list = []
 95
 96        family = 0
 97        if ip_version == 4:  # noqa: PLR2004
 98            family = socket.AF_INET
 99        if ip_version == 6:  # noqa: PLR2004
100            family = socket.AF_INET6
101
102        try:
103            for ip_addr in socket.getaddrinfo(host, None, family=family):
104                ip_list.append(str(ip_addr[4][0]))
105        except OSError:
106            pass
107        return list(set(ip_list))
108
109    @cached(cache=TLRUCache(maxsize=32, ttu=lambda key, value, now: now + 180))
110    @staticmethod
111    def expr_reverse_dns(ip_addr: str) -> str:
112        """Perform a reverse DNS lookup."""
113        try:
114            return socket.getfqdn(ip_addr)
115        except OSError:
116            return ip_addr
117
118    @staticmethod
119    def expr_flatten(value: list[Any] | Any) -> Any | None:
120        """Flatten `value` if its a list"""
121        if isinstance(value, list):
122            if len(value) < 1:
123                return None
124            return value[0]
125        return value
126
127    @staticmethod
128    def expr_regex_match(value: Any, regex: str) -> bool:
129        """Expression Filter to run re.search"""
130        return re.search(regex, value) is not None
131
132    @staticmethod
133    def expr_regex_replace(value: Any, regex: str, repl: str) -> str:
134        """Expression Filter to run re.sub"""
135        return re.sub(regex, repl, value)
136
137    @staticmethod
138    def expr_is_group_member(user: User, **group_filters) -> bool:
139        """Check if `user` is member of group with name `group_name`"""
140        return user.all_groups().filter(**group_filters).exists()
141
142    @staticmethod
143    def expr_user_by(**filters) -> User | None:
144        """Get user by filters"""
145        try:
146            users = User.objects.filter(**filters)
147            if users:
148                return users.first()
149            return None
150        except FieldError:
151            return None
152
153    @staticmethod
154    def expr_func_user_has_authenticator(user: User, device_type: str | None = None) -> bool:
155        """Check if a user has any authenticator devices, optionally matching *device_type*"""
156        user_devices = devices_for_user(user)
157        if device_type:
158            for device in user_devices:
159                device_class = device.__class__.__name__.lower().replace("device", "")
160                if device_class == device_type:
161                    return True
162            return False
163        return len(list(user_devices)) > 0
164
165    def expr_event_create(self, action: str, **kwargs):
166        """Create event with supplied data and try to extract as much relevant data
167        from the context"""
168        context = self._context.copy()
169        # If the result was a complex variable, we don't want to reuse it
170        context.pop("result", None)
171        context.pop("handler", None)
172        event_kwargs = context
173        event_kwargs.update(kwargs)
174        event = Event.new(
175            action,
176            app=self._filename,
177            **event_kwargs,
178        )
179        if "request" in context and isinstance(context["request"], PolicyRequest):
180            policy_request: PolicyRequest = context["request"]
181            if policy_request.http_request:
182                event.from_http(policy_request.http_request)
183                return
184        event.save()
185
186    def expr_func_call_policy(self, name: str, **kwargs) -> PolicyResult:
187        """Call policy by name, with current request"""
188        policy = Policy.objects.filter(name=name).select_subclasses().first()
189        if not policy:
190            raise ValueError(f"Policy '{name}' not found.")
191        user = self._context.get("user", get_anonymous_user())
192        req = PolicyRequest(user)
193        if "request" in self._context:
194            req = self._context["request"]
195        req.context.update(kwargs)
196        proc = PolicyProcess(PolicyBinding(policy=policy), request=req, connection=None)
197        return proc.profiling_wrapper()
198
199    def expr_create_jwt(
200        self,
201        user: User,
202        provider: OAuth2Provider | str,
203        scopes: list[str],
204        validity: str = "seconds=60",
205    ) -> str | None:
206        """Issue a JWT for a given provider"""
207        request: HttpRequest | None = self._context.get("http_request")
208        if not request:
209            return None
210        if not isinstance(provider, OAuth2Provider):
211            provider = OAuth2Provider.objects.get(name=provider)
212        session = None
213        if hasattr(request, "session") and request.session.session_key:
214            session = request.session["authenticatedsession"]
215        access_token = AccessToken(
216            provider=provider,
217            user=user,
218            expires=now() + timedelta_from_string(validity),
219            scope=scopes,
220            auth_time=now(),
221            session=session,
222        )
223        access_token.id_token = IDToken.new(provider, access_token, request)
224        access_token.save()
225        return access_token.token
226
227    def expr_create_jwt_raw(
228        self, provider: OAuth2Provider | str, validity: str = "seconds=60", **kwargs
229    ) -> str:
230        """Issue a JWT for a given provider with completely customized data"""
231        if not isinstance(provider, OAuth2Provider):
232            provider = OAuth2Provider.objects.get(name=provider)
233        kwargs["exp"] = int((now() + timedelta_from_string(validity)).timestamp())
234        kwargs["aud"] = provider.client_id
235        return provider.encode(kwargs)
236
237    def expr_send_email(  # noqa: PLR0913
238        self,
239        address: str | list[str],
240        subject: str,
241        body: str | None = None,
242        stage: EmailStage | None = None,
243        template: str | None = None,
244        context: dict | None = None,
245        cc: str | list[str] | None = None,
246        bcc: str | list[str] | None = None,
247    ) -> bool:
248        """Send an email using authentik's email system
249
250        Args:
251            address: Email address(es) to send to. Can be:
252                - Single email: "user@example.com"
253                - List of emails: ["user1@example.com", "user2@example.com"]
254            subject: Email subject
255            body: Email body (plain text/HTML). Mutually exclusive with template.
256            stage: EmailStage instance to use for settings. If None, uses global settings.
257            template: Template name to render. Mutually exclusive with body.
258            context: Additional context variables for template rendering.
259            cc: Email address(es) to CC. Same format as address.
260            bcc: Email address(es) to BCC. Same format as address.
261
262        Returns:
263            bool: True if email was queued successfully, False otherwise
264        """
265        # Deferred imports to avoid circular import issues
266        from authentik.stages.email.tasks import send_mails
267
268        if body and template:
269            raise ValueError("body and template parameters are mutually exclusive")
270
271        if not body and not template:
272            raise ValueError("Either body or template parameter must be provided")
273
274        to_addresses = normalize_addresses(address)
275        cc_addresses = normalize_addresses(cc)
276        bcc_addresses = normalize_addresses(bcc)
277
278        try:
279            if template is not None:
280                # Use all available context from the evaluator for template rendering
281                template_context = self._context.copy()
282                # Add any custom context passed to the function
283                if context:
284                    template_context.update(context)
285
286                # Use template rendering
287                message = TemplateEmailMessage(
288                    subject=subject,
289                    to=to_addresses,
290                    cc=cc_addresses,
291                    bcc=bcc_addresses,
292                    template_name=template,
293                    template_context=template_context,
294                )
295            else:
296                # Use plain body
297                message = TemplateEmailMessage(
298                    subject=subject,
299                    to=to_addresses,
300                    cc=cc_addresses,
301                    bcc=bcc_addresses,
302                    body=body,
303                )
304
305            send_mails(stage, message)
306            return True
307
308        except (SMTPException, ConnectionError, ValidationError, ValueError) as exc:
309            LOGGER.warning("Failed to send email", exc=exc, addresses=to_addresses, subject=subject)
310            return False
311
312    def wrap_expression(self, expression: str) -> str:
313        """Wrap expression in a function, call it, and save the result as `result`"""
314        handler_signature = ",".join(
315            [x for x in [sanitize_arg(x) for x in self._context.keys()] if x]
316        )
317        full_expression = ""
318        full_expression += f"def handler({handler_signature}):\n"
319        full_expression += indent(expression, "    ")
320        full_expression += f"\nresult = handler({handler_signature})"
321        return full_expression
322
323    def compile(self, expression: str) -> CodeType:
324        """Parse expression. Raises SyntaxError or ValueError if the syntax is incorrect."""
325        expression = self.wrap_expression(expression)
326        return compile(expression, self._filename, "exec")
327
328    def evaluate(self, expression_source: str) -> Any:
329        """Parse and evaluate expression. If the syntax is incorrect, a SyntaxError is raised.
330        If any exception is raised during execution, it is raised.
331        The result is returned without any type-checking."""
332        with start_span(op="authentik.lib.evaluator.evaluate") as span:
333            span: Span
334            span.description = self._filename
335            span.set_data("expression", expression_source)
336            try:
337                ast_obj = self.compile(expression_source)
338            except (SyntaxError, ValueError) as exc:
339                self.handle_error(exc, expression_source)
340                raise exc
341            try:
342                _locals = {sanitize_arg(x): y for x, y in self._context.items()}
343                # Yes this is an exec, yes it is potentially bad. Since we limit what variables are
344                # available here, and these policies can only be edited by admins, this is a risk
345                # we're willing to take.
346
347                exec(ast_obj, self._globals, _locals)  # nosec # noqa
348                result = _locals["result"]
349            except Exception as exc:
350                # So, this is a bit questionable. Essentially, we are edit the stacktrace
351                # so the user only sees information relevant to them
352                # and none of our surrounding error handling
353                if exc.__traceback__ is not None:
354                    exc.__traceback__ = exc.__traceback__.tb_next
355                if not isinstance(exc, ControlFlowException):
356                    self.handle_error(exc, expression_source)
357                raise exc
358            return result
359
360    def handle_error(self, exc: Exception, expression_source: str):  # pragma: no cover
361        """Exception Handler"""
362        LOGGER.warning("Expression error", exc=exc)
363
364    def validate(self, expression: str) -> bool:
365        """Validate expression's syntax, raise ValidationError if Syntax is invalid"""
366        try:
367            self.compile(expression)
368            return True
369        except (ValueError, SyntaxError) as exc:
370            raise ValidationError(f"Expression Syntax Error: {str(exc)}") from exc