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

131 lines
2.8 KiB
Python

"""
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.
Status:
- 400: `kiwi-vpn` not installed
"""
# 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.
Status:
- 403: invalid auth token, or user not found
"""
# 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,
) -> User:
"""
Get a user by name.
Status:
- 404: user not found
"""
# fail if device doesn't exist
if (user := User.get(user_name)) is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
return user
async def get_device_by_id(
device_id: int,
) -> Device:
"""
Get a device by ID.
Status:
- 404: device not found
"""
# 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