Merge 45512d5 into 52cfc68

backend/Makefile

@@ -49,7 +49,7 @@ install:
 	$(VENV)/bin/pip install -e ".[dev]"
 
 openapi:
-	$(VENV)/bin/python -m capellacollab.cli openapi generate /tmp/openapi.json
+	CAPELLACOLLAB_SKIP_OPENAPI_ERROR_RESPONSES=1 $(VENV)/bin/python -m capellacollab.cli openapi generate /tmp/openapi.json
 
 dev: database app
 

backend/capellacollab/__main__.py

@@ -9,7 +9,7 @@
 import fastapi_pagination
 import starlette_prometheus
 import uvicorn
-from fastapi import exception_handlers, middleware, responses, routing
+from fastapi import middleware, responses, routing
 from fastapi.middleware import cors
 
 import capellacollab.projects.toolmodels.backups.runs.interface as pipeline_runs_interface
@@ -18,7 +18,6 @@
 
 # This import statement is required and should not be removed! (Alembic will not work otherwise)
 from capellacollab.config import config
-from capellacollab.core import exceptions as core_exceptions
 from capellacollab.core import logging as core_logging
 from capellacollab.core.database import engine, migration
 from capellacollab.routes import router
@@ -146,30 +145,6 @@ def use_route_names_as_operation_ids() -> None:
 
 use_route_names_as_operation_ids()
 
-
-async def exception_handler(
-    request: fastapi.Request, exc: core_exceptions.BaseError
-) -> fastapi.Response:
-    return await exception_handlers.http_exception_handler(
-        request,
-        fastapi.HTTPException(
-            status_code=exc.status_code,
-            detail={
-                "title": exc.title,
-                "reason": exc.reason,
-                "err_code": exc.err_code,
-            },
-        ),
-    )
-
-
-def register_exceptions():
-    for exc in core_exceptions.BaseError.__subclasses__():
-        app.add_exception_handler(exc, exception_handler)  # type: ignore[arg-type]
-
-
-register_exceptions()
-
 if __name__ == "__main__":
     if os.getenv("FASTAPI_AUTO_RELOAD", "").lower() in ("1", "true", "t"):
         logging.getLogger("watchfiles.main").setLevel(logging.WARNING)

backend/capellacollab/core/authentication/basic_auth.py

@@ -5,12 +5,14 @@
 import logging
 
 import fastapi
-from fastapi import security, status
+from fastapi import security
 
 from capellacollab.core import database
 from capellacollab.users import crud as user_crud
 from capellacollab.users.tokens import crud as token_crud
 
+from . import exceptions
+
 logger = logging.getLogger(__name__)
 
 
@@ -23,11 +25,7 @@ async def __call__(  # type: ignore
         )
         if not credentials:
             if self.auto_error:
-                raise fastapi.HTTPException(
-                    status_code=status.HTTP_401_UNAUTHORIZED,
-                    detail="Not authenticated",
-                    headers={"WWW-Authenticate": "Bearer, Basic"},
-                )
+                raise exceptions.UnauthenticatedError()
             return None
         with database.SessionLocal() as session:
             user = user_crud.get_user_by_name(session, credentials.username)
@@ -41,27 +39,13 @@ async def __call__(  # type: ignore
             if not db_token:
                 logger.info("Token invalid for user %s", credentials.username)
                 if self.auto_error:
-                    raise fastapi.HTTPException(
-                        status_code=status.HTTP_401_UNAUTHORIZED,
-                        detail={
-                            "err_code": "BASIC_TOKEN_INVALID",
-                            "reason": "The used token is not valid.",
-                        },
-                        headers={"WWW-Authenticate": "Bearer, Basic"},
-                    )
+                    raise exceptions.InvalidPersonalAccessTokenError()
                 return None
 
             if db_token.expiration_date < datetime.date.today():
                 logger.info("Token expired for user %s", credentials.username)
                 if self.auto_error:
-                    raise fastapi.HTTPException(
-                        status_code=status.HTTP_401_UNAUTHORIZED,
-                        detail={
-                            "err_code": "token_exp",
-                            "reason": "The Signature of the token is expired. Please request a new access token.",
-                        },
-                        headers={"WWW-Authenticate": "Bearer, Basic"},
-                    )
+                    raise exceptions.TokenSignatureExpired()
                 return None
         return self.get_username(credentials)
 

backend/capellacollab/core/authentication/exceptions.py

@@ -0,0 +1,113 @@
+# SPDX-FileCopyrightText: Copyright DB InfraGO AG and contributors
+# SPDX-License-Identifier: Apache-2.0
+
+from fastapi import status
+
+from capellacollab.core import exceptions as core_exceptions
+from capellacollab.projects.users import models as projects_users_models
+from capellacollab.users import models as users_models
+
+
+class RequiredRoleNotMetError(core_exceptions.BaseError):
+    def __init__(self, required_role: users_models.Role):
+        super().__init__(
+            status_code=status.HTTP_403_FORBIDDEN,
+            title="Minumum role not met",
+            reason=f"The role {required_role.value} is required for this transaction.",
+            err_code="REQUIRED_ROLE_NOT_MET",
+        )
+
+
+class RequiredProjectRoleNotMetError(core_exceptions.BaseError):
+    def __init__(
+        self,
+        required_role: projects_users_models.ProjectUserRole,
+        project_slug: str,
+    ):
+        super().__init__(
+            status_code=status.HTTP_403_FORBIDDEN,
+            title="Minumum project role not met",
+            reason=f"The role {required_role.value} in the project '{project_slug}' is required for this transaction.",
+            err_code="REQUIRED_PROJECT_ROLE_NOT_MET",
+        )
+
+
+class RequiredProjectPermissionNotMetError(core_exceptions.BaseError):
+    def __init__(
+        self,
+        required_permission: projects_users_models.ProjectUserPermission,
+        project_slug: str,
+    ):
+        super().__init__(
+            status_code=status.HTTP_403_FORBIDDEN,
+            title="Minumum project permission not met",
+            reason=f"The permission {required_permission.value} in the project '{project_slug}' is required for this transaction.",
+            err_code="REQUIRED_PROJECT_PERMISSION_NOT_MET",
+        )
+
+
+class UnknownScheme(core_exceptions.BaseError):
+    def __init__(self, scheme: str):
+        super().__init__(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            title="Invalid scheme detected",
+            reason=(
+                f"The scheme '{scheme}' is not supported. "
+                "Use 'basic' or 'bearer' instead"
+            ),
+            err_code="UNKNOWN_SCHEME",
+            headers={"WWW-Authenticate": "Bearer, Basic"},
+        )
+
+
+class TokenSignatureExpired(core_exceptions.BaseError):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            title="Token signature expired",
+            reason="The Signature of the token is expired. Please request a new access token.",
+            err_code="TOKEN_SIGNATURE_EXPIRED",
+            headers={"WWW-Authenticate": "Bearer, Basic"},
+        )
+
+
+class JWTValidationFailed(core_exceptions.BaseError):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            title="Token validation failed",
+            reason="The validation of the access token failed. Please contact your administrator.",
+            err_code="JWT_TOKEN_VALIDATION_FAILED",
+        )
+
+
+class JWTInvalidToken(core_exceptions.BaseError):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            title="Access token not valid",
+            reason="The used token is not valid.",
+            err_code="JWT_TOKEN_INVALID",
+        )
+
+
+class UnauthenticatedError(core_exceptions.BaseError):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            title="Unauthenticated",
+            reason="Not authenticated",
+            err_code="UNAUTHENTICATED",
+            headers={"WWW-Authenticate": "Bearer, Basic"},
+        )
+
+
+class InvalidPersonalAccessTokenError(core_exceptions.BaseError):
+    def __init__(self):
+        super().__init__(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            title="Personal access token not valid.",
+            reason="The used token is not valid.",
+            err_code="BASIC_TOKEN_INVALID",
+            headers={"WWW-Authenticate": "Bearer, Basic"},
+        )

backend/capellacollab/core/authentication/injectables.py

@@ -7,7 +7,6 @@
 import logging
 
 import fastapi
-from fastapi import status
 from fastapi.openapi import models as openapi_models
 from fastapi.security import base as security_base
 from fastapi.security import utils as security_utils
@@ -16,12 +15,16 @@
 from capellacollab.core import database
 from capellacollab.core.authentication import basic_auth, jwt_bearer
 from capellacollab.projects import crud as projects_crud
+from capellacollab.projects import exceptions as projects_exceptions
 from capellacollab.projects import models as projects_models
 from capellacollab.projects.users import crud as projects_users_crud
 from capellacollab.projects.users import models as projects_users_models
 from capellacollab.users import crud as users_crud
+from capellacollab.users import exceptions as users_exceptions
 from capellacollab.users import models as users_models
 
+from . import exceptions
+
 logger = logging.getLogger(__name__)
 
 
@@ -76,17 +79,16 @@ async def get_username(
     authorization = request.headers.get("Authorization")
     scheme, _ = security_utils.get_authorization_scheme_param(authorization)
     username = None
+
     match scheme.lower():
         case "basic":
             username = await basic_auth.HTTPBasicAuth()(request)
         case "bearer":
             username = await jwt_bearer.JWTBearer()(request)
+        case "":
+            raise exceptions.UnauthenticatedError()
         case _:
-            raise fastapi.HTTPException(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                detail="Token is none and username cannot be derived",
-                headers={"WWW-Authenticate": "Bearer, Basic"},
-            )
+            raise exceptions.UnknownScheme(scheme)
 
     assert username
     return username
@@ -104,22 +106,16 @@ def __call__(
     ) -> bool:
         if not (user := users_crud.get_user_by_name(db, username)):
             if self.verify:
-                raise fastapi.HTTPException(
-                    status_code=status.HTTP_404_NOT_FOUND,
-                    detail={"reason": f"User {username} was not found"},
-                )
+                raise users_exceptions.UserNotFoundError(username)
             return False
 
         if (
             user.role != users_models.Role.ADMIN
             and self.required_role == users_models.Role.ADMIN
         ):
             if self.verify:
-                raise fastapi.HTTPException(
-                    status_code=status.HTTP_403_FORBIDDEN,
-                    detail={
-                        "reason": "You need to be administrator for this transaction.",
-                    },
+                raise exceptions.RequiredRoleNotMetError(
+                    users_models.Role.ADMIN
                 )
             return False
         return True
@@ -169,11 +165,8 @@ def __call__(
             project_user.role
         ) < self.roles.index(self.required_role):
             if self.verify:
-                raise fastapi.HTTPException(
-                    status_code=403,
-                    detail={
-                        "reason": f"The role '{self.required_role.value}' in the project '{project_slug}' is required.",
-                    },
+                raise exceptions.RequiredProjectRoleNotMetError(
+                    self.required_role, project_slug
                 )
             return False
 
@@ -187,21 +180,15 @@ def _get_user_and_check(
     ) -> users_models.DatabaseUser | None:
         user = users_crud.get_user_by_name(db, username)
         if not user and self.verify:
-            raise fastapi.HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail={"reason": f"User {username} was not found"},
-            )
+            raise users_exceptions.UserNotFoundError(username)
         return user
 
     def _get_project_and_check(
         self, project_slug: str, db: orm.Session
     ) -> projects_models.DatabaseProject | None:
         project = projects_crud.get_project_by_slug(db, project_slug)
         if not project and self.verify:
-            raise fastapi.HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail={"reason": f"The project {project_slug} was not found"},
-            )
+            raise projects_exceptions.ProjectNotFoundError(project_slug)
         return project
 
     def _is_internal_project_accessible(
@@ -228,11 +215,8 @@ def _has_user_required_project_permissions(
             == projects_users_models.ProjectUserPermission.WRITE
         ):
             if self.verify:
-                raise fastapi.HTTPException(
-                    status_code=403,
-                    detail={
-                        "reason": f"You need to have '{self.required_permission.value}'-access in the project!",
-                    },
+                raise exceptions.RequiredProjectPermissionNotMetError(
+                    self.required_permission, project_user.project.slug
                 )
             return False
         return True