api: add basic jwt support with required scope

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>

authentik/api/authentication.py

@@ -10,6 +10,8 @@
 from authentik.core.middleware import KEY_AUTH_VIA, LOCAL
 from authentik.core.models import Token, TokenIntents, User
 from authentik.outposts.models import Outpost
+from authentik.providers.oauth2.constants import SCOPE_AUTHENTIK_API
+from authentik.providers.oauth2.models import RefreshToken
 
 LOGGER = get_logger()
 
@@ -34,14 +36,30 @@ def bearer_auth(raw_header: bytes) -> Optional[User]:
     auth_credentials = validate_auth(raw_header)
     if not auth_credentials:
         return None
+    if not hasattr(LOCAL, "authentik"):
+        LOCAL.authentik = {}
+    LOCAL.authentik[KEY_AUTH_VIA] = "api_token"
     # first, check traditional tokens
-    token = Token.filter_not_expired(key=auth_credentials, intent=TokenIntents.INTENT_API).first()
-    if hasattr(LOCAL, "authentik"):
-        LOCAL.authentik[KEY_AUTH_VIA] = "api_token"
-    if token:
-        return token.user
+    key_token = Token.filter_not_expired(
+        key=auth_credentials, intent=TokenIntents.INTENT_API
+    ).first()
+    if key_token:
+        return key_token.user
+    # then try to auth via JWT
+    jwt_token = RefreshToken.filter_not_expired(
+        refresh_token=auth_credentials, _scope__icontains=SCOPE_AUTHENTIK_API
+    ).first()
+    if jwt_token:
+        # Double-check scopes, since they are saved in a single string
+        # we want to check the parsed version too
+        if SCOPE_AUTHENTIK_API not in jwt_token.scope:
+            raise AuthenticationFailed("Token invalid/expired")
+        LOCAL.authentik[KEY_AUTH_VIA] = "jwt"
+        return jwt_token.user
+    # then try to auth via secret key (for embedded outpost/etc)
     user = token_secret_key(auth_credentials)
     if user:
+        LOCAL.authentik[KEY_AUTH_VIA] = "secret_key"
         return user
     raise AuthenticationFailed("Token invalid/expired")
 
@@ -56,8 +74,6 @@ def token_secret_key(value: str) -> Optional[User]:
     outposts = Outpost.objects.filter(managed=MANAGED_OUTPOST)
     if not outposts:
         return None
-    if hasattr(LOCAL, "authentik"):
-        LOCAL.authentik[KEY_AUTH_VIA] = "secret_key"
     outpost = outposts.first()
     return outpost.user
 

authentik/providers/oauth2/constants.py

@@ -18,6 +18,8 @@
 SCOPE_OPENID_PROFILE = "profile"
 SCOPE_OPENID_EMAIL = "email"
 
+SCOPE_AUTHENTIK_API = "goauthentik.io/api"
+
 # Read/write full user (including email)
 SCOPE_GITHUB_USER = "user"
 # Read user (without email)

authentik/providers/oauth2/views/userinfo.py

@@ -10,6 +10,7 @@
 from authentik.core.exceptions import PropertyMappingExpressionException
 from authentik.events.models import Event, EventAction
 from authentik.providers.oauth2.constants import (
+    SCOPE_AUTHENTIK_API,
     SCOPE_GITHUB_ORG_READ,
     SCOPE_GITHUB_USER,
     SCOPE_GITHUB_USER_EMAIL,
@@ -35,15 +36,16 @@ def get_scope_descriptions(self, scopes: list[str]) -> list[dict[str, str]]:
                 scope_descriptions.append({"id": scope.scope_name, "name": scope.description})
         # GitHub Compatibility Scopes are handled differently, since they required custom paths
         # Hence they don't exist as Scope objects
-        github_scope_map = {
+        special_scope_map = {
             SCOPE_GITHUB_USER: ("GitHub Compatibility: Access your User Information"),
             SCOPE_GITHUB_USER_READ: ("GitHub Compatibility: Access your User Information"),
             SCOPE_GITHUB_USER_EMAIL: ("GitHub Compatibility: Access you Email addresses"),
             SCOPE_GITHUB_ORG_READ: ("GitHub Compatibility: Access your Groups"),
+            SCOPE_AUTHENTIK_API: ("authentik API Access on behalf of your user"),
         }
         for scope in scopes:
-            if scope in github_scope_map:
-                scope_descriptions.append({"id": scope, "name": github_scope_map[scope]})
+            if scope in special_scope_map:
+                scope_descriptions.append({"id": scope, "name": special_scope_map[scope]})
         return scope_descriptions
 
     def get_claims(self, token: RefreshToken) -> dict[str, Any]: