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,36 @@ 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.objects.get(name=provider)
+        session = None
+        if hasattr(request, "session") and request.session.session_key:
+            session = AuthenticatedSession.objects.filter(
+                session_key=request.session.session_key
+            ).first()
+        access_token = AccessToken(
+            provider=provider,
+            user=user,
+            expires=now() + timedelta_from_string(validity),
+            scope=scopes,
+            auth_time=now(),
+            session=session,
+        )
+        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())

authentik/lib/tests/test_evaluator.py

@@ -1,11 +1,15 @@
 """Test Evaluator base functions"""
 
-from django.test import TestCase
+from django.test import RequestFactory, TestCase
+from django.urls import reverse
+from jwt import decode
 
-from authentik.core.tests.utils import create_test_admin_user
+from authentik.blueprints.tests import apply_blueprint
+from authentik.core.tests.utils import create_test_admin_user, create_test_flow, create_test_user
 from authentik.events.models import Event
 from authentik.lib.expression.evaluator import BaseEvaluator
 from authentik.lib.generators import generate_id
+from authentik.providers.oauth2.models import OAuth2Provider, ScopeMapping
 
 
 class TestEvaluator(TestCase):
@@ -41,3 +45,35 @@ def test_expr_event_create(self):
         event = Event.objects.filter(action="custom_foo").first()
         self.assertIsNotNone(event)
         self.assertEqual(event.context, {"bar": "baz", "foo": "bar"})
+
+    @apply_blueprint("system/providers-oauth2.yaml")
+    def test_expr_create_jwt(self):
+        """Test expr_create_jwt"""
+        rf = RequestFactory()
+        user = create_test_user()
+        provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            authorization_flow=create_test_flow(),
+        )
+        provider.property_mappings.set(
+            ScopeMapping.objects.filter(
+                managed__in=[
+                    "goauthentik.io/providers/oauth2/scope-openid",
+                    "goauthentik.io/providers/oauth2/scope-email",
+                    "goauthentik.io/providers/oauth2/scope-profile",
+                ]
+            )
+        )
+        evaluator = BaseEvaluator(generate_id())
+        evaluator._context = {
+            "http_request": rf.get(reverse("authentik_core:root-redirect")),
+            "user": user,
+            "provider": provider.name,
+        }
+        jwt = evaluator.evaluate(
+            "return ak_create_jwt(user, provider, ['openid', 'email', 'profile'])"
+        )
+        decoded = decode(
+            jwt, provider.client_secret, algorithms=["HS256"], audience=provider.client_id
+        )
+        self.assertEqual(decoded["preferred_username"], user.username)