kiwi-vpn/api/kiwi_vpn_api/routers/_common.py

"""
Common dependencies for routers.
"""


from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

from ..config import Config
from ..db import Connection
from ..db.schemas import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="user/authenticate")


class Responses:
    """
    Just a namespace.

    Describes API response status codes.
    """

    OK = {
        "content": None,
    }
    INSTALLED = {
        "description": "kiwi-vpn already installed",
        "content": None,
    }
    NOT_INSTALLED = {
        "description": "kiwi-vpn not installed",
        "content": None,
    }
    NEEDS_USER = {
        "description": "Must be logged in",
        "content": None,
    }
    NEEDS_ADMIN = {
        "description": "Must be admin",
        "content": None,
    }
    ENTRY_EXISTS = {
        "description": "Entry exists in database",
        "content": None,
    }


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session | None = Depends(Connection.get),
    current_config: Config | None = Depends(Config.load),
):
    """
    Get the currently logged-in user from the database.
    """

    # can't connect to an unconfigured database
    if current_config is None:
        return None

    username = await current_config.jwt.decode_token(token)
    user = User.from_db(db, username)

    return user