Coverage for server / authentication / auth0_bearer.py: 88%
179 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-10-04 09:33 +0000
1"""
2Auth0 Bearer Authentication Dependency
4FastAPI dependency for Auth0 RS256 JWT token validation with profile auto-creation
5and role-based access control.
7On first Auth0 login, auto-creates a MongoDB user profile from token claims.
8Subsequent logins look up the existing profile by auth0_user_id.
10Sets request.state.user_details with a standardized shape compatible with
11existing service layer code.
13Usage:
14 from server.authentication.auth0_bearer import Auth0Bearer
16 @app.get("/protected", dependencies=[Depends(Auth0Bearer(access_levels=["teacher"]))])
17 async def protected_route(request: Request):
18 user_info = request.state.user_details
19 return {"user": user_info}
20"""
22import logging
23import os
24from typing import List, Optional
26from fastapi import HTTPException, Request, status
27from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
29from .auth0_handler import auth0_handler
30from .auth0_config import get_auth0_settings
31from .cookie_session import validate_csrf
33logger = logging.getLogger(__name__)
35# This API serves teacher/student apps only — not admin or staff portals.
36APP_ALLOWED_ROLES = frozenset({"teacher", "student"})
38# Roles that require a `school_id` on their user_collection document. Without
39# `school_id` the GrowthBook scope chain (school → district) can't resolve, so
40# feature flags fall back to defaults — see PR #157 (`feature/growthbook`).
41_ROLES_REQUIRING_SCHOOL = frozenset({"teacher", "student"})
44async def _autorecover_school_id(
45 *,
46 user_doc: dict,
47 role: str,
48 auth0_user_id: str,
49 email: Optional[str],
50) -> Optional[str]:
51 """If `user_doc` lacks `school_id`, try to assign one and persist it.
53 PR #157 (`feature/growthbook`, 2026-06-13) added a hard 403 when a
54 teacher/student `user_collection` doc was missing `school_id`. That broke
55 legacy users (real ones — not just E2E test accounts) who pre-dated the
56 school-scope requirement: they had a valid Auth0 identity + profile, but
57 no school link, and got bounced at `/v1/auth/me`.
59 Recovery strategy (first match wins):
60 1. Same-district pick — if user has `district_id`, find any school in
61 that district. Preserves tenant.
62 2. Configured fallback — `AUTH0_FALLBACK_SCHOOL_ID` env var. Lets
63 deploys nominate a default school for orphaned legacy users.
64 3. Soft-warn — emit `logger.warning` and return `None`. Caller MAY let
65 auth proceed; staff can clean up via the admin portal.
67 Persists the recovered `school_id` so subsequent logins are fast. The
68 persistence is best-effort — a write failure is logged but does NOT block
69 the login.
70 """
71 if role not in _ROLES_REQUIRING_SCHOOL:
72 return None
73 existing = user_doc.get("school_id")
74 if existing:
75 return str(existing)
77 from server.models.users import User
78 from bson import ObjectId
80 resolved_school_id = None
81 source = None
83 # 1. Same-district pick.
84 district_id = user_doc.get("district_id")
85 if district_id:
86 try:
87 from server.connection.database import staff_admin_db
89 school_doc = None
90 if staff_admin_db is not None:
91 school_doc = await staff_admin_db["school_collection"].find_one(
92 {"district_id": district_id}, {"_id": 1}
93 )
94 if school_doc is None:
95 ts_db = User.get_pymongo_collection().database
96 school_doc = await ts_db["school_collection"].find_one(
97 {"district_id": district_id}, {"_id": 1}
98 )
99 if school_doc:
100 resolved_school_id = str(school_doc["_id"])
101 source = "same-district"
102 except Exception as e: # pragma: no cover — defense in depth
103 logger.warning(
104 "auto-recover school_id: same-district lookup failed for "
105 "auth0_user_id=%s district_id=%s err=%s",
106 auth0_user_id,
107 district_id,
108 e,
109 )
111 # 2. Configured fallback.
112 if resolved_school_id is None:
113 fallback = os.environ.get("AUTH0_FALLBACK_SCHOOL_ID")
114 if fallback:
115 try:
116 ObjectId(fallback)
117 resolved_school_id = fallback
118 source = "fallback-env"
119 except Exception:
120 logger.warning(
121 "auto-recover school_id: AUTH0_FALLBACK_SCHOOL_ID is set "
122 "but is not a valid ObjectId: %r",
123 fallback,
124 )
126 # 3. Soft-warn + return None.
127 if resolved_school_id is None:
128 logger.warning(
129 "auto-recover school_id: NO recovery source found for legacy user "
130 "auth0_user_id=%s email=%s role=%s user_doc_id=%s. Auth proceeds "
131 "but feature-flag scope will degrade to defaults. Staff should "
132 "set this user's school_id via the admin portal.",
133 auth0_user_id,
134 email,
135 role,
136 user_doc.get("_id"),
137 )
138 return None
140 # Persist (best effort).
141 try:
142 collection = User.get_pymongo_collection()
143 await collection.update_one(
144 {"_id": user_doc["_id"]},
145 {"$set": {"school_id": ObjectId(resolved_school_id)}},
146 )
147 logger.warning(
148 "auto-recover school_id: assigned %s (via %s) to legacy user "
149 "auth0_user_id=%s email=%s user_doc_id=%s",
150 resolved_school_id,
151 source,
152 auth0_user_id,
153 email,
154 user_doc.get("_id"),
155 )
156 except Exception as e:
157 logger.warning(
158 "auto-recover school_id: failed to persist for "
159 "auth0_user_id=%s err=%s (returning resolved id anyway so this "
160 "request can complete; next login will retry)",
161 auth0_user_id,
162 e,
163 )
165 return resolved_school_id
168def _coerce_roles(raw_roles) -> list[str]:
169 """Normalize Auth0 roles claim to a list of role strings."""
170 if isinstance(raw_roles, str):
171 return [r.strip() for r in raw_roles.replace(",", " ").split() if r.strip()]
172 if not raw_roles:
173 return []
174 return [str(role) for role in raw_roles]
177def app_role_from_token_roles(roles: list[str]) -> Optional[str]:
178 """Return the teacher/student role for this API, or None if not allowed."""
179 if "teacher" in roles:
180 return "teacher"
181 if "student" in roles:
182 return "student"
183 return None
186class Auth0Bearer(HTTPBearer):
187 """
188 FastAPI dependency for Auth0 JWT authentication with profile auto-creation
189 and role-based access control.
191 After token validation, looks up the user profile in MongoDB by auth0_user_id.
192 If no profile exists (first login), auto-creates one from token claims.
194 Sets request.state.user_details to:
195 {
196 "uuid": "auth0|abc123",
197 "name": "John Doe",
198 "role": "teacher",
199 "email": "teacher@example.com",
200 "auth_provider": "auth0",
201 "auth0_user_id": "auth0|abc123",
202 "mongodb_id": "507f1f77bcf86cd799439011",
203 }
204 """
206 def __init__(
207 self,
208 access_levels: Optional[List[str]] = None,
209 auto_error: bool = True,
210 ):
211 self.access_levels = access_levels
212 super().__init__(auto_error=auto_error)
214 async def __call__(self, request: Request) -> str:
215 # ------------------------------------------------------------------
216 # Phase 1 (ADR-001): cookie-based auth path.
217 # Only activated when there is NO Authorization header — existing
218 # Bearer-header clients fall through to the original path unchanged.
219 # ------------------------------------------------------------------
220 cookie_token = request.cookies.get("access_token")
221 has_auth_header = bool(request.headers.get("Authorization"))
223 if cookie_token and not has_auth_header:
224 # CSRF double-submit check for unsafe methods (POST/PUT/PATCH/DELETE).
225 # Safe methods (GET/HEAD/OPTIONS) pass through — cross-origin
226 # responses are not readable due to CORS.
227 if not validate_csrf(request):
228 raise HTTPException(
229 status_code=status.HTTP_403_FORBIDDEN,
230 detail="CSRF token missing or invalid.",
231 )
232 token = cookie_token
233 else:
234 # Original header-based path — untouched.
235 credentials: HTTPAuthorizationCredentials = await super().__call__(request)
237 if not credentials:
238 raise HTTPException(
239 status_code=status.HTTP_401_UNAUTHORIZED,
240 detail="Invalid authorization code",
241 )
243 if credentials.scheme.lower() != "bearer":
244 raise HTTPException(
245 status_code=status.HTTP_401_UNAUTHORIZED,
246 detail="Invalid authentication scheme. Expected 'Bearer'",
247 )
249 token = credentials.credentials
251 # Validate token with Auth0
252 try:
253 payload = await auth0_handler.verify_token(token)
254 except HTTPException:
255 raise
256 except Exception as e:
257 logger.error(f"Token validation failed: {e}")
258 raise HTTPException(
259 status_code=status.HTTP_401_UNAUTHORIZED,
260 detail="Invalid authentication credentials",
261 )
263 # Extract user info from validated payload
264 settings = get_auth0_settings()
265 auth0_sub = payload.get("sub")
266 if not auth0_sub:
267 raise HTTPException(
268 status_code=status.HTTP_401_UNAUTHORIZED,
269 detail="Token missing subject claim",
270 )
272 roles = _coerce_roles(payload.get(settings.AUTH0_ROLES_CLAIM, []))
273 email = payload.get("email") or payload.get("https://eruditiontx.com/email")
274 name = payload.get("name", "")
276 app_role = app_role_from_token_roles(roles)
277 if app_role is None:
278 logger.warning(
279 "Blocked authentication for non-teacher/student roles: %s", roles
280 )
281 raise HTTPException(
282 status_code=status.HTTP_403_FORBIDDEN,
283 detail="This application is for teachers and students only.",
284 )
286 # Enforce route-level role access (teacher-only vs student-only routes).
287 if self.access_levels and app_role not in self.access_levels:
288 logger.warning(
289 "Role authorization failed. Required: %s, has: %s",
290 self.access_levels,
291 app_role,
292 )
293 raise HTTPException(
294 status_code=status.HTTP_403_FORBIDDEN,
295 detail="Insufficient permissions",
296 )
298 role = app_role
300 # Profile lookup / auto-creation
301 mongodb_id = await self._resolve_profile(
302 auth0_user_id=auth0_sub,
303 email=email,
304 name=name,
305 role=role,
306 )
308 # Reject deactivated accounts. The admin-staff API soft-deactivates users
309 # by setting status="inactive" on the user_collection doc. Once resolved,
310 # we check the status so a deactivated student's existing session is
311 # rejected on the next /auth/me probe without requiring an Auth0 change.
312 # The profile is the authority for the display name; the access token is
313 # not. Auth0 puts `name` on the ID TOKEN, and this API validates the
314 # ACCESS token — whose claims are aud/azp/exp/gty/iat/iss/scope/sub plus
315 # this tenant's two namespaced claims. So `payload.get("name", "")` above
316 # ALWAYS returned "", and /v1/auth/me advertised a field it never filled,
317 # for every account and both roles (found by Zephyr EI-T436, 2026-09-22).
318 profile_name = await self._check_inactive_status(mongodb_id)
319 name = profile_name or name
321 # Set standardized user_details on request state
322 # uuid is set to mongodb_id (ObjectId string) for backward compatibility
323 # with existing service code that does ObjectId(user_details["uuid"])
324 request.state.user_details = {
325 "uuid": mongodb_id,
326 "name": name,
327 "role": role,
328 "email": email,
329 "auth_provider": "auth0",
330 "auth0_user_id": auth0_sub,
331 "mongodb_id": mongodb_id,
332 }
334 return token
336 async def _check_inactive_status(self, mongodb_id: str) -> str:
337 """Raise HTTP 403 if the resolved user doc has status='inactive', and
338 return the caller's display name.
340 Uses a minimal projection against the already-resolved ObjectId so this
341 is a single indexed lookup. Only an explicit 'inactive' value blocks the
342 request — missing/null/other values pass through.
344 THE NAME COMES BACK FROM HERE ON PURPOSE. It is the only document read
345 on the authenticated path, so taking first_name/last_name from the same
346 projection costs nothing: no extra round trip, no second query. The
347 alternative — changing _resolve_profile's return type — would have meant
348 touching six return points for a value only this caller wants.
350 Returns:
351 "First Last" from the profile, or "" when the lookup fails or the
352 document carries no name. "" is what the old behaviour always
353 produced, so every caller already handles it.
355 Raises:
356 HTTPException 403 when the account is deactivated.
357 """
358 from server.models.users import User
359 from bson import ObjectId
361 try:
362 collection = User.get_pymongo_collection()
363 doc = await collection.find_one(
364 {"_id": ObjectId(mongodb_id)},
365 {"status": 1, "first_name": 1, "middle_name": 1, "last_name": 1},
366 )
367 except Exception as e:
368 logger.warning(
369 "_check_inactive_status lookup failed for %s: %s", mongodb_id, e
370 )
371 return "" # fail-open: don't block a request on a lookup error
373 if doc and doc.get("status") == "inactive":
374 raise HTTPException(
375 status_code=status.HTTP_403_FORBIDDEN,
376 detail="This account has been deactivated. Contact your administrator.",
377 )
379 if not doc:
380 return ""
382 # first + MIDDLE + last, matching how the rest of the product composes a
383 # display name — SearchProfileWidget.jsx does
384 # [first_name, middle_name, last_name].filter(Boolean).join(" ")
385 # so this is the same filter-and-join and /auth/me agrees with the name
386 # shown elsewhere rather than inventing a shorter one.
387 parts = [
388 str(doc.get("first_name") or "").strip(),
389 str(doc.get("middle_name") or "").strip(),
390 str(doc.get("last_name") or "").strip(),
391 ]
392 return " ".join(part for part in parts if part)
394 async def _resolve_profile(
395 self,
396 auth0_user_id: str,
397 email: str,
398 name: str,
399 role: str,
400 ) -> str:
401 """Look up or auto-create user profile in MongoDB. Returns mongodb_id string."""
402 from server.models.users import User
403 from server.services.growthbook.user_scope import find_best_user_document
405 user_doc = await find_best_user_document(
406 auth0_user_id=auth0_user_id,
407 email=email,
408 )
409 if user_doc:
410 # Legacy users without school_id are auto-recovered instead of
411 # hard-403'd (see `_autorecover_school_id` for the policy).
412 await _autorecover_school_id(
413 user_doc=user_doc,
414 role=role,
415 auth0_user_id=auth0_user_id,
416 email=email,
417 )
418 return str(user_doc["_id"])
420 collection = User.get_pymongo_collection()
422 # Legacy Beanie path with repair for malformed documents
423 try:
424 user = await User.find_one(User.auth0_user_id == auth0_user_id)
425 if user:
426 return str(user.id)
427 except Exception:
428 raw_doc = await collection.find_one({"auth0_user_id": auth0_user_id})
429 if raw_doc:
430 updates = {}
431 if not raw_doc.get("last_name"):
432 updates["last_name"] = (
433 name.split(" ", 1)[-1] if name and " " in name else "User"
434 )
435 if not raw_doc.get("first_name"):
436 updates["first_name"] = name.split(" ", 1)[0] if name else "User"
437 if updates:
438 await collection.update_one(
439 {"_id": raw_doc["_id"]}, {"$set": updates}
440 )
441 logger.info("Repaired user profile %s: %s", raw_doc["_id"], updates)
442 await _autorecover_school_id(
443 user_doc=raw_doc,
444 role=role,
445 auth0_user_id=auth0_user_id,
446 email=email,
447 )
448 return str(raw_doc["_id"])
450 # Email fallback — links staff-provisioned profile on first Auth0 login
451 if email:
452 try:
453 user = await User.find_one(User.email == email)
454 except Exception:
455 user = None
456 if user:
457 user.auth0_user_id = auth0_user_id
458 user.auth_provider = "auth0"
459 await user.save()
460 _user_dict = {
461 "_id": user.id,
462 "school_id": getattr(user, "school_id", None),
463 "district_id": getattr(user, "district_id", None),
464 }
465 await _autorecover_school_id(
466 user_doc=_user_dict,
467 role=role,
468 auth0_user_id=auth0_user_id,
469 email=email,
470 )
471 return str(user.id)
472 raw_doc = await collection.find_one({"email": email})
473 if raw_doc:
474 await collection.update_one(
475 {"_id": raw_doc["_id"]},
476 {
477 "$set": {
478 "auth0_user_id": auth0_user_id,
479 "auth_provider": "auth0",
480 }
481 },
482 )
483 await _autorecover_school_id(
484 user_doc=raw_doc,
485 role=role,
486 auth0_user_id=auth0_user_id,
487 email=email,
488 )
489 return str(raw_doc["_id"])
491 # Teachers/students must be staff-provisioned with school_id
492 if role in ("teacher", "student"):
493 logger.warning(
494 "Auth0 login for unprovisioned %s: auth0_user_id=%s email=%s",
495 role,
496 auth0_user_id,
497 email,
498 )
499 raise HTTPException(
500 status_code=status.HTTP_403_FORBIDDEN,
501 detail={
502 "message": "Account not provisioned.",
503 "reason": "missing_school_id",
504 "hint": (
505 "Your school administrator must create your account in the "
506 "staff portal and assign a school before you can sign in."
507 ),
508 },
509 )
511 # Auto-create profile on first Auth0 login (teacher/student only).
512 if role not in APP_ALLOWED_ROLES:
513 raise HTTPException(
514 status_code=status.HTTP_403_FORBIDDEN,
515 detail="This application is for teachers and students only.",
516 )
518 name_parts = name.split(" ", 1) if name else ["", ""]
519 first_name = name_parts[0] or "User"
520 last_name = name_parts[1] if len(name_parts) > 1 else first_name
521 last_name = last_name or first_name
523 new_user = User(
524 first_name=first_name,
525 last_name=last_name,
526 email=email or f"{auth0_user_id.replace('|', '_')}@auth0.placeholder",
527 role=role,
528 status="active",
529 auth0_user_id=auth0_user_id,
530 auth_provider="auth0",
531 )
532 await new_user.insert()
533 logger.info("Auto-created profile for Auth0 user %s", auth0_user_id)
534 return str(new_user.id)