""" Common dependencies for routers. """ from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from ..config import Config, Settings from ..db import Device, User oauth2_scheme = OAuth2PasswordBearer( tokenUrl=f"{Settings._.api_v1_prefix}/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_PERMISSION = { "description": "Operation not permitted", "content": None, } ENTRY_ADDED = { "description": "Entry added to database", "content": None, } ENTRY_EXISTS = { "description": "Entry exists in database", "content": None, } ENTRY_DOESNT_EXIST = { "description": "Entry does not exist in database", "content": None, } async def get_current_config( current_config: Config | None = Depends(Config.load), ) -> Config: """ Get the current configuration if it exists. """ # fail if not configured if current_config is None: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST) return current_config async def get_current_user( token: str = Depends(oauth2_scheme), current_config: Config = Depends(get_current_config), ) -> User: """ Get the currently logged-in user if it exists. """ # don't use error 404 here - possible user enumeration # fail if not requested by a user if (username := await current_config.jwt.decode_token(token)) is None: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN) if (user := User.get(username)) is None: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN) return user async def get_user_by_name( user_name: str, _: Config = Depends(get_current_config), ) -> User | None: """ Get a user by name. """ return User.get(user_name) async def get_device_by_id( device_id: int, current_config: Config = Depends(get_current_config), ) -> Device | None: # can't connect to an unconfigured database if current_config is None: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST) # fail if device doesn't exist if (device := Device.get(device_id)) is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND) return device