Coverage for apps/accounts/auth.py: 88%
303 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 17:24 +0000
1"""Authentication + authorization: principals, pluggable verifier, Ninja guards.
3Endpoints declare a single required permission via ``requires("...")``; the guard
4authenticates (staff vs borrower realm), resolves the principal, and checks the
5permission. Scope (org-unit / own-loan) is enforced through the helpers here.
6"""
8from __future__ import annotations
10import logging
11from dataclasses import dataclass
12from datetime import timedelta
14import jwt
15from django.conf import settings
16from django.db.models import Q
17from django.utils import timezone
18from jwt import PyJWKClient
19from ninja.security import HttpBearer
21from apps.accounts.models import RolePermission, User, UserRole
22from apps.application_1003.models import Borrower
23from domain.result import AppError, DomainError
25logger = logging.getLogger(__name__)
28@dataclass(frozen=True)
29class StaffPrincipal:
30 user_id: int
31 email: str
32 permissions: frozenset[str]
33 org_unit_ids: frozenset[int]
34 is_hq: bool
35 kind: str = "staff"
36 # The real admin's user id when this principal is an Act As impersonation
37 # (add-staff-act-as); None for an ordinary login. Not consumed anywhere
38 # yet — a cheap hook for a future audit-log feature, same as other
39 # nullable "connect this later" fields in this codebase.
40 acting_admin_id: int | None = None
42 def has(self, permission: str) -> bool:
43 return permission in self.permissions
46@dataclass(frozen=True)
47class BorrowerPrincipal:
48 borrower_email: str
49 loan_ids: frozenset[int]
50 cognito_sub: str | None = None
51 kind: str = "borrower"
54@dataclass(frozen=True)
55class ServicePrincipal:
56 """A system-to-system caller (Cognito client-credentials), not a human user."""
58 client_id: str
59 scopes: frozenset[str]
60 kind: str = "service"
62 def has_scope(self, scope: str) -> bool:
63 return scope in self.scopes
66def effective_permissions(user: User) -> frozenset[str]:
67 """Union of permission codes across all of the user's roles."""
68 role_ids = UserRole.objects.filter(user=user).values_list("role_id", flat=True)
69 codes = RolePermission.objects.filter(role_id__in=role_ids).values_list(
70 "permission__code", flat=True
71 )
72 return frozenset(codes)
75def build_staff_principal(user: User, *, acting_admin_id: int | None = None) -> StaffPrincipal:
76 bindings = list(UserRole.objects.filter(user=user).select_related("org_unit"))
77 org_unit_ids = {b.org_unit_id for b in bindings if b.org_unit_id is not None}
78 # HQ (org-wide) is conferred only by a binding to a headquarters org-unit — NOT by a
79 # null-org (unscoped) binding. A coarse role bound unscoped (e.g. a freshly provisioned
80 # Cognito user with no branch yet) therefore grants no loan scope until a branch is added.
81 is_hq = any(b.org_unit and b.org_unit.is_headquarters for b in bindings)
82 return StaffPrincipal(
83 user_id=user.pk,
84 email=user.email,
85 permissions=effective_permissions(user),
86 org_unit_ids=frozenset(org_unit_ids),
87 is_hq=is_hq,
88 acting_admin_id=acting_admin_id,
89 )
92def build_borrower_principal(email: str, cognito_sub: str | None = None) -> BorrowerPrincipal:
93 # Case-insensitive: borrower accounts store emails lower-case, staff may type them any way.
94 loan_ids = set(Borrower.objects.filter(email__iexact=email).values_list("loan_id", flat=True))
95 return BorrowerPrincipal(
96 borrower_email=email, loan_ids=frozenset(loan_ids), cognito_sub=cognito_sub
97 )
100# --- Pluggable verifiers -------------------------------------------------------
101class DevVerifier:
102 """Dev/test only. Token ``staff:<email>`` or ``borrower:<email>`` maps to a user."""
104 def verify_staff(self, token: str) -> StaffPrincipal | None:
105 if not token or not token.startswith("staff:"):
106 return None
107 user = User.objects.filter(email=token.split(":", 1)[1], is_active=True).first()
108 return build_staff_principal(user) if user else None
110 def verify_borrower(self, token: str) -> BorrowerPrincipal | None:
111 if not token or not token.startswith("borrower:"):
112 return None
113 email = token.split(":", 1)[1]
114 from apps.borrower_portal.models import BorrowerAccount
116 # A disabled borrower account is refused here too, as with local tokens.
117 if BorrowerAccount.objects.filter(email__iexact=email, is_active=False).exists():
118 return None
119 return build_borrower_principal(email)
121 def verify_service(self, token: str, required_scope: str | None = None):
122 """Dev/test M2M. Token ``service:<space-separated scopes>``."""
123 if not token or not token.startswith("service:"):
124 return None
125 scopes = frozenset(token.split(":", 1)[1].split())
126 if required_scope and required_scope not in scopes:
127 return None
128 return ServicePrincipal(client_id="dev-service", scopes=scopes)
131def mint_local_token(user: User, *, acting_admin_id: int | None = None) -> str:
132 """Issue a Lumen-signed login token (AUTH_VERIFIER=local; add-staff-auth-login).
134 ``acting_admin_id`` embeds the real admin's id as a separate ``act_as``
135 claim for an Act As token (add-staff-act-as) — ``sub`` stays the
136 *effective* user (``user``), so permission resolution is unchanged.
137 """
138 now = timezone.now()
139 ttl = timedelta(hours=getattr(settings, "AUTH_LOCAL_TOKEN_TTL_HOURS", 24))
140 payload = {"sub": str(user.pk), "email": user.email, "iat": now, "exp": now + ttl}
141 if acting_admin_id is not None:
142 payload["act_as"] = str(acting_admin_id)
143 return jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")
146BORROWER_REALM = "borrower"
149def mint_borrower_token(account) -> str:
150 """Issue a Lumen-signed borrower token (add-borrower-portal-api design.md D2).
152 Same key and TTL as staff tokens, but carries ``realm: "borrower"`` and the
153 ``BorrowerAccount`` id as ``sub``. ``LocalVerifier.verify_staff`` rejects any token with a
154 realm claim, so a borrower id can never be read as a staff user id.
155 """
156 now = timezone.now()
157 ttl = timedelta(hours=getattr(settings, "AUTH_LOCAL_TOKEN_TTL_HOURS", 24))
158 payload = {
159 "sub": str(account.pk),
160 "email": account.email,
161 "realm": BORROWER_REALM,
162 "iat": now,
163 "exp": now + ttl,
164 }
165 return jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")
168def _decode_local_token(token: str) -> dict | None:
169 try:
170 return jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
171 except Exception as e:
172 logger.warning("Local token verification failed: %s", e)
173 return None
176class LocalVerifier:
177 """Verifies Lumen-signed local login tokens for staff and borrowers. The service realm
178 stays on Cognito/dev."""
180 def verify_staff(self, token: str) -> StaffPrincipal | None:
181 claims = _decode_local_token(token)
182 if not claims:
183 return None
184 # Staff tokens carry no realm. Any realm-tagged token (a borrower's) is refused, or a
185 # borrower account id in ``sub`` would be looked up as a staff user id.
186 if claims.get("realm"):
187 return None
188 sub = claims.get("sub")
189 if not sub: 189 ↛ 190line 189 didn't jump to line 190 because the condition on line 189 was never true
190 return None
191 user = User.objects.filter(pk=sub, is_active=True).first()
192 if user is None:
193 return None
194 act_as = claims.get("act_as")
195 return build_staff_principal(user, acting_admin_id=int(act_as) if act_as else None)
197 def verify_borrower(self, token: str) -> BorrowerPrincipal | None:
198 claims = _decode_local_token(token)
199 if not claims or claims.get("realm") != BORROWER_REALM:
200 return None
201 sub = claims.get("sub")
202 if not sub or not str(sub).isdigit(): 202 ↛ 203line 202 didn't jump to line 203 because the condition on line 202 was never true
203 return None
204 from apps.borrower_portal.models import BorrowerAccount
206 account = BorrowerAccount.objects.filter(pk=int(sub), is_active=True).first()
207 if account is None:
208 return None
209 return build_borrower_principal(account.email, cognito_sub=account.cognito_sub)
211 def verify_service(self, token: str, required_scope: str | None = None):
212 return None
215COGNITO_ROLE_MAP: dict[str, str] = {
216 "SUPER_ADMIN": "Owner",
217 "Admins": "Company Admin",
218 "MLO": "Loan Officer",
219 "Processor": "Processor",
220}
223def sync_user_roles_from_cognito(user: User, groups: list[str]) -> None:
224 """Assign the Cognito-mapped coarse role to the user — **DB-first branch scoping**.
226 Identity + coarse role come from Cognito; branch (org-unit) membership lives in the Lumen
227 DB (``OrgUnit`` + ``UserRole``) and is **not** defaulted to headquarters. The coarse role
228 is bound **unscoped** (``org_unit=None``), which — since a null-org binding no longer
229 confers HQ (see ``build_staff_principal``) — grants the role's permissions but **no loan
230 scope**. A Lumen admin then assigns one or more branches (``UserRole`` rows), each of which
231 adds loan scope. This is the safe default: no company-wide visibility on provisioning.
232 """
233 from apps.accounts.models import Role, UserRole
235 for group in groups:
236 role_name = COGNITO_ROLE_MAP.get(group)
237 if role_name: 237 ↛ 235line 237 didn't jump to line 235 because the condition on line 237 was always true
238 role, _ = Role.objects.get_or_create(name=role_name)
239 UserRole.objects.get_or_create(user=user, role=role, org_unit=None)
242def assign_user_branch(user: User, role_name: str, org_unit) -> None:
243 """Admin branch-assignment: bind ``role_name`` for ``user`` within ``org_unit``.
245 This is how a provisioned user gains loan scope (DB-first). Callable from the Django
246 admin, an invite/onboarding flow, or a future admin API.
247 """
248 from apps.accounts.models import Role, UserRole
250 role, _ = Role.objects.get_or_create(name=role_name)
251 UserRole.objects.get_or_create(user=user, role=role, org_unit=org_unit)
254class CognitoVerifier:
255 """Verifies AWS Cognito ID tokens (RS256 via JWKS, issuer + client ID), maps email and
256 cognito:groups to staff roles, and handles JIT user provisioning.
257 """
259 def __init__(self, jwks_client: PyJWKClient | None = None):
260 self.region = getattr(settings, "COGNITO_AWS_REGION", "us-east-1")
261 self.user_pool_id = getattr(settings, "COGNITO_USER_POOL_ID", "")
262 self.client_id = getattr(settings, "COGNITO_APP_CLIENT_ID", "")
263 self.issuer = (
264 getattr(settings, "COGNITO_ISSUER", "")
265 or f"https://cognito-idp.{self.region}.amazonaws.com/{self.user_pool_id}"
266 )
267 self.jwks_url = f"{self.issuer}/.well-known/jwks.json"
268 self._jwks = jwks_client or PyJWKClient(self.jwks_url)
270 def _claims(self, token: str, *, verify_audience: bool = True) -> dict | None:
271 has_iss = bool(self.user_pool_id or getattr(settings, "COGNITO_ISSUER", ""))
272 # Production MUST verify issuer, and (for user ID tokens) audience. Fail closed if the
273 # verifier is active but misconfigured, rather than silently skipping verification.
274 if not settings.DEBUG and (not has_iss or (verify_audience and not self.client_id)):
275 logger.error(
276 "Cognito verifier misconfigured for production: issuer and audience are required"
277 )
278 return None
279 try:
280 signing_key = self._jwks.get_signing_key_from_jwt(token)
281 # Service (client-credentials) access tokens carry no `aud`; skip audience there.
282 verify_aud = verify_audience and bool(self.client_id)
283 options = {"verify_aud": verify_aud, "verify_iss": has_iss}
284 kwargs: dict = {"algorithms": ["RS256"], "options": options}
285 if has_iss: 285 ↛ 287line 285 didn't jump to line 287 because the condition on line 285 was always true
286 kwargs["issuer"] = self.issuer
287 if verify_aud: 287 ↛ 289line 287 didn't jump to line 289 because the condition on line 287 was always true
288 kwargs["audience"] = self.client_id
289 return jwt.decode(token, signing_key.key, **kwargs)
290 except Exception as e:
291 logger.warning("Cognito token verification failed: %s", e)
292 return None
294 def verify_staff(self, token: str) -> StaffPrincipal | None:
295 claims = self._claims(token)
296 if not claims:
297 return None
298 sub = claims.get("sub")
299 email = claims.get("email")
300 if not sub: 300 ↛ 301line 300 didn't jump to line 301 because the condition on line 300 was never true
301 logger.warning("Cognito token has no sub claim")
302 return None
304 groups = claims.get("cognito:groups", [])
305 if isinstance(groups, str): 305 ↛ 306line 305 didn't jump to line 306 because the condition on line 305 was never true
306 groups = [groups]
308 # Prevent pure applicant/borrower tokens from the staff realm.
309 has_staff_group = any(g in COGNITO_ROLE_MAP for g in groups)
310 if not has_staff_group and ("Applicants" in groups or not groups):
311 logger.warning("Cognito token lacks staff groups for %s: %s", email, groups)
312 return None
314 # Resolve by the immutable `sub`; fall back to email (pre-Cognito users), then JIT.
315 user = User.objects.filter(cognito_sub=sub).first()
316 if user is None and email:
317 user = User.objects.filter(email=email).first()
318 if user is None:
319 if not email: 319 ↛ 320line 319 didn't jump to line 320 because the condition on line 319 was never true
320 logger.warning("Cannot provision a Cognito user without an email claim")
321 return None
322 name = claims.get("name", "") or ""
323 first_name = claims.get("given_name") or (name.split(" ")[0] if name else "Staff")
324 last_name = claims.get("family_name") or (
325 name.split(" ", 1)[1] if " " in name else "User"
326 )
327 user = User.objects.create_user(
328 email=email, first_name=first_name, last_name=last_name, is_staff=True
329 )
330 elif not user.is_active: 330 ↛ 331line 330 didn't jump to line 331 because the condition on line 330 was never true
331 return None
333 # Persist / backfill the `sub` link — the cross-DB key to Tireless's `users.cognito_id`.
334 if user.cognito_sub != sub:
335 user.cognito_sub = sub
336 user.save(update_fields=["cognito_sub"])
338 if groups: 338 ↛ 341line 338 didn't jump to line 341 because the condition on line 338 was always true
339 sync_user_roles_from_cognito(user, groups)
341 return build_staff_principal(user)
343 def verify_borrower(self, token: str) -> BorrowerPrincipal | None:
344 claims = self._claims(token)
345 if not claims: 345 ↛ 346line 345 didn't jump to line 346 because the condition on line 345 was never true
346 return None
347 groups = claims.get("cognito:groups", [])
348 if isinstance(groups, str): 348 ↛ 349line 348 didn't jump to line 349 because the condition on line 348 was never true
349 groups = [groups]
350 # Borrower realm: the token must be an Applicant and carry no staff group.
351 if "Applicants" not in groups or any(g in COGNITO_ROLE_MAP for g in groups):
352 return None
353 email = claims.get("email")
354 if not email: 354 ↛ 355line 354 didn't jump to line 355 because the condition on line 354 was never true
355 return None
356 return build_borrower_principal(email, cognito_sub=claims.get("sub"))
358 def verify_service(self, token: str, required_scope: str | None = None):
359 """Verify a Cognito client-credentials access token (M2M): no user, scope-checked."""
360 claims = self._claims(token, verify_audience=False)
361 if not claims or claims.get("token_use") != "access":
362 return None
363 scopes = frozenset((claims.get("scope") or "").split())
364 if required_scope and required_scope not in scopes:
365 logger.warning("Service token missing required scope %s", required_scope)
366 return None
367 return ServicePrincipal(
368 client_id=claims.get("client_id") or claims.get("sub", ""), scopes=scopes
369 )
372_cognito_verifier: CognitoVerifier | None = None
373_local_verifier: LocalVerifier | None = None
376def get_verifier():
377 global _cognito_verifier, _local_verifier
378 mode = getattr(settings, "AUTH_VERIFIER", "dev")
379 if mode == "cognito": 379 ↛ 380line 379 didn't jump to line 380 because the condition on line 379 was never true
380 if _cognito_verifier is None:
381 _cognito_verifier = CognitoVerifier()
382 return _cognito_verifier
383 if mode == "local":
384 if _local_verifier is None:
385 _local_verifier = LocalVerifier()
386 return _local_verifier
387 return DevVerifier()
390# --- Ninja guards --------------------------------------------------------------
391class StaffAuth(HttpBearer):
392 def __init__(self, permission: str | None = None):
393 self.permission = permission
394 super().__init__()
396 def authenticate(self, request, token):
397 principal = get_verifier().verify_staff(token)
398 if principal is None:
399 return None # -> 401
400 if self.permission and not principal.has(self.permission):
401 raise DomainError([AppError("FORBIDDEN", f"Missing permission: {self.permission}")])
402 request.principal = principal
403 return principal
406class BorrowerAuth(HttpBearer):
407 def authenticate(self, request, token):
408 principal = get_verifier().verify_borrower(token)
409 if principal is None: 409 ↛ 410line 409 didn't jump to line 410 because the condition on line 409 was never true
410 return None
411 request.principal = principal
412 return principal
415def requires(permission: str) -> StaffAuth:
416 return StaffAuth(permission)
419def borrower() -> BorrowerAuth:
420 return BorrowerAuth()
423class PortalAuth(HttpBearer):
424 """Borrower-only portal endpoints (add-borrower-portal-api design.md D3).
426 A borrower principal with a portal account is let through and ``request.borrower_account``
427 is set. A valid *staff* token is refused with FORBIDDEN (the portal contract: staff get
428 403); anything else is 401. Cognito borrowers get an account on first use.
429 """
431 def authenticate(self, request, token):
432 verifier = get_verifier()
433 principal = verifier.verify_borrower(token)
434 if principal is None:
435 if verifier.verify_staff(token) is not None:
436 raise DomainError(
437 [AppError("FORBIDDEN", "The borrower portal is for borrowers only.")]
438 )
439 return None
440 from apps.borrower_portal.services import account_for_principal
442 account = account_for_principal(principal)
443 if account is None:
444 return None
445 request.principal = principal
446 request.borrower_account = account
447 return principal
450def portal() -> PortalAuth:
451 return PortalAuth()
454def borrower_loan(request):
455 """The signed-in borrower's portal loan, or None (design.md D3).
457 The newest loan on which they are the primary borrower, else the newest on which they are
458 any borrower. Only loans in ``principal.loan_ids`` are ever considered, so a borrower can
459 never reach another borrower's loan.
460 """
461 from apps.core_los.models import Loan
463 principal = request.principal
464 if not principal.loan_ids:
465 return None
466 loans = Loan.objects.filter(pk__in=principal.loan_ids).order_by("-created_at", "-pk")
467 primary = loans.filter(
468 borrowers__is_primary=True, borrowers__email__iexact=principal.borrower_email
469 ).first()
470 return primary or loans.first()
473class ServiceAuth(HttpBearer):
474 """Authenticates a system-to-system caller (Cognito client-credentials M2M)."""
476 def __init__(self, scope: str | None = None):
477 self.scope = scope or getattr(settings, "COGNITO_M2M_SCOPE", "") or None
478 super().__init__()
480 def authenticate(self, request, token):
481 verify_service = getattr(get_verifier(), "verify_service", None)
482 if verify_service is None:
483 return None # verifier has no M2M path -> 401
484 principal = verify_service(token, self.scope)
485 if principal is None:
486 return None
487 request.principal = principal
488 return principal
491def requires_service(scope: str | None = None) -> ServiceAuth:
492 """Guard for service-only endpoints (loan handoff, condition/fraud sync callbacks)."""
493 return ServiceAuth(scope)
496# --- Scope enforcement ---------------------------------------------------------
497def staff_loan_queryset(principal: StaffPrincipal, scope: str = "all"):
498 """Loans visible to ``principal``, optionally narrowed by pipeline ``scope``.
500 ``all`` (default) is the existing org-unit/HQ visibility rule, unchanged for
501 every caller that doesn't pass ``scope``. ``team`` and ``all`` compute the same
502 base set today (see add-loan-pipeline-api design.md); ``me`` narrows that base to
503 loans where the principal holds one of the loan-team roles.
504 """
505 from apps.core_los.models import Loan
507 if principal.is_hq or "VIEW_ALL_PIPELINE" in principal.permissions:
508 base = Loan.objects.all()
509 else:
510 base = Loan.objects.filter(org_unit_id__in=principal.org_unit_ids)
511 if scope == "me":
512 return base.filter(
513 Q(loan_officer_id=principal.user_id)
514 | Q(additional_lo_id=principal.user_id)
515 | Q(processor_id=principal.user_id)
516 | Q(loa_id=principal.user_id)
517 )
518 return base
521def ensure_staff_scope(principal: StaffPrincipal, loan) -> None:
522 if principal.is_hq or "VIEW_ALL_PIPELINE" in principal.permissions:
523 return
524 if loan.org_unit_id not in principal.org_unit_ids: 524 ↛ exitline 524 didn't return from function 'ensure_staff_scope' because the condition on line 524 was always true
525 raise DomainError([AppError("FORBIDDEN", "Loan is outside your org-unit scope.")])