Function get_current_user

apps/api/src/security/auth.py:345–452

(
    request: Request,
    db_session: AsyncSession = Depends(get_db_session),
)

async def get_current_user(
    request: Request,
    db_session: AsyncSession = Depends(get_db_session),
) -> Union[PublicUser, APITokenUser, SuperadminAPITokenUser, AnonymousUser]:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

    # Step 1: Check for API token (Bearer lh_... or Bearer lh_sa_...)
    auth_header = request.headers.get("Authorization", "").strip()
    auth_lower = auth_header.lower()

    # ORDER MATTERS: the superadmin prefix "lh_sa_" also starts with "lh_",
    # so the broader org-token branch below would shadow it. Keep this
    # branch above the org-token branch — do not reorder for "alphabetical
    # cleanliness" — or every superadmin token will silently fall through
    # to org-token validation and fail.
    if auth_lower.startswith("bearer lh_sa_"):
        token = auth_header[7:].strip()  # strip "Bearer "
        sa_user = await validate_superadmin_api_token(token, db_session)
        if sa_user:
            # No _verify_api_token_org_boundary call: superadmin tokens are
            # cross-org by design and have no org to bind to.
            request.state.user = sa_user
            request.state.is_api_token = True
            request.state.is_superadmin_api_token = True
            return sa_user
        raise credentials_exception

    # Case-insensitive check for "Bearer " prefix with lh_ token (org-scoped)
    if auth_lower.startswith("bearer lh_"):
        token = auth_header[7:].strip()  # Remove "Bearer " prefix and trim
        api_token_user = await validate_api_token(token, db_session)
        if api_token_user:
            # Verify org boundary: if the URL contains an org_id or org_slug,
            # ensure it matches the token's organization
            await _verify_api_token_org_boundary(request, api_token_user, db_session)
            request.state.user = api_token_user
            request.state.is_api_token = True
            return api_token_user
        raise credentials_exception

    # Step 2: Fall back to JWT logic using PyJWT
    token = extract_jwt_from_request(request)
    username = None

    if token:
        payload = decode_jwt(token)
        if payload:
            # Reject tokens minted for a single-purpose flow (e.g. magic-link
            # one-time sign-in, password reset, email verification). Those tokens
            # are only valid at their specific consume endpoint — allowing them as
            # session tokens would let an intercepted single-use token act as a
            # full session for its entire TTL.
            token_purpose = payload.get("purpose")
            SINGLE_USE_PURPOSES = {