Add application fetching from Keycloak

Author: thcrt

pyproject.toml

@@ -14,6 +14,7 @@ dependencies = [
     "pydantic-settings>=2.6.1",
     "waitress>=3.0.1",
     "authentik-client>=2024.10.4.post1733219872",
+    "mantelo>=2.1.1",
 ]
 
 [build-system]

src/blobdash/__init__.py

@@ -4,7 +4,7 @@
 
 
 from .settings import Settings
-from .auth import AuthentikAuthProvider
+from .auth import AuthentikAuthProvider, KeycloakAuthProvider
 from .applications import ApplicationProvider
 
 
@@ -21,14 +21,21 @@ def create_app():
         secho(e)
         raise SystemExit(1)
 
-    if settings.auth.apps.enabled:
-        match settings.auth.apps.provider:
-            case "authentik":
-                auth_provider = AuthentikAuthProvider(
-                    settings.auth.apps.host, settings.auth.apps.token
-                )
-    else:
-        auth_provider = None
+    match settings.auth.fetch.provider:
+        case "authentik":
+            auth_provider = AuthentikAuthProvider(
+                settings.auth.fetch.host, settings.auth.fetch.token
+            )
+        case "keycloak":
+            auth_provider = KeycloakAuthProvider(
+                settings.auth.fetch.host,
+                settings.auth.fetch.realm,
+                settings.auth.fetch.auth_realm,
+                settings.auth.fetch.id,
+                settings.auth.fetch.secret,
+            )
+        case None:
+            auth_provider = None
 
     app_provider = ApplicationProvider(auth_provider, settings.apps)
 

src/blobdash/auth.py

@@ -1,27 +1,64 @@
 import json
 import urllib.parse
 from abc import ABC, abstractmethod
+
 import authentik_client
+from mantelo import KeycloakAdmin
+
 
 from .applications import Application
 
 
 class AuthProvider(ABC):
-    def __init__(self, host: str, token: str) -> None:
-        self.host = host
-        self.token = token
-
     @abstractmethod
     def get_applications(self, username):
         pass
 
 
+class KeycloakAuthProvider(AuthProvider):
+    def __init__(self, host, realm, auth_realm, id, secret):
+        self.host = host
+        self.realm = realm
+        self.auth_realm = realm if auth_realm is None else auth_realm
+        self.id = id
+        self.secret = secret
+
+        self._api = KeycloakAdmin.from_client_credentials(
+            server_url=self.host,
+            realm_name=self.realm,
+            authentication_realm_name=self.auth_realm,
+            client_id=self.id,
+            client_secret=self.secret,
+        )
+
+    def get_applications(self, username):
+        # Get user ID from username. Returns a list of user objects, so choose the first (and only)
+        # entry and grab its ID.
+        user = self._api.users.get(username=username, exact=True)[0]["id"]
+
+        # Get a list of all clients the user has logged into, by cycling through consents for the
+        # user ID and grabbing the client object associated with that client ID
+        clients = [
+            self._api.clients.get(clientId=consent["clientId"])[0]
+            for consent in self._api.users(user).consents.get()
+        ]
+
+        return {
+            client["clientId"]: Application(
+                name=client["name"],
+                url=client["baseUrl"],
+                icon=client["attributes"]["logoUri"],
+                desc=client["description"],
+            )
+            for client in clients
+        }
+
+
 class AuthentikAuthProvider(AuthProvider):
     def __init__(self, host, token):
         # Base path for Authentik API
-        host = urllib.parse.urljoin(host, "/api/v3")
-
-        super().__init__(host, token)
+        self.host = urllib.parse.urljoin(host, "/api/v3")
+        self.token = token
 
         # Set up API client. The `CoreAPI` is the only one we need to view users and applications
         self._api = authentik_client.CoreApi(

src/blobdash/settings.py

@@ -12,19 +12,27 @@
 from .applications import Application
 
 
-class AuthApplicationSettings(BaseModel):
-    enabled: bool = False
-    provider: Literal["authentik"] = "authentik"
-    host: str = "https://auth.example.com"
-    token: str = "changeme"
+class AuthentikSettings(BaseModel):
+    provider: Literal["authentik"]
+    host: str
+    token: str
+
+
+class KeycloakSettings(BaseModel):
+    provider: Literal["keycloak"]
+    host: str
+    auth_realm: Optional[str] = None
+    realm: str
+    id: str
+    secret: str
 
 
 class AuthSettings(BaseModel):
     enabled: bool = False
     header: str = "X-App-User"
     logout_url: str = "/flows/-/default/invalidation"
     default_user: Optional[str] = None
-    apps: AuthApplicationSettings = AuthApplicationSettings()
+    fetch: AuthentikSettings | KeycloakSettings | None = None
 
 
 class DashdotSettings(BaseModel):