Coverage for app/utils/keycloak_auth.py: 46.10%
154 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-25 15:51 +0000
1"""
2Keycloak Authentication & Authorization
3Handles token validation and user management with Keycloak.
4"""
5import logging
6import threading
8from fastapi import Depends, HTTPException, status
9from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
10from jose import JWTError, jwt
11from keycloak import KeycloakAdmin, KeycloakAuthenticationError, KeycloakOpenID
12from sqlalchemy.orm import Session
14from app.config import settings
15from app.database import get_db
16from app.models import User, UserRole
18logger = logging.getLogger(__name__)
20security = HTTPBearer()
23# ----------------------------------------------------------------
24# KEYCLOAK CLIENTS
25# ----------------------------------------------------------------
26def get_keycloak_client() -> KeycloakOpenID:
27 """Get Keycloak OpenID client instance"""
28 return KeycloakOpenID(
29 server_url=settings.KEYCLOAK_SERVER_URL,
30 client_id=settings.KEYCLOAK_CLIENT_ID,
31 realm_name=settings.KEYCLOAK_REALM,
32 client_secret_key=settings.KEYCLOAK_CLIENT_SECRET,
33 verify=True,
34 )
37def get_keycloak_admin() -> KeycloakAdmin:
38 """Get Keycloak Admin client (uses service-account client_credentials grant)."""
39 return KeycloakAdmin(
40 server_url=settings.KEYCLOAK_SERVER_URL,
41 realm_name=settings.KEYCLOAK_REALM,
42 client_id=settings.KEYCLOAK_CLIENT_ID,
43 client_secret_key=settings.KEYCLOAK_CLIENT_SECRET,
44 verify=True,
45 )
48# ----------------------------------------------------------------
49# PUBLIC KEY CACHE
50# ----------------------------------------------------------------
51# python-keycloak's public_key() does a fresh HTTP GET on every call,
52# which we make on every authenticated request. The realm signing key
53# is stable for the process lifetime, so we cache the PEM-formatted key
54# after the first fetch; restart the process on key rotation.
55_public_key_pem: str | None = None
56_public_key_lock = threading.Lock()
59def _get_realm_public_key_pem() -> str:
60 global _public_key_pem
61 if _public_key_pem is not None:
62 return _public_key_pem
63 with _public_key_lock:
64 if _public_key_pem is not None:
65 return _public_key_pem
66 raw = get_keycloak_client().public_key()
67 _public_key_pem = (
68 "-----BEGIN PUBLIC KEY-----\n" + raw + "\n-----END PUBLIC KEY-----"
69 )
70 return _public_key_pem
73# ----------------------------------------------------------------
74# TOKEN VALIDATION
75# ----------------------------------------------------------------
76def verify_keycloak_token(token: str) -> dict:
77 """Validate token via Keycloak introspection (slow, server round-trip)."""
78 try:
79 keycloak_client = get_keycloak_client()
80 token_info = keycloak_client.introspect(token)
81 if not token_info.get("active"):
82 raise HTTPException(
83 status_code=status.HTTP_401_UNAUTHORIZED,
84 detail="Token is not active or has expired",
85 headers={"WWW-Authenticate": "Bearer"},
86 )
87 return token_info
88 except KeycloakAuthenticationError:
89 raise HTTPException(
90 status_code=status.HTTP_401_UNAUTHORIZED,
91 detail="Authentication failed",
92 headers={"WWW-Authenticate": "Bearer"},
93 )
94 except HTTPException:
95 raise
96 except Exception:
97 raise HTTPException(
98 status_code=status.HTTP_401_UNAUTHORIZED,
99 detail="Could not validate credentials",
100 headers={"WWW-Authenticate": "Bearer"},
101 )
104def verify_keycloak_token_offline(token: str) -> dict:
105 """Validate JWT signature against Keycloak public key (fast, no server round-trip)."""
106 try:
107 public_key = _get_realm_public_key_pem()
108 return jwt.decode(
109 token,
110 public_key,
111 algorithms=["RS256"],
112 options={"verify_signature": True, "verify_aud": False, "verify_exp": True},
113 )
114 except JWTError:
115 raise HTTPException(
116 status_code=status.HTTP_401_UNAUTHORIZED,
117 detail="Invalid token",
118 headers={"WWW-Authenticate": "Bearer"},
119 )
120 except Exception:
121 raise HTTPException(
122 status_code=status.HTTP_401_UNAUTHORIZED,
123 detail="Token validation failed",
124 headers={"WWW-Authenticate": "Bearer"},
125 )
128# ----------------------------------------------------------------
129# ROLE MAPPING
130# ----------------------------------------------------------------
131def map_keycloak_roles_to_app_role(keycloak_roles: list) -> UserRole:
132 """Priority: admin > teacher > student"""
133 if "admin" in keycloak_roles:
134 return UserRole.ADMIN
135 if "teacher" in keycloak_roles:
136 return UserRole.TEACHER
137 return UserRole.STUDENT
140# ----------------------------------------------------------------
141# USER SYNC (Just-in-Time Provisioning)
142# ----------------------------------------------------------------
143def sync_user_from_keycloak(db: Session, keycloak_user_data: dict) -> User:
144 """
145 Synchronize a Keycloak user into the local DB.
146 Creates the user if missing, updates role/name if changed.
147 Expects keys: id (or sub), username, email, roles (or realm_access.roles),
148 firstName/given_name, lastName/family_name.
149 """
150 keycloak_id = keycloak_user_data.get("id") or keycloak_user_data.get("sub")
151 email = keycloak_user_data.get("email")
152 username = keycloak_user_data.get("username") or keycloak_id
153 keycloak_roles = (
154 keycloak_user_data.get("roles")
155 or keycloak_user_data.get("realm_access", {}).get("roles", [])
156 )
157 app_role = map_keycloak_roles_to_app_role(keycloak_roles)
158 first_name = keycloak_user_data.get("firstName") or keycloak_user_data.get("given_name")
159 last_name = keycloak_user_data.get("lastName") or keycloak_user_data.get("family_name")
161 if not email:
162 raise HTTPException(
163 status_code=status.HTTP_400_BAD_REQUEST,
164 detail="Keycloak user has no email; refusing to provision.",
165 )
167 user = db.query(User).filter(User.keycloak_id == keycloak_id).first()
168 if not user:
169 user = User(
170 keycloak_id=keycloak_id,
171 email=email,
172 username=username,
173 role=app_role,
174 firstName=first_name,
175 lastName=last_name,
176 )
177 db.add(user)
178 db.commit()
179 db.refresh(user)
180 return user
182 updated = False
183 # Email is the identifier used for display, search, and
184 # notifications; follow the Keycloak record when it changes.
185 if email and user.email != email:
186 user.email = email
187 updated = True
188 # Username is the login identifier shown in the UI and in
189 # Terraform-issued credentials; keep it in sync too.
190 if username and user.username != username:
191 user.username = username
192 updated = True
193 if user.role != app_role:
194 user.role = app_role
195 updated = True
196 if first_name and user.firstName != first_name:
197 user.firstName = first_name
198 updated = True
199 if last_name and user.lastName != last_name:
200 user.lastName = last_name
201 updated = True
202 if updated:
203 db.commit()
204 db.refresh(user)
205 return user
208# ----------------------------------------------------------------
209# AUTH DEPENDENCY
210# ----------------------------------------------------------------
211def get_current_user_keycloak(
212 credentials: HTTPAuthorizationCredentials = Depends(security),
213 db: Session = Depends(get_db),
214) -> User:
215 """
216 Validate the bearer token and return the local User record.
217 JIT-provisions the user from Keycloak claims on first sight.
218 """
219 token_info = verify_keycloak_token_offline(credentials.credentials)
221 keycloak_id = token_info.get("sub")
222 if not keycloak_id:
223 raise HTTPException(
224 status_code=status.HTTP_401_UNAUTHORIZED,
225 detail="Token missing user ID (sub)",
226 )
228 return sync_user_from_keycloak(
229 db,
230 {
231 "id": keycloak_id,
232 "email": token_info.get("email"),
233 "username": token_info.get("preferred_username"),
234 "roles": token_info.get("realm_access", {}).get("roles", []),
235 "firstName": token_info.get("given_name"),
236 "lastName": token_info.get("family_name"),
237 },
238 )
241# ----------------------------------------------------------------
242# KEYCLOAK USER LOOKUP (admin API)
243# ----------------------------------------------------------------
244def _project_keycloak_user(u: dict, *, include_enabled: bool = False) -> dict:
245 """Project a raw Keycloak user record into our simplified dict shape.
247 Includes the ``enabled`` flag only when ``include_enabled`` is set,
248 preserving the differing output shapes of the two call sites.
249 """
250 projected = {
251 "id": u.get("id"),
252 "username": u.get("username"),
253 "email": u.get("email"),
254 "firstName": u.get("firstName", ""),
255 "lastName": u.get("lastName", ""),
256 }
257 if include_enabled:
258 projected["enabled"] = u.get("enabled", True)
259 return projected
262def search_keycloak_users(search_query: str, max_results: int = 10) -> list[dict]:
263 """Search Keycloak users by username/email/name (uses service account)."""
264 try:
265 keycloak_admin = get_keycloak_admin()
266 users = keycloak_admin.get_users({"search": search_query, "max": max_results})
267 return [
268 _project_keycloak_user(u, include_enabled=True)
269 for u in users
270 ]
271 except Exception:
272 raise HTTPException(
273 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
274 detail="Failed to search Keycloak users",
275 )
278def get_keycloak_users_by_ids(ids: list[str]) -> dict:
279 """Resolve a list of Keycloak IDs to simplified user dicts."""
280 result: dict = {}
281 if not ids:
282 return result
283 try:
284 kc = get_keycloak_admin()
285 for kid in ids:
286 try:
287 user = kc.get_user(kid)
288 except Exception:
289 try:
290 users = kc.get_users({"search": kid, "max": 1})
291 user = users[0] if users else None
292 except Exception:
293 user = None
294 if not user:
295 continue
296 projected = _project_keycloak_user(user)
297 projected["id"] = user.get("id") or kid
298 result[user.get("id") or kid] = projected
299 return result
300 except Exception:
301 raise HTTPException(
302 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
303 detail="Failed to fetch Keycloak users",
304 )
307# ----------------------------------------------------------------
308# JIT REFRESH FOR OUTGOING SIDE-CHANNELS (e.g. mail)
309# ----------------------------------------------------------------
310def refresh_user_from_keycloak(db: Session, user: User) -> User:
311 """Pull the latest Keycloak record for ``user`` and reconcile our DB row.
313 Used by paths that address a user outside of an active HTTP request
314 (most importantly the mail notifier) so credentials go to the
315 current email even if it changed since the wizard pick.
317 Best-effort by design:
319 * If the user has no ``keycloak_id``, there is nothing to refresh —
320 return the row unchanged.
321 * If the Admin API is down or the user was deleted in Keycloak, log
322 at WARNING and return the row unchanged so a flaky KC doesn't
323 block notifications.
325 Reuses :func:`sync_user_from_keycloak` for the DB write.
326 """
327 if not user or not getattr(user, "keycloak_id", None):
328 return user
330 try:
331 kc = get_keycloak_admin()
332 kc_user = kc.get_user(user.keycloak_id)
333 except Exception as e:
334 logger.warning(
335 "refresh_user_from_keycloak: get_user(%s) failed (%s); "
336 "falling back to DB record for %s",
337 user.keycloak_id, e, user.email,
338 )
339 return user
341 if not kc_user:
342 logger.warning(
343 "refresh_user_from_keycloak: keycloak returned no user for id=%s; "
344 "user may have been deleted upstream — keeping DB record for %s",
345 user.keycloak_id, user.email,
346 )
347 return user
349 # ``get_user`` doesn't include realm roles; pre-seed with the
350 # current role mapping so re-running role mapping doesn't cause a
351 # spurious downgrade. Role rotation goes through the auth dependency
352 # on next login.
353 current_role_token = []
354 if user.role == UserRole.ADMIN:
355 current_role_token = ["admin"]
356 elif user.role == UserRole.TEACHER:
357 current_role_token = ["teacher"]
359 try:
360 return sync_user_from_keycloak(
361 db,
362 {
363 "id": kc_user.get("id") or user.keycloak_id,
364 "email": kc_user.get("email"),
365 "username": kc_user.get("username"),
366 "firstName": kc_user.get("firstName"),
367 "lastName": kc_user.get("lastName"),
368 "roles": current_role_token,
369 },
370 )
371 except HTTPException as e:
372 # ``sync_user_from_keycloak`` raises 400 when KC returns a user
373 # without an email; defend in depth so a bad record doesn't
374 # block the notifier.
375 logger.warning(
376 "refresh_user_from_keycloak: sync rejected KC payload for %s: %s",
377 user.keycloak_id, e.detail,
378 )
379 return user