We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
DNS Resolution in Expressions
**Is your feature request related to a problem? Please describe.**
I would like to be able to resolve a hostname in expressions. In particular, my docker container resolves `host.docker.internal` to network address of the machine the container is running on. I would like to be able to check for requests from this hostname, but there is no way to detect this in expressions.

**Describe the solution you'd like**
I would like a function to be exposed in expressions that resolves a hostname to an address or list of addresses. Such a function could be implemented in `authentik/lib/expression/evaluator.py` like this:
```py
    def expr_resolve_host(host: str) -> List[str]:
        """Resolve a hostname to a list of IP addresses."""
        
        return [
            sockaddr[0]
            for family, type, proto, canonname, sockaddr in 
            socket.getaddrinfo(host, None)
        ]
```

**Describe alternatives you've considered**
It would be possible for me to set up the docker network statically, and then hardcode the address, but this is fragile, and I would prefer for Docker to manage its network allocation.

**Additional context**
I currently have an expression that checks for requests for my local/trusted networks. This works well, but if I access Authentik directly from the docker host, the source address is the "host gateway" address of the docker container's network, which is not in my trusted networks. (The host address of the docker internal network.) I can imagine this functionality would also be useful for getting the addresses other containers the docker container is connected to.

At my scale and use case, it doesn't really matter, but it may be worth caching this information for larger installations. I'm not sure if `socket.getaddrinfo` does any caching of it's own, but if it decides to do real DNS resolution it could be quite slow. If an expression is getting hit tens or hundreds of times a minute this could be a serious issue. (This could alternatively be resolved with a local DNS caching server on the host.)

Any caching should be conservatively short, at most 60 seconds or so, since an change in address could also cause serious problems. A timeout may also be in order, since some use cases would prefer a fast empty response over an always correct answer. Ideally, these would be configurable with function parameters so users can determine their own caching and timeout needs.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `authentik/lib/expression/evaluator.py`
Content:
```
1 """authentik expression policy evaluator"""
2 import re
3 from ipaddress import ip_address, ip_network
4 from textwrap import indent
5 from typing import Any, Iterable, Optional
6 
7 from django.core.exceptions import FieldError
8 from django_otp import devices_for_user
9 from rest_framework.serializers import ValidationError
10 from sentry_sdk.hub import Hub
11 from sentry_sdk.tracing import Span
12 from structlog.stdlib import get_logger
13 
14 from authentik.core.models import User
15 from authentik.events.models import Event
16 from authentik.lib.utils.http import get_http_session
17 from authentik.policies.types import PolicyRequest
18 
19 LOGGER = get_logger()
20 
21 
22 class BaseEvaluator:
23     """Validate and evaluate python-based expressions"""
24 
25     # Globals that can be used by function
26     _globals: dict[str, Any]
27     # Context passed as locals to exec()
28     _context: dict[str, Any]
29 
30     # Filename used for exec
31     _filename: str
32 
33     def __init__(self, filename: Optional[str] = None):
34         self._filename = filename if filename else "BaseEvaluator"
35         # update website/docs/expressions/_objects.md
36         # update website/docs/expressions/_functions.md
37         self._globals = {
38             "regex_match": BaseEvaluator.expr_regex_match,
39             "regex_replace": BaseEvaluator.expr_regex_replace,
40             "list_flatten": BaseEvaluator.expr_flatten,
41             "ak_is_group_member": BaseEvaluator.expr_is_group_member,
42             "ak_user_by": BaseEvaluator.expr_user_by,
43             "ak_user_has_authenticator": BaseEvaluator.expr_func_user_has_authenticator,
44             "ak_create_event": self.expr_event_create,
45             "ak_logger": get_logger(self._filename).bind(),
46             "requests": get_http_session(),
47             "ip_address": ip_address,
48             "ip_network": ip_network,
49         }
50         self._context = {}
51 
52     @staticmethod
53     def expr_flatten(value: list[Any] | Any) -> Optional[Any]:
54         """Flatten `value` if its a list"""
55         if isinstance(value, list):
56             if len(value) < 1:
57                 return None
58             return value[0]
59         return value
60 
61     @staticmethod
62     def expr_regex_match(value: Any, regex: str) -> bool:
63         """Expression Filter to run re.search"""
64         return re.search(regex, value) is not None
65 
66     @staticmethod
67     def expr_regex_replace(value: Any, regex: str, repl: str) -> str:
68         """Expression Filter to run re.sub"""
69         return re.sub(regex, repl, value)
70 
71     @staticmethod
72     def expr_is_group_member(user: User, **group_filters) -> bool:
73         """Check if `user` is member of group with name `group_name`"""
74         return user.ak_groups.filter(**group_filters).exists()
75 
76     @staticmethod
77     def expr_user_by(**filters) -> Optional[User]:
78         """Get user by filters"""
79         try:
80             users = User.objects.filter(**filters)
81             if users:
82                 return users.first()
83             return None
84         except FieldError:
85             return None
86 
87     @staticmethod
88     def expr_func_user_has_authenticator(user: User, device_type: Optional[str] = None) -> bool:
89         """Check if a user has any authenticator devices, optionally matching *device_type*"""
90         user_devices = devices_for_user(user)
91         if device_type:
92             for device in user_devices:
93                 device_class = device.__class__.__name__.lower().replace("device", "")
94                 if device_class == device_type:
95                     return True
96             return False
97         return len(list(user_devices)) > 0
98 
99     def expr_event_create(self, action: str, **kwargs):
100         """Create event with supplied data and try to extract as much relevant data
101         from the context"""
102         # If the result was a complex variable, we don't want to re-use it
103         self._context.pop("result", None)
104         self._context.pop("handler", None)
105         kwargs["context"] = self._context
106         event = Event.new(
107             action,
108             app=self._filename,
109             **kwargs,
110         )
111         if "request" in self._context and isinstance(self._context["request"], PolicyRequest):
112             policy_request: PolicyRequest = self._context["request"]
113             if policy_request.http_request:
114                 event.from_http(policy_request)
115                 return
116         event.save()
117 
118     def wrap_expression(self, expression: str, params: Iterable[str]) -> str:
119         """Wrap expression in a function, call it, and save the result as `result`"""
120         handler_signature = ",".join(params)
121         full_expression = ""
122         full_expression += f"def handler({handler_signature}):\n"
123         full_expression += indent(expression, "    ")
124         full_expression += f"\nresult = handler({handler_signature})"
125         return full_expression
126 
127     def evaluate(self, expression_source: str) -> Any:
128         """Parse and evaluate expression. If the syntax is incorrect, a SyntaxError is raised.
129         If any exception is raised during execution, it is raised.
130         The result is returned without any type-checking."""
131         with Hub.current.start_span(op="authentik.lib.evaluator.evaluate") as span:
132             span: Span
133             span.description = self._filename
134             span.set_data("expression", expression_source)
135             param_keys = self._context.keys()
136             try:
137                 ast_obj = compile(
138                     self.wrap_expression(expression_source, param_keys),
139                     self._filename,
140                     "exec",
141                 )
142             except (SyntaxError, ValueError) as exc:
143                 self.handle_error(exc, expression_source)
144                 raise exc
145             try:
146                 _locals = self._context
147                 # Yes this is an exec, yes it is potentially bad. Since we limit what variables are
148                 # available here, and these policies can only be edited by admins, this is a risk
149                 # we're willing to take.
150                 # pylint: disable=exec-used
151                 exec(ast_obj, self._globals, _locals)  # nosec # noqa
152                 result = _locals["result"]
153             except Exception as exc:
154                 # So, this is a bit questionable. Essentially, we are edit the stacktrace
155                 # so the user only sees information relevant to them
156                 # and none of our surrounding error handling
157                 exc.__traceback__ = exc.__traceback__.tb_next
158                 self.handle_error(exc, expression_source)
159                 raise exc
160             return result
161 
162     def handle_error(self, exc: Exception, expression_source: str):  # pragma: no cover
163         """Exception Handler"""
164         LOGGER.warning("Expression error", exc=exc)
165 
166     def validate(self, expression: str) -> bool:
167         """Validate expression's syntax, raise ValidationError if Syntax is invalid"""
168         param_keys = self._context.keys()
169         try:
170             compile(
171                 self.wrap_expression(expression, param_keys),
172                 self._filename,
173                 "exec",
174             )
175             return True
176         except (ValueError, SyntaxError) as exc:
177             raise ValidationError(f"Expression Syntax Error: {str(exc)}") from exc
178
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/authentik/lib/expression/evaluator.py b/authentik/lib/expression/evaluator.py
--- a/authentik/lib/expression/evaluator.py
+++ b/authentik/lib/expression/evaluator.py
@@ -1,9 +1,11 @@
 """authentik expression policy evaluator"""
 import re
+import socket
 from ipaddress import ip_address, ip_network
 from textwrap import indent
 from typing import Any, Iterable, Optional
 
+from cachetools import TLRUCache, cached
 from django.core.exceptions import FieldError
 from django_otp import devices_for_user
 from rest_framework.serializers import ValidationError
@@ -41,6 +43,8 @@
             "ak_is_group_member": BaseEvaluator.expr_is_group_member,
             "ak_user_by": BaseEvaluator.expr_user_by,
             "ak_user_has_authenticator": BaseEvaluator.expr_func_user_has_authenticator,
+            "resolve_dns": BaseEvaluator.expr_resolve_dns,
+            "reverse_dns": BaseEvaluator.expr_reverse_dns,
             "ak_create_event": self.expr_event_create,
             "ak_logger": get_logger(self._filename).bind(),
             "requests": get_http_session(),
@@ -49,6 +53,39 @@
         }
         self._context = {}
 
+    @cached(cache=TLRUCache(maxsize=32, ttu=lambda key, value, now: now + 180))
+    @staticmethod
+    def expr_resolve_dns(host: str, ip_version: Optional[int] = None) -> list[str]:
+        """Resolve host to a list of IPv4 and/or IPv6 addresses."""
+        # Although it seems to be fine (raising OSError), docs warn
+        # against passing `None` for both the host and the port
+        # https://docs.python.org/3/library/socket.html#socket.getaddrinfo
+        host = host or ""
+
+        ip_list = []
+
+        family = 0
+        if ip_version == 4:
+            family = socket.AF_INET
+        if ip_version == 6:
+            family = socket.AF_INET6
+
+        try:
+            for ip_addr in socket.getaddrinfo(host, None, family=family):
+                ip_list.append(str(ip_addr[4][0]))
+        except OSError:
+            pass
+        return list(set(ip_list))
+
+    @cached(cache=TLRUCache(maxsize=32, ttu=lambda key, value, now: now + 180))
+    @staticmethod
+    def expr_reverse_dns(ip_addr: str) -> str:
+        """Perform a reverse DNS lookup."""
+        try:
+            return socket.getfqdn(ip_addr)
+        except OSError:
+            return ip_addr
+
     @staticmethod
     def expr_flatten(value: list[Any] | Any) -> Optional[Any]:
         """Flatten `value` if its a list"""

{"golden_diff": "diff --git a/authentik/lib/expression/evaluator.py b/authentik/lib/expression/evaluator.py\n--- a/authentik/lib/expression/evaluator.py\n+++ b/authentik/lib/expression/evaluator.py\n@@ -1,9 +1,11 @@\n \"\"\"authentik expression policy evaluator\"\"\"\n import re\n+import socket\n from ipaddress import ip_address, ip_network\n from textwrap import indent\n from typing import Any, Iterable, Optional\n \n+from cachetools import TLRUCache, cached\n from django.core.exceptions import FieldError\n from django_otp import devices_for_user\n from rest_framework.serializers import ValidationError\n@@ -41,6 +43,8 @@\n             \"ak_is_group_member\": BaseEvaluator.expr_is_group_member,\n             \"ak_user_by\": BaseEvaluator.expr_user_by,\n             \"ak_user_has_authenticator\": BaseEvaluator.expr_func_user_has_authenticator,\n+            \"resolve_dns\": BaseEvaluator.expr_resolve_dns,\n+            \"reverse_dns\": BaseEvaluator.expr_reverse_dns,\n             \"ak_create_event\": self.expr_event_create,\n             \"ak_logger\": get_logger(self._filename).bind(),\n             \"requests\": get_http_session(),\n@@ -49,6 +53,39 @@\n         }\n         self._context = {}\n \n+    @cached(cache=TLRUCache(maxsize=32, ttu=lambda key, value, now: now + 180))\n+    @staticmethod\n+    def expr_resolve_dns(host: str, ip_version: Optional[int] = None) -> list[str]:\n+        \"\"\"Resolve host to a list of IPv4 and/or IPv6 addresses.\"\"\"\n+        # Although it seems to be fine (raising OSError), docs warn\n+        # against passing `None` for both the host and the port\n+        # https://docs.python.org/3/library/socket.html#socket.getaddrinfo\n+        host = host or \"\"\n+\n+        ip_list = []\n+\n+        family = 0\n+        if ip_version == 4:\n+            family = socket.AF_INET\n+        if ip_version == 6:\n+            family = socket.AF_INET6\n+\n+        try:\n+            for ip_addr in socket.getaddrinfo(host, None, family=family):\n+                ip_list.append(str(ip_addr[4][0]))\n+        except OSError:\n+            pass\n+        return list(set(ip_list))\n+\n+    @cached(cache=TLRUCache(maxsize=32, ttu=lambda key, value, now: now + 180))\n+    @staticmethod\n+    def expr_reverse_dns(ip_addr: str) -> str:\n+        \"\"\"Perform a reverse DNS lookup.\"\"\"\n+        try:\n+            return socket.getfqdn(ip_addr)\n+        except OSError:\n+            return ip_addr\n+\n     @staticmethod\n     def expr_flatten(value: list[Any] | Any) -> Optional[Any]:\n         \"\"\"Flatten `value` if its a list\"\"\"\n", "issue": "DNS Resolution in Expressions\n**Is your feature request related to a problem? Please describe.**\r\nI would like to be able to resolve a hostname in expressions. In particular, my docker container resolves `host.docker.internal` to network address of the machine the container is running on. I would like to be able to check for requests from this hostname, but there is no way to detect this in expressions.\r\n\r\n**Describe the solution you'd like**\r\nI would like a function to be exposed in expressions that resolves a hostname to an address or list of addresses. Such a function could be implemented in `authentik/lib/expression/evaluator.py` like this:\r\n```py\r\n    def expr_resolve_host(host: str) -> List[str]:\r\n        \"\"\"Resolve a hostname to a list of IP addresses.\"\"\"\r\n        \r\n        return [\r\n            sockaddr[0]\r\n            for family, type, proto, canonname, sockaddr in \r\n            socket.getaddrinfo(host, None)\r\n        ]\r\n```\r\n\r\n**Describe alternatives you've considered**\r\nIt would be possible for me to set up the docker network statically, and then hardcode the address, but this is fragile, and I would prefer for Docker to manage its network allocation.\r\n\r\n**Additional context**\r\nI currently have an expression that checks for requests for my local/trusted networks. This works well, but if I access Authentik directly from the docker host, the source address is the \"host gateway\" address of the docker container's network, which is not in my trusted networks. (The host address of the docker internal network.) I can imagine this functionality would also be useful for getting the addresses other containers the docker container is connected to.\r\n\r\nAt my scale and use case, it doesn't really matter, but it may be worth caching this information for larger installations. I'm not sure if `socket.getaddrinfo` does any caching of it's own, but if it decides to do real DNS resolution it could be quite slow. If an expression is getting hit tens or hundreds of times a minute this could be a serious issue. (This could alternatively be resolved with a local DNS caching server on the host.)\r\n\r\nAny caching should be conservatively short, at most 60 seconds or so, since an change in address could also cause serious problems. A timeout may also be in order, since some use cases would prefer a fast empty response over an always correct answer. Ideally, these would be configurable with function parameters so users can determine their own caching and timeout needs.\n", "before_files": [{"content": "\"\"\"authentik expression policy evaluator\"\"\"\nimport re\nfrom ipaddress import ip_address, ip_network\nfrom textwrap import indent\nfrom typing import Any, Iterable, Optional\n\nfrom django.core.exceptions import FieldError\nfrom django_otp import devices_for_user\nfrom rest_framework.serializers import ValidationError\nfrom sentry_sdk.hub import Hub\nfrom sentry_sdk.tracing import Span\nfrom structlog.stdlib import get_logger\n\nfrom authentik.core.models import User\nfrom authentik.events.models import Event\nfrom authentik.lib.utils.http import get_http_session\nfrom authentik.policies.types import PolicyRequest\n\nLOGGER = get_logger()\n\n\nclass BaseEvaluator:\n    \"\"\"Validate and evaluate python-based expressions\"\"\"\n\n    # Globals that can be used by function\n    _globals: dict[str, Any]\n    # Context passed as locals to exec()\n    _context: dict[str, Any]\n\n    # Filename used for exec\n    _filename: str\n\n    def __init__(self, filename: Optional[str] = None):\n        self._filename = filename if filename else \"BaseEvaluator\"\n        # update website/docs/expressions/_objects.md\n        # update website/docs/expressions/_functions.md\n        self._globals = {\n            \"regex_match\": BaseEvaluator.expr_regex_match,\n            \"regex_replace\": BaseEvaluator.expr_regex_replace,\n            \"list_flatten\": BaseEvaluator.expr_flatten,\n            \"ak_is_group_member\": BaseEvaluator.expr_is_group_member,\n            \"ak_user_by\": BaseEvaluator.expr_user_by,\n            \"ak_user_has_authenticator\": BaseEvaluator.expr_func_user_has_authenticator,\n            \"ak_create_event\": self.expr_event_create,\n            \"ak_logger\": get_logger(self._filename).bind(),\n            \"requests\": get_http_session(),\n            \"ip_address\": ip_address,\n            \"ip_network\": ip_network,\n        }\n        self._context = {}\n\n    @staticmethod\n    def expr_flatten(value: list[Any] | Any) -> Optional[Any]:\n        \"\"\"Flatten `value` if its a list\"\"\"\n        if isinstance(value, list):\n            if len(value) < 1:\n                return None\n            return value[0]\n        return value\n\n    @staticmethod\n    def expr_regex_match(value: Any, regex: str) -> bool:\n        \"\"\"Expression Filter to run re.search\"\"\"\n        return re.search(regex, value) is not None\n\n    @staticmethod\n    def expr_regex_replace(value: Any, regex: str, repl: str) -> str:\n        \"\"\"Expression Filter to run re.sub\"\"\"\n        return re.sub(regex, repl, value)\n\n    @staticmethod\n    def expr_is_group_member(user: User, **group_filters) -> bool:\n        \"\"\"Check if `user` is member of group with name `group_name`\"\"\"\n        return user.ak_groups.filter(**group_filters).exists()\n\n    @staticmethod\n    def expr_user_by(**filters) -> Optional[User]:\n        \"\"\"Get user by filters\"\"\"\n        try:\n            users = User.objects.filter(**filters)\n            if users:\n                return users.first()\n            return None\n        except FieldError:\n            return None\n\n    @staticmethod\n    def expr_func_user_has_authenticator(user: User, device_type: Optional[str] = None) -> bool:\n        \"\"\"Check if a user has any authenticator devices, optionally matching *device_type*\"\"\"\n        user_devices = devices_for_user(user)\n        if device_type:\n            for device in user_devices:\n                device_class = device.__class__.__name__.lower().replace(\"device\", \"\")\n                if device_class == device_type:\n                    return True\n            return False\n        return len(list(user_devices)) > 0\n\n    def expr_event_create(self, action: str, **kwargs):\n        \"\"\"Create event with supplied data and try to extract as much relevant data\n        from the context\"\"\"\n        # If the result was a complex variable, we don't want to re-use it\n        self._context.pop(\"result\", None)\n        self._context.pop(\"handler\", None)\n        kwargs[\"context\"] = self._context\n        event = Event.new(\n            action,\n            app=self._filename,\n            **kwargs,\n        )\n        if \"request\" in self._context and isinstance(self._context[\"request\"], PolicyRequest):\n            policy_request: PolicyRequest = self._context[\"request\"]\n            if policy_request.http_request:\n                event.from_http(policy_request)\n                return\n        event.save()\n\n    def wrap_expression(self, expression: str, params: Iterable[str]) -> str:\n        \"\"\"Wrap expression in a function, call it, and save the result as `result`\"\"\"\n        handler_signature = \",\".join(params)\n        full_expression = \"\"\n        full_expression += f\"def handler({handler_signature}):\\n\"\n        full_expression += indent(expression, \"    \")\n        full_expression += f\"\\nresult = handler({handler_signature})\"\n        return full_expression\n\n    def evaluate(self, expression_source: str) -> Any:\n        \"\"\"Parse and evaluate expression. If the syntax is incorrect, a SyntaxError is raised.\n        If any exception is raised during execution, it is raised.\n        The result is returned without any type-checking.\"\"\"\n        with Hub.current.start_span(op=\"authentik.lib.evaluator.evaluate\") as span:\n            span: Span\n            span.description = self._filename\n            span.set_data(\"expression\", expression_source)\n            param_keys = self._context.keys()\n            try:\n                ast_obj = compile(\n                    self.wrap_expression(expression_source, param_keys),\n                    self._filename,\n                    \"exec\",\n                )\n            except (SyntaxError, ValueError) as exc:\n                self.handle_error(exc, expression_source)\n                raise exc\n            try:\n                _locals = self._context\n                # Yes this is an exec, yes it is potentially bad. Since we limit what variables are\n                # available here, and these policies can only be edited by admins, this is a risk\n                # we're willing to take.\n                # pylint: disable=exec-used\n                exec(ast_obj, self._globals, _locals)  # nosec # noqa\n                result = _locals[\"result\"]\n            except Exception as exc:\n                # So, this is a bit questionable. Essentially, we are edit the stacktrace\n                # so the user only sees information relevant to them\n                # and none of our surrounding error handling\n                exc.__traceback__ = exc.__traceback__.tb_next\n                self.handle_error(exc, expression_source)\n                raise exc\n            return result\n\n    def handle_error(self, exc: Exception, expression_source: str):  # pragma: no cover\n        \"\"\"Exception Handler\"\"\"\n        LOGGER.warning(\"Expression error\", exc=exc)\n\n    def validate(self, expression: str) -> bool:\n        \"\"\"Validate expression's syntax, raise ValidationError if Syntax is invalid\"\"\"\n        param_keys = self._context.keys()\n        try:\n            compile(\n                self.wrap_expression(expression, param_keys),\n                self._filename,\n                \"exec\",\n            )\n            return True\n        except (ValueError, SyntaxError) as exc:\n            raise ValidationError(f\"Expression Syntax Error: {str(exc)}\") from exc\n", "path": "authentik/lib/expression/evaluator.py"}], "after_files": [{"content": "\"\"\"authentik expression policy evaluator\"\"\"\nimport re\nimport socket\nfrom ipaddress import ip_address, ip_network\nfrom textwrap import indent\nfrom typing import Any, Iterable, Optional\n\nfrom cachetools import TLRUCache, cached\nfrom django.core.exceptions import FieldError\nfrom django_otp import devices_for_user\nfrom rest_framework.serializers import ValidationError\nfrom sentry_sdk.hub import Hub\nfrom sentry_sdk.tracing import Span\nfrom structlog.stdlib import get_logger\n\nfrom authentik.core.models import User\nfrom authentik.events.models import Event\nfrom authentik.lib.utils.http import get_http_session\nfrom authentik.policies.types import PolicyRequest\n\nLOGGER = get_logger()\n\n\nclass BaseEvaluator:\n    \"\"\"Validate and evaluate python-based expressions\"\"\"\n\n    # Globals that can be used by function\n    _globals: dict[str, Any]\n    # Context passed as locals to exec()\n    _context: dict[str, Any]\n\n    # Filename used for exec\n    _filename: str\n\n    def __init__(self, filename: Optional[str] = None):\n        self._filename = filename if filename else \"BaseEvaluator\"\n        # update website/docs/expressions/_objects.md\n        # update website/docs/expressions/_functions.md\n        self._globals = {\n            \"regex_match\": BaseEvaluator.expr_regex_match,\n            \"regex_replace\": BaseEvaluator.expr_regex_replace,\n            \"list_flatten\": BaseEvaluator.expr_flatten,\n            \"ak_is_group_member\": BaseEvaluator.expr_is_group_member,\n            \"ak_user_by\": BaseEvaluator.expr_user_by,\n            \"ak_user_has_authenticator\": BaseEvaluator.expr_func_user_has_authenticator,\n            \"resolve_dns\": BaseEvaluator.expr_resolve_dns,\n            \"reverse_dns\": BaseEvaluator.expr_reverse_dns,\n            \"ak_create_event\": self.expr_event_create,\n            \"ak_logger\": get_logger(self._filename).bind(),\n            \"requests\": get_http_session(),\n            \"ip_address\": ip_address,\n            \"ip_network\": ip_network,\n        }\n        self._context = {}\n\n    @cached(cache=TLRUCache(maxsize=32, ttu=lambda key, value, now: now + 180))\n    @staticmethod\n    def expr_resolve_dns(host: str, ip_version: Optional[int] = None) -> list[str]:\n        \"\"\"Resolve host to a list of IPv4 and/or IPv6 addresses.\"\"\"\n        # Although it seems to be fine (raising OSError), docs warn\n        # against passing `None` for both the host and the port\n        # https://docs.python.org/3/library/socket.html#socket.getaddrinfo\n        host = host or \"\"\n\n        ip_list = []\n\n        family = 0\n        if ip_version == 4:\n            family = socket.AF_INET\n        if ip_version == 6:\n            family = socket.AF_INET6\n\n        try:\n            for ip_addr in socket.getaddrinfo(host, None, family=family):\n                ip_list.append(str(ip_addr[4][0]))\n        except OSError:\n            pass\n        return list(set(ip_list))\n\n    @cached(cache=TLRUCache(maxsize=32, ttu=lambda key, value, now: now + 180))\n    @staticmethod\n    def expr_reverse_dns(ip_addr: str) -> str:\n        \"\"\"Perform a reverse DNS lookup.\"\"\"\n        try:\n            return socket.getfqdn(ip_addr)\n        except OSError:\n            return ip_addr\n\n    @staticmethod\n    def expr_flatten(value: list[Any] | Any) -> Optional[Any]:\n        \"\"\"Flatten `value` if its a list\"\"\"\n        if isinstance(value, list):\n            if len(value) < 1:\n                return None\n            return value[0]\n        return value\n\n    @staticmethod\n    def expr_regex_match(value: Any, regex: str) -> bool:\n        \"\"\"Expression Filter to run re.search\"\"\"\n        return re.search(regex, value) is not None\n\n    @staticmethod\n    def expr_regex_replace(value: Any, regex: str, repl: str) -> str:\n        \"\"\"Expression Filter to run re.sub\"\"\"\n        return re.sub(regex, repl, value)\n\n    @staticmethod\n    def expr_is_group_member(user: User, **group_filters) -> bool:\n        \"\"\"Check if `user` is member of group with name `group_name`\"\"\"\n        return user.ak_groups.filter(**group_filters).exists()\n\n    @staticmethod\n    def expr_user_by(**filters) -> Optional[User]:\n        \"\"\"Get user by filters\"\"\"\n        try:\n            users = User.objects.filter(**filters)\n            if users:\n                return users.first()\n            return None\n        except FieldError:\n            return None\n\n    @staticmethod\n    def expr_func_user_has_authenticator(user: User, device_type: Optional[str] = None) -> bool:\n        \"\"\"Check if a user has any authenticator devices, optionally matching *device_type*\"\"\"\n        user_devices = devices_for_user(user)\n        if device_type:\n            for device in user_devices:\n                device_class = device.__class__.__name__.lower().replace(\"device\", \"\")\n                if device_class == device_type:\n                    return True\n            return False\n        return len(list(user_devices)) > 0\n\n    def expr_event_create(self, action: str, **kwargs):\n        \"\"\"Create event with supplied data and try to extract as much relevant data\n        from the context\"\"\"\n        # If the result was a complex variable, we don't want to re-use it\n        self._context.pop(\"result\", None)\n        self._context.pop(\"handler\", None)\n        kwargs[\"context\"] = self._context\n        event = Event.new(\n            action,\n            app=self._filename,\n            **kwargs,\n        )\n        if \"request\" in self._context and isinstance(self._context[\"request\"], PolicyRequest):\n            policy_request: PolicyRequest = self._context[\"request\"]\n            if policy_request.http_request:\n                event.from_http(policy_request)\n                return\n        event.save()\n\n    def wrap_expression(self, expression: str, params: Iterable[str]) -> str:\n        \"\"\"Wrap expression in a function, call it, and save the result as `result`\"\"\"\n        handler_signature = \",\".join(params)\n        full_expression = \"\"\n        full_expression += f\"def handler({handler_signature}):\\n\"\n        full_expression += indent(expression, \"    \")\n        full_expression += f\"\\nresult = handler({handler_signature})\"\n        return full_expression\n\n    def evaluate(self, expression_source: str) -> Any:\n        \"\"\"Parse and evaluate expression. If the syntax is incorrect, a SyntaxError is raised.\n        If any exception is raised during execution, it is raised.\n        The result is returned without any type-checking.\"\"\"\n        with Hub.current.start_span(op=\"authentik.lib.evaluator.evaluate\") as span:\n            span: Span\n            span.description = self._filename\n            span.set_data(\"expression\", expression_source)\n            param_keys = self._context.keys()\n            try:\n                ast_obj = compile(\n                    self.wrap_expression(expression_source, param_keys),\n                    self._filename,\n                    \"exec\",\n                )\n            except (SyntaxError, ValueError) as exc:\n                self.handle_error(exc, expression_source)\n                raise exc\n            try:\n                _locals = self._context\n                # Yes this is an exec, yes it is potentially bad. Since we limit what variables are\n                # available here, and these policies can only be edited by admins, this is a risk\n                # we're willing to take.\n                # pylint: disable=exec-used\n                exec(ast_obj, self._globals, _locals)  # nosec # noqa\n                result = _locals[\"result\"]\n            except Exception as exc:\n                # So, this is a bit questionable. Essentially, we are edit the stacktrace\n                # so the user only sees information relevant to them\n                # and none of our surrounding error handling\n                exc.__traceback__ = exc.__traceback__.tb_next\n                self.handle_error(exc, expression_source)\n                raise exc\n            return result\n\n    def handle_error(self, exc: Exception, expression_source: str):  # pragma: no cover\n        \"\"\"Exception Handler\"\"\"\n        LOGGER.warning(\"Expression error\", exc=exc)\n\n    def validate(self, expression: str) -> bool:\n        \"\"\"Validate expression's syntax, raise ValidationError if Syntax is invalid\"\"\"\n        param_keys = self._context.keys()\n        try:\n            compile(\n                self.wrap_expression(expression, param_keys),\n                self._filename,\n                \"exec\",\n            )\n            return True\n        except (ValueError, SyntaxError) as exc:\n            raise ValidationError(f\"Expression Syntax Error: {str(exc)}\") from exc\n", "path": "authentik/lib/expression/evaluator.py"}]}