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

"""
/user endpoints.
"""

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel

from ..config import Config
from ..db import TagValue, User, UserCreate, UserRead
from ._common import (Responses, get_current_config, get_current_user,
                      get_user_by_name)

router = APIRouter(prefix="/user", tags=["user"])


class Token(BaseModel):
    """
    Response model for issuing tokens.
    """

    access_token: str
    token_type: str


@router.post(
    "/authenticate",
    responses={
        status.HTTP_200_OK: Responses.OK,
        status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
        status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
    },
    response_model=Token,
)
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    current_config: Config = Depends(get_current_config),
):
    """
    POST ./authenticate: Authenticate a user. Issues a bearer token.

    Status:

    - 401: username/password is incorrect
    """

    # try logging in
    if (user := User.authenticate(
        name=form_data.username,
        password=form_data.password,
    )) is None:
        # authentication failed
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

    # authentication succeeded
    access_token = await current_config.jwt.create_token(user.name)
    return {"access_token": access_token, "token_type": "bearer"}


@router.get(
    "/current",
    responses={
        status.HTTP_200_OK: Responses.OK,
        status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
        status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
        status.HTTP_403_FORBIDDEN: Responses.NEEDS_USER,
    },
    response_model=UserRead,
)
async def get_current_user_route(
    current_user: User = Depends(get_current_user),
):
    """
    GET ./current: Respond with the currently logged-in user.
    """

    return current_user


@router.post(
    "",
    responses={
        status.HTTP_201_CREATED: Responses.ENTRY_ADDED,
        status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
        status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
        status.HTTP_409_CONFLICT: Responses.ENTRY_EXISTS,
    },
    response_model=UserRead,
    status_code=status.HTTP_201_CREATED,
)
async def add_user(
    user: UserCreate,
    current_user: User = Depends(get_current_user),
) -> User:
    """
    POST ./: Create a new user in the database.

    Status:

    - 403: no user permission to create user
    - 409: user could not be created
    """

    # check permissions
    try:
        current_user.check_create(User)

    except PermissionError as e:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN) from e

    # create the new user
    new_user = User.create(user=user)

    # fail if creation was unsuccessful
    if new_user is None:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT)

    new_user.add_tags([TagValue.login])
    new_user.update()

    # return the created user on success
    return new_user


@router.delete(
    "/{user_name}",
    responses={
        status.HTTP_200_OK: Responses.OK,
        status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
        status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
    },
    response_model=User,
)
async def remove_user(
    current_user: User = Depends(get_current_user),
    user: User = Depends(get_user_by_name),
):
    """
    DELETE ./{user_name}: Remove a user from the database.

    Status:

    - 403: no user permission to admin user
    """

    # check permissions
    try:
        current_user.check_admin(user)

    except PermissionError as e:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN) from e

    # delete user
    user.delete()


@router.post(
    "/{user_name}/tags",
    responses={
        status.HTTP_201_CREATED: Responses.ENTRY_ADDED,
        status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
        status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
    },
    status_code=status.HTTP_201_CREATED,
)
async def extend_tags(
    tags: list[TagValue],
    current_user: User = Depends(get_current_user),
    user: User = Depends(get_user_by_name),
):
    """
    POST ./{user_name}/tags: Add tags to a user.

    Status:

    - 403: no user permission to admin user
    """

    # check permissions
    try:
        current_user.check_admin(user)

    except PermissionError as e:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN) from e

    # change user
    user.add_tags(tags)
    user.update()


@router.delete(
    "/{user_name}/tags",
    responses={
        status.HTTP_200_OK: Responses.OK,
        status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
        status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
    },
)
async def remove_tags(
    tags: list[TagValue],
    current_user: User = Depends(get_current_user),
    user: User = Depends(get_user_by_name),
):
    """
    DELETE ./{user_name}/tags: Remove tags from a user.

    Status:

    - 403: no user permission to admin user
    """

    # check permissions
    try:
        current_user.check_admin(user)

    except PermissionError as e:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN) from e

    # change user
    user.remove_tags(tags)
    user.update()