internal.auth.middleware

  1"""Middlewares to include in routes for auto authorisation."""
  2
  3from typing import Annotated
  4
  5from fastapi import HTTPException, Security, status
  6from fastapi.security import (
  7    HTTPAuthorizationCredentials,
  8    HTTPBasic,
  9    HTTPBasicCredentials,
 10    HTTPBearer,
 11)
 12from pydantic import BaseModel
 13
 14from internal.auth.security import check_password
 15from internal.database.dependency import database_dependency
 16from internal.queries.admin import AsyncQuerier as AdminQuerier
 17from internal.queries.models import UserRole
 18from internal.queries.token import AsyncQuerier as TokenQuerier
 19from internal.queries.token import GetSessionByTokenRow
 20from internal.queries.user import AsyncQuerier as UserQuerier
 21from internal.settings.env import auth_settings
 22
 23
 24class BasicAuthResponse(BaseModel):
 25    """Response when user got authorised with email and password."""
 26
 27    user_id: int
 28    role: UserRole
 29
 30
 31async def basic_auth(
 32    conn: database_dependency, credentials: HTTPBasicCredentials = Security(HTTPBasic())
 33) -> BasicAuthResponse:
 34    """Fetches user id for endpoint with basic auth.
 35
 36    Args:
 37      conn: database connection
 38      credentials: credentials (email(username), password) passed from request header
 39
 40    Returns:
 41      user id and role if user authenticated
 42
 43    Raises:
 44        HTTPException: if user wasn't found in the database or password incorrect
 45    """
 46    user = await UserQuerier(conn).get_user_login(email=credentials.username)
 47    if not user:
 48        raise HTTPException(
 49            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials"
 50        )
 51    if not check_password(credentials.password, user.pw_hash):
 52        raise HTTPException(
 53            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials"
 54        )
 55
 56    if user.role == UserRole.ADMIN:
 57        admin_info = await AdminQuerier(conn).get_admin(user_id=user.user_id)
 58        if admin_info and not admin_info.active:
 59            raise HTTPException(
 60                status_code=status.HTTP_403_FORBIDDEN,
 61                detail="Admin account deactivated",
 62            )
 63
 64    return BasicAuthResponse(user_id=user.user_id, role=user.role)
 65
 66
 67async def bearer_auth(
 68    conn: database_dependency,
 69    credentials: HTTPAuthorizationCredentials = Security(HTTPBearer()),
 70) -> GetSessionByTokenRow:
 71    """Fetches user session from given token credentials.
 72
 73    Args:
 74      conn: database connectionn
 75      credentials: credentials (token) gotten from headers
 76
 77    Returns:
 78      user session with user and session information
 79
 80    Raises:
 81      HTTPException: if user wasn't found in the database
 82    """
 83    session = await TokenQuerier(conn).get_session_by_token(
 84        token=credentials.credentials
 85    )
 86    if not session:
 87        raise HTTPException(
 88            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid session"
 89        )
 90    return session
 91
 92
 93def consumer_auth(
 94    session: GetSessionByTokenRow = Security(bearer_auth),
 95) -> GetSessionByTokenRow:
 96    """Authentisate consumer in a middleware.
 97
 98    Args:
 99      session: user session from bearer authentication
100
101    Returns:
102      user session if user was successfully authenticated
103
104    Raises:
105      HTTPException: if user is not a consumer
106    """
107    if session.role == UserRole.CONSUMER:
108        return session
109    raise HTTPException(
110        status_code=status.HTTP_403_FORBIDDEN, detail="Not authorised as consumer"
111    )
112
113
114def seller_auth(
115    session: GetSessionByTokenRow = Security(bearer_auth),
116) -> GetSessionByTokenRow:
117    """Authentisate seller in a middleware.
118
119    Args:
120      session: user session from bearer authentication
121
122    Returns:
123      user session if user was successfully authenticated
124
125    Raises:
126      HTTPException: if user is not a seller
127    """
128    if session.role == UserRole.SELLER:
129        return session
130    raise HTTPException(
131        status_code=status.HTTP_403_FORBIDDEN, detail="Not authorised as seller"
132    )
133
134
135async def admin_auth(
136    conn: database_dependency, session: GetSessionByTokenRow = Security(bearer_auth)
137) -> GetSessionByTokenRow:
138    """Authentisate admin in a middleware.
139
140    Args:
141      conn: database connection
142      session: user session from bearer authentication
143
144    Returns:
145      user session if user was successfully authenticated
146
147    Raises:
148      HTTPException: if user is not a admin
149    """
150    if session.role != UserRole.ADMIN:
151        raise HTTPException(
152            status_code=status.HTTP_403_FORBIDDEN, detail="Not authorised as admin"
153        )
154
155    admin_info = await AdminQuerier(conn).get_admin(user_id=session.user_id)
156    if not admin_info or not admin_info.active:
157        raise HTTPException(
158            status_code=status.HTTP_403_FORBIDDEN, detail="Admin account deactivated"
159        )
160
161    return session
162
163
164def root_auth(credentials: HTTPBasicCredentials = Security(HTTPBasic())) -> None:
165    """Root user authentication.
166
167    Args:
168        credentials: root user credentials
169
170    Raises:
171        HTTPException: if not root user
172    """
173    if ("" in {auth_settings.root_username, auth_settings.root_password}) or not (
174        credentials.username == auth_settings.root_username
175        and credentials.password == auth_settings.root_password
176    ):
177        raise HTTPException(
178            status.HTTP_401_UNAUTHORIZED, "Not authorised as root admin"
179        )
180
181
182BasicAuthDep = Annotated[BasicAuthResponse, Security(basic_auth)]
183BearerAuthDep = Annotated[GetSessionByTokenRow, Security(bearer_auth)]
184ConsumerAuthDep = Annotated[GetSessionByTokenRow, Security(consumer_auth)]
185SellerAuthDep = Annotated[GetSessionByTokenRow, Security(seller_auth)]
186AdminAuthDep = Annotated[GetSessionByTokenRow, Security(admin_auth)]
187RootAuthDep = Annotated[None, Security(root_auth)]