lib: add expression helper ak_create_jwt to create JWTs

authentik/core/expression/evaluator.py

@@ -58,6 +58,7 @@ def set_context(
             self._context["user"] = user
         if request:
             req.http_request = request
+            self._context["http_request"] = request
         req.context.update(**kwargs)
         self._context["request"] = req
         self._context.update(**kwargs)

authentik/lib/expression/evaluator.py

@@ -9,20 +9,25 @@
 
 from cachetools import TLRUCache, cached
 from django.core.exceptions import FieldError
+from django.http import HttpRequest
 from django.utils.text import slugify
+from django.utils.timezone import now
 from guardian.shortcuts import get_anonymous_user
 from rest_framework.serializers import ValidationError
 from sentry_sdk import start_span
 from sentry_sdk.tracing import Span
 from structlog.stdlib import get_logger
 
-from authentik.core.models import User
+from authentik.core.models import AuthenticatedSession, User
 from authentik.events.models import Event
 from authentik.lib.expression.exceptions import ControlFlowException
 from authentik.lib.utils.http import get_http_session
+from authentik.lib.utils.time import timedelta_from_string
 from authentik.policies.models import Policy, PolicyBinding
 from authentik.policies.process import PolicyProcess
 from authentik.policies.types import PolicyRequest, PolicyResult
+from authentik.providers.oauth2.id_token import IDToken
+from authentik.providers.oauth2.models import AccessToken, OAuth2Provider
 from authentik.stages.authenticator import devices_for_user
 
 LOGGER = get_logger()
@@ -56,6 +61,7 @@ def __init__(self, filename: str | None = None):
             "ak_logger": get_logger(self._filename).bind(),
             "ak_user_by": BaseEvaluator.expr_user_by,
             "ak_user_has_authenticator": BaseEvaluator.expr_func_user_has_authenticator,
+            "ak_create_jwt": self.expr_create_jwt,
             "ip_address": ip_address,
             "ip_network": ip_network,
             "list_flatten": BaseEvaluator.expr_flatten,
@@ -182,6 +188,33 @@ def expr_func_call_policy(self, name: str, **kwargs) -> PolicyResult:
         proc = PolicyProcess(PolicyBinding(policy=policy), request=req, connection=None)
         return proc.profiling_wrapper()
 
+    def expr_create_jwt(
+        self,
+        user: User,
+        provider: OAuth2Provider | str,
+        scopes: list[str],
+        validity: str = "seconds=60",
+    ) -> str | None:
+        """Issue a JWT for a given provider"""
+        request: HttpRequest = self._context.get("http_request")
+        if not request:
+            return None
+        if not isinstance(provider, OAuth2Provider):
+            provider = OAuth2Provider(name=provider)
+        access_token = AccessToken(
+            provider=provider,
+            user=user,
+            expires=now() + timedelta_from_string(validity),
+            scope=scopes,
+            auth_time=now(),
+            session=AuthenticatedSession.objects.filter(
+                session_key=request.session.session_key
+            ).first(),
+        )
+        access_token.id_token = IDToken.new(provider, access_token, request)
+        access_token.save()
+        return access_token.token
+
     def wrap_expression(self, expression: str) -> str:
         """Wrap expression in a function, call it, and save the result as `result`"""
         handler_signature = ",".join(sanitize_arg(x) for x in self._context.keys())

website/docs/expressions/_functions.md

@@ -101,6 +101,30 @@ Example:
 ak_create_event("my_custom_event", foo=request.user)
 ```
 
+### `ak_create_jwt(user: User, provider: OAuth2Provider | str, scopes: list[str], validity = "seconds=60") -> str | None` <span class="badge badge--version">authentik 2025.2+</span>
+
+Create a new JWT signed by the given `provider` for `user`.
+
+The `provider` parameter can either be an instance of `OAuth2Provider` or a the name of a provider instance as a string. Scopes is an array of all scopes that the JWT should have.
+
+The JWT is valid for 60 seconds by default, this can be customized using the `validity` parameter. The syntax of the parameter is `hours=1,minutes=2,seconds=3`. The following keys are allowed:
+
+    - Microseconds
+    - Milliseconds
+    - Seconds
+    - Minutes
+    - Hours
+    - Days
+    - Weeks
+
+    All values accept floating-point values.
+
+Example:
+
+```python
+jwt = ak_create_jwt(request.user, "my-oauth2-provider-name", ["openid", "profile", "email"])
+```
+
 ## Comparing IP Addresses
 
 To compare IP Addresses or check if an IP Address is within a given subnet, you can use the functions `ip_address('192.0.2.1')` and `ip_network('192.0.2.0/24')`. With these objects you can do [arithmetic operations](https://docs.python.org/3/library/ipaddress.html#operators).