from typing import Optional
from datetime import timedelta, datetime
from pydantic import BaseModel
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status as http_status, Header
from fastapi.security import OAuth2PasswordBearer
from nacsos_data.models.users import UserModel
from nacsos_data.models.projects import ProjectPermissionsModel, ProjectPermission
from nacsos_data.db.crud.users import read_user_by_name as crud_get_user_by_name, read_user_by_id, read_user_by_name
from nacsos_data.db.crud.projects import read_project_permissions_for_user as crud_get_project_permissions_for_user
from server.api.errors import MissingInformationError
from server.data import db_engine
from server.util.config import settings
from server.util.logging import get_logger
logger = get_logger('nacsos.util.security')
class InsufficientPermissions(Exception):
status = http_status.HTTP_403_FORBIDDEN
class UserPermissions(BaseModel):
user: UserModel
permissions: ProjectPermissionsModel
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: Optional[str] = None
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='api/login/token', auto_error=False)
def create_access_token(data: dict[str, str | datetime], expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({'exp': expire})
encoded_jwt = jwt.encode(to_encode, settings.SERVER.SECRET_KEY, algorithm=settings.SERVER.HASH_ALGORITHM)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=http_status.HTTP_401_UNAUTHORIZED,
detail='Could not validate credentials',
headers={'WWW-Authenticate': 'Bearer'},
)
user = None
if settings.USERS.DEFAULT_USER is None:
try:
if token is None:
raise credentials_exception
payload = jwt.decode(token, settings.SERVER.SECRET_KEY, algorithms=[settings.SERVER.HASH_ALGORITHM])
username: str = payload.get('sub')
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except JWTError:
raise credentials_exception
token_user = token_data.username
if token_user is not None:
user = await crud_get_user_by_name(username=token_user, engine=db_engine)
else:
user = await read_user_by_name(username=settings.USERS.DEFAULT_USER, engine=db_engine)
logger.warning('Authentication using fake user!')
if user is None:
raise credentials_exception
logger.debug(f'Current user: user_id: {user.user_id} {user.username}')
return user
async def get_current_active_user(current_user: UserModel = Depends(get_current_user)) -> UserModel:
if not current_user.is_active:
raise HTTPException(status_code=http_status.HTTP_400_BAD_REQUEST, detail='Inactive user')
return current_user
def get_current_active_superuser(current_user: UserModel = Depends(get_current_active_user)) -> UserModel:
if not current_user.is_superuser:
raise HTTPException(
status_code=http_status.HTTP_400_BAD_REQUEST, detail="The user doesn't have enough privileges"
)
return current_user
async def get_project_permissions_for_user(project_id: str, current_user: UserModel) -> ProjectPermissionsModel | None:
if current_user.user_id is None:
raise MissingInformationError('The `current_user` is missing the (here) required `user_id` field.')
if current_user.is_superuser:
# admin gets to do anything always, so return with simulated full permissions
return ProjectPermissionsModel.get_virtual_admin(project_id=project_id,
user_id=str(current_user.user_id))
return await crud_get_project_permissions_for_user(user_id=current_user.user_id,
project_id=project_id,
engine=db_engine)
class UserPermissionChecker:
def __init__(self,
permissions: list[ProjectPermission] | ProjectPermission | None = None,
fulfill_all: bool = True):
self.permissions = permissions
self.fulfill_all = fulfill_all
# convert singular permission to list for unified processing later
if type(self.permissions) is str:
self.permissions = [self.permissions]
async def __call__(self,
x_project_id: str = Header(),
current_user: UserModel = Depends(get_current_active_user)) -> UserPermissions:
"""
This function checks the whether a set of required permissions is fulfilled
for the given project for the currently active user.
The list of `permissions` corresponds to boolean fields in the respective `ProjectPermissions` instance.
If left empty, only the existence of such an instance is checked – meaning whether or not the user
is allowed to see or access the project in one way or another.
If at least one permission is not fulfilled or no instance exists, this function raises a 403 HTTPException
:return: `ProjectPermissions` if permissions are fulfilled, exception otherwise
:raises HTTPException if permissions are not fulfilled
"""
project_permissions = await get_project_permissions_for_user(project_id=x_project_id,
current_user=current_user)
user_permissions = UserPermissions(user=current_user, permissions=project_permissions)
if project_permissions is not None:
# no specific permissions were required (only basic access to the project) -> permitted!
if self.permissions is None:
return user_permissions
any_permission_fulfilled = False
# check that each required permission is fulfilled
for permission in self.permissions:
p_permission = getattr(project_permissions, permission, False)
if self.fulfill_all and not p_permission:
raise InsufficientPermissions(
f'User does not have permission "{permission}" for project "{x_project_id}".'
)
any_permission_fulfilled = any_permission_fulfilled or p_permission
if not any_permission_fulfilled and not self.fulfill_all:
raise InsufficientPermissions(
f'User does not have any of the required permissions ({self.permissions}) '
f'for project "{x_project_id}".'
)
return user_permissions
raise HTTPException(
status_code=http_status.HTTP_403_FORBIDDEN,
detail=f'User does not have permission to access project "{x_project_id}".',
)