FastAPI Depends() Plumbing for Auth & RBAC (get_current_user)

What? (Concept Overview)

FastAPI’s Depends() plumbing composes auth as a function argument rather than middleware: an OAuth2PasswordBearer URL-declares the token endpoint, a get_current_user dependency parses the Authorization: Bearer … header, decodes the JWT, and returns the authenticated user to the handler. Every protected route just adds Depends(get_current_active_user) and the security policy is enforced uniformly.

Project Context

The FCA Support Agent’s app/api/deps.py exposes get_current_user and get_current_active_user. RBAC scopes (read:accounts, write:messages, etc.) are embedded in the JWT payload at issue time (app/api/routes/auth.py) and validated by SecurityService.verify_token — single source of truth across /api/v1/messages and /api/v1/admin routers.

How? (Quick Reference Blocks)

3.1 OAuth2Bearer Token URL Declaration

# app/api/deps.py
from fastapi.security import OAuth2PasswordBearer
 
# `tokenUrl` is used by the OpenAPI docs to render the "Authorize" button.
# Must match the actual login endpoint in app/api/routes/auth.py.
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

3.2 get_current_user — Parse → Verify → Lookup

# app/api/deps.py
from fastapi import Depends, HTTPException, status
from jose import JWTError, jwt
from app.services.security_service import SecurityService, oauth2_scheme
 
async def get_current_user(
    token: str = Depends(oauth2_scheme),
) -> "Customer":
    payload = SecurityService().verify_token(token)
    if payload is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    user_id = payload.get("sub")
    if user_id is None:
        raise HTTPException(status_code=401, detail="Invalid token payload")
 
    # Lazy-fetch the user — services own the DB session lifecycle.
    async with CustomerService() as svc:
        user = await svc.repo.get_by_id(user_id)
    if user is None:
        raise HTTPException(status_code=401, detail="User not found")
    return user

3.3 get_current_active_user — Active-Account Gate

# app/api/deps.py
async def get_current_active_user(
    current_user = Depends(get_current_user),
) -> "Customer":
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

3.4 RBAC Scope Enforcement

# app/api/deps.py (illustrative Security dependency factory)
from fastapi import Security
from typing import List
 
def require_scopes(required: List[str]):
    async def _check(
        token: str = Depends(oauth2_scheme),
    ):
        payload = SecurityService().verify_token(token)
        scopes = (payload or {}).get("scopes", [])
        if not all(s in scopes for s in required):
            raise HTTPException(
                status_code=403,
                detail=f"Missing required scopes: {required}",
            )
        return payload
    return _check
 
# Usage in a router:
@router.post("/admin/scary-action", dependencies=[Security(require_scopes(["admin"]))])
async def scary_action(): ...

Why? (Parameter Breakdown

• OAuth2PasswordBearer instead of raw header parsing — Generates the OpenAPI “Authorize” UI for free; clients can paste a Bearer token and exercise protected endpoints from the docs page.
• Returning the Customer (or None) instead of the token string — Callers receive a typed object. current_user.id is enough; no payload["sub"] ceremony per handler.
• Lazy DB lookup inside get_current_user — Verifying the JWT signature is fast (HMAC); verifying the user is still active is the DB round-trip. The two-stage pipeline mirrors Express.js / Spring Security patterns.
• WWW-Authenticate: Bearer header on 401** — RFC 6750 says clients should be told how to authenticate. Without it, clients can’t distinguish “no token” from “expired token”.
• Security(...) vs Depends(...) for RBAC — Security() is the documented way to attach security-scoped dependencies; it surfaces them correctly in OpenAPI under securitySchemes.
• Scopes as List[str] in the JWT payload — Allow coarse-grained authorisation: “admin”, “operator”, “user”. A scope-upgrade is a token re-issue with no code/deploy needed.

Common Pitfalls

1. Reading payload["sub"] without type-checking. A legacy token issued before the schema change may have sub as an integer; comparison breaks silently. Cast + validate: user_id = int(payload["sub"]) inside a try/except (KeyError, TypeError, ValueError).
2. Not including the lookup error in WWW-Authenticate — RFC 6750 expects error="invalid_token" (and error_description=) so OAuth-aware clients can show “session expired, please log in again” instead of “auth failed”.

Real-World Interview Prep

Q1: Why split get_current_user from get_current_active_user?

A: Two-stage validation lets you opt out at the right level. get_current_user enforces “the token is valid and the user exists” (every protected endpoint needs this). get_current_active_user adds “the user isn’t disabled” — useful for routes that need it (e.g., /api/v1/messages) but unnecessary for the token-refresh endpoint itself. Without the split, every disabled user is locked out of token refresh → forced logout loop.

Q2: How do you invalidate a token issued to a former employee without rotating the JWT secret?

A: Use a token-blacklist (Redis SET of revoked jti claims) checked inside get_current_user. Pattern: at logout, push the JWT’s jti into revoked_tokens with TTL = remaining token lifetime. get_current_user does if payload["jti"] in redis.smembers("revoked_tokens"): raise 401. Alternative: short access tokens (e.g., 5 min) with refresh tokens that DO check the blacklist; rotation frequency reduces blast radius.

Q3: What does OAuth2PasswordBearer actually do behind the scenes?

A: Just declares a tokenUrl for OpenAPI. It does NOT enforce any auth. The enforcement is whatever you put in your dependency function. Common misconception: people think installing the scheme on a route magically rejects unauthenticated calls. It doesn’t — you still need Depends(oauth2_scheme) somewhere to extract the bearer token.

Top-to-Bottom Code Walkthrough (app/api/deps.py)

This file holds the plumbing that every protected endpoint reuses. FastAPI’s Depends() runs these once per request; if they raise, the endpoint is never called.

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

• A Security() dependency that tells FastAPI: “the Authorization: Bearer <token> header on every request is parsed here. The token-endpoint is /api/v1/auth/login (used for OpenAPI docs only).”
• It exposes a token: str = Depends(oauth2_scheme) parameter inside get_current_user.

get_current_user(token, db) -> Customer

1. Decode the JWT: payload = jwt.decode(token, settings.secret_key, algorithms=[settings.jwt_algorithm]).
   • The algorithms=[...] argument prevents algorithm-confusion attacks (e.g., a token forged with alg=none).
   • If the signature is invalid OR the token is expired, jwt.decode raises JWTError which propagates as a 401.
2. Lookup the user: user_id = int(payload.get("sub")) — sub is the standard JWT “subject” claim, set to the Customer’s primary key at login time. user = await db.execute(select(Customer).where(Customer.id == user_id)).
3. Authorization hallucination guard: security_service.verify_scopes(payload.get("scopes", []), required_scopes) — checks that each scope used in Depends(..., scopes=["write:messages"]) is present in the JWT payload. This is RBAC at the dependency level, no per-endpoint if/else.

get_current_active_user(user)

• Tiny wrapper: if not user.is_active: raise HTTPException(400, "Inactive user"). Some endpoints (like /health) need the user but not the active check; this keeps the choice with the endpoint, not the dependency.

How endpoints consume these

@router.post("/messages")
async def send_message(
    body: MessageBody,
    current_user: Customer = Depends(get_current_active_user),
    db: AsyncSession = Depends(get_db),
):
    if body.customer_id != current_user.customer_id:
        raise HTTPException(403, "IDOR")  # IDOR guard

The endpoint’s only authentication concern is current_user.customer_id != body.customer_id — that’s the IDOR check.

Common Pitfalls

Reading the JWT with jwt.decode(token, SECRET_KEY) (no algorithms=) in 2024+. Always pass algorithms=[settings.jwt_algorithm] — without it, an attacker can submit a forged token signed with alg=none and bypass everything.

Storing the password hashed in the JWT payload — tokens are readable on the client; secrets belong on the server.

Forgetting to mark the token type="access" lets a refresh token act as an access token. Always include a type claim.

Real-World Interview Prep

Q1: Why is current_user: Customer = Depends(get_current_active_user) better than getting the user inside the route body?

A: It’s testable (swap the dependency in pytest with a mock user), DRY (every endpoint gets the same logic), and declarative (route signatures read like contracts).

Q2: When would you choose HTTP cookies over OAuth2PasswordBearer?

A: Cookies for browser-only SPAs where same-origin applies; Bearer for mobile + CLI + 3rd-party API. Bearer travels correctly across hosts; cookies don’t.

Q3: How do you handle token refresh without forcing the client to re-login?

A: Emit two tokens at login: short-lived (5 min) access_token and longer (7 day) refresh_token. Client calls /auth/refresh when the access token expires. Both are JWTs; the refresh token is stored server-side so it can be revoked.